Persistence read side

The write-side storage SPIs are deliberately separate from Persistence Query. Read-side consumers ask for a capability trait, receive a QueryStream<T>, and may lift it into the normal Datum stream DSL.

Query capabilities

Trait	Output	Ordering contract
`EventsByPersistenceIdQuery`	`EventEnvelope`	Inclusive sequence-number range for one persistence id
`EventsByTagQuery`	`EventEnvelope`	Backend-ordered events carrying one tag, resumed from an `Offset`
`PersistenceIdsQuery`	`PersistenceId`	Each known id once in backend discovery order
`DurableStateChangesQuery`	`DurableStateChange`	Coalesced latest durable-state changes by tag

EventEnvelope contains the offset, persistence id, sequence number, timestamp, event-adapter manifest, serialized payload, optional metadata, and tags. It intentionally stays byte-oriented; the caller chooses the serializer catalog and event adapter at the read boundary.

DurableStateChange is either Updated(UpdatedDurableState) or Deleted(DeletedDurableState). Rapid revisions for one id may be coalesced, but the latest revision must eventually be emitted by a conformant live implementation.

Live versus current

Every capability takes QueryMode:

QueryMode::Current captures a materialization boundary, emits only that snapshot, and completes.
QueryMode::Live emits the same boundary snapshot and then stays open for later changes. A finite events_by_persistence_id upper bound completes when that sequence number is reached.

Choose Current for backfills, tests, and bounded reconciliation. Choose Live for projections, and give the surrounding stream a cancellation/shutdown path.

Offsets are backend-local

Never compare or move offsets between backend instances or databases.

Offset	Intended backend semantics
`Offset::NoOffset`	No previously observed resume position.
`Offset::Sequence(n)`	In-memory and RocksDB per-database append position; resume strictly after `n`.
`Offset::Timestamp { epoch_micros, seq_nr }`	PostgreSQL database timestamp plus per-id journal sequence for events. Durable-state changes place their revision in `seq_nr`. Resume is inclusive at an equal pair, so exact-position duplicates may be re-emitted.

The timestamp pair is lexicographically ordered and not globally unique: different persistence ids may share both fields. The seq_nr field name is therefore journal-shaped even when the envelope is a durable-state change. Store the offset with the same backend identity and make projection writes idempotent. The inclusive PostgreSQL contract intentionally prefers duplicate delivery over a gap.

Shipped capability implementations

All three first-party backends implement the capability traits appropriate to their journal and durable-state stores:

Backend	Event queries	Persistence ids	Durable changes	Live-query mechanism
`MemoryJournal` / `MemoryDurableStateStore`	By persistence id and tag	Yes	Yes	In-process notifications; `Offset::Sequence`
`RocksDbBackend`	By persistence id and tag	Yes	Yes	A watch notification wakes the single-owner query cursor; `Offset::Sequence`
`PostgresPersistence`	By persistence id and tag	Yes	Yes	Adaptive polling from `min_interval` to `max_interval`; `Offset::Timestamp`

Do not pass a concrete offset across families: PostgreSQL requires a timestamp offset from the same database, while memory and RocksDB use sequence offsets from the same backend instance. PostgreSQL and RocksDB explicitly reject the wrong family; Offset::NoOffset starts from the beginning everywhere. Query through the narrow capability trait your projection needs so application code is not coupled to the backend's write-side surface.

Lift a query into Datum streams

into_source turns a QueryStream<T> into Source<T>. A later query error becomes StreamError::Failed. The cursor is a single-consumer resource: clones of the resulting source share that cursor rather than opening independent queries.

decode_envelopes(catalog, adapter) returns Flow<EventEnvelope, D>. It dispatches by serializer id, then calls the event adapter. Recovery-side 0..n expansion/filtering is preserved.

The following current query projects tagged counter deltas into a total. It uses the serializer helper from the event-sourcing quickstart.

rust

use datum_persistence::{IdentityAdapter, Tag};
use datum_persistence::query::{
    EventsByTagQuery, Offset, QueryMode, decode_envelopes, into_source,
};

let query = journal
    .events_by_tag(&Tag::new("counter"), Offset::NoOffset, QueryMode::Current)
    .await
    .unwrap();

let projection = into_source(query)
    .via(decode_envelopes(counter_catalog(), IdentityAdapter))
    .run_fold(0_i64, |total, delta| total + delta)
    .unwrap();
let total = tokio::task::spawn_blocking(move || projection.wait())
    .await
    .unwrap()
    .unwrap();

For a durable live read model, use the managed projection runtime. It persists provider-complete cursors, reopens a fresh query on restart, and keeps the envelope available to the handler for idempotency or exactly-once work. decode_envelopes intentionally outputs only domain events, so a manually composed flow still needs to retain the original envelope when an update requires both the event and its checkpoint.

Query error handling

Opening a query can fail immediately with PersistenceError; handle that before materializing the source. Once opened, storage failures arrive as error items in QueryStream and fail the Datum source. Managed projections apply their recovery ladder around a SourceProvider that reopens the query from its stored cursor. In a manually composed stream, apply ordinary restart/recovery operators around an equivalent fresh-query factory—do not rematerialize clones of the same into_source cursor as a substitute for reopening it.

Persistence read side ​

Query capabilities ​

Live versus current ​

Offsets are backend-local ​

Shipped capability implementations ​

Lift a query into Datum streams ​

Query error handling ​

See also ​