cadrumo.adapters.persistence.operations.journal module

Atomic filesystem implementation of the credential-free operation journal.

class OperationJournalRepository(*, storage_root)[source]

Bases: OperationJournal, OperationEventStream, OperationObservationReader

Async operation-journal port over the atomic filesystem substrate.

Every port call runs the synchronous substrate – its lock wait and its file reads and writes – on a worker thread, so a frontend’s event loop stays responsive while a journal operation waits on contention or disk. The thread inherits the caller’s context, so the settings and storage route in effect for the caller are the ones the substrate reads.

Parameters:

storage_root (Path)

async load(operation_id)[source]

Load the latest credential-free snapshot for one operation.

Return type:

OperationPersistedSnapshot

Parameters:

operation_id (str)

async resolve_idempotency(claim)[source]

Resolve a claim only when its matching operation journal exists.

Return type:

str | None

Parameters:

claim (OperationIdempotencyClaim)

async create(snapshot, *, lease)[source]

Create a complete initial journal before making an idempotency replay visible.

Return type:

str

Parameters:
async read_after(operation_id, cursor, *, limit)[source]

Return one bounded, exclusive page from retained event history.

Return type:

OperationReplayPage

Parameters:

  • operation_id (str)

  • cursor (OperationEventCursor)

  • limit (OperationReplayLimit)

async read_observation(operation_id, after_cursor, *, limit)[source]

Return snapshot, replay, and progress facts anchored to one locked record.

The caller is typically a poll worker on an interface event loop, so the lock wait and the read both run on a worker thread; cancelling the awaiting task returns control at once while the bounded read finishes.

Return type:

OperationObservationMaterialization

Parameters:

  • operation_id (str)

  • after_cursor (OperationEventCursor)

  • limit (OperationReplayLimit)

async commit(snapshot, *, expected_revision, lease)[source]

Atomically advance an existing snapshot through the typed substrate.

Return type:

None

Parameters:
is_absent(operation_id)[source]

Distinguish an absent record from a present but unreadable record.

Return type:

bool

Parameters:

operation_id (str)