baldur.adapters.memory — Core State Repositories

In-memory implementations of the highest-traffic repository interfaces: failed-operation persistence and circuit-breaker state. For testing, standalone (non-framework) usage, and prototyping.

InMemoryFailedOperationRepository

InMemoryFailedOperationRepository()

Bases: FailedOperationRepository

In-memory implementation of FailedOperationRepository.

Thread-safe storage for DLQ entries in memory. Data is lost when the process exits.

Maintains status/domain indexes for O(1) lookup instead of O(n) scan.

create

create(
    domain: str,
    failure_type: str,
    error_message: str = "",
    error_code: str = "",
    entity_type: str | None = None,
    entity_id: str | None = None,
    entity_refs: dict[str, Any] | None = None,
    user_id: int | None = None,
    snapshot_data: dict[str, Any] | None = None,
    request_data: dict[str, Any] | None = None,
    response_data: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None,
    retry_count: int = 0,
    max_retries: int = 2,
    next_action_hint: str = "",
    recommended_action: str = "",
    expires_at: datetime | None = None,
) -> FailedOperationData

Create a new failed operation record (domain-neutral).

get_by_id

get_by_id(id: str) -> FailedOperationData | None

Get a failed operation by ID.

get_pending_by_domain

get_pending_by_domain(
    domain: str, limit: int = 100
) -> list[FailedOperationData]

Get pending operations for a specific domain.

get_pending_count_by_domain

get_pending_count_by_domain(domain: str) -> int

Get count of pending operations for a domain.

update_status

update_status(
    id: str,
    status: str,
    resolution_type: str = "",
    resolution_note: str = "",
    resolved_by_id: int | None = None,
    recommended_action: str = "",
) -> bool

Update the status of a failed operation.

increment_retry_count

increment_retry_count(id: str) -> bool

Increment retry count and update last_retry_at.

mark_as_resolved

mark_as_resolved(
    id: str,
    resolution_type: str,
    resolution_note: str = "",
    resolved_by_id: int | None = None,
) -> bool

Mark a failed operation as resolved.

get_expired_operations

get_expired_operations(
    before_date: datetime, limit: int = 100
) -> list[FailedOperationData]

Get operations that have expired.

bulk_update_status

bulk_update_status(ids: list[str], status: str) -> int

Bulk update status for multiple operations.

find_by_status

find_by_status(
    status: str,
    domain: str | None = None,
    failure_type: str | None = None,
    limit: int = 100,
) -> list[FailedOperationData]

Find operations by status with optional filters.

find

find(
    *,
    status: str | None = None,
    domain: str | None = None,
    failure_type: str | None = None,
    offset: int = 0,
    limit: int = 100
) -> list[FailedOperationData]

Paginated cross-status query ordered by created_at DESC.

count

count(
    *,
    status: str | None = None,
    domain: str | None = None,
    failure_type: str | None = None
) -> int

Count operations matching filters (pre-slice set size).

count_created_in_window

count_created_in_window(
    start: datetime, end: datetime
) -> int

Count entries whose created_at is within the inclusive [start, end].

find_replayable_page

find_replayable_page(
    *,
    max_retries: int,
    domain: str | None = None,
    failure_type: str | None = None,
    source: str | None = None,
    limit: int = 100,
    cursor: str | None = None
) -> ReplayablePage

Select replayable entries from cursor onward, oldest first.

One O(n) pass over the status (or status+domain) index, keeping the limit smallest by (created_at, id) with heapq.nsmallest — O(n log limit) and no materialized sort. This backing is the zero-config default and the pass runs under the same lock create() takes, so sorting a full domain on every call would block capture for the length of each sort.

The pass has no scan bound: the index is a Python set with no useful iteration order, so stopping partway through would leave the walk unable to say what it had examined, and next_cursor would skip the members it never looked at.

find_sla_breached

find_sla_breached(
    current_time: datetime,
    sla_thresholds: dict[str, timedelta],
) -> list[FailedOperationData]

Find operations that have breached their SLA.

find_expired

find_expired(
    current_time: datetime,
) -> list[FailedOperationData]

Find operations past their retention period.

get_statistics

get_statistics() -> dict[str, Any]

Get statistics about failed operations.

Adds pending-specific breakdowns for the daily report: - pending_by_domain: {domain: pending_count} (required by update_dlq_pending_gauges — pre-existing bug fix) - pending_by_domain_and_failure_type: {domain: {failure_type: count}} (powers DLQPendingBreakdown in daily report)

