aion_store/

store.rs

//! Event-store traits and single-writer capability.

use aion_core::{
    Event, RunId, TimerId, WorkflowFilter, WorkflowId, WorkflowStatus, WorkflowSummary,
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};

use crate::{OutboxRow, StoreError, TimerEntry, TimerRetirement};

mod write_capability {
    /// Capability required to append workflow events.
    ///
    /// This token enforces Aion's single-writer durability invariant at the type level: only the
    /// recorder append path may hold write authority for a workflow. `SequenceConflict` remains the
    /// runtime defense-in-depth signal for any internal misuse or future bypass that attempts to
    /// append with a stale head.
    #[derive(Clone, Copy, Debug)]
    pub struct WriteToken {
        _private: (),
    }

    impl WriteToken {
        /// Constructs a write token for Aion's recorder path.
        #[must_use]
        pub fn recorder() -> Self {
            Self { _private: () }
        }
    }

    pub(crate) fn conformance() -> WriteToken {
        WriteToken { _private: () }
    }
}

pub use write_capability::WriteToken;

/// Summary of one concrete run in a workflow's continuation chain.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RunSummary {
    /// Concrete run identifier for this chain entry.
    pub run_id: RunId,
    /// Parent run that continued as this run, or `None` for the first run.
    pub parent_run_id: Option<RunId>,
    /// Status projected from this run's slice of lifecycle events.
    pub status: WorkflowStatus,
    /// Timestamp of this run's `WorkflowStarted` event.
    pub started_at: DateTime<Utc>,
    /// Workflow sequence of this run's `WorkflowStarted` event — the boundary
    /// at which this generation opens.
    ///
    /// SEQUENCE, NOT TIMESTAMP, IS THE AXIS THAT LOCATES AN EVENT'S GENERATION.
    /// A reader holding an event at sequence `n` answers "which run is this?" by
    /// taking the last boundary at or before `n`; `started_at` is a recorded
    /// value and cannot order events against it. Without this field the only way
    /// to answer that question is to scan history for `WorkflowStarted` — which
    /// is exactly what the ops console did, over whatever window it happened to
    /// have loaded, so on any workflow longer than one window it reported
    /// "run unknown" for every attempt it showed.
    pub started_seq: u64,
    /// Timestamp of this run's terminal lifecycle event, when closed.
    pub closed_at: Option<DateTime<Utc>>,
}

