Projections

A projection turns the persistence query side into a managed read-side processor. It pairs a restartable SourceProvider, a fresh handler factory, and a fenced offset store, then owns cursor commits, recovery, restart backoff, lifecycle, and local scale-out. The API mirrors the core Akka Projection model with Rust-native builders.

A Projection is an inert blueprint. Building or configuring it opens no query and starts no handler, actor, or task. spawn or run_on materializes the blueprint on a ProjectionRuntime and returns a ProjectionHandle.

The Rust examples on this page are imported from crates/datum-persistence/tests/docs_projection.rs and compile and run as integration tests.

Choose delivery semantics

Delivery guarantees apply at the handler boundary and cursor store. Select the weakest mode that meets the read model's loss and replay requirements.

Mode	Handler and cursor order	Commit cadence	Failure boundary
At-least-once (ALO)	Handle, then save	Single-envelope ALO saves after 100 envelopes or 500 ms by default, whichever comes first	A crash can replay the unsaved cadence tail plus duplicates permitted at an inclusive provider boundary. The handler must be idempotent.
At-most-once (AMO)	Save, then handle	One fenced save before every envelope; no batching	A crash can lose the in-flight envelope, but a committed cursor prevents its replay.
Exactly-once PostgreSQL (EO)	Lock and validate the offset row, run the handler, write the cursor, and commit in one SQL transaction	One transaction per envelope	User writes and the cursor commit or roll back together, except for the documented Skip case below.
Grouped ALO	Buffer, handle one complete group, then save its final cursor	One save after every successful group; the ALO save cadence does not apply	The whole group is the retry or skip unit. A partial buffer at stop or pause is dropped without cursor advance and is read again.
Grouped EO PostgreSQL	Handle the complete group and write its final cursor in one SQL transaction	One transaction per group	All user writes in the group and its cursor commit or roll back together, except for Skip.

Grouped builders close a group after 20 envelopes or 500 ms by default, whichever happens first. Override those thresholds with GroupConfig. Use SaveOffsetConfig only for single-envelope ALO; AMO and grouped modes deliberately ignore it.

Exactly-once is PostgreSQL-only. The runtime supplies a PostgresProjectionTx to ExactlyOncePostgresHandler or GroupedExactlyOncePostgresHandler; its query helpers keep commit and rollback under runtime control. Cassandra cannot atomically combine user-table writes with the offset row and therefore rejects EO configuration.

The exactly-once Skip exception

RecoveryStrategy::Skip and exhausted RetryAndSkip must advance past failed work. In EO mode the failed handler transaction first rolls back, then the runtime commits the cursor alone in a separate transaction. Read-model state and the stored cursor intentionally disagree for that skipped envelope or group. This is the documented exception to the EO state/cursor agreement.

The Cassandra AMO cost

Cassandra uses lightweight transactions (LWTs) for generation and expected-position fencing. ALO pays that Paxos cost at its configured save cadence and grouped ALO pays once per successful group. AMO saves before every envelope by definition, so AMO on Cassandra costs one LWT per envelope.

Make ALO handlers idempotent

ALO is the normal choice for rebuildable read models, but a successful handler can run again when the process dies before its cursor is saved. Treat (persistence_id, sequence_nr) as a natural event identity and make the read-model mutation an upsert or conditional insert:

• Put a unique constraint on the event identity and update the aggregate in the same database transaction as inserting that identity.
• For document or key-value models, use a conditional write, compare-and-set, or an applied-event set in the same atomic record/batch as the model update.
• Make external calls idempotent with the same event identity as the idempotency key. If the destination cannot provide that contract, use an outbox or accept duplicate effects explicitly.
• Do not use an additive update such as total = total + delta without deduplication. A replay would count the delta twice.

The managed tag recipe in the cookbook demonstrates an event-identity upsert.

Set up an offset store

ProjectionRuntime::new takes an Arc<dyn ProjectionOffsetStore>. Projection support introduces no new Cargo feature: durable stores use the existing postgres, cassandra, and rocksdb features.

