Struct LibSqlStore 

pub struct LibSqlStore { /* private fields */ }

Durable EventStore backed by a shared libSQL connection.

Implementations

impl LibSqlStore

pub async fn connect(config: LibSqlConfig) -> Result<Self, StoreError>

Open a store from operator-provided libSQL configuration.

Errors

Returns StoreError::Backend when the connection cannot be opened or when the idempotent schema DDL cannot be applied.

pub async fn open(path: impl Into<PathBuf>) -> Result<Self, StoreError>

Open an embedded local-file store at path.

Operator tunables remain unset; this convenience constructor only selects embedded mode.

Errors

Returns StoreError::Backend when the connection cannot be opened or when the idempotent schema DDL cannot be applied.

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

Validate stored event blobs against the current Aion event schema.

Errors

Returns StoreError::Serialization when any stored event cannot be decoded by the current event schema, or StoreError::Backend for libSQL scan failures.

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

Trigger and await a libSQL replica synchronization cycle.

Errors

Returns StoreError::Backend when the current libSQL database mode does not support sync or when the replica sync operation fails.

pub async fn append_with_outbox( &self, _token: WriteToken, workflow_id: &WorkflowId, events: &[Event], expected_seq: u64, outbox_rows: Option<&[OutboxRow]>, ) -> Result<(), StoreError>

Atomically append events and, when outbox_rows is Some, the outbox rows in the same IMMEDIATE transaction, under the expected-head sequence guard.

This is the durable fan-out write: the scheduling events and their outbox rows commit together or not at all. Passing None is equivalent to WritableEventStore::append.

Errors

Returns StoreError::SequenceConflict when the stored head differs from expected_seq, StoreError::Serialization when an event or outbox payload cannot be serialized, and StoreError::Backend for libSQL boundary failures.

pub async fn outbox_row_state( &self, dispatch_key: &str, ) -> Result<Option<OutboxRowState>, StoreError>

Read the persisted lifecycle state of one outbox row by its dispatch key.

Returns the (status, attempt, visible_after) triple, or None when no row carries dispatch_key (the dedup guard may have ignored it). This is an out-of-band inspection helper — the OutboxStore dispatch contract itself keys terminal transitions off dispatch_key and never needs to read a row back — used by the dispatcher’s tests and by operators auditing dead-lettered (failed) rows.

Errors

Returns StoreError::Backend for libSQL boundary failures and StoreError::Serialization when the stored status token or timestamp cannot be decoded.

Trait Implementations

impl Clone for LibSqlStore

fn clone(&self) -> LibSqlStore

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl NamespaceStore for LibSqlStore

fn register_namespace<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, origin: NamespaceOrigin, ) -> Pin<Box<dyn Future<Output = Result<MintOutcome, StoreError>> + Send + 'async_trait>>

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

Idempotent minted-on-use upsert.

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

where Self: 'async_trait, 'life0: 'async_trait,

Explicit upsert (POST /namespaces).

fn list_namespaces<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<NamespaceRecord>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns the live durable set, ascending by created_at (ties broken by name).

fn get_namespace<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<NamespaceRecord>, StoreError>> + Send + 'async_trait>>

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

Looks up a single namespace by name.

fn set_namespace_placement<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, placement: NamespacePlacement, ) -> Pin<Box<dyn Future<Output = Result<Option<()>, StoreError>> + Send + 'async_trait>>

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

Idempotent quorum-CAS update of an existing namespace’s placement directive (Control-Plane Phase 2, P2-P2).

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

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

Transitions a namespace from NamespaceState::Active to NamespaceState::Deprecated (deprecate-before-delete).

impl OutboxStore for LibSqlStore

fn append_outbox_batch<'life0, 'life1, 'async_trait>( &'life0 self, rows: &'life1 [OutboxRow], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Inserts rows into the outbox, silently ignoring any whose dispatch_key already exists.

