Skip to content

baldur.services — Service Access

The re-export facade for service getters and the core service classes. Resolve the circuit-breaker and replay services here; the SLA-threshold helper exposes the configured breach thresholds.

Service getters

get_circuit_breaker_service 

get_circuit_breaker_service() -> CircuitBreakerService

Return the runtime-scoped CircuitBreakerService singleton.

Delegates to the active :class:~baldur.runtime.BaldurRuntime — test isolation, copy_context() scoping, and runtime swap-in work transparently through the runtime's singleton store.

Explicit-def wrapper (instead of tuple-unpacking the make_singleton_factory return value directly) so the symbol is statically discoverable by docs tooling (mkdocstrings/griffe) and carries a Public-surface docstring per the reference page contract.

get_replay_service 

get_replay_service() -> ReplayService

Get the singleton replay service instance.

get_sla_thresholds 

get_sla_thresholds() -> SLASettings

Get SLA thresholds configuration.

Core service classes

CircuitBreakerService 

CircuitBreakerService(
    config: CircuitBreakerConfig | None = None,
    repository: CircuitBreakerStateRepository | None = None,
)

Bases: EventEmitterMixin, ProtectionMixin, ManualControlMixin

Circuit Breaker Service.

Provides management operations for circuit breaker states. Designed for manual (toggle-based) control by operators.

Usage

service = CircuitBreakerService()

Force open (block requests)

result = service.force_open( service_name="external_api", reason="External service maintenance", controlled_by=admin_user )

Force close (allow requests)

result = service.force_close( service_name="external_api", reason="Service recovered", controlled_by=admin_user, trigger_replay=True )

Check if requests should be allowed

if service.should_allow("external_api"): # proceed with request

For testing with mock repository

mock_repo = Mock(spec=CircuitBreakerStateRepository) service = CircuitBreakerService(repository=mock_repo)

Initialize the circuit breaker service.

Parameters:

Name Type Description Default
config CircuitBreakerConfig | None

Optional configuration, loads from settings if None

 None
repository CircuitBreakerStateRepository | None

Optional repository for DI, uses Django adapter if None

 None

repository property 

repository: CircuitBreakerStateRepository

Get the repository using ProviderRegistry with fallback policy.

is_enabled property 

is_enabled: bool

Check if circuit breaker is enabled.

register_downstream_checker 

register_downstream_checker(
    checker: Callable[[str], bool],
) -> None

Register a should_allow() pre-check hook.

checker(service_name) → False triggers a preemptive Fallback. checker MUST only perform local in-memory lookups (no external I/O).

apply_threshold_override 

apply_threshold_override(
    service_name: str, override: Any
) -> None

Apply a threshold override set by the mesh coordinator.

While the override is active, the service's failure_threshold and recovery_timeout use the override values.

remove_threshold_override 

remove_threshold_override(service_name: str) -> None

Remove the threshold override and revert to the original config.

get_effective_config 

get_effective_config(
    service_name: str,
) -> CircuitBreakerConfig

Return the effective config with overrides applied.

Lookup happens in the L1 local cache, so there is no external I/O. Without an override, returns the base config; otherwise replaces only the overridden fields.

get_or_create_state 

get_or_create_state(
    service_name: str,
) -> CircuitBreakerStateData

Get or create a circuit breaker state for a service.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
CircuitBreakerStateData

CircuitBreakerStateData instance

get_state 

get_state(service_name: str) -> str

Get the current state of a circuit breaker.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
str

Current state (closed, open, half_open)

should_allow 

should_allow(service_name: str) -> bool

Check if requests should be allowed through the circuit breaker.

Post-476: HALF_OPEN slot acquisition is delegated to the repository's atomic try_acquire_half_open_slot so the per-service counter is cluster-wide accurate (Redis Lua) instead of per-process best-effort.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
bool

True if requests should be allowed, False if blocked

should_allow_with_state 

should_allow_with_state(
    service_name: str,
) -> CircuitBreakerDecision

Companion API to should_allow that returns the admission decision and the resolved state in a single call.

Closes the redundant get_or_create_state lookup that CircuitBreakerPolicy.execute() previously incurred on the reject path: the policy can now read decision.allowed for branching and decision.state.state for the rejection metadata without a second repository RLock acquire.

For is_enabled=False callers we return CircuitBreakerDecision with allowed=True and a freshly-fetched state — direct callers of the companion API contract receive a non-None state regardless of feature-flag posture. CircuitBreakerPolicy short-circuits on is_enabled before invoking this method, so the disabled-CB fetch is off the hot path.