/// Read and durable-timer contract for Aion event stores.
#[async_trait]
pub trait ReadableEventStore: Send + Sync + 'static {
    /// Reads the complete event history for `workflow_id` in ascending sequence order.
    ///
    /// A workflow with no recorded events is observed as an empty history. This includes unknown
    /// workflow identifiers: because the first append with `expected_seq == 0` creates a workflow
    /// implicitly, "unknown workflow" and "empty history" are the same observable state for reads.
    /// This method must not return [`StoreError::NotFound`] for absent workflows.
    async fn read_history(&self, workflow_id: &WorkflowId) -> Result<Vec<Event>, StoreError>;

    /// Reads the event history for `workflow_id` restricted to events with sequence number
    /// greater than or equal to `from_seq`, in ascending sequence order.
    ///
    /// This is the range-read primitive behind O(delta) WS resume: callers replaying from a
    /// cursor must not pay for the full history. Semantics:
    ///
    /// - `from_seq <= 1` is equivalent to [`Self::read_history`]: sequence numbers start at 1,
    ///   so every recorded event satisfies the bound.
    /// - `from_seq` beyond the current head returns an empty vector, never an error. Whether a
    ///   beyond-head cursor is *valid* is protocol judgment, not store judgment: the WS resume
    ///   protocol rejects `resume_from_seq > head + 1` as an invalid cursor
    ///   (`ResumeCursorAheadOfHistory`), but it makes that call by comparing the cursor against
    ///   the head it observes — the store only answers which events exist at or after the
    ///   requested sequence.
    /// - Unknown workflows behave exactly like [`Self::read_history`] for unknown workflows:
    ///   empty history, never [`StoreError::NotFound`], because "unknown workflow" and "empty
    ///   history" are the same observable state for reads.
    ///
    /// There is deliberately no default implementation: a read-all-then-filter fallback would
    /// silently reintroduce O(history) behavior. Every backend must implement this as a real
    /// range read (for SQL backends, an indexed `seq >= ?` range scan).
    async fn read_history_from(
        &self,
        workflow_id: &WorkflowId,
        from_seq: u64,
    ) -> Result<Vec<Event>, StoreError>;

    /// Reads the concrete run chain for `workflow_id` in continuation order.
    async fn read_run_chain(&self, workflow_id: &WorkflowId)
    -> Result<Vec<RunSummary>, StoreError>;

    /// Lists every workflow identifier that has at least one event in history.
    ///
    /// Unlike [`Self::list_active`], this includes terminal workflows and exists to let projection
    /// repair jobs reconcile derived indexes against the authoritative event history.
    async fn list_workflow_ids(&self) -> Result<Vec<WorkflowId>, StoreError>;

    /// Lists workflow identifiers whose projected status is exactly
    /// [`WorkflowStatus::Running`](aion_core::WorkflowStatus::Running).
    async fn list_active(&self) -> Result<Vec<WorkflowId>, StoreError>;

    /// Lists workflow identifiers whose projected status is exactly
    /// [`WorkflowStatus::Paused`](aion_core::WorkflowStatus::Paused).
    ///
    /// Mirrors [`Self::list_active`] with a `== Paused` exact-equality filter: it
    /// is the durable source the dispatch-hold set is rebuilt from at startup and
    /// at shard adoption, so a run paused before a `kill -9` keeps its outbox rows
    /// held after restart. A paused run is excluded from [`Self::list_active`]
    /// (which filters `== Running`), so nothing else would repopulate the hold.
    async fn list_paused(&self) -> Result<Vec<WorkflowId>, StoreError>;

    /// Returns workflow summaries matching `filter`.
    async fn query(&self, filter: &WorkflowFilter) -> Result<Vec<WorkflowSummary>, StoreError>;

    /// Persists a durable timer for `workflow_id` that is due at `fire_at`.
    ///
    /// Timer scheduling remains on the public store surface because timers are not workflow-history
    /// appends and are used by the timer subsystem after the recorder has written `TimerStarted`.
    ///
    /// `armed_seq` is the workflow-history sequence of the arming's
    /// `TimerStarted` event and becomes part of the row's identity
    /// ([`TimerEntry::armed_seq`]): together with `fire_at` it is what a
    /// [`Self::retire_timer`] compare matches, so a re-arm to the IDENTICAL
    /// instant still writes a distinguishable row. An engine-internal arming
    /// that records no `TimerStarted` (the schedule coordinator) passes `0` —
    /// history sequences start at 1, so `0` is unambiguous, and such rows are
    /// keyed per trigger and never alias a workflow arming.
    async fn schedule_timer(
        &self,
        workflow_id: &WorkflowId,
        timer_id: &TimerId,
        fire_at: DateTime<Utc>,
        armed_seq: u64,
    ) -> Result<(), StoreError>;

    /// Returns durable timers whose `fire_at` is less than or equal to `as_of`.
    async fn expired_timers(&self, as_of: DateTime<Utc>) -> Result<Vec<TimerEntry>, StoreError>;

    /// 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.
    ///
    /// Called by the timer subsystem once the row's consuming fact is durably
    /// recorded — the fire path after `TimerFired` lands (or is refused by a
    /// terminal run), the cancel path after `TimerCancelled` lands, and the
    /// boot sweep's reconciliation of rows whose consumption predates this
    /// method. Without retirement every timer a workflow ever armed leaves a
    /// permanent row, [`Self::expired_timers`] returns the workflow's whole
    /// consumed past on every call, and the startup sweep walks it — the
    /// 2026-08-24 estate outage's 90-minute boot.
    ///
    /// The `(fire_at, armed_seq)` condition is what makes retirement safe
    /// against re-arms: a named timer re-armed after its fire OVERWRITES the
    /// same row with the replacement arming's values
    /// ([`Self::schedule_timer`]), and a retirement decided against the
    /// consumed arming must never delete the replacement's row — that row is
    /// the re-armed timer's only durable claim to a recovery fire. The
    /// `armed_seq` half is load-bearing for a re-arm to the IDENTICAL
    /// instant: `fire_at` alone cannot tell those two armings apart, and the
    /// replacement's `TimerStarted` always carries a strictly higher
    /// sequence. A caller always knows the arming it is retiring (the sweep
    /// row it walked, or the `TimerStarted` it read), so the condition costs
    /// nothing; a mismatch means "already re-armed, nothing left to retire"
    /// and is the [`TimerRetirement::Superseded`] SUCCESS, not an error.
    ///
    /// The condition must hold under CONCURRENT re-arming, not merely against
    /// a stale caller decision: a `schedule_timer` racing this call must
    /// either land before the compare (mismatch, `Superseded`) or after the
    /// delete (its row survives) — never inside it. A backend without an
    /// atomic conditional delete must serialize this method against
    /// [`Self::schedule_timer`] itself.
    ///
    /// Idempotent: retiring an absent (never scheduled, or already retired)
    /// row succeeds as [`TimerRetirement::Retired`] — the boot sweep and a
    /// racing live fire may both retire the same row, and the second act must
    /// be a no-op, not an error. Replay never calls this: replay is read-only
    /// on the timer keyspace.
    ///
    /// A distributed backend must ride retirement on the same stamped,
    /// replicated write path as [`Self::schedule_timer`], routed onto the
    /// workflow's shard — an adopted shard must not resurrect retired rows.
    async fn retire_timer(
        &self,
        workflow_id: &WorkflowId,
        timer_id: &TimerId,
        fire_at: DateTime<Utc>,
        armed_seq: u64,
    ) -> Result<TimerRetirement, StoreError>;

    /// 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`.
    ///
    /// This is the engine-lifecycle hook behind a multi-shard deployment: the
    /// boot path tells the store which shards this node serves so recovery and
    /// enumeration see only that node's slice of the cluster's state. The
    /// default implementation is a deliberate no-op — the single-shard in-memory
    /// backend owns everything unconditionally, so a `None` or any shard set
    /// leaves its behaviour byte-identical. The sharded haematite backend
    /// overrides this to scope its enumeration. Decorators that wrap another
    /// store must forward this call to their inner store.
    fn set_owned_shards(&self, shards: Option<&[usize]>) {
        let _ = shards;
    }

    /// 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.
    ///
    /// This is the SS-2 election hook the engine boot path calls right after
    /// [`Self::set_owned_shards`] and BEFORE startup recovery: a distributed
    /// backend wins the per-shard election and becomes the live owner, so the
    /// subsequent recovery reads see the full committed history for its shards.
    ///
    /// The default implementation is a deliberate no-op returning `Ok(())` —
    /// non-distributed backends (the in-memory store and single-node haematite)
    /// own everything unconditionally and elect nothing, so boot stays
    /// byte-identical. Only a DISTRIBUTED sharded backend
    /// overrides this to run the election. Decorators that wrap another store
    /// must forward this call to their inner store.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Backend`] when a distributed backend cannot win the
    /// election or become the live owner of one of `shards`; the node must not
    /// serve those shards in that case (fail-closed).
    fn acquire_owned_shards(&self, shards: &[usize]) -> Result<(), StoreError> {
        let _ = shards;
        Ok(())
    }

    /// 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).
    ///
    /// The default implementation is a deliberate no-op returning `Ok(())` —
    /// single-shard / non-distributed backends own everything unconditionally and
    /// elect nothing. Only a DISTRIBUTED sharded backend (haematite) overrides it.
    /// Decorators that wrap another store must forward this call.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::NotOwner`] when a strictly higher ballot deposed this
    /// candidate (a clean, droppable election loss), and [`StoreError::Backend`]
    /// for a quorum-unavailable election or any transport fault (retryable).
    fn acquire_owned_shard(&self, shard: usize) -> Result<(), StoreError> {
        let _ = shard;
        Ok(())
    }

    /// 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.
    ///
    /// This is the residual-window re-assertion the failover path uses to exclude
    /// a survivor that lost its epoch between winning acquire+publish and widening
    /// its enumeration scope (ADR-021 clean-partial). It is a POINT-IN-TIME
    /// ADVISORY, not a durable lock — the authoritative gate remains the per-write
    /// CAS fence.
    ///
    /// The default implementation returns `true` — single-shard / non-distributed
    /// backends own everything unconditionally, so the failover path's
    /// re-assertion is a no-op there and behaviour stays byte-identical. Only a
    /// DISTRIBUTED sharded backend (haematite) overrides it. Decorators that wrap
    /// another store must forward this call.
    fn is_current_owner(&self, shard: usize) -> bool {
        let _ = shard;
        true
    }

    /// Add `shards` to this node's owned-enumeration scope, UNIONING them with
    /// the shards it already owns rather than replacing the set.
    ///
    /// This is the SS-5 failover hook: when a live node absorbs a dead peer's
    /// shards it must KEEP serving its own shards while ALSO enumerating the
    /// adopted ones. [`Self::set_owned_shards`] replaces the scope (the boot
    /// path's one-shot assignment); this widens it in place. The boot path uses
    /// `set_owned_shards`; the failover path uses this.
    ///
    /// The default implementation is a deliberate no-op — single-shard backends
    /// own everything unconditionally, so widening their scope is meaningless and
    /// leaves their behaviour byte-identical. Only a sharded backend (haematite)
    /// overrides this. Decorators that wrap another store must forward this call.
    ///
    /// When the store currently owns ALL shards (the `None` / single-node
    /// default), it already enumerates `shards`, so a sharded backend leaves the
    /// own-all scope untouched.
    fn extend_owned_shards(&self, shards: &[usize]) {
        let _ = shards;
    }

    /// 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).
    ///
    /// This is the failover-publish hook the engine calls from `adopt_shards`
    /// right after it has won `shard`'s election: it records, durably and
    /// cluster-visibly, that this node has adopted `shard`, so a request reaching
    /// a DIFFERENT survivor routes to this adopter rather than mis-resolving to
    /// the dead declared owner (gap #2).
    ///
    /// The default implementation is a deliberate no-op returning `Ok(())` —
    /// single-shard / non-distributed backends own everything unconditionally and
    /// have no peers to coordinate, so boot and adoption stay byte-identical. Only
    /// a DISTRIBUTED sharded backend overrides this. Decorators that wrap another
    /// store must forward this call.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::NotOwner`] when a distributed backend's fenced
    /// directory write is out-voted (this node is not actually the owner), and
    /// [`StoreError::Backend`] for any other replication/transport failure.
    fn publish_shard_owner(&self, shard: usize) -> Result<(), StoreError> {
        let _ = shard;
        Ok(())
    }
}