Memory adapter iterates the pending index (O(N) in-memory).

get_facet_counts

get_facet_counts(
    *, status: str | None = None, domain: str | None = None
) -> dict[str, dict[str, int]]

Faceted status×domain counts via the 1D/2D indexes.

by_status is scoped by domain; by_domain is scoped by status (faceted-search semantics). Empty buckets are dropped explicitly with if ids: _remove_from_index discards ids without deleting an emptied set, so a fully-drained status/domain key lingers as an empty set and would otherwise surface as :0, breaking zero-drop parity with the SQL/Redis adapters.

try_acquire_for_replay

try_acquire_for_replay(
    id: str, max_retries: int, force: bool = False
) -> FailedOperationData | None

Atomically acquire a DLQ entry for replay.

force=True bypasses the cap gate (operator cap-override): it accepts a {PENDING, REQUIRES_REVIEW} source, resets retry_count to a fresh budget (this redrive is attempt 1), and stamps the metadata history scar before the reset. See FailedOperationRepository.try_acquire_for_replay.

complete_replay

complete_replay(
    id: str,
    success: bool,
    resolution_type: str = "",
    note: str = "",
    resolved_by_id: int | None = None,
    error_details: dict[str, Any] | None = None,
) -> bool

Complete a replay operation by updating the final status.

release_stale_replaying

release_stale_replaying(
    older_than_minutes: int = 30,
) -> int

Release DLQ entries stuck in REPLAYING state.

clear

clear() -> None

Clear all entries (for testing).

archive_old_resolved

archive_old_resolved(older_than_days: int = 30) -> int

Archive resolved entries older than N days.

purge_archived

purge_archived(
    ids: list[str] | None = None,
    older_than_days: int | None = None,
) -> int

Permanently delete archived entries.

count_archived_older_than

count_archived_older_than(older_than_days: int) -> int

Count archived entries older than N days.

count_all

count_all() -> int

Return active DLQ item count (excludes resolved/rejected/archived).

Matches Redis adapter semantics where resolved entries are removed from the PENDING_KEY sorted set.

count_by_domain

count_by_domain(domain: str) -> int

Return active DLQ item count for a domain (same population as :meth:count_all, scoped to one domain).

The per-domain cap is enforced against this number, so it has to exclude the same finished entries the global count does — otherwise a domain's cap trips on entries that were resolved days ago.

get_oldest_ids

get_oldest_ids(
    count: int, domain: str | None = None
) -> list[str]

Return IDs of the oldest evictable items (by created_at).

Evictable = neither finished nor in flight; see :data:_NON_EVICTABLE_STATUSES for why the candidate set is narrowed rather than filtered after the fact.

delete

delete(entry_id: str) -> bool

Delete a single entry by ID. Returns True if deleted.

evict_oldest

evict_oldest(count: int, domain: str | None = None) -> int

Delete the oldest items, skipping entries in protected statuses.

get_cleanup_stats

get_cleanup_stats() -> dict[str, Any]

Get statistics for cleanup operations.

compress_and_evict_oldest

compress_and_evict_oldest(
    count: int, domain: str | None = None
) -> int

Compress then evict oldest entries (in-memory implementation).

Same logical flow as Redis adapter but uses Python dict/list.

store_compressed_entry

store_compressed_entry(entry: DLQCompressedEntry) -> bool

Store compressed entry in memory dict.

get_compressed_entries

get_compressed_entries(
    domain: str | None = None,
    status: str | None = None,
    limit: int = 100,
) -> list[DLQCompressedEntry]

Query compressed entries from memory, newest first.

get_compressed_entry

get_compressed_entry(
    entry_id: str,
) -> DLQCompressedEntry | None

Return a single compressed entry by id, or None if absent.

get_compressed_entries_before

get_compressed_entries_before(
    *,
    status: str,
    before: datetime,
    limit: int = 100,
    offset: int = 0,
    after: datetime | None = None
) -> list[DLQCompressedEntry]

Query compressed entries in a cutoff window, oldest first.

get_compressed_summary

get_compressed_summary() -> dict[str, Any]

Aggregate statistics of compressed entries.

update_compressed_status

update_compressed_status(
    entry_id: str, new_status: str
) -> bool

Transition compressed entry lifecycle status.

