Struct MemoryStores 

pub struct MemoryStores { /* private fields */ }

A complete, deterministic set of stores backed by one in-process state.

Implements ConversationStore, InteractionStore, CommandJournal, EventJournal, OutboxStore, ReplayStore and CommitStore over the same state, so a card written through the interaction trait is the card a commit bundle settles.

Wrap it in Stores::from_memory, or take the shortcut Stores::in_memory.

Implementations

impl MemoryStores

pub fn new() -> MemoryStores

An empty store on the wall clock.

pub fn with_clock(clock: Arc<dyn Clock>) -> MemoryStores

An empty store that stamps its own columns from clock.

Keep a clone of the clock to drive it: with a ManualClock the test decides when a claim goes stale or a phase marker moves.

pub fn now(&self) -> DateTime<Utc>

The instant this store would stamp a write with right now.

pub fn fail_next( &self, point: FailurePoint, error: StoreError, ) -> Result<(), StoreError>

Arms one failure: the next call reaching point returns error instead of its normal result.

Whether the write that precedes the boundary survives is part of the point’s meaning; see FailurePoint. Arming several failures is allowed, including several at the same point, and they fire in arming order. Inside CommitStore::commit a fired failure discards the whole bundle.

Errors

• Corrupt when the internal lock is poisoned.

pub fn armed_failures(&self) -> Result<Vec<FailurePoint>, StoreError>

The points still armed, in arming order.

Errors

• Corrupt when the internal lock is poisoned.

pub fn clear_failures(&self) -> Result<(), StoreError>

Disarms every pending failure.

Errors

• Corrupt when the internal lock is poisoned.

Trait Implementations

impl CommandJournalReader for MemoryStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, ) -> Pin<Box<dyn Future<Output = Result<CommandJournalEntry, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads one entry.

fn for_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<Vec<CommandJournalEntry>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Every entry of a turn, ordered by created_at then command_id.

fn pending_for_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<Vec<CommandJournalEntry>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Entries of a turn whose status is Pending or Executing, ordered by created_at then command_id (spec §23.1: resume by idempotency key).

impl CommandJournalWriter for MemoryStores

fn begin<'life0, 'async_trait>( &'life0 self, entry: CommandJournalEntry, ) -> Pin<Box<dyn Future<Output = Result<JournalAdmission, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Admits a command under UNIQUE (account_id, idempotency_key).

fn mark_executing<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Moves Pending → Executing. Calling it on an Executing entry is accepted without change.

fn complete<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, outcome: JournalOutcome, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Records the outcome and moves the entry to JournalOutcome::status. Recording the same outcome again on an entry already in that status is accepted without change.

fn fail<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, error: &'life3 ExecutionError, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: 'async_trait,

Records an execution error as the outcome, through JournalOutcome::from_execution_error. Same rules as Self::complete.

impl CommitStore for MemoryStores

fn commit<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, bundle: CommitBundle, ) -> Pin<Box<dyn Future<Output = Result<CommitReceipt, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Applies the bundle for account atomically.

impl ConversationReader for MemoryStores

fn load_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 ConversationId, ) -> Pin<Box<dyn Future<Output = Result<ConversationRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads a conversation of account.

fn load_recent_turns<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredTurn>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads the most recent limit turns of a conversation in chronological order (oldest of the selected first). Ordering is by received_at, then turn_id.

fn load_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<StoredTurn, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads one turn.

fn turn_phase<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<TurnPhaseMarker, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Reads the phase marker of a turn.