/// Write authority for appending workflow-history events.
///
/// `append` requires a [`WriteToken`], so having an `Arc<dyn EventStore>` or
/// `Arc<dyn ReadableEventStore>` is not sufficient to write events.
#[async_trait]
pub trait WritableEventStore: Send + Sync + 'static {
    /// Atomically appends `events` to `workflow_id` when the stored history head equals
    /// `expected_seq`.
    ///
    /// Implementations must apply every event in `events` or none of them. If the current stored
    /// head for `workflow_id` differs from `expected_seq`, this method must return
    /// [`StoreError::SequenceConflict`] and leave history unchanged. A first append with
    /// `expected_seq == 0` creates the workflow history implicitly.
    async fn append(
        &self,
        token: WriteToken,
        workflow_id: &WorkflowId,
        events: &[Event],
        expected_seq: u64,
    ) -> Result<(), StoreError>;

    /// 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`].
    ///
    /// This is the durable fan-out write: the `ActivityScheduled`/`ActivityStarted` scheduling
    /// events and their matching outbox rows commit together or not at all. Atomicity is
    /// load-bearing — a committed fan-out batch always carries both, or neither — so the out-of-band
    /// dispatcher can never observe an outbox row whose scheduling events were rolled back, and a
    /// re-issued append cannot leave events without their dispatch rows.
    ///
    /// # Default implementation (safe for outbox-unaware backends)
    ///
    /// The default delegates to [`Self::append`] when `outbox_rows` is empty (byte-for-byte
    /// equivalent to an event-only append), and otherwise returns [`StoreError::Backend`] **rather
    /// than silently dropping the outbox rows**. Dropping them would be the dangerous failure mode:
    /// the events would commit, the workflow would believe its fan-out is durably staged, and the
    /// rows would never be dispatched. A hard error forces a backend to opt in to durable-outbox
    /// support by overriding this method (as the haematite store does) before any caller can route a
    /// fan-out batch through it.
    ///
    /// # 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 backend boundary failures or when an outbox-unaware backend is
    /// asked to persist a non-empty `outbox_rows` slice.
    async fn append_with_outbox(
        &self,
        token: WriteToken,
        workflow_id: &WorkflowId,
        events: &[Event],
        expected_seq: u64,
        outbox_rows: &[OutboxRow],
    ) -> Result<(), StoreError> {
        if outbox_rows.is_empty() {
            return self.append(token, workflow_id, events, expected_seq).await;
        }
        Err(StoreError::Backend(String::from(
            "this event store does not support durable-outbox appends; \
             refusing to drop outbox rows (override WritableEventStore::append_with_outbox)",
        )))
    }

    /// Returns the outbox rows for `rows`' `dispatch_key`s to `Pending`, re-staging them for the
    /// out-of-band dispatcher.
    ///
    /// This is the crash-recovery re-arm: on first arrival after a restart, an activity whose
    /// `ActivityScheduled` is recorded but which has no terminal event lost its in-flight dispatch
    /// when the previous engine process died. Under the durable-outbox model the recovering workflow
    /// re-stages the dispatch by flipping its outbox row back to claimable `Pending` (an UPSERT — a
    /// brand-new `dispatch_key` with no prior row is inserted as `Pending`) instead of driving an
    /// in-process completion task. Redelivery is safe: the completion dedup
    /// (`record_fan_out_completion`) ignores a terminal for an already-resolved ordinal, so re-arm is
    /// at-least-once.
    ///
    /// The dispatch retry budget is preserved across re-arm: a backend must NOT reset an existing
    /// row's `attempt` to zero, so a workflow that reliably crashes the server still eventually
    /// dead-letters rather than re-dispatching forever.
    ///
    /// # Default implementation (safe for outbox-unaware backends)
    ///
    /// An empty `rows` slice is `Ok(())`. A non-empty slice returns [`StoreError::Backend`] **rather
    /// than silently no-op'ing the re-arm**: a store without durable-outbox support cannot re-stage a
    /// dispatch, and silently dropping the request would strand the recovered activity. A hard error
    /// forces a backend to opt in (as the haematite store does) before any caller can route a re-arm
    /// through it.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Serialization`] when an outbox payload cannot be serialized, and
    /// [`StoreError::Backend`] for backend boundary failures or when an outbox-unaware backend is
    /// asked to re-arm a non-empty `rows` slice.
    async fn rearm_outbox_pending(&self, rows: &[OutboxRow]) -> Result<(), StoreError> {
        if rows.is_empty() {
            return Ok(());
        }
        Err(StoreError::Backend(String::from(
            "this event store does not support durable-outbox re-arm; \
             refusing to drop a non-empty re-arm (override WritableEventStore::rearm_outbox_pending)",
        )))
    }

    /// Idempotently settles one outbox row to cancelled when this writer is backed by an outbox.
    ///
    /// Outbox-aware backends override this and delegate to [`crate::OutboxStore`]. The default is a
    /// no-op so non-outbox test stores and legacy backends can still record `ActivityCancelled`
    /// history without requiring an outbox table.
    ///
    /// # Errors
    ///
    /// Outbox-aware overrides return [`StoreError::Backend`] for backend boundary failures.
    async fn settle_outbox_row_cancelled(&self, dispatch_key: &str) -> Result<(), StoreError> {
        let _ = dispatch_key;
        Ok(())
    }

    /// Idempotently settles EVERY live ([`crate::OutboxStatus::Pending`] or
    /// [`crate::OutboxStatus::Claimed`]) outbox row of `workflow_id` to
    /// [`crate::OutboxStatus::Cancelled`], returning the settled `dispatch_key`s (#253).
    ///
    /// This is the workflow-terminal settle the Recorder runs after durably recording a workflow
    /// terminal (`WorkflowCompleted`/`WorkflowFailed`/`WorkflowCancelled`): a terminal workflow's
    /// staged dispatches must never be redelivered, so its live rows are retired to the terminal
    /// `Cancelled` disposition that re-arm and claim contractually never touch. Rows already in
    /// `Done`/`Failed`/`Cancelled` are left untouched, so the settle is idempotent. Reopen still
    /// supersedes it: [`Self::rearm_outbox_pending`] forcibly returns any existing row — including a
    /// `Cancelled` one — to `Pending`, so a reopened workflow's re-dispatches deliver again.
    ///
    /// The default is a no-op returning no settled keys, mirroring
    /// [`Self::settle_outbox_row_cancelled`], so non-outbox test stores and legacy backends record
    /// workflow terminals without requiring an outbox table. Outbox-aware backends override it and
    /// share the implementation with [`crate::OutboxStore::cancel_outbox_rows_for_workflow`].
    ///
    /// # Errors
    ///
    /// Outbox-aware overrides return [`StoreError::Backend`] for backend boundary failures and
    /// [`StoreError::Serialization`] when a stored row cannot be decoded.
    async fn settle_workflow_outbox_rows_cancelled(
        &self,
        workflow_id: &WorkflowId,
    ) -> Result<Vec<String>, StoreError> {
        let _ = workflow_id;
        Ok(Vec::new())
    }
}

/// Convenience trait for concrete stores that support reads/timers, recorder
/// writes, and deployed-package persistence.
///
/// [`crate::PackageStore`] is part of the contract, not an optional add-on:
/// runtime-deployed packages share the durability promise of event history
/// (a recovered run is pinned to a recorded package version, and a backend
/// that dropped the archive would strand it).
pub trait EventStore: ReadableEventStore + WritableEventStore + crate::PackageStore {}

impl<T> EventStore for T where
    T: ReadableEventStore + WritableEventStore + crate::PackageStore + ?Sized
{
}

pub(crate) fn conformance_write_token() -> WriteToken {
    write_capability::conformance()
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use super::{EventStore, ReadableEventStore, WritableEventStore};

    #[test]
    fn event_store_traits_are_object_safe() {
        let _: Option<Arc<dyn ReadableEventStore>> = None;
        let _: Option<Arc<dyn WritableEventStore>> = None;
        let _: Option<Arc<dyn EventStore>> = None;
    }
}