should_allow_with_fallback 

should_allow_with_fallback(
    service_name: str,
    cache_key: str | None = None,
    default_response: Any | None = None,
    request_data: dict[str, Any] | None = None,
) -> CircuitBreakerFallbackResult

Check if requests should be allowed with fallback strategy support.

.. deprecated:: This method is deprecated. Use a CircuitBreakerPolicy + FallbackPolicy combination instead.

When CB is open, instead of simply blocking, this method can: 1. Return cached (stale) data 2. Queue the request to DLQ for later retry 3. Return a default/static response

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required
cache_key str | None

Optional Redis key for cached data lookup

 None
default_response Any | None

Optional default response to return

 None
request_data dict[str, Any] | None

Optional request data for DLQ queueing

 None

Returns:

Type Description
CircuitBreakerFallbackResult

CircuitBreakerFallbackResult with decision and optional fallback data

get_total_calls 

get_total_calls(service_name: str) -> int

Get the repository counter total for a service (failure + success).

These are the stored counters, not a call count: failure_count is the consecutive-failure count since the last success or reset, and success_count accrues on HALF_OPEN recovery trials. Successful CLOSED calls are not counted here — the rate trigger's denominator comes from :meth:get_window_evidence instead.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
int

Sum of the stored failure and success counters.

get_all_states 

get_all_states() -> list[dict[str, Any]]

Get all circuit breaker states.

Returns:

Type Description
list[dict[str, Any]]

List of state dictionaries

get_open_states 

get_open_states(
    limit: int | None = None,
) -> list[CircuitBreakerStateData]

Get circuit breaker states currently in OPEN state.

More efficient than get_all_states() for watchdog recovery which only needs OPEN states. Delegates to repository.get_open_states() which uses SCAN instead of KEYS in Redis.

Parameters:

Name Type Description Default
limit int | None

Maximum number of results. None means no limit.

 None

Returns:

Type Description
list[CircuitBreakerStateData]

List of CircuitBreakerStateData with state == OPEN,
list[CircuitBreakerStateData]

ordered by opened_at ascending (oldest first).

get_window_evidence 

get_window_evidence(service_name: str) -> tuple[int, int]

Return (failures, total) recorded in this worker's outcome window.

The evidence the failure-rate trigger decides on: outcomes observed while the circuit was CLOSED, bounded by sliding_window_size and cleared at every observed state transition. (0, 0) means no evidence, which is not the same as a 0% failure rate.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required

Returns:

Type Description
tuple[int, int]

Failure count and total call count over the window.

get_aggregate_failure_rate 

get_aggregate_failure_rate() -> float

Return the system-wide circuit-breaker failure fraction (0.0-1.0).

Aggregates the recorded call outcomes across every tracked service as sum(window_failures) / sum(window_total), returning 0.0 when no calls have been observed on any circuit.

This is a system-wide mean error fraction, not a fixed-time-window or per-service rate: a single failing service among many healthy ones is averaged below threshold, while a broad multi-service failure raises the mean. That makes it suited to a system-wide stability gate, where each individual service is still protected by its own circuit breaker.

Evidence comes from this service object's outcome windows, so the fraction covers the traffic admitted through this instance. A process that also protects calls through protect() or @circuit_breaker gives those policies their own service instances, whose outcomes this method does not see — read the aggregate from the same instance that recorded the traffic.

Returns:

Type Description
float

Failure fraction in the range 0.0-1.0.

record_failure 

record_failure(
    service_name: str,
    error_context: dict[str, Any] | None = None,
    hint_state: CircuitBreakerStateData | None = None,
) -> None

Record a failure for a service.

This is used for automatic circuit breaker mode. If the threshold is exceeded AND minimum_calls is met, the circuit opens automatically.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required
error_context dict[str, Any] | None

Optional context about the failure (for snapshot)

 None
hint_state CircuitBreakerStateData | None

Optional pre-fetched state — when the caller already loaded the state via should_allow_with_state, passing it here skips a redundant repository lookup. The hint is used only when its service_name matches; otherwise we fall through to a fresh fetch. Failures always increment counters, so the hint cannot fully skip the repository write.

 None

record_success 

record_success(
    service_name: str,
    hint_state: CircuitBreakerStateData | None = None,
) -> None

Record a success for a service.

