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() -> Self

An empty store on the wall clock.

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

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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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

Formats the value using the given formatter.

impl Default for MemoryStores

fn default() -> Self

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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: '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 Self: 'async_trait, 'life0: 'async_trait,

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