Trait StateBackend 

pub trait StateBackend:
    Send
    + Sync
    + 'static {
    // Required methods
    fn key_group_capacity(&self) -> u32;
    fn write_partial<'life0, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        vnode: u32,
        assignment_version: u64,
        bytes: Bytes,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn read_partial<'life0, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        vnode: u32,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn write_commit_descriptor<'life0, 'life1, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        key: &'life1 str,
        bytes: Bytes,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn read_commit_descriptor<'life0, 'life1, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        key: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn read_sealed_commit_descriptor_bounded<'life0, 'life1, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        sealed: &'life1 SealedCommitDescriptor,
        max_bytes: u64,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn seal_checkpoint<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        assignment_fence: Option<&'life1 CheckpointAssignmentFence>,
        vnodes: &'life2 [u32],
        required_descriptors: &'life3 [String],
    ) -> Pin<Box<dyn Future<Output = Result<bool, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn checkpoint_seal_inventory<'life0, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
    ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointSealInventory>, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn prune_before<'life0, 'async_trait>(
        &'life0 self,
        before: u64,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn bind_state_namespace<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        deployment_id: &'life1 str,
        pipeline_identity: &'life2 PipelineIdentity,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn write_certified_partial<'life0, 'life1, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        vnode: u32,
        assignment_fence: &'life1 CheckpointAssignmentFence,
        writer_node_id: u64,
        _bytes: Bytes,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn write_certified_commit_descriptor<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        key: &'life1 str,
        assignment_fence: &'life2 CheckpointAssignmentFence,
        writer_node_id: u64,
        leader_proof: &'life3 LeaderProof,
        _bytes: Bytes,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait { ... }
    fn read_commit_descriptor_bounded<'life0, 'life1, 'async_trait>(
        &'life0 self,
        attempt: CheckpointAttempt,
        key: &'life1 str,
        max_bytes: u64,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn verify_checkpoint_artifact_metadata<'life0, 'life1, 'async_trait>(
        &'life0 self,
        inventory: &'life1 CheckpointSealInventory,
    ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn durability_scope(&self) -> StateBackendDurability { ... }
    fn uses_exact_object_store(&self, _expected: &Arc<dyn ObjectStore>) -> bool { ... }
    fn set_authoritative_version(&self, _version: u64) { ... }
    fn authoritative_version(&self) -> u64 { ... }
}

A pluggable state store used by streaming operators for partial aggregates and watermarks.

Object Safety

The trait is deliberately object-safe:

• No generic methods.
• No Self-returning methods.
• All async methods are async_trait boxed futures.

This lets the engine hold Arc<dyn StateBackend> and swap implementations at construction time without touching call sites.

Concurrency

Implementations must be Send + Sync + 'static. The engine expects to share a single backend across many worker tasks concurrently.

Idempotence

write_partial must be idempotent for a given (attempt, vnode) pair. An identical retry succeeds; conflicting bytes return StateBackendError::Conflict and must never overwrite the winner.

Required Methods

fn key_group_capacity(&self) -> u32

Number of stable key groups this backend can address.

Hosts must validate this value against the runtime’s VnodeRegistry before installing the backend. The raw representation is intentional: custom backends can report an invalid value and be rejected at admission instead of forcing construction-time panics.

fn write_partial<'life0, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, vnode: u32, assignment_version: u64, bytes: Bytes, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Persist a partial aggregate for (vnode, epoch).

assignment_version is the VnodeRegistry::assignment_version the writer observed when it started this write. Backends that implement the assignment fence compare it against their exact authoritative version and reject both stale and future writers. Backends that opt out of fencing accept any version.

fn read_partial<'life0, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, vnode: u32, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>

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

Read the partial aggregate for (attempt, vnode), if any.

fn write_commit_descriptor<'life0, 'life1, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, key: &'life1 str, bytes: Bytes, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Persist a local-runtime coordinated-commit descriptor for attempt under key.

key is opaque and unique within the attempt. Backends with an installed assignment authority must reject this uncertified path; cluster writers use Self::write_certified_commit_descriptor.

fn read_commit_descriptor<'life0, 'life1, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, key: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>

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

Read one exact coordinated-commit descriptor, if it exists.

Recovery and external commit paths already know the immutable keys from the checkpoint seal. Implementations must use a direct lookup here: a prefix listing per sink turns recovery into O(sinks x descriptors) I/O and unnecessarily materializes other sinks’ potentially large payloads.

fn read_sealed_commit_descriptor_bounded<'life0, 'life1, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, sealed: &'life1 SealedCommitDescriptor, max_bytes: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>

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

Read a descriptor admitted by an exact checkpoint seal.

Implementations must compare the currently stored provenance, length, and payload digest with sealed before returning the payload. Validating only the stored object’s own envelope is insufficient: a self-consistent replacement after _SEAL must not change the recovery cut or an external sink commit.

