laminar_core/state/
backend.rs

1//! The `StateBackend` trait persists per-checkpoint-attempt vnode artifacts
2//! and exposes an exact-attempt durability seal. It is not the hot keyed-state
3//! implementation used while operators process records.
4
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use bytes::Bytes;
9use object_store::ObjectStore;
10use sha2::{Digest, Sha256};
11
12use crate::checkpoint::{
13    CheckpointAssignmentFence, CheckpointParticipant, LeaderProof, PipelineIdentity,
14    PIPELINE_IDENTITY_VERSION,
15};
16
17pub(crate) const STATE_NAMESPACE_RESOURCE: &str = "state-v2/_NAMESPACE";
18
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub(crate) struct StateNamespaceBinding {
21    pub(crate) deployment_id: String,
22    pub(crate) pipeline_identity: PipelineIdentity,
23}
24
25impl StateNamespaceBinding {
26    pub(crate) fn try_new(
27        deployment_id: &str,
28        pipeline_identity: &PipelineIdentity,
29    ) -> Result<Self, StateBackendError> {
30        let deployment =
31            uuid::Uuid::parse_str(deployment_id).map_err(|error| StateBackendError::Conflict {
32                resource: STATE_NAMESPACE_RESOURCE.into(),
33                message: format!("deployment ID is not a canonical UUID: {error}"),
34            })?;
35        if deployment.is_nil() || deployment.to_string() != deployment_id {
36            return Err(StateBackendError::Conflict {
37                resource: STATE_NAMESPACE_RESOURCE.into(),
38                message: "deployment ID must be a canonical non-nil UUID".into(),
39            });
40        }
41        if pipeline_identity.canonical_version != PIPELINE_IDENTITY_VERSION
42            || pipeline_identity.sha256.len() != 64
43            || !pipeline_identity
44                .sha256
45                .bytes()
46                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
47        {
48            return Err(StateBackendError::Conflict {
49                resource: STATE_NAMESPACE_RESOURCE.into(),
50                message: "pipeline identity is not canonical".into(),
51            });
52        }
53        Ok(Self {
54            deployment_id: deployment_id.to_owned(),
55            pipeline_identity: pipeline_identity.clone(),
56        })
57    }
58}
59
60/// Exact identity of one checkpoint attempt.
61///
62/// Runtime-generated and persisted attempts use one durable order: `epoch == checkpoint_id`.
63/// The duplicated fields remain in the current wire/storage layout and are validated at every
64/// boundary.
65#[derive(
66    Debug,
67    Clone,
68    Copy,
69    PartialEq,
70    Eq,
71    Hash,
72    serde::Serialize,
73    serde::Deserialize,
74    rkyv::Archive,
75    rkyv::Serialize,
76    rkyv::Deserialize,
77)]
78pub struct CheckpointAttempt {
79    /// Logical pipeline epoch represented by this checkpoint.
80    pub epoch: u64,
81    /// Globally unique, never-reused checkpoint attempt ID.
82    pub checkpoint_id: u64,
83}
84
85/// Explicit relation between two checkpoint attempts.
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub enum CheckpointAttemptRelation {
88    /// Both epoch and checkpoint ID are identical.
89    Exact,
90    /// Both epoch and checkpoint ID are lower than the compared attempt.
91    Older,
92    /// Both epoch and checkpoint ID are higher than the compared attempt.
93    Newer,
94    /// The dimensions move in different directions, or only one dimension changes.
95    Conflict,
96}
97
98impl CheckpointAttempt {
99    /// Construct an exact checkpoint attempt identity.
100    #[must_use]
101    pub const fn new(epoch: u64, checkpoint_id: u64) -> Self {
102        Self {
103            epoch,
104            checkpoint_id,
105        }
106    }
107
108    /// Construct a runtime attempt from the single durable checkpoint order.
109    #[must_use]
110    pub const fn canonical(checkpoint_id: u64) -> Self {
111        Self::new(checkpoint_id, checkpoint_id)
112    }
113
114    /// Whether both persisted fields represent the same durable checkpoint identity.
115    #[must_use]
116    pub const fn is_canonical(self) -> bool {
117        self.epoch != 0 && self.epoch == self.checkpoint_id
118    }
119
120    /// Relate this attempt to `other` without inventing a lexicographic order.
121    #[must_use]
122    pub const fn relation_to(self, other: Self) -> CheckpointAttemptRelation {
123        if self.epoch == other.epoch && self.checkpoint_id == other.checkpoint_id {
124            CheckpointAttemptRelation::Exact
125        } else if self.epoch < other.epoch && self.checkpoint_id < other.checkpoint_id {
126            CheckpointAttemptRelation::Older
127        } else if self.epoch > other.epoch && self.checkpoint_id > other.checkpoint_id {
128            CheckpointAttemptRelation::Newer
129        } else {
130            CheckpointAttemptRelation::Conflict
131        }
132    }
133}
134
135/// Current on-storage checkpoint-seal payload format.
136pub(crate) const CHECKPOINT_SEAL_VERSION: u32 = 7;
137
138/// Cluster identity that produced one immutable vnode partial.
139///
140/// The certificate digest binds the writer to the exact assignment version, ordered owner map,
141/// and boot-incarnation roster. Local-runtime partials do not carry this structure.
142#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
143pub struct SealedVnodeWriter {
144    /// Stable logical node identifier.
145    pub node_id: u64,
146    /// Boot-unique process identity certified for `node_id`.
147    pub boot_incarnation: uuid::Uuid,
148    /// SHA-256 of the exact checkpoint assignment certificate.
149    #[serde(with = "hex_digest_serde")]
150    pub assignment_certificate_digest: [u8; 32],
151}
152
153impl SealedVnodeWriter {
154    #[must_use]
155    pub(crate) fn from_fence(fence: &CheckpointAssignmentFence, node_id: u64) -> Option<Self> {
156        let boot_incarnation = fence.participant_incarnation(node_id)?;
157        Some(Self {
158            node_id,
159            boot_incarnation,
160            assignment_certificate_digest: fence.digest(),
161        })
162    }
163
164    /// Whether this writer is the exact process certified for its node by `fence`.
165    #[must_use]
166    pub fn matches_fence(&self, fence: &CheckpointAssignmentFence) -> bool {
167        self.node_id != 0
168            && !self.boot_incarnation.is_nil()
169            && fence.participant_incarnation(self.node_id) == Some(self.boot_incarnation)
170            && self.assignment_certificate_digest == fence.digest()
171    }
172}
173
174/// Immutable provenance and content attestation for one vnode partial admitted by a seal.
175///
176/// The assignment generation and exact writer certificate are the durable split-brain fence,
177/// while the payload digest lets recovery detect corruption without placing large state blobs in
178/// the seal itself.
179#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
180pub struct SealedVnodePartial {
181    /// Vnode whose partial was sealed.
182    pub vnode: u32,
183    /// Assignment generation observed by the writer.
184    pub assignment_version: u64,
185    /// Exact cluster writer, or `None` for a fence-free local-runtime partial.
186    pub writer: Option<SealedVnodeWriter>,
187    /// Raw partial payload length, excluding the storage envelope.
188    pub payload_len: u64,
189    /// SHA-256 of the raw partial payload.
190    pub payload_sha256: String,
191}
192
193impl SealedVnodePartial {
194    pub(crate) fn new(
195        vnode: u32,
196        assignment_version: u64,
197        writer: Option<SealedVnodeWriter>,
198        payload: &[u8],
199    ) -> Self {
200        Self {
201            vnode,
202            assignment_version,
203            writer,
204            payload_len: payload.len() as u64,
205            payload_sha256: digest_hex(&sha256(payload)),
206        }
207    }
208}
209
210/// Exact process and leader term that produced one immutable commit descriptor.
211#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
212pub struct SealedCommitDescriptorWriter {
213    /// Boot-exact process that wrote the descriptor.
214    pub participant: CheckpointParticipant,
215    /// SHA-256 of the exact checkpoint assignment certificate.
216    #[serde(with = "hex_digest_serde")]
217    pub assignment_certificate_digest: [u8; 32],
218    /// Durable leader term that initiated the checkpoint attempt.
219    pub leader_proof: LeaderProof,
220}
221
222impl SealedCommitDescriptorWriter {
223    #[must_use]
224    pub(crate) fn from_fence(
225        fence: &CheckpointAssignmentFence,
226        writer_node_id: u64,
227        leader_proof: &LeaderProof,
228    ) -> Option<Self> {
229        let boot_incarnation = fence.participant_incarnation(writer_node_id)?;
230        if !fence.is_canonical()
231            || !leader_proof.is_canonical()
232            || fence.participant_incarnation(leader_proof.owner.node_id)
233                != Some(leader_proof.owner.boot_id)
234        {
235            return None;
236        }
237        Some(Self {
238            participant: CheckpointParticipant {
239                node_id: writer_node_id,
240                boot_incarnation,
241            },
242            assignment_certificate_digest: fence.digest(),
243            leader_proof: leader_proof.clone(),
244        })
245    }
246
247    /// Whether this writer and checkpoint authority exactly match `fence`.
248    #[must_use]
249    pub fn matches_fence(&self, fence: &CheckpointAssignmentFence) -> bool {
250        self.participant.node_id != 0
251            && !self.participant.boot_incarnation.is_nil()
252            && fence.participant_incarnation(self.participant.node_id)
253                == Some(self.participant.boot_incarnation)
254            && self.assignment_certificate_digest == fence.digest()
255            && self.leader_proof.is_canonical()
256            && fence.participant_incarnation(self.leader_proof.owner.node_id)
257                == Some(self.leader_proof.owner.boot_id)
258    }
259}
260
261/// Immutable provenance and content attestation for one commit descriptor admitted by a seal.
262#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
263pub struct SealedCommitDescriptor {
264    /// Exact descriptor key within the checkpoint attempt.
265    pub key: String,
266    /// Assignment generation observed by the writer, or zero for a local-runtime descriptor.
267    pub assignment_version: u64,
268    /// Exact cluster writer and authority, or `None` for a local-runtime descriptor.
269    pub writer: Option<SealedCommitDescriptorWriter>,
270    /// Raw descriptor payload length, excluding its storage envelope.
271    pub payload_len: u64,
272    /// SHA-256 of the raw descriptor payload.
273    pub payload_sha256: String,
274}
275
276impl SealedCommitDescriptor {
277    pub(crate) fn new(
278        key: String,
279        assignment_version: u64,
280        writer: Option<SealedCommitDescriptorWriter>,
281        payload: &[u8],
282    ) -> Self {
283        Self {
284            key,
285            assignment_version,
286            writer,
287            payload_len: payload.len() as u64,
288            payload_sha256: digest_hex(&sha256(payload)),
289        }
290    }
291}
292
293pub(crate) fn sha256(bytes: &[u8]) -> [u8; 32] {
294    Sha256::digest(bytes).into()
295}
296
297pub(crate) fn digest_hex(digest: &[u8; 32]) -> String {
298    const HEX: &[u8; 16] = b"0123456789abcdef";
299    let mut encoded = String::with_capacity(digest.len() * 2);
300    for byte in digest {
301        encoded.push(HEX[(byte >> 4) as usize] as char);
302        encoded.push(HEX[(byte & 0x0f) as usize] as char);
303    }
304    encoded
305}
306
307mod hex_digest_serde {
308    use serde::{Deserialize, Deserializer, Serializer};
309
310    pub(super) fn serialize<S>(digest: &[u8; 32], serializer: S) -> Result<S::Ok, S::Error>
311    where
312        S: Serializer,
313    {
314        serializer.serialize_str(&super::digest_hex(digest))
315    }
316
317    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<[u8; 32], D::Error>
318    where
319        D: Deserializer<'de>,
320    {
321        let encoded = String::deserialize(deserializer)?;
322        if encoded.len() != 64
323            || !encoded
324                .bytes()
325                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
326        {
327            return Err(serde::de::Error::custom(
328                "SHA-256 digest must be 64 lowercase hexadecimal characters",
329            ));
330        }
331        let mut digest = [0_u8; 32];
332        for (index, byte) in digest.iter_mut().enumerate() {
333            let offset = index * 2;
334            *byte = u8::from_str_radix(&encoded[offset..offset + 2], 16)
335                .map_err(serde::de::Error::custom)?;
336        }
337        Ok(digest)
338    }
339}
340
341/// Canonical artifact inventory bound into an immutable checkpoint seal.
342///
343/// A seal is meaningful only for the exact set that was proven durable. The
344/// inventory is exposed so recovery-side protocols (notably the designated
345/// external-sink committer) can independently validate the participant markers
346/// that the seal admitted.
347#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
348pub struct CheckpointSealInventory {
349    /// Exact attempt named by the seal.
350    pub attempt: CheckpointAttempt,
351    /// Exact cluster assignment certificate, or `None` for a local-runtime seal.
352    pub assignment_fence: Option<CheckpointAssignmentFence>,
353    /// Assignment generation attested by every vnode partial, including local-runtime seals.
354    pub assignment_version: u64,
355    /// Sorted, duplicate-free vnode partials required by the seal.
356    pub required_vnodes: Vec<u32>,
357    /// Per-vnode writer, assignment-generation, length, and digest attestations.
358    pub sealed_partials: Vec<SealedVnodePartial>,
359    /// Sorted, duplicate-free commit-descriptor keys required by the seal.
360    pub required_descriptors: Vec<String>,
361    /// Per-descriptor writer, authority, length, and digest attestations.
362    pub sealed_descriptors: Vec<SealedCommitDescriptor>,
363}
364
365impl CheckpointSealInventory {
366    /// Attestation for one exact descriptor key.
367    #[must_use]
368    pub fn sealed_descriptor(&self, key: &str) -> Option<&SealedCommitDescriptor> {
369        self.sealed_descriptors
370            .binary_search_by(|descriptor| descriptor.key.as_str().cmp(key))
371            .ok()
372            .map(|index| &self.sealed_descriptors[index])
373    }
374
375    /// Exact leader term certified by every cluster descriptor in this seal.
376    ///
377    /// An empty descriptor inventory has no descriptor authority. Local descriptors return
378    /// `None`; a mixed, incomplete, or assignment-mismatched inventory is rejected.
379    ///
380    /// # Errors
381    ///
382    /// Returns an error when the inventory is noncanonical, incomplete, or mixes authorities.
383    pub fn descriptor_leader_proof(&self) -> Result<Option<&LeaderProof>, String> {
384        if !self.attempt.is_canonical() {
385            return Err(
386                "checkpoint seal inventory must use one nonzero canonical checkpoint ID".into(),
387            );
388        }
389        if self.sealed_descriptors.len() != self.required_descriptors.len() {
390            return Err(
391                "checkpoint seal descriptor attestations do not exactly cover its keys".into(),
392            );
393        }
394        let descriptor_keys: Vec<&str> = self
395            .sealed_descriptors
396            .iter()
397            .map(|descriptor| descriptor.key.as_str())
398            .collect();
399        let required_keys: Vec<&str> = self
400            .required_descriptors
401            .iter()
402            .map(String::as_str)
403            .collect();
404        if descriptor_keys != required_keys {
405            return Err(
406                "checkpoint seal descriptor attestations do not exactly cover its keys".into(),
407            );
408        }
409
410        let mut authority = None;
411        for descriptor in &self.sealed_descriptors {
412            let valid_digest = descriptor.payload_sha256.len() == 64
413                && descriptor
414                    .payload_sha256
415                    .bytes()
416                    .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte));
417            if !valid_digest {
418                return Err(format!(
419                    "checkpoint seal has invalid descriptor digest for '{}'",
420                    descriptor.key
421                ));
422            }
423            match (&self.assignment_fence, &descriptor.writer) {
424                (Some(fence), Some(writer))
425                    if descriptor.assignment_version == fence.assignment_version
426                        && writer.matches_fence(fence) =>
427                {
428                    if authority.is_some_and(|proof| proof != &writer.leader_proof) {
429                        return Err(
430                            "checkpoint seal descriptors name different leader terms".into()
431                        );
432                    }
433                    authority = Some(&writer.leader_proof);
434                }
435                (None, None) if descriptor.assignment_version == 0 => {}
436                _ => {
437                    return Err(format!(
438                        "checkpoint seal has invalid writer certificate for descriptor '{}'",
439                        descriptor.key
440                    ));
441                }
442            }
443        }
444        Ok(authority)
445    }
446}
447
448/// Structured body of an exact-attempt state seal.
449///
450/// This remains crate-private because callers consume only the canonical
451/// [`CheckpointSealInventory`]; the additional fields are durability audit and
452/// fencing metadata owned by backend implementations.
453#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
454pub(crate) struct CheckpointSeal {
455    pub version: u32,
456    pub attempt: CheckpointAttempt,
457    pub instance_id: String,
458    pub execution_id: uuid::Uuid,
459    pub assignment_fence: Option<CheckpointAssignmentFence>,
460    pub assignment_version: u64,
461    pub required_vnodes: Vec<u32>,
462    pub sealed_partials: Vec<SealedVnodePartial>,
463    pub required_descriptors: Vec<String>,
464    pub sealed_descriptors: Vec<SealedCommitDescriptor>,
465}
466
467impl CheckpointSeal {
468    pub(crate) fn new(
469        instance_id: String,
470        execution_id: uuid::Uuid,
471        mut inventory: CheckpointSealInventory,
472    ) -> Self {
473        inventory.required_vnodes.sort_unstable();
474        inventory.required_vnodes.dedup();
475        inventory.required_descriptors.sort_unstable();
476        inventory.required_descriptors.dedup();
477        inventory
478            .sealed_descriptors
479            .sort_unstable_by(|left, right| left.key.cmp(&right.key));
480        inventory
481            .sealed_descriptors
482            .dedup_by(|left, right| left.key == right.key);
483        inventory
484            .sealed_partials
485            .sort_unstable_by_key(|partial| partial.vnode);
486        inventory
487            .sealed_partials
488            .dedup_by_key(|partial| partial.vnode);
489        Self {
490            version: CHECKPOINT_SEAL_VERSION,
491            attempt: inventory.attempt,
492            instance_id,
493            execution_id,
494            assignment_fence: inventory.assignment_fence,
495            assignment_version: inventory.assignment_version,
496            required_vnodes: inventory.required_vnodes,
497            sealed_partials: inventory.sealed_partials,
498            required_descriptors: inventory.required_descriptors,
499            sealed_descriptors: inventory.sealed_descriptors,
500        }
501    }
502
503    pub(crate) fn inventory(&self) -> CheckpointSealInventory {
504        CheckpointSealInventory {
505            attempt: self.attempt,
506            assignment_fence: self.assignment_fence.clone(),
507            assignment_version: self.assignment_version,
508            required_vnodes: self.required_vnodes.clone(),
509            sealed_partials: self.sealed_partials.clone(),
510            required_descriptors: self.required_descriptors.clone(),
511            sealed_descriptors: self.sealed_descriptors.clone(),
512        }
513    }
514
515    pub(crate) fn validate(&self) -> Result<(), String> {
516        if self.version != CHECKPOINT_SEAL_VERSION {
517            return Err(format!(
518                "unsupported checkpoint seal version {}; expected {CHECKPOINT_SEAL_VERSION}",
519                self.version
520            ));
521        }
522        if self.instance_id.is_empty() || self.execution_id.is_nil() {
523            return Err("checkpoint seal has an empty writer identity".into());
524        }
525        if !self.attempt.is_canonical() {
526            return Err("checkpoint seal must use one nonzero canonical checkpoint ID".into());
527        }
528        if self
529            .assignment_fence
530            .as_ref()
531            .is_some_and(|fence| !fence.is_canonical())
532        {
533            return Err("checkpoint seal has a non-canonical assignment certificate".into());
534        }
535        if self
536            .assignment_fence
537            .as_ref()
538            .is_some_and(|fence| fence.assignment_version != self.assignment_version)
539        {
540            return Err("checkpoint seal assignment version does not match its certificate".into());
541        }
542
543        let mut canonical_vnodes = self.required_vnodes.clone();
544        canonical_vnodes.sort_unstable();
545        canonical_vnodes.dedup();
546        if canonical_vnodes != self.required_vnodes {
547            return Err("checkpoint seal vnode inventory is not canonical".into());
548        }
549        let partial_vnodes: Vec<u32> = self
550            .sealed_partials
551            .iter()
552            .map(|partial| partial.vnode)
553            .collect();
554        if partial_vnodes != canonical_vnodes {
555            return Err(
556                "checkpoint seal partial attestations do not exactly cover its vnodes".into(),
557            );
558        }
559        for partial in &self.sealed_partials {
560            let valid_digest = |digest: &str| {
561                digest.len() == 64
562                    && digest
563                        .bytes()
564                        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
565            };
566            if partial.assignment_version != self.assignment_version
567                || !valid_digest(&partial.payload_sha256)
568            {
569                return Err(format!(
570                    "checkpoint seal has invalid provenance for vnode {}",
571                    partial.vnode
572                ));
573            }
574            match (&self.assignment_fence, &partial.writer) {
575                (Some(fence), Some(writer)) if writer.matches_fence(fence) => {}
576                (None, None) => {}
577                _ => {
578                    return Err(format!(
579                        "checkpoint seal has invalid writer certificate for vnode {}",
580                        partial.vnode
581                    ));
582                }
583            }
584        }
585
586        let mut canonical_descriptors = self.required_descriptors.clone();
587        canonical_descriptors.sort_unstable();
588        canonical_descriptors.dedup();
589        if canonical_descriptors != self.required_descriptors
590            || canonical_descriptors.iter().any(String::is_empty)
591        {
592            return Err("checkpoint seal descriptor inventory is not canonical".into());
593        }
594        self.inventory().descriptor_leader_proof()?;
595        Ok(())
596    }
597}
598
599/// Failure scope survived by a state backend.
600///
601/// The variants are ordered by strength so admission can require a minimum
602/// scope without collapsing node-local persistence and cluster-shared storage
603/// into the same boolean.
604#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
605pub enum StateBackendDurability {
606    /// State is lost when the current process exits.
607    Volatile,
608    /// State survives a process restart on the same node, but peers are not
609    /// guaranteed to be able to read it after that node fails.
610    NodeDurable,
611    /// State is durably stored in a namespace reachable by every cluster node.
612    ClusterShared,
613}
614
615impl StateBackendDurability {
616    /// Whether this scope is at least as strong as `required`.
617    #[must_use]
618    pub fn satisfies(self, required: Self) -> bool {
619        self >= required
620    }
621
622    /// Classify a filesystem/object-store URL by the failure domain it can
623    /// survive. Unknown and process-local schemes fail closed as volatile.
624    #[must_use]
625    pub fn for_storage_url(url: &str) -> Self {
626        match url.split_once("://").map(|(scheme, _)| scheme) {
627            Some("file")
628                if crate::checkpoint::object_store_builder::is_absolute_local_file_url(url) =>
629            {
630                Self::NodeDurable
631            }
632            Some("s3" | "gs" | "az" | "abfs" | "abfss") => Self::ClusterShared,
633            _ => Self::Volatile,
634        }
635    }
636}
637
638/// Errors a [`StateBackend`] can raise.
639#[derive(Debug, thiserror::Error)]
640pub enum StateBackendError {
641    /// Underlying I/O failure (filesystem, `object_store`, network).
642    #[error("I/O error: {0}")]
643    Io(String),
644
645    /// Serialization or framing error in the stored partial bytes.
646    #[error("serialization error: {0}")]
647    Serialization(String),
648
649    /// The partial for `(vnode, epoch)` is not present.
650    #[error("not found: vnode={vnode} epoch={epoch}")]
651    NotFound {
652        /// Virtual node ID.
653        vnode: u32,
654        /// Epoch number.
655        epoch: u64,
656    },
657
658    /// The caller's assignment version is older than the backend's
659    /// authoritative version. Thrown by [`StateBackend::write_partial`]
660    /// when a stale writer (for example, a deposed vnode owner)
661    /// attempts to persist state at a version that has since been
662    /// superseded. The caller should abandon the write, refresh its
663    /// assignment snapshot, and retry at the new version.
664    #[error("stale assignment version: caller={caller} < authoritative={authoritative}")]
665    StaleVersion {
666        /// Version the writer believes is current.
667        caller: u64,
668        /// Authoritative version seen by the backend.
669        authoritative: u64,
670    },
671
672    /// The caller claims an assignment generation this backend has not adopted. Accepting it
673    /// would let an unverified future owner bypass the local assignment fence.
674    #[error("future assignment version: caller={caller} > authoritative={authoritative}")]
675    FutureVersion {
676        /// Version the writer claims is current.
677        caller: u64,
678        /// Exact authoritative version adopted by this backend.
679        authoritative: u64,
680    },
681
682    /// A create-once artifact or seal already exists with different bytes or
683    /// fencing metadata. Retrying the identical write is accepted; conflicting
684    /// content is never overwritten.
685    #[error("immutable state conflict at {resource}: {message}")]
686    Conflict {
687        /// Object path or in-process key that conflicted.
688        resource: String,
689        /// Human-readable conflict detail.
690        message: String,
691    },
692}
693
694/// A pluggable state store used by streaming operators for partial
695/// aggregates and watermarks.
696///
697/// ## Object Safety
698///
699/// The trait is deliberately object-safe:
700///
701/// - No generic methods.
702/// - No `Self`-returning methods.
703/// - All async methods are `async_trait` boxed futures.
704///
705/// This lets the engine hold `Arc<dyn StateBackend>` and swap
706/// implementations at construction time without touching call sites.
707///
708/// ## Concurrency
709///
710/// Implementations must be `Send + Sync + 'static`. The engine expects
711/// to share a single backend across many worker tasks concurrently.
712///
713/// ## Idempotence
714///
715/// [`write_partial`](Self::write_partial) must be idempotent for a given
716/// `(attempt, vnode)` pair. An identical retry succeeds; conflicting bytes
717/// return [`StateBackendError::Conflict`] and must never overwrite the winner.
718#[async_trait]
719pub trait StateBackend: Send + Sync + 'static {
720    /// Number of stable key groups this backend can address.
721    ///
722    /// Hosts must validate this value against the runtime's [`VnodeRegistry`] before installing
723    /// the backend. The raw representation is intentional: custom backends can report an invalid
724    /// value and be rejected at admission instead of forcing construction-time panics.
725    ///
726    /// [`VnodeRegistry`]: crate::state::VnodeRegistry
727    fn key_group_capacity(&self) -> u32;
728
729    /// Bind this storage root to one deployment and logical pipeline before recovery or writes.
730    ///
731    /// Durable custom backends must override this with an atomic create-once binding. Volatile
732    /// custom backends have no restart-visible namespace, so the default only validates the
733    /// supplied identity for them and fails closed for every durable scope.
734    async fn bind_state_namespace(
735        &self,
736        deployment_id: &str,
737        pipeline_identity: &PipelineIdentity,
738    ) -> Result<(), StateBackendError> {
739        StateNamespaceBinding::try_new(deployment_id, pipeline_identity)?;
740        if self.durability_scope() == StateBackendDurability::Volatile {
741            return Ok(());
742        }
743        Err(StateBackendError::Conflict {
744            resource: STATE_NAMESPACE_RESOURCE.into(),
745            message: "durable backend does not implement immutable namespace binding".into(),
746        })
747    }
748
749    /// Persist a partial aggregate for `(vnode, epoch)`.
750    ///
751    /// `assignment_version` is the [`VnodeRegistry::assignment_version`]
752    /// the writer observed when it started this write. Backends that
753    /// implement the assignment fence compare it against their exact
754    /// authoritative version and reject both stale and future writers.
755    /// Backends that opt out of fencing accept any version.
756    ///
757    /// [`VnodeRegistry::assignment_version`]: crate::state::VnodeRegistry::assignment_version
758    async fn write_partial(
759        &self,
760        attempt: CheckpointAttempt,
761        vnode: u32,
762        assignment_version: u64,
763        bytes: Bytes,
764    ) -> Result<(), StateBackendError>;
765
766    /// Persist a cluster partial certified by the exact active assignment and writer process.
767    ///
768    /// Cluster coordinators must use this method. The default fails closed so a custom backend
769    /// cannot silently discard provenance and later publish an apparently certified seal.
770    async fn write_certified_partial(
771        &self,
772        attempt: CheckpointAttempt,
773        vnode: u32,
774        assignment_fence: &CheckpointAssignmentFence,
775        writer_node_id: u64,
776        _bytes: Bytes,
777    ) -> Result<(), StateBackendError> {
778        Err(StateBackendError::Conflict {
779            resource: format!(
780                "state-v2/epoch={}/checkpoint={}/vnode={vnode}/partial.bin",
781                attempt.epoch, attempt.checkpoint_id
782            ),
783            message: format!(
784                "backend cannot persist certified vnode partials for assignment {} writer {}",
785                assignment_fence.assignment_version, writer_node_id
786            ),
787        })
788    }
789
790    /// Read the partial aggregate for `(attempt, vnode)`, if any.
791    async fn read_partial(
792        &self,
793        attempt: CheckpointAttempt,
794        vnode: u32,
795    ) -> Result<Option<Bytes>, StateBackendError>;
796
797    /// Persist a local-runtime coordinated-commit descriptor for `attempt` under `key`.
798    ///
799    /// `key` is opaque and unique within the attempt. Backends with an installed assignment
800    /// authority must reject this uncertified path; cluster writers use
801    /// [`Self::write_certified_commit_descriptor`].
802    async fn write_commit_descriptor(
803        &self,
804        attempt: CheckpointAttempt,
805        key: &str,
806        bytes: Bytes,
807    ) -> Result<(), StateBackendError>;
808
809    /// Persist a cluster descriptor certified by the exact assignment, writer process, and
810    /// checkpoint-initiating leader term.
811    ///
812    /// Cluster coordinators must use this method. The default fails closed so a custom backend
813    /// cannot silently discard provenance and later publish an apparently certified seal.
814    async fn write_certified_commit_descriptor(
815        &self,
816        attempt: CheckpointAttempt,
817        key: &str,
818        assignment_fence: &CheckpointAssignmentFence,
819        writer_node_id: u64,
820        leader_proof: &LeaderProof,
821        _bytes: Bytes,
822    ) -> Result<(), StateBackendError> {
823        Err(StateBackendError::Conflict {
824            resource: format!(
825                "state-v2/epoch={}/checkpoint={}/commit/{key}",
826                attempt.epoch, attempt.checkpoint_id
827            ),
828            message: format!(
829                "backend cannot persist certified commit descriptors for assignment {} writer {} leader token {}",
830                assignment_fence.assignment_version,
831                writer_node_id,
832                leader_proof.fencing_token
833            ),
834        })
835    }
836
837    /// Read one exact coordinated-commit descriptor, if it exists.
838    ///
839    /// Recovery and external commit paths already know the immutable keys from
840    /// the checkpoint seal. Implementations must use a direct lookup here: a
841    /// prefix listing per sink turns recovery into O(sinks x descriptors) I/O
842    /// and unnecessarily materializes other sinks' potentially large payloads.
843    async fn read_commit_descriptor(
844        &self,
845        attempt: CheckpointAttempt,
846        key: &str,
847    ) -> Result<Option<Bytes>, StateBackendError>;
848
849    /// Read one descriptor while enforcing a private control-plane allocation bound.
850    ///
851    /// Durable object-store implementations should reject from object metadata before loading the
852    /// body. The default preserves custom backend compatibility while still validating the result;
853    /// cluster deployments fail closed unless their backend supports certified partial writes.
854    async fn read_commit_descriptor_bounded(
855        &self,
856        attempt: CheckpointAttempt,
857        key: &str,
858        max_bytes: u64,
859    ) -> Result<Option<Bytes>, StateBackendError> {
860        let bytes = self.read_commit_descriptor(attempt, key).await?;
861        if bytes
862            .as_ref()
863            .is_some_and(|bytes| bytes.len() as u64 > max_bytes)
864        {
865            return Err(StateBackendError::Conflict {
866                resource: format!(
867                    "state-v2/epoch={}/checkpoint={}/commit/{key}",
868                    attempt.epoch, attempt.checkpoint_id
869                ),
870                message: format!("commit descriptor exceeds the {max_bytes}-byte read bound"),
871            });
872        }
873        Ok(bytes)
874    }
875
876    /// Read a descriptor admitted by an exact checkpoint seal.
877    ///
878    /// Implementations must compare the currently stored provenance, length, and payload digest
879    /// with `sealed` before returning the payload. Validating only the stored object's own envelope
880    /// is insufficient: a self-consistent replacement after `_SEAL` must not change the recovery
881    /// cut or an external sink commit.
882    async fn read_sealed_commit_descriptor_bounded(
883        &self,
884        attempt: CheckpointAttempt,
885        sealed: &SealedCommitDescriptor,
886        max_bytes: u64,
887    ) -> Result<Option<Bytes>, StateBackendError>;
888
889    /// Durability barrier: true once every `vnode` partial and every
890    /// `required_descriptors` key for `attempt` is persisted, sealing that exact attempt.
891    /// Sinks do not commit until it returns `Ok(true)`. In cluster mode,
892    /// `required_descriptors` also binds every participant's final readiness attestation,
893    /// including participants with no vnodes or coordinated sinks.
894    async fn seal_checkpoint(
895        &self,
896        attempt: CheckpointAttempt,
897        assignment_fence: Option<&CheckpointAssignmentFence>,
898        vnodes: &[u32],
899        required_descriptors: &[String],
900    ) -> Result<bool, StateBackendError>;
901
902    /// Read the canonical artifact inventory for an exact sealed attempt.
903    /// Returns `None` when that attempt has no live seal or is below the durable prune floor.
904    async fn checkpoint_seal_inventory(
905        &self,
906        attempt: CheckpointAttempt,
907    ) -> Result<Option<CheckpointSealInventory>, StateBackendError>;
908
909    /// Prove from storage metadata that every artifact named by an exact seal still exists with
910    /// its sealed length, without reading artifact payloads.
911    ///
912    /// Cluster retention invokes this only while advancing the shared GC floor. Durable custom
913    /// backends must override the fail-closed default with equivalent metadata evidence.
914    async fn verify_checkpoint_artifact_metadata(
915        &self,
916        inventory: &CheckpointSealInventory,
917    ) -> Result<(), StateBackendError> {
918        Err(StateBackendError::Conflict {
919            resource: format!(
920                "state-v2/epoch={}/checkpoint={}",
921                inventory.attempt.epoch, inventory.attempt.checkpoint_id
922            ),
923            message: "backend does not implement metadata-only sealed-artifact verification".into(),
924        })
925    }
926
927    /// Garbage-collect every partial and state seal whose epoch is
928    /// strictly less than `before`. Called by the checkpoint
929    /// coordinator after a successful checkpoint commit so the backend
930    /// does not retain state for epochs that can never be recovered.
931    ///
932    /// Required — there is intentionally no default. Without it an
933    /// in-memory backend leaks a `Bytes` per vnode per checkpoint
934    /// forever, and an object-store backend leaves `state-v2/epoch=N/…` objects
935    /// forever. Implementations must durably make the retired attempt unreadable before deleting
936    /// any artifact it attests. Test backends that truly do not accumulate state should
937    /// implement `Ok(())` explicitly so the choice is visible.
938    async fn prune_before(&self, before: u64) -> Result<(), StateBackendError>;
939
940    /// Failure scope survived by this backend.
941    ///
942    /// Implementations default to [`StateBackendDurability::Volatile`]. A
943    /// backend must report [`StateBackendDurability::ClusterShared`] only when
944    /// every runtime node can reach the same durable namespace; merely using
945    /// the `object_store` API or a `file://` URL is not sufficient.
946    fn durability_scope(&self) -> StateBackendDurability {
947        StateBackendDurability::Volatile
948    }
949
950    /// Whether this backend uses the exact object-store handle admitted by cluster startup.
951    ///
952    /// The default fails closed. Object-store-backed cluster implementations override this with
953    /// handle identity, not URL or namespace-string equivalence.
954    fn uses_exact_object_store(&self, _expected: &Arc<dyn ObjectStore>) -> bool {
955        false
956    }
957
958    /// Raise the backend's authoritative assignment version — the
959    /// exact [`VnodeRegistry::assignment_version`] it will accept on
960    /// partial and descriptor writes. Hosts call this on boot
961    /// after adopting an `AssignmentSnapshot` and on each subsequent
962    /// rotation so stale writers from a deposed leader are fenced out.
963    ///
964    /// Default is a no-op — backends that opt out of fencing (e.g. the
965    /// in-process backend used for single-node deployments) inherit it
966    /// unchanged. Monotonic on implementations that do fence: a call
967    /// with `version <= current` is a no-op.
968    ///
969    /// [`VnodeRegistry::assignment_version`]: crate::state::VnodeRegistry::assignment_version
970    fn set_authoritative_version(&self, _version: u64) {}
971
972    /// Current authoritative assignment version. `0` means the fence is
973    /// disabled — every caller version is accepted. Backends that do
974    /// not fence return `0` unconditionally.
975    fn authoritative_version(&self) -> u64 {
976        0
977    }
978}
979
980const _: Option<&dyn StateBackend> = None;
981
982#[cfg(test)]
983mod tests {
984    use super::*;
985
986    #[test]
987    fn durability_scopes_are_ordered_by_failure_domain() {
988        assert!(
989            StateBackendDurability::ClusterShared.satisfies(StateBackendDurability::NodeDurable)
990        );
991        assert!(StateBackendDurability::NodeDurable.satisfies(StateBackendDurability::NodeDurable));
992        assert!(
993            !StateBackendDurability::NodeDurable.satisfies(StateBackendDurability::ClusterShared)
994        );
995        assert!(!StateBackendDurability::Volatile.satisfies(StateBackendDurability::NodeDurable));
996    }
997
998    #[test]
999    fn checkpoint_attempt_supports_serde_rkyv_hash_and_explicit_relation() {
1000        let attempt = CheckpointAttempt::canonical(42);
1001        let json = serde_json::to_vec(&attempt).unwrap();
1002        assert_eq!(
1003            serde_json::from_slice::<CheckpointAttempt>(&json).unwrap(),
1004            attempt
1005        );
1006
1007        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&attempt).unwrap();
1008        assert_eq!(
1009            rkyv::from_bytes::<CheckpointAttempt, rkyv::rancor::Error>(&bytes).unwrap(),
1010            attempt
1011        );
1012
1013        let mut hashed = rustc_hash::FxHashSet::default();
1014        hashed.insert(attempt);
1015        assert!(hashed.contains(&attempt));
1016
1017        use CheckpointAttemptRelation::{Conflict, Exact, Newer, Older};
1018        assert_eq!(attempt.relation_to(attempt), Exact);
1019        assert_eq!(CheckpointAttempt::canonical(41).relation_to(attempt), Older);
1020        assert_eq!(CheckpointAttempt::canonical(43).relation_to(attempt), Newer);
1021
1022        let malformed = CheckpointAttempt::new(41, 43);
1023        assert!(!malformed.is_canonical());
1024        assert_eq!(malformed.relation_to(attempt), Conflict);
1025    }
1026}
laminar_core/state/backend.rs

laminar_core/state/
backend.rs