This is used for automatic circuit breaker mode. In half-open state, enough successes will close the circuit.

Parameters:

Name Type Description Default
service_name str

Name of the external service

 required
hint_state CircuitBreakerStateData | None

Optional pre-fetched state — when the caller already loaded the state via should_allow_with_state, passing it here unlocks two optimizations: 1. Fast path: when the hint indicates steady-state CLOSED (manually_controlled=False, failure_count=0), the call returns immediately without touching the repository — the eventual update_state(failure_count=0) is a no-op. 2. Slow path: skips the redundant get_or_create_state repository read when the hint's service_name matches. Stale-hint cases are observably equivalent to the no-hint path: a missed reset is corrected by the next call's slow path.

 None

check_recovery_transitions 

check_recovery_transitions() -> dict

Check for circuit breakers that should transition from OPEN to HALF_OPEN.

This method should be called periodically (e.g., every minute) to check if any OPEN circuits have exceeded the recovery timeout and should transition to HALF_OPEN for testing.

Returns:

Type Description
dict

Dictionary with transitioned service names and count

manual_control 

manual_control(
    service_name: str,
    action: str,
    reason: str = "",
    controlled_by: Any = None,
) -> CircuitBreakerResult

Manually control a circuit breaker state.

Parameters:

Name Type Description Default
service_name str

Name of the service

 required
action str

'open', 'close', or 'auto'

 required
reason str

Reason for the control action

 ''
controlled_by Any

User who initiated the action

 None

Returns:

Type Description
CircuitBreakerResult

CircuitBreakerResult with operation details

reconcile_cb_cell_mapping 

reconcile_cb_cell_mapping() -> dict[str, Any]

Reconcile CB-to-Cell mapping consistency after a Ring Resize.

  1. Iterate over all CBs and extract cell_id from the Composite Key
  2. Compare against the correct cell_id per the current Hash Ring
  3. On mismatch: archive + delete the orphan CB (no state transition)
  4. CBs for new Cells are lazily created by get_or_create()

Returns:

Type Description
dict[str, Any]

{"archived": [...], "errors": [...]}

ReplayService 

ReplayService(
    repository: FailedOperationRepository | None = None,
    cache: CacheProviderInterface | None = None,
)

Bases: EventEmitterMixin

DLQ Replay Service.

Orchestrates replay operations for failed operations.

Usage

service = ReplayService()

Single replay

result = service.replay_single(dlq_id="web-1:112:a1b2c3d4e5f60708:5")

Batch replay

batch_result = service.replay_batch( failure_type="PG_TIMEOUT", max_items=50 )

For testing with mock repository

mock_repo = Mock(spec=FailedOperationRepository) service = ReplayService(repository=mock_repo)

Initialize the replay service.

Parameters:

Name Type Description Default
repository FailedOperationRepository | None

Optional repository for DI, uses Django adapter if None

 None
cache CacheProviderInterface | None

Optional cache provider for the per-service inflight lock guarding replay_on_circuit_close. If omitted, the provider is lazy-resolved via ProviderRegistry on first use. If resolution fails or the resolved provider does not support get_lock(), the guard fails open with a WARNING log.

 None

repository property 

repository: FailedOperationRepository

Get the repository using ProviderRegistry with fallback policy.

cache property 

cache: CacheProviderInterface | None

Lazy-resolve the cache provider for the circuit-close inflight lock.

Returns None if no provider can be resolved — caller falls open in that case. The inflight lock uses cache.get_lock() (owner-fenced DistributedLock), so adapter-level lock support is validated at acquire-time rather than via a separate setnx gate.

getattr with defaults handles test fixtures that bypass __init__ via ReplayService.__new__(...). A bypassed-init instance is observationally identical to a fresh instance with cache=None for this fail-open guard.

replay_single 

replay_single(
    dlq_id: str,
    trigger: (
        ResolutionTrigger | str
    ) = ResolutionTrigger.MANUAL_REPLAY,
    actor_id: str | None = None,
) -> ReplayResult

Replay a single DLQ entry.

This method uses atomic acquisition to prevent race conditions when multiple workers try to replay the same entry simultaneously.

Safety Checks (via check_all_governance): 1. Kill Switch - system-wide deactivation check 2. Emergency Level - blocked at LEVEL_2+ to protect resources 3. Error budget - automation blocked when the budget is exhausted

Audit Logging: - Blocks are automatically recorded in the AuditLog

Parameters:

Name Type Description Default
dlq_id str