fn list_unfinished_turns<'life0, 'async_trait>( &'life0 self, scope: RecoveryScope, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<TurnPhaseMarker>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Lists up to limit markers of turns whose phase is not terminal, oldest first (by received_at, then turn_id).

impl ConversationWriter for MemoryStores

fn create_conversation<'life0, 'async_trait>( &'life0 self, record: ConversationRecord, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Creates a conversation.

fn append_user_turn<'life0, 'async_trait>( &'life0 self, turn: StoredUserTurn, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Appends a user turn to its conversation and creates its phase marker at TurnPhase::Received.

fn append_assistant_turn<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn: AssistantTurn, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Persists the assistant turn that answers a user turn, exactly as returned to the client.

fn set_turn_phase<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, phase: TurnPhase, ) -> Pin<Box<dyn Future<Output = Result<TurnPhaseMarker, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Writes the phase marker of a turn and returns the updated marker.

impl Debug for MemoryStores

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl Default for MemoryStores

fn default() -> MemoryStores

Returns the “default value” for a type.

impl EventJournalReader for MemoryStores

fn list_since<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, since: CaseRevision, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredEvent>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Events of one case with case_revision > since, in append order, at most limit. since = CaseRevision::ZERO lists the whole history.

fn read_from<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, after: EventCursor, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<EventPage, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

The account’s next events after after in the journal’s total order, across every case, at most limit, as an EventPage.

fn get_by_ids<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, ids: &'life2 [EventId], ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredEvent>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

The events with the given identifiers that exist for account, in the order requested; unknown or foreign identifiers are omitted. Receipt verification reads events back through this method (ADR-012 point 6).

fn count<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Number of events of a case.

impl EventJournalWriter for MemoryStores

fn append<'life0, 'async_trait>( &'life0 self, batch: EventBatch, ) -> Pin<Box<dyn Future<Output = Result<Vec<EventId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Appends a batch atomically and returns the event identifiers in order.

fn redact_payload<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, account: &'life1 AccountId, event_id: &'life2 EventId, authority: &'life3 RedactionAuthority, ) -> Pin<Box<dyn Future<Output = Result<EventRedaction, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, MemoryStores: 'async_trait,

Erases one event’s payload in place, and records that it happened.

impl InteractionReader for MemoryStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads an interaction of account.

fn list_open_for_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, ) -> Pin<Box<dyn Future<Output = Result<Vec<Interaction>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Lists the open (Active or Resolving) interactions of a conversation, ordered by created_at then id.

fn list_open_for_case<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, ) -> Pin<Box<dyn Future<Output = Result<Vec<Interaction>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Lists the open (Active or Resolving) interactions of a case, ordered by created_at then id.

fn blocking_answered_at<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, revision: CaseRevision, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Whether a blocking card of case_key bound to revision has already been answered by the user.

impl InteractionWriter for MemoryStores

fn insert<'life0, 'async_trait>( &'life0 self, interaction: Interaction, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Inserts a new Active interaction.

fn insert_replacing_blocking<'life0, 'async_trait>( &'life0 self, interaction: Interaction, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Inserts a new Active interaction, invalidating the Active blocking occupant of the same case when there is one. Returns the identifiers it invalidated (empty when the slot was free or the new card is not blocking).

fn begin_resolution<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, expected_status: InteractionStatus, option_id: OptionId, resolved_by: TurnId, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Compare-and-swap expected_status → Resolving, recording the chosen option and the resolving turn. The only legal source status is Active; the explicit parameter makes a lost race visible to the caller instead of hiding it behind a re-read.

fn finish_resolution<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, outcome: ResolutionOutcome, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Settles a Resolving interaction.

fn invalidate_for_case<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, new_revision: CaseRevision, reason: InvalidationReason, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Invalidates every Active interaction of the case that is bound to a revision other than new_revision and does not declare revision independence. Returns the invalidated identifiers in list order.

fn invalidate_case_cards<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, reason: InvalidationReason, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Invalidates every Active interaction of the case, whatever revision it is bound to, and whether or not it declares revision independence. Returns the invalidated identifiers in list order.

fn expire_due<'life0, 'async_trait>( &'life0 self, now: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Moves every Active interaction whose expires_at <= now to Expired, across all accounts. Returns the expired identifiers in list order.

impl OutboxReader for MemoryStores

fn get<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, ) -> Pin<Box<dyn Future<Output = Result<OutboxRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Loads one record.

fn list_for_command<'life0, 'life1, 'async_trait>( &'life0 self, command_id: &'life1 CommandId, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRecord>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Every record produced by a command, ordered by created_at then id.

impl OutboxWriter for MemoryStores

fn enqueue<'life0, 'async_trait>( &'life0 self, entry: OutboxEntry, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Enqueues a Pending entry.

fn claim_due<'life0, 'life1, 'async_trait>( &'life0 self, now: DateTime<Utc>, limit: usize, worker_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxEntry>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Claims up to limit due Pending entries for worker_id (see the module documentation), oldest first by created_at then id, and returns them already in Dispatching with attempt_count incremented.

fn mark_completed<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Dispatching | OutcomeUnknown → Completed; clears the claim and stamps completed_at. Accepted without change when already Completed.

fn mark_failed<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, reason: String, retry_at: Option<DateTime<Utc>>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Records a failure. reason is a stable code, never free text, and is stored on the record either way.

fn mark_outcome_unknown<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, remote_ref: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Dispatching → OutcomeUnknown, releasing the claim and recording the remote reference when the remote returned one (I15).

fn reschedule<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, next_attempt_at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MemoryStores: 'async_trait,

Returns the entry to Pending with next_attempt_at, releasing any claim. Legal from Dispatching, OutcomeUnknown and Pending (which only moves the time).

fn release_expired_claims<'life0, 'async_trait>( &'life0 self, claimed_before: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxId>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Releases every Dispatching entry whose claim was taken strictly before claimed_before back to Pending, due immediately. Returns the released identifiers. Meant for a reaper that recovers entries a crashed dispatcher left behind.

impl ReplayReader for MemoryStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<ReplayRecord, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

Loads the record of a turn.

fn list_for_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ReplayRecord>, StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, MemoryStores: 'async_trait,

The most recent limit records of a conversation in chronological order (by recorded_at, then turn_id).

impl ReplayWriter for MemoryStores

fn put<'life0, 'async_trait>( &'life0 self, record: ReplayRecord, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where 'life0: 'async_trait, MemoryStores: 'async_trait,

Inserts or replaces the record of (record.account_id, record.turn_id).