Backend	Construction and schema bootstrap	Durability and constraints
Memory	MemoryProjectionOffsetStore::new(); always compiled; no schema	Semantic reference for tests. Clones share process-local state, but dropping the process loses cursors, generations, and pause flags.
PostgreSQL	Enable postgres; use PostgresProjectionOffsetStoreConfig::new(database_url) with PostgresProjectionOffsetStore::connect, or for_pool with from_pool. connect/initialize executes idempotent create-and-migrate DDL.	Creates datum_projection_offset_store, datum_projection_timestamp_offset_store, and datum_projection_management by default. Supports ALO, AMO, grouped ALO, EO, and grouped EO.
Cassandra / ScyllaDB	Enable cassandra; create CassandraProjectionOffsetStoreConfig::new(contact_point, keyspace) and call CassandraProjectionOffsetStore::connect. Initialization creates the keyspace, datum_projection_offset_store, and datum_projection_management, then prepares statements.	Durable ALO, AMO, and grouped ALO. Cursor fencing uses LWT; EO is unsupported.
RocksDB	Enable rocksdb; RocksDbBackend::open(path) is the combined persistence and projection offset store.	Opening a new database creates the projection column family. Reopening an older Datum database adds that column family through the create-missing-column-families path. This embedded backend is Datum-only.

PostgreSQL and Cassandra expose schema_ddl() when operations need to inspect the exact bootstrap statements. The configured database principal must be allowed to create or migrate the selected schema/keyspace. RocksDB opening is synchronous and establishes the database before returning.

Only the durable stores preserve cursor, owner generation, and pause state across process death. The public conformance testkit distinguishes semantic conformance (all stores) from close/reopen, schema-migration, and crash-save durability (PostgreSQL, Cassandra, and RocksDB).

Source providers and source completeness

A SourceProvider owns the durable cursor format and provider manifest, opens a fresh query for every restart, extracts a complete cursor from each envelope, and filters already-consumed inclusive-boundary events. Never reuse or clone one already materialized query stream: clones of query::into_source share a cursor and are not a restart factory.

EventsBySlicesSourceProvider is the first-party PostgreSQL provider. Other query shapes can implement the public SourceProvider trait; the cookbook includes a sequence-offset events_by_tag provider for memory and RocksDB.

The source-completeness limits are part of the delivery contract:

Delivery semantics are guarantees about envelopes the provider emits — they cannot repair source-side omissions. The two known omission classes are carried verbatim into projection docs and rustdoc: PostgreSQL Live tag/slice queries can permanently skip events committed by transactions open longer than visibility_lag; a Cassandra journal write can succeed while its tag-index write is lost until external repair. Durable-state changes projections additionally inherit the coalescing (latest-revision-per-id) semantics. QueryMode::Current completion is a successful terminal state (final cursor flush, handle resolves), distinct from failure — Live completion, Current completion, and source failure do not share one restart policy.

Size PostgreSQL visibility_lag for the deployment's maximum expected transaction duration, and monitor transactions that exceed it. Repairing a missing Cassandra tag-index write is an external operational action; replay semantics cannot manufacture an envelope the query never emitted.

Recovery and restart backoff

Failure handling has two tiers.

The per-invocation tier applies to one envelope for a single handler and one whole group for a grouped handler:

RecoveryStrategy	Behavior
Fail	Escalate immediately to projection restart. This is the default.
Skip	Advance the cursor past the failed invocation and continue.
RetryAndFail { retries, delay }	Retry the same invocation in order, then restart on exhaustion.
RetryAndSkip { retries, delay }	Retry the same invocation in order, then advance past it on exhaustion.

RecoveryStrategy::retry_and_fail() and retry_and_skip() use five retries one second apart. Retries keep source order and reuse the same envelope or whole group.

Unhandled or exhausted failures enter the projection restart tier. The delay formula is min(min_backoff · 2^consecutive_failures, max_backoff) · (1 + U(0, random_factor)). Jitter is applied after the cap, so the effective upper bound is max_backoff · (1 + random_factor). RestartBackoff defaults to 3 seconds minimum, 30 seconds maximum, a 0.2 random factor, and unlimited restarts. max_restarts limits restarts in the max_restarts_within tracking window.