InMemoryCircuitBreakerStateRepository

InMemoryCircuitBreakerStateRepository()

Bases: CircuitBreakerStateRepository

In-memory implementation of CircuitBreakerStateRepository.

Thread-safe storage for circuit breaker states in memory.

failure_count and success_count are plain cumulative counters, so a reset written by update_state sticks — matching the Redis and SQL repositories. Rate evidence for the failure-rate trigger lives in the circuit breaker service's outcome window, not here.

get_by_service_name

get_by_service_name(
    service_name: str,
) -> CircuitBreakerStateData | None

Get circuit breaker state by service name.

get_or_create

get_or_create(service_name: str) -> CircuitBreakerStateData

Get or create a circuit breaker state.

Double-checked locking — steady-state callers (entry exists) bypass the RLock entirely and pay only a GIL-atomic dict.get. Only the first-call create needs the lock. Mirrors the precedent in protect.py and the project-wide policy in factory/base.py ("read-only dict operations rely on CPython GIL atomicity").

Removes the surviving read-path acquire that was the contention floor under high concurrency.

hydrate_snapshot

hydrate_snapshot(
    snapshot: CircuitBreakerStateData,
    skip_if_local_pin_active: bool = False,
) -> bool

Create-or-replace a row from a durable snapshot, pin fields included.

The single primitive for wholesale restores from the durable layer: the layered repository's construction-time load and its miss-hydration lane. It carries the manual-control fields that a per-field state copy drops, so an operator's Block/Allow survives into a process that did not take it.

Copied from the snapshot: state, failure_count, success_count, opened_at, last_failure_at, manually_controlled, controlled_by_id, control_reason, manual_override_expires_at, metadata.

Preserved (layer-local, never carried by a snapshot): id, created_at, and the half-open window fields, which belong to the atomic slot-acquire primitives rather than to bulk transfer.

Deliberately NOT the two-step set_manual_control + update_state: that primitive restamps opened_at on an OPEN row, which would corrupt the discriminator that decides whether a pin's lift is due, and the two writes are not atomic against a concurrent admission read. One locked replace avoids both.

Parameters:

Name	Type	Description	Default
snapshot	CircuitBreakerStateData	the durable row to restore.	required
skip_if_local_pin_active	bool	decline the restore, inside the same lock hold that would perform it, when the local row already carries an override still in force. Convergence lanes acting on a deliberately older remote read pass it so a pin placed here after that read is not overwritten by the stale remote view. The wholesale restore lanes leave it off: an absent or never-hydrated local row has no local decision to protect.	False

Returns:

Type	Description
bool	True when the row was written, False when the active-pin guard
bool	declined it.

update_state

update_state(
    service_name: str,
    state: str,
    failure_count: int | None = None,
    success_count: int | None = None,
    opened_at: datetime | None = None,
    last_failure_at: datetime | None = None,
    half_open_request_count: int | None = None,
    reset_half_open_count: bool = False,
    clear_opened_at: bool = False,
    skip_if_pinned: bool = False,
    keep_open: bool = False,
) -> bool

Update circuit breaker state.

clear_opened_at, skip_if_pinned and keep_open are all evaluated inside the single lock hold that performs the write, so none can be decided against a row an operator or a concurrent trip replaced in between.

increment_failure_count

increment_failure_count(
    service_name: str,
    last_failure_at: datetime | None = None,
) -> int

Increment failure count.

reset_counts

reset_counts(service_name: str) -> bool

Reset failure/success counts and clear the OPEN-era timestamp.

opened_at is cleared alongside the counters so a CLOSED DTO does not carry a stale OPEN-era timestamp. The L2-authoritative close path in LayeredCircuitBreakerStateRepository invokes this before transitioning L1 to CLOSED.

converge_to_closed_unless_pinned

converge_to_closed_unless_pinned(service_name: str) -> bool

Move an existing row to CLOSED in one lock hold, active pins excepted.

The write half of a convergence lane that has already established, from a healthy shared store, that this service is closed cluster-wide. The pin check and the write must not be split into two calls: an operator forcing the breaker open between them would keep its manual-control fields while the state flipped back to closed, leaving a row that is pinned closed and can no longer trip for the pin's whole lifetime, because outcome recording is suppressed while a pin is in force.

Writes the same fields the store-authoritative close writeback does — counters cleared, OPEN-era timestamp dropped, half-open window discarded — and leaves the manual-control fields exactly as it found them.

