spate_coordination/store/
mod.rs

1//! Store primitives the coordination protocol runs on.
2//!
3//! The protocol needs exactly six operations — create-if-absent,
4//! CAS-on-revision update, point read, guarded delete, prefix watch, and a
5//! reconcile listing — over two keyspaces: a **durable** one (split
6//! records, the plan record — survives owner death) and an **ephemeral**
7//! one whose keys expire a fixed TTL after their last write (leases —
8//! every heartbeat rewrite re-arms the clock). [`CoordinationStore`] is
9//! that contract, public so deployments can bring their own backend
10//! (Redis, etcd) next to the built-in NATS and in-memory stores.
11//!
12//! # Contract notes for implementors
13//!
14//! - [`Revision`]s are store-assigned and **strictly increase per key**
15//!   across its write history (bucket-wide sequences satisfy this).
16//! - `update` on an [`Ephemeral`](Keyspace::Ephemeral) key re-arms its
17//!   TTL; expiry surfaces to watchers as [`WatchEvent::Delete`].
18//! - A watch delivers a snapshot of live keys, then [`WatchEvent::
19//!   SnapshotDone`], then live updates. A broken watch stream is
20//!   [`StoreError::Retryable`]: consumers re-watch and apply `Put`s only
21//!   at a revision above the last one they saw, so replays are idempotent.
22//! - `list` is the loss-proof backstop for missed watch events: a key the
23//!   consumer believes live but absent from a listing is treated as
24//!   deleted. (The protocol never depends on *which* marker a watch
25//!   delivered — graceful release vs expiry is decided from durable
26//!   record state, not from watch semantics.)
27
28use std::future::Future;
29use std::time::Duration;
30
31pub mod memory;
32pub(crate) mod metered;
33#[cfg(feature = "nats")]
34pub mod nats;
35
36/// Which keyspace an operation targets.
37#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
38pub enum Keyspace {
39    /// Never expires; survives owner death (split records, plan record).
40    Durable,
41    /// Every write re-arms expiry [`lease_ttl`](CoordinationStore::lease_ttl)
42    /// from write time on the store's clock; expiry surfaces to watchers
43    /// as a delete.
44    Ephemeral,
45}
46
47/// Store-assigned version token, strictly increasing per key —
48/// content-independent, so it carries none of the ABA hazards a
49/// content-hash token would.
50#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
51pub struct Revision(pub u64);
52
53/// One key's current value and revision.
54#[derive(Clone, Debug, PartialEq, Eq)]
55pub struct Entry {
56    /// The key, relative to the keyspace.
57    pub key: String,
58    /// The value bytes.
59    pub value: Vec<u8>,
60    /// Revision of the write that produced this value.
61    pub revision: Revision,
62}
63
64/// Outcome of a conditional write. Losing is a protocol outcome (someone
65/// else won the race), never an error.
66#[derive(Clone, Copy, Debug, PartialEq, Eq)]
67#[must_use]
68pub enum CasOutcome {
69    /// The write landed at this revision.
70    Won(Revision),
71    /// The precondition failed: the key existed (create), or moved past
72    /// the expected revision (update/delete).
73    Lost,
74}
75
76impl CasOutcome {
77    /// The winning revision, if the write landed.
78    pub fn won(self) -> Option<Revision> {
79        match self {
80            CasOutcome::Won(rev) => Some(rev),
81            CasOutcome::Lost => None,
82        }
83    }
84}
85
86/// One observation from a keyspace watch.
87#[derive(Clone, Debug, PartialEq, Eq)]
88#[non_exhaustive]
89pub enum WatchEvent {
90    /// A key's current value (snapshot replay or live update).
91    Put(Entry),
92    /// The key is gone — explicit delete or ephemeral expiry.
93    Delete {
94        /// The key, relative to the keyspace.
95        key: String,
96        /// Revision of the deletion itself, strictly greater than every
97        /// revision the key held before it. Deletes and puts for one key
98        /// are only ordered through these revisions: a consumer that
99        /// rewrote the key must ignore a delete whose revision is below
100        /// its own write's (the stale echo of an older deletion).
101        revision: Revision,
102    },
103    /// The initial snapshot is fully delivered; everything after is live.
104    SnapshotDone,
105}
106
107/// A store operation failed.
108#[derive(Debug, thiserror::Error)]
109#[non_exhaustive]
110pub enum StoreError {
111    /// Transient: the operation may succeed if retried.
112    #[error("retryable store error: {0}")]
113    Retryable(String),
114    /// Unrecoverable: misconfiguration, unsupported server, corrupt state.
115    #[error("fatal store error: {0}")]
116    Fatal(String),
117}
118
119/// Boxed watch stream: ordered per key, snapshot-then-live.
120pub type WatchStream = futures_util::stream::BoxStream<'static, Result<WatchEvent, StoreError>>;
121
122/// The minimal primitives the coordination protocol needs; see the
123/// [module docs](self) for the cross-implementation contract.
124///
125/// # Implementor notes
126///
127/// - [`StoreCoordinator`](crate::StoreCoordinator) additionally requires
128///   the store to be `Clone`: implement the trait on a cheap shared
129///   handle (wrap your connection state in an `Arc`, as the built-in
130///   backends do).
131/// - The trait uses `impl Future` returns and is therefore not
132///   dyn-compatible (`Box<dyn CoordinationStore>` will not compile).
133///   Deployments selecting a backend at runtime branch on the concrete
134///   store and box the [`SplitCoordinator`](spate_core::coordination::
135///   SplitCoordinator) instead — that trait is the dyn-compatible seam.
136pub trait CoordinationStore: Send + Sync + 'static {
137    /// The TTL every [`Ephemeral`](Keyspace::Ephemeral) write re-arms.
138    /// Fixed at store construction (per-write TTLs are not portable).
139    fn lease_ttl(&self) -> Duration;
140
141    /// Create `key` if absent.
142    fn create(
143        &self,
144        ks: Keyspace,
145        key: &str,
146        value: Vec<u8>,
147    ) -> impl Future<Output = Result<CasOutcome, StoreError>> + Send;
148
149    /// Replace `key` iff it is currently at `expected`.
150    fn update(
151        &self,
152        ks: Keyspace,
153        key: &str,
154        value: Vec<u8>,
155        expected: Revision,
156    ) -> impl Future<Output = Result<CasOutcome, StoreError>> + Send;
157
158    /// Read `key`'s current value, `None` when absent (deleted and
159    /// expired keys read as absent).
160    fn get(
161        &self,
162        ks: Keyspace,
163        key: &str,
164    ) -> impl Future<Output = Result<Option<Entry>, StoreError>> + Send;
165
166    /// Delete `key`; with `expected`, only if still at that revision.
167    /// Deleting an absent key wins vacuously.
168    fn delete(
169        &self,
170        ks: Keyspace,
171        key: &str,
172        expected: Option<Revision>,
173    ) -> impl Future<Output = Result<CasOutcome, StoreError>> + Send;
174
175    /// Snapshot + live tail of every key under `prefix`.
176    fn watch(
177        &self,
178        ks: Keyspace,
179        prefix: &str,
180    ) -> impl Future<Output = Result<WatchStream, StoreError>> + Send;
181
182    /// Point-in-time listing of every live key under `prefix`.
183    fn list(
184        &self,
185        ks: Keyspace,
186        prefix: &str,
187    ) -> impl Future<Output = Result<Vec<Entry>, StoreError>> + Send;
188}
spate_coordination/store/mod.rs

spate_coordination/store/
mod.rs
