Struct InstrumentedEventStore 

pub struct InstrumentedEventStore { /* private fields */ }

Event-store wrapper that observes operation latency and lifecycle events without changing engine crates.

Implementations

impl InstrumentedEventStore

pub fn new( inner: Arc<dyn EventStore>, metrics: Metrics, namespace: impl Into<String>, ) -> Self

Wrap an event store with server-side metrics.

The store is given a private, never-pulsed outbox wake; callers that share the engine’s stage seam with the dispatcher install the shared handle with Self::with_outbox_wake.

pub fn with_outbox_wake(self, outbox_wake: Arc<Notify>) -> Self

Install the shared advisory outbox wake (LSUB-2).

The supplied Notify is the same handle the OutboxDispatcher awaits, so a committed outbox-row batch wakes the dispatcher’s run loop directly.

Trait Implementations

impl Debug for InstrumentedEventStore

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

Formats the value using the given formatter.

impl PackageStore for InstrumentedEventStore

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 InstrumentedEventStore

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

Forward owned-shard scoping to the inner store; this decorator adds only metrics, never shard policy, so the inner backend remains the sole authority on enumeration scope.

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

Forward the SS-2 shard election to the inner store; this decorator adds only metrics, never ownership policy, so the inner backend runs the election (or no-ops in single-node mode).

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

Forward the per-shard (ADR-021 clean-partial) election to the inner store. MUST be forwarded: the adoption fence (Engine::adopt_shards) drives the SINGULAR per-shard seam, and the trait default is a silent no-op that would let a survivor “adopt” a shard WITHOUT winning the election — its in-memory live epoch is then never seeded, so every recovery write is fenced by the surviving quorum and cross-node failover stalls (#157).

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

Forward the SS-5 failover scope-widening to the inner store; this decorator adds only metrics, never ownership policy.

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

Forward the residual-window ownership re-assertion (ADR-021). MUST be forwarded: the trait default returns true, which would make the adoption planner treat a shard it never actually won as a survivor (#157).

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

Forward the SS-3 shard-owner directory publish (fenced by the election just won). MUST be forwarded: the trait default is a silent no-op, so a request reaching a different survivor would mis-resolve to the dead declared owner instead of this adopter (#157).

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 exactly WorkflowStatus::Running.

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 stream_heads<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<StreamHead>, StoreError>> + Send + 'async_trait>>

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

Lists every workflow’s event stream with the seq of its last event — the stream-side half of the handshake that lets a boot trust a visibility row instead of opening the history behind it (crate::visibility::head).

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>, armed_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,

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 retire_timer<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, workflow_id: &'life1 WorkflowId, timer_id: &'life2 TimerId, fire_at: DateTime<Utc>, armed_seq: u64, ) -> Pin<Box<dyn Future<Output = Result<TimerRetirement, StoreError>> + Send + 'async_trait>>

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

Removes the durable timer row for (workflow_id, timer_id) — but only while the row still carries exactly the (fire_at, armed_seq) identity of the arming being retired.

impl WritableEventStore for InstrumentedEventStore

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,

Forward the atomic durable-outbox append to the inner store.

The default trait method REFUSES a non-empty outbox_rows slice (to stop an outbox-unaware backend silently dropping fan-out rows). Without this override the engine — which writes through this decorator — would never reach the inner haematite store’s outbox-capable append, so a commissioned (outbox.enabled) server could not stage a single fan-out member. We delegate to the inner store so its atomicity guarantee (events + rows commit together) holds, and observe the same append latency bucket and lifecycle metrics as a plain 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,

Forward the crash-recovery outbox re-arm to the inner store.

As with Self::append_with_outbox, the refusing default would strand a recovered fan-out member because the engine re-arms through this decorator.

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,

Forward the fan-out cancellation settle to the inner store.

MUST be forwarded: the trait default is a SILENT Ok(()) no-op, so without this override a cancelled fan-out ordinal’s outbox row is never settled on an outbox.enabled server — it stays claimable and the dispatcher re-dispatches the cancelled activity (the same silent-default forwarding hazard as the per-shard failover seam, #157). Timed under the shared write bucket like the sibling outbox re-arm.

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,

Forward the workflow-terminal outbox settle (#253) to the inner store.

MUST be forwarded for the same reason as Self::settle_outbox_row_cancelled: the trait default is a silent empty-Ok no-op, and the Recorder settles a terminal workflow’s rows through this decorator — inheriting the default would leave a dead workflow’s rows claimable and redeliverable. Timed under the shared write bucket like the sibling settle.

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.