fn claim_outbox_rows<'life0, 'async_trait>( &'life0 self, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically claims up to limit pending rows whose visible_after has passed.

fn claim_outbox_rows_excluding<'life0, 'life1, 'async_trait>( &'life0 self, limit: u32, held: &'life1 HashSet<WorkflowId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

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

Atomically claims up to limit due pending rows whose owning workflow is NOT in held (the pause dispatch-hold, #204).

fn claim_outbox_rows_scoped<'life0, 'life1, 'async_trait>( &'life0 self, scope: &'life1 ClaimScope, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

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

Atomically claims up to limit pending rows that are due AND in scope (LSUB-1a).

fn claim_outbox_rows_scoped_excluding<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, scope: &'life1 ClaimScope, limit: u32, held: &'life2 HashSet<WorkflowId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

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

Atomically claims up to limit due pending rows that are in scope AND whose owning workflow is NOT in held (the pause dispatch-hold, #204).

fn rearm_stale_claimed_outbox_rows<'life0, 'async_trait>( &'life0 self, older_than: DateTime<Utc>, visible_after: DateTime<Utc>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Re-arms stale claimed rows so a live dispatcher can claim them again without restart.

fn list_stale_claimed_outbox_rows<'life0, 'async_trait>( &'life0 self, older_than: DateTime<Utc>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRow>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns up to limit STALE claimed rows — status is OutboxStatus::Claimed and the durable claimed_at is older than older_than — WITHOUT transitioning them (#253).

fn list_unsettled_outbox_workflow_ids<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowId>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns the distinct workflow ids owning at least one UNSETTLED row — status is OutboxStatus::Pending or OutboxStatus::Claimed — scoped to this node’s owned shards like every other outbox enumeration (#253).

fn cancel_outbox_rows_for_workflow<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>

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

Idempotently settles EVERY live (OutboxStatus::Pending or OutboxStatus::Claimed) row of workflow_id to OutboxStatus::Cancelled, returning the settled dispatch_keys (#253).

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

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

Marks the row identified by dispatch_key as OutboxStatus::Done.

fn retry_outbox_row<'life0, 'life1, 'async_trait>( &'life0 self, dispatch_key: &'life1 str, next_attempt: u32, visible_after: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Returns the row identified by dispatch_key to OutboxStatus::Pending for retry.

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

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

Marks the row identified by dispatch_key as OutboxStatus::Failed (dead letter).

fn count_inflight_outbox_rows<'life0, 'life1, 'async_trait>( &'life0 self, namespace: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

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

Returns the count of in-flight outbox rows for namespace (CP2-Q1.5).

fn count_claimed_outbox_rows<'life0, 'life1, 'async_trait>( &'life0 self, namespace: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

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

Returns the count of CLAIMED outbox rows for namespace (CP2-Q2).

fn count_claimed_outbox_rows_by_namespace<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, namespaces: &'life1 [&'life2 str], ) -> Pin<Box<dyn Future<Output = Result<BTreeMap<String, u64>, StoreError>> + Send + 'async_trait>>

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

Counts the CLAIMED outbox rows for each namespace in namespaces, in ONE pass (CP2-Q2 perf).

fn pending_outbox_routes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<ClaimScope>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Enumerates the distinct (namespace, task_queue, node) routes that currently have at least one CLAIMABLE pending row — a row whose status is OutboxStatus::Pending and whose visible_after fence has passed (CP2-Q2).

impl PackageStore for LibSqlStore

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

where Self: 'async_trait, 'life0: 'async_trait,

Persists record and atomically points the type’s route at it.

fn put_package_with_routes<'life0, 'life1, 'async_trait>( &'life0 self, record: PackageRecord, route_workflow_types: &'life1 [String], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Persists record and atomically points every member workflow type at its content hash.

fn list_packages<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<PackageRecord>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists every persisted package in ascending deployed_at order (ties broken by (workflow_type, content_hash) text order), so startup reload re-applies deploys deterministically.

fn delete_package<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_type: &'life1 str, content_hash: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Deletes the persisted archive for (workflow_type, content_hash).

fn put_package_route<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_type: &'life1 str, content_hash: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Upserts the route pointer for workflow_type to content_hash.

fn list_package_routes<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<PackageRouteRecord>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists every persisted route pointer in workflow_type text order.

impl ReadableEventStore for LibSqlStore

fn read_history<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, StoreError>> + Send + 'async_trait>>

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

Reads the complete event history for workflow_id in ascending sequence order.

fn read_history_from<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, from_seq: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, StoreError>> + Send + 'async_trait>>

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

Reads the event history for workflow_id restricted to events with sequence number greater than or equal to from_seq, in ascending sequence order.

fn read_run_chain<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, ) -> Pin<Box<dyn Future<Output = Result<Vec<RunSummary>, StoreError>> + Send + 'async_trait>>

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

Reads the concrete run chain for workflow_id in continuation order.

fn list_workflow_ids<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowId>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists every workflow identifier that has at least one event in history.

fn list_active<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowId>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists workflow identifiers whose projected status is non-terminal.

fn list_paused<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowId>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists workflow identifiers whose projected status is exactly WorkflowStatus::Paused.

fn query<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 WorkflowFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<WorkflowSummary>, StoreError>> + Send + 'async_trait>>

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

Returns workflow summaries matching filter.

fn schedule_timer<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, timer_id: &'life2 TimerId, fire_at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Persists a durable timer for workflow_id that is due at fire_at.

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

where Self: 'async_trait, 'life0: 'async_trait,

Returns durable timers whose fire_at is less than or equal to as_of.

fn set_owned_shards(&self, shards: Option<&[usize]>)

Restrict every per-workflow enumeration (active workflows, timers, outbox rows) to the named set of distribution shards this node owns, or restore the own-all-shards default when shards is None.

fn acquire_owned_shards(&self, shards: &[usize]) -> Result<(), StoreError>

Acquire-and-serve ownership of each named distribution shard BEFORE the boot path recovers or enumerates over them, so the node is the fenced owner and its replicated state is union-merged locally first.

fn acquire_owned_shard(&self, shard: usize) -> Result<(), StoreError>

Acquire-and-serve ownership of a SINGLE distribution shard — the per-shard primitive Self::acquire_owned_shards is a loop over, exposed so the failover path can drive a per-shard abort seam: a clean election loss on one shard (StoreError::NotOwner) drops only that shard rather than failing the whole adoption batch (ADR-021 clean-partial).

fn is_current_owner(&self, shard: usize) -> bool

Whether this node currently holds LIVE serve-authority for shard — it won the per-shard election THIS process lifetime and has not been deposed in-process.

fn extend_owned_shards(&self, shards: &[usize])

Add shards to this node’s owned-enumeration scope, UNIONING them with the shards it already owns rather than replacing the set.

fn publish_shard_owner(&self, shard: usize) -> Result<(), StoreError>

Publish THIS node as the current owner of shard in the cluster’s shard-owner directory, so other nodes’ request-routing edges resolve shard to this node (SS-3).

impl VisibilityStore for LibSqlStore

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

where Self: 'async_trait, 'life0: 'async_trait,

Upserts a workflow visibility record.

fn list_workflows<'life0, 'async_trait>( &'life0 self, filter: ListWorkflowsFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<VisibilityWorkflowSummary>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists workflow visibility summaries matching filter.

fn count_workflows<'life0, 'async_trait>( &'life0 self, filter: ListWorkflowsFilter, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Counts workflow visibility summaries matching filter.

impl WritableEventStore for LibSqlStore

fn append<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _token: WriteToken, workflow_id: &'life1 WorkflowId, events: &'life2 [Event], expected_seq: u64, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Atomically appends events to workflow_id when the stored history head equals expected_seq.

fn append_with_outbox<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _token: WriteToken, workflow_id: &'life1 WorkflowId, events: &'life2 [Event], expected_seq: u64, outbox_rows: &'life3 [OutboxRow], ) -> 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,

Atomically appends events and the durable-outbox outbox_rows for workflow_id in a single transaction, under the same expected-head sequence guard as Self::append.

fn rearm_outbox_pending<'life0, 'life1, 'async_trait>( &'life0 self, rows: &'life1 [OutboxRow], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Returns the outbox rows for rows’ dispatch_keys to Pending, re-staging them for the out-of-band dispatcher.

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

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

Idempotently settles one outbox row to cancelled when this writer is backed by an outbox.

fn settle_workflow_outbox_rows_cancelled<'life0, 'life1, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>

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

Idempotently settles EVERY live (crate::OutboxStatus::Pending or crate::OutboxStatus::Claimed) outbox row of workflow_id to crate::OutboxStatus::Cancelled, returning the settled dispatch_keys (#253).