Returns:

Type	Description
bool	True when the row was transitioned. False when there is no local
bool	row, or an override is in force, in which case nothing is written.

set_manual_control

set_manual_control(
    service_name: str,
    state: str,
    controlled_by_id: int | None = None,
    reason: str = "",
    expires_at: datetime | None = None,
) -> bool

Set manual control override.

clear_manual_control

clear_manual_control(
    service_name: str, preserve_reason: bool = False
) -> bool

Clear manual control override.

Only the manual-control flag is cleared. state and the counters (failure_count, success_count) are not modified. If a state transition is needed, the caller must invoke update_state first.

record_failure

record_failure(
    service_name: str,
) -> CircuitBreakerStateData

Record a failure and return updated state.

Increments the cumulative failure counter. A preceding update_state(failure_count=0) therefore sticks, so the counter reads as failures since the last success — the consecutive-failure evidence the count trigger expects, identical to Redis and SQL.

record_success

record_success(
    service_name: str,
) -> CircuitBreakerStateData

Record a success and return updated state.

Increments the cumulative success counter, leaving the failure counter untouched — resetting it is the service's decision, written through update_state.

record_success_with_close_check

record_success_with_close_check(
    service_name: str, success_threshold: int
) -> CircuitBreakerCloseAttempt

Atomic record-success + threshold-check + close transition.

Whole sequence executes under self._lock: success increment, threshold check, and (if crossed from HALF_OPEN) the close transition are one critical section. Closes the TOCTOU race where multiple stale-view callers each pass the threshold check and emit duplicate CIRCUIT_BREAKER_CLOSED events for the same logical recovery.

record_failure_with_open_check

record_failure_with_open_check(
    service_name: str,
) -> CircuitBreakerOpenAttempt

Atomic record-failure + HALF_OPEN -> OPEN re-open transition.

Whole sequence executes under self._lock: state read and (if the state is HALF_OPEN) the re-open transition are one critical section. Closes the TOCTOU race where multiple stale-view callers each read HALF_OPEN and emit duplicate CIRCUIT_BREAKER_OPENED events for the same logical re-open. Symmetric mirror of record_success_with_close_check; a single HALF_OPEN failure re-opens unconditionally (no threshold).

trip_to_open

trip_to_open(
    service_name: str, failure_count: int
) -> CircuitBreakerOpenAttempt

Atomic CLOSED -> OPEN trip under a single lock acquire.

The pin check, the state read and the OPEN write are one critical section, so the row an operator pinned cannot be overwritten between the check and the write, and two threads that both decided to trip produce exactly one did_open=True. Branch contract: see CircuitBreakerStateRepository.trip_to_open.

get_all_states

get_all_states() -> list[CircuitBreakerStateData]

Get all circuit breaker states.

reset

reset(service_name: str) -> bool

Reset circuit breaker to initial closed state.

atomic_force_open

atomic_force_open(
    service_name: str,
    reason: str = "",
    controlled_by_id: int | None = None,
    ttl_minutes: int | None = None,
) -> tuple[bool, str, str]

Atomically force open a circuit breaker.

atomic_force_close

atomic_force_close(
    service_name: str,
    reason: str = "",
    controlled_by_id: int | None = None,
    ttl_minutes: int | None = None,
) -> tuple[bool, str, str]

Atomically force close a circuit breaker.

atomic_reset

atomic_reset(
    service_name: str,
    reason: str = "",
    controlled_by_id: int | None = None,
) -> tuple[bool, str, str]

Atomically reset a circuit breaker to initial state.

try_acquire_half_open_slot

try_acquire_half_open_slot(
    service_name: str,
    limit: int,
    stuck_timeout_seconds: int,
) -> tuple[bool, str, str]

Atomic HALF_OPEN slot acquisition under RLock.

reset_half_open_count

reset_half_open_count(service_name: str) -> None

Reset HALF_OPEN counter and clear window watermark.

get_open_states

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

Get OPEN circuit breaker states, oldest-first.

get_all_open

get_all_open() -> list[CircuitBreakerStateData]

Get all open circuit breakers.

delete

delete(service_name: str) -> bool

Delete a circuit breaker state.

delete_state

delete_state(service_name: str) -> bool

Delete circuit breaker state (alias for delete).

clear

clear() -> None

Clear all entries (for testing).