Each restart reloads the stored cursor, opens a fresh source, and constructs a fresh handler. Restartable source and offset-store I/O failures use this ladder. Fatal configuration, cursor codec, provider-manifest mismatch, and fencing errors bypass backoff and stop with ProjectionStatus::Failed.

Management and shutdown

All management calls serialize through the owning actor, after demand has stopped where a mutation requires quiescence:

• get_offset() loads the current provider-complete cursor.
• clear_offset() clears it and restarts from the beginning.
• update_offset(cursor) installs a provider-compatible cursor and restarts from it.
• pause() quiesces, flushes, and writes the paused flag. resume() clears the flag and starts a fresh incarnation.
• status() and status_receiver() expose point-in-time and change-driven lifecycle state.
• stop_and_wait() stops demand, lets the in-flight invocation finish, flushes according to the active mode, joins the source worker, and then resolves.

Use with_status_observer on the blueprint for informational lifecycle, handler-attempt, and cursor progress callbacks. Observer callbacks do not change delivery semantics.

Pause state survives restarts and process death only with a durable offset store. Clear removes the cursor but does not remove the independent pause state. Clear/update and a racing handler save are protected by owner generation and expected-position checks.

Dropping ProjectionHandle is best-effort signal-only and does not guarantee a final flush. Keep the handle and await stop_and_wait() during graceful shutdown.

Local scale-out with ProjectionGroup

ProjectionGroup runs independent projection instances in one local system. Each key has its own ProjectionId, generation claim, cursor, source worker, handler, and restart ladder. A failure or restart in one key does not restart its siblings.

The explicit-key constructor is useful for tag, tenant, or partition lists:

use std::{
    collections::BTreeMap,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
};

use datum_persistence::{
    PersistenceId, SequenceNr,
    projection::{
        Handler, MemoryProjectionOffsetStore, Projection, ProjectionGroup, ProjectionResult,
        ProjectionRuntime, ProjectionStatus, TestEnvelope, TestSourceProvider,
    },
    query::QueryMode,
};

struct CountHandler(Arc<AtomicUsize>);

#[async_trait::async_trait]
impl Handler<TestEnvelope<&'static str>> for CountHandler {
    async fn process(
        &mut self,
        _envelope: &TestEnvelope<&'static str>,
    ) -> ProjectionResult<()> {
        self.0.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }
}

let sources = Arc::new(BTreeMap::from([
    (
        "blue".to_owned(),
        TestSourceProvider::new(vec![TestEnvelope::new(
            1,
            PersistenceId::from_unique("paint|blue").unwrap(),
            SequenceNr::new(1),
            "blue",
        )]),
    ),
    (
        "green".to_owned(),
        TestSourceProvider::new(vec![TestEnvelope::new(
            1,
            PersistenceId::from_unique("paint|green").unwrap(),
            SequenceNr::new(1),
            "green",
        )]),
    ),
]));
let processed = Arc::new(AtomicUsize::new(0));
let group = ProjectionGroup::for_keys("paint-index", ["blue", "green"], {
    let sources = Arc::clone(&sources);
    let processed = Arc::clone(&processed);
    move |projection_id| {
        let source = sources[projection_id.key()].clone();
        let processed = Arc::clone(&processed);
        Projection::at_least_once(projection_id, source, move || {
            CountHandler(Arc::clone(&processed))
        })
        .with_query_mode(QueryMode::Current)
    }
});

let store = Arc::new(MemoryProjectionOffsetStore::new());
let handles = group.spawn(ProjectionRuntime::new(store)).await.unwrap();
for handle in handles.instances() {
    assert_eq!(handle.wait().await.unwrap(), ProjectionStatus::Completed);
}
assert_eq!(processed.load(Ordering::Relaxed), 2);

The group validates and reserves every identity before starting any actor, so an overlapping live standalone projection or group fails without partially materializing the new group. A ProjectionGroupHandle supports per-key lookup plus statuses, pause_all, resume_all, and stop_and_wait.

Scale-out is local, not a cluster distributor. Cluster relocation additionally needs placement, remembered/keep-alive startup, reconciliation, fencing leases, and quiesced handoff.

Slices and events_by_slices