fn seal_checkpoint<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, assignment_fence: Option<&'life1 CheckpointAssignmentFence>, vnodes: &'life2 [u32], required_descriptors: &'life3 [String], ) -> Pin<Box<dyn Future<Output = Result<bool, StateBackendError>> + Send + 'async_trait>>

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

Durability barrier: true once every vnode partial and every required_descriptors key for attempt is persisted, sealing that exact attempt. Sinks do not commit until it returns Ok(true). In cluster mode, required_descriptors also binds every participant’s final readiness attestation, including participants with no vnodes or coordinated sinks.

fn checkpoint_seal_inventory<'life0, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointSealInventory>, StateBackendError>> + Send + 'async_trait>>

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

Read the canonical artifact inventory for an exact sealed attempt. Returns None when that attempt has no live seal or is below the durable prune floor.

fn prune_before<'life0, 'async_trait>( &'life0 self, before: u64, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Garbage-collect every partial and state seal whose epoch is strictly less than before. Called by the checkpoint coordinator after a successful checkpoint commit so the backend does not retain state for epochs that can never be recovered.

Required — there is intentionally no default. Without it an in-memory backend leaks a Bytes per vnode per checkpoint forever, and an object-store backend leaves state-v2/epoch=N/… objects forever. Implementations must durably make the retired attempt unreadable before deleting any artifact it attests. Test backends that truly do not accumulate state should implement Ok(()) explicitly so the choice is visible.

Provided Methods

fn bind_state_namespace<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, deployment_id: &'life1 str, pipeline_identity: &'life2 PipelineIdentity, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Bind this storage root to one deployment and logical pipeline before recovery or writes.

Durable custom backends must override this with an atomic create-once binding. Volatile custom backends have no restart-visible namespace, so the default only validates the supplied identity for them and fails closed for every durable scope.

fn write_certified_partial<'life0, 'life1, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, vnode: u32, assignment_fence: &'life1 CheckpointAssignmentFence, writer_node_id: u64, _bytes: Bytes, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Persist a cluster partial certified by the exact active assignment and writer process.

Cluster coordinators must use this method. The default fails closed so a custom backend cannot silently discard provenance and later publish an apparently certified seal.

fn write_certified_commit_descriptor<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, key: &'life1 str, assignment_fence: &'life2 CheckpointAssignmentFence, writer_node_id: u64, leader_proof: &'life3 LeaderProof, _bytes: Bytes, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Persist a cluster descriptor certified by the exact assignment, writer process, and checkpoint-initiating leader term.

Cluster coordinators must use this method. The default fails closed so a custom backend cannot silently discard provenance and later publish an apparently certified seal.

fn read_commit_descriptor_bounded<'life0, 'life1, 'async_trait>( &'life0 self, attempt: CheckpointAttempt, key: &'life1 str, max_bytes: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, StateBackendError>> + Send + 'async_trait>>

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

Read one descriptor while enforcing a private control-plane allocation bound.

Durable object-store implementations should reject from object metadata before loading the body. The default preserves custom backend compatibility while still validating the result; cluster deployments fail closed unless their backend supports certified partial writes.

fn verify_checkpoint_artifact_metadata<'life0, 'life1, 'async_trait>( &'life0 self, inventory: &'life1 CheckpointSealInventory, ) -> Pin<Box<dyn Future<Output = Result<(), StateBackendError>> + Send + 'async_trait>>

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

Prove from storage metadata that every artifact named by an exact seal still exists with its sealed length, without reading artifact payloads.

Cluster retention invokes this only while advancing the shared GC floor. Durable custom backends must override the fail-closed default with equivalent metadata evidence.

fn durability_scope(&self) -> StateBackendDurability

Failure scope survived by this backend.

Implementations default to StateBackendDurability::Volatile. A backend must report StateBackendDurability::ClusterShared only when every runtime node can reach the same durable namespace; merely using the object_store API or a file:// URL is not sufficient.

fn uses_exact_object_store(&self, _expected: &Arc<dyn ObjectStore>) -> bool

Whether this backend uses the exact object-store handle admitted by cluster startup.

The default fails closed. Object-store-backed cluster implementations override this with handle identity, not URL or namespace-string equivalence.

fn set_authoritative_version(&self, _version: u64)

Raise the backend’s authoritative assignment version — the exact VnodeRegistry::assignment_version it will accept on partial and descriptor writes. Hosts call this on boot after adopting an AssignmentSnapshot and on each subsequent rotation so stale writers from a deposed leader are fenced out.

Default is a no-op — backends that opt out of fencing (e.g. the in-process backend used for single-node deployments) inherit it unchanged. Monotonic on implementations that do fence: a call with version <= current is a no-op.

fn authoritative_version(&self) -> u64

Current authoritative assignment version. 0 means the fence is disabled — every caller version is accepted. Backends that do not fence return 0 unconditionally.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl StateBackend for InProcessBackend

impl StateBackend for ObjectStoreBackend