ID of the FailedOperation to replay

 required
trigger ResolutionTrigger | str

Provenance trigger stamped into resolution_type on success (default: manual replay)

 MANUAL_REPLAY
actor_id str | None

Acting principal for the audit trail (default: ambient ActorContext / system)

 None

Returns:

Type Description
ReplayResult

ReplayResult indicating success or failure

replay_batch 

replay_batch(
    domain: str | None = None,
    failure_type: str | None = None,
    max_items: int = 100,
    use_adaptive: bool | None = None,
    use_priority: bool | None = None,
    trigger: (
        ResolutionTrigger | str
    ) = ResolutionTrigger.MANUAL_REPLAY,
    actor_id: str | None = None,
) -> BatchReplayResult

Replay multiple DLQ entries matching criteria.

Safety Checks (via check_all_governance): 1. Kill Switch - system-wide deactivation check 2. Emergency Level - blocked at LEVEL_2+ to protect resources 3. Error budget - automation blocked when the budget is exhausted

Adaptive Mode: - When adaptive_enabled=True in RuntimeConfig, batch size is dynamic - High failure rate (>=20%) reduces batch size by 20% - 3 consecutive perfect batches increases batch size by 5

Priority Mode: - When priority_enabled=True in RuntimeConfig, domains are processed by priority - Critical domains are processed first, then normal, then low - Respects domain-specific max_retries overrides

Audit Logging: - Blocks are automatically recorded in the AuditLog

Parameters:

Name Type Description Default
domain str | None

Filter by domain (optional, ignored in priority mode)

 None
failure_type str | None

Filter by failure type (optional)

 None
max_items int

Maximum number of items to replay (ignored in adaptive mode)

 100
use_adaptive bool | None

Override adaptive mode setting (None = use RuntimeConfig)

 None
use_priority bool | None

Override priority mode setting (None = use RuntimeConfig)

 None

Returns:

Type Description
BatchReplayResult

BatchReplayResult with summary and individual results

replay_on_circuit_close 

replay_on_circuit_close(
    service_name: str,
    max_items: int = 50,
    escalate_failures: bool = True,
    service_failure_type_map: (
        dict[str, list[str]] | None
    ) = None,
) -> BatchReplayResult

Replay entries when circuit breaker closes.

This is triggered when an external service recovers. Only replays entries related to the recovered service.

IMPORTANT: When triggered by force_close with trigger_replay=True, any replay failures are escalated to REQUIRES_REVIEW status. This is because operator-initiated recovery implies the operator intended to resolve these items, so failures need explicit attention.

Parameters:

Name Type Description Default
service_name str

Name of the service that recovered

 required
max_items int

Maximum number of items to replay

 50
escalate_failures bool

If True, mark failed replays as REQUIRES_REVIEW

 True
service_failure_type_map dict[str, list[str]] | None

Custom mapping of service names to failure types. If None, uses RuntimeConfig fallback. Example: {"my_service": ["TIMEOUT", "CONNECTION_ERROR"]}

 None

Returns:

Type Description
BatchReplayResult

BatchReplayResult with summary. inflight_skipped=True indicates
BatchReplayResult

the per-service inflight lock rejected this call as a duplicate.

BatchReplayResult dataclass 

BatchReplayResult(
    total: int = 0,
    success_count: int = 0,
    failed_count: int = 0,
    skipped_count: int = 0,
    results: list[ReplayResult] = list(),
    governance_blocked: bool = False,
    governance_block_reason: str = "",
    inflight_skipped: bool = False,
    capped: bool = False,
    priority_used: bool = False,
    domains_processed: list[str] | None = None,
)

Result of a batch replay operation.

ReplayResult dataclass 

ReplayResult(
    success: bool,
    dlq_id: str,
    message: str = "",
    error: str | None = None,
    data: dict[str, Any] | None = None,
    skipped: bool = False,
)

Result of a replay operation.

succeeded classmethod 

succeeded(
    dlq_id: str, message: str = "", data: dict | None = None
) -> ReplayResult

Factory for successful replay.

failed classmethod 

failed(dlq_id: str, error: str) -> ReplayResult

Factory for failed replay.

skipped_result classmethod 

skipped_result(
    dlq_id: str, reason: str = ""
) -> ReplayResult

Factory for idempotency-skipped replay.

blocked classmethod 

blocked(
    dlq_id: str, governance_result: GovernanceCheckResult
) -> ReplayResult

Factory for governance-blocked replay.