PostgreSQL journals assign every full persistence id to one of exactly 1,024 slices using the already-persisted FNV-1a hash. slice_for(full_persistence_id) exposes that frozen mapping. slice_ranges(n) divides 0..=1023 into equal contiguous ranges; n must be a whole-number divisor of 1,024.

EventsBySlicesQuery::events_by_slices(entity_type, min, max, offset, mode) supports Live and Current queries. Build one EventsBySlicesSourceProvider per range, using a stable, non-secret query/database identity and a positive range version. The provider derives the mandatory key slices-{min}-{max}@v{version} and rejects a projection id with any other key.

To run slices as a local group, compute slice_ranges(n), construct the providers, index them by their projection_key(), and pass those keys to ProjectionGroup::for_keys. The template returns the projection using the matching provider, exactly like the explicit-key example above.

Reshard explicitly

Changing n changes range ownership and keys. Never start a new range count as an implicit key remap. Use this procedure:

1. Stop every old instance and await the group-wide final flush.
2. Load each old range's final cursor and pair it with its range/version in SliceRangeCursor. An old range with no cursor must remain represented as None.
3. Compute the new ranges, increment the range version, and construct the new EventsBySlicesSourceProvider and range-derived ProjectionId for each.
4. For every new range, collect old ranges whose intersections cover it exactly and call seed_resharded_slice_range. The function validates manifests and coverage and refuses to overwrite a previously seeded key.
5. Seed all new keys before starting any new projection instance, then spawn the new group.

The seed uses the minimum boundary position among the covering old ranges. That can replay the bounded position spread of a range that had advanced further, so the ALO idempotency rule still applies, but taking the minimum prevents loss. Equal-minimum PostgreSQL boundary identities are merged and filtered to the new range.

Test projections in process

TestSourceProvider supplies deterministic envelopes without an external service. Use inclusive to exercise exact-position duplicate filtering. with_manual_pace returns a ManualPaceProbe; call request_one or request(n) to control when source demand can emit. push, complete, and the source/materialization failure hooks cover live and restart paths.

ProjectionTestKit runs a blueprint and polls a user assertion within a bounded window. Empty sources cannot deadlock the test: an unsatisfied assertion becomes ProjectionError::TestKitTimeout, while a terminal projection failure is returned immediately.

use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};

use datum_persistence::{
    PersistenceId, SequenceNr,
    projection::{
        Handler, MemoryProjectionOffsetStore, Projection, ProjectionId, ProjectionResult,
        ProjectionRuntime, ProjectionTestKit, TestEnvelope, TestSourceProvider,
    },
    query::QueryMode,
};

struct TestHandler(Arc<AtomicUsize>);

#[async_trait::async_trait]
impl Handler<TestEnvelope<u64>> for TestHandler {
    async fn process(&mut self, _envelope: &TestEnvelope<u64>) -> ProjectionResult<()> {
        self.0.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }
}

let provider = TestSourceProvider::inclusive(vec![
    TestEnvelope::new(
        1,
        PersistenceId::from_unique("account|a").unwrap(),
        SequenceNr::new(1),
        10,
    ),
    TestEnvelope::new(
        1,
        PersistenceId::from_unique("account|b").unwrap(),
        SequenceNr::new(1),
        20,
    ),
]);
let processed = Arc::new(AtomicUsize::new(0));
let handler_count = Arc::clone(&processed);
let projection = Projection::at_least_once(
    ProjectionId::new("account-index-test", "all"),
    provider,
    move || TestHandler(Arc::clone(&handler_count)),
)
.with_query_mode(QueryMode::Current);

let store = Arc::new(MemoryProjectionOffsetStore::new());
let testkit = ProjectionTestKit::new(ProjectionRuntime::new(store));
testkit
    .run(projection, || processed.load(Ordering::Relaxed) == 2)
    .await
    .unwrap();

For a custom offset store, also run assert_projection_offset_store_semantic_conformance. Durable implementations additionally use assert_projection_offset_store_durable_conformance through a DurableProjectionOffsetStoreHarness.

See also

• Persistence overview
• Persistence read side
• Persistence backends
• Managed events_by_tag recipe