Event sourcing

An EventSourcedBehavior<C, E, S> separates decisions from state transitions:

• The command handler receives &S and an owned C, then returns an Effect<C, E, S>.
• The event handler is the only function that changes state. It receives &S and &E, returning either S or PersistenceResult<S>.
• A PersistenceId identifies one journal stream. Its sequence numbers are gapless and local to that id.

The blueprint does no I/O. spawn(EventSourcedRuntime) creates the actor and completes recovery before returning an EventSourcedHandle.

Counter quickstart

This serializer is deliberately small so the behavior example does not hide serialization behind a framework macro. Production code may use ProstSerializer instead.

use datum_persistence::{
    PersistenceError, PersistenceResult, PersistenceSerializer, SerializerCatalog, SerializerId,
};

#[derive(Clone, Copy)]
struct I64Serializer;

impl PersistenceSerializer<i64> for I64Serializer {
    fn serializer_id(&self) -> SerializerId {
        SerializerId::new(41)
    }

    fn manifest(&self, _value: &i64) -> String {
        "counter-i64-v1".into()
    }

    fn serialize(&self, value: &i64) -> PersistenceResult<Vec<u8>> {
        Ok(value.to_be_bytes().to_vec())
    }

    fn deserialize(&self, manifest: &str, payload: &[u8]) -> PersistenceResult<i64> {
        if manifest != "counter-i64-v1" || payload.len() != 8 {
            return Err(PersistenceError::Serialization("invalid counter frame".into()));
        }
        let mut bytes = [0_u8; 8];
        bytes.copy_from_slice(payload);
        Ok(i64::from_be_bytes(bytes))
    }
}

fn counter_catalog() -> SerializerCatalog<i64> {
    SerializerCatalog::new(I64Serializer)
}

The command owns Datum's actor ReplyPort. with_enforced_replies makes every match arm return a ReplyEffect; then_reply is scheduled only after persistence succeeds and sees the confirmed resulting state.

use std::{sync::Arc, time::Duration};

use datum::{ActorFlow, ReplyPort, Sink, Source};
use datum_persistence::{
    Effect, EventSourcedBehavior, EventSourcedMessage, EventSourcedRuntime, IdentityAdapter,
    MemoryJournal, PersistenceId,
};

enum CounterCommand {
    Add { amount: i64, reply_to: ReplyPort<i64> },
    Get { reply_to: ReplyPort<i64> },
}

let behavior = EventSourcedBehavior::with_enforced_replies(
    PersistenceId::new("counter", "counter-1").unwrap(),
    0_i64,
    |state, command| match command {
        CounterCommand::Add { amount, reply_to } => {
            Effect::persist(amount).then_reply(reply_to, |state| *state)
        }
        CounterCommand::Get { reply_to } => Effect::reply(reply_to, *state),
    },
    |state: &i64, event: &i64| state + event,
)
.with_event_adapter(IdentityAdapter, counter_catalog());

let handle = behavior
    .spawn(EventSourcedRuntime::new(Arc::new(MemoryJournal::new())))
    .await
    .unwrap();

let completion = Source::from_iter([2_i64, 3])
    .via(ActorFlow::ask(
        handle.actor_ref().clone(),
        1,
        Duration::from_secs(2),
        |amount, reply_to| {
            EventSourcedMessage::command(CounterCommand::Add { amount, reply_to })
        },
    ))
    .run_with(Sink::collect())
    .unwrap();
let replies = tokio::task::spawn_blocking(move || completion.wait())
    .await
    .unwrap()
    .unwrap();
assert_eq!(replies, vec![2, 5]);

let current = handle.state().await.unwrap();
assert_eq!(current.state, 5);
assert_eq!(current.sequence_nr.as_u64(), 2);
handle.stop();

EventSourcedHandle::tell(command) is the fire-and-forget path. actor_ref() exposes the same ordered ingress for ActorFlow::ask; request builders must wrap commands with EventSourcedMessage::command. state().await asks through that mailbox and returns only confirmed state.

The Effect DSL

One main effect may be followed by ordered side effects:

Effect	Result
Effect::persist(event)	Persist one event.
Effect::persist_all(events)	Persist a batch as one atomic journal write.
persist_with_metadata / persist_all_with_metadata	Attach already-serialized metadata frames.
Effect::none()	Handle without persistence.
Effect::unhandled()	Record that the command was not handled.
Effect::stash(command)	Put the owned command in the bounded manual stash.
Effect::unstash_all()	Begin a bounded drain of the current manual stash.
Effect::stop()	Stop after this command.
Effect::async_effect(future)	Await a future whose output is another Effect. Later commands are stashed while it resolves.

Chain then_run, then_reply, then_stop, or then_unstash_all. Callbacks run in registration order only after the main effect succeeds. async_reply is the enforced-reply form of async_effect. The manual stash defaults to 1,024 commands and can be changed with with_stash_capacity; overflow drops the newest command.

For ordinary handlers, then_reply returns ReplyEffect but is accepted because the enforced type is only required by with_enforced_replies. Use then_no_reply or Effect::no_reply() to make an intentional no-reply branch explicit.

Signals and failure supervision

Install on_signal on the blueprint. The enum is non-exhaustive, so keep a wildcard arm.

let behavior = behavior.on_signal(|state, signal| match signal {
    datum_persistence::EventSourcedSignal::RecoveryCompleted => {
        eprintln!("recovered state={state}");
    }
    datum_persistence::EventSourcedSignal::PersistRejected { reason, .. } => {
        eprintln!("write rejected: {reason}");
    }
    datum_persistence::EventSourcedSignal::PersistFailed { error, .. } => {
        eprintln!("storage failed: {error}");
    }
    _ => {}
});

RecoveryFailed, snapshot save/load signals, and retention deletion signals use the same callback. A snapshot save failure is non-fatal; a recovery failure stops spawn. A persist rejection always stops. A storage failure stops by default, or can enter delayed recovery with on_persist_failure(RestartSettings); commands are automatically stashed during that backoff.

Snapshots and retention

Snapshots require both a snapshot store in EventSourcedRuntime and a state adapter/catalog on the behavior. A predicate snapshot is evaluated after each staged event, but a batch produces at most one snapshot and the snapshot contains the post-batch state.

use std::sync::Arc;

use datum_persistence::{
    EventSourcedRuntime, IdentityAdapter, MemoryJournal, MemorySnapshotStore, RetentionCriteria,
};

let retention = RetentionCriteria::snapshot_every(1_000, 3)
    .unwrap()
    .with_delete_events_on_snapshot();

let behavior = behavior
    .with_state_adapter(IdentityAdapter, counter_catalog())
    .snapshot_when(|state, _event, _sequence_nr| *state >= 10_000)
    .with_retention(retention);

let runtime = EventSourcedRuntime::new(Arc::new(MemoryJournal::new()))
    .with_snapshot_store(Arc::new(MemorySnapshotStore::new()));
let handle = behavior.spawn(runtime).await.unwrap();

snapshot_every(every, keep) validates both values are non-zero. At each boundary it saves a snapshot; cleanup retains keep windows. with_delete_events_on_snapshot additionally deletes the older journal prefix, but only after the retention snapshot succeeds. Cleanup is asynchronous and serialized; its completion/failure arrives through signals.

Recovery defaults to the latest snapshot. Use Recovery::with_snapshot_selection with SnapshotSelection::None or explicit SnapshotCriteria when you need a full replay or bounded selection.

Taggers and the read side

Tags are stored with each confirmed event and drive events_by_tag projections. A simple tagger sees the event; a state tagger also sees the post-event state.

use datum_persistence::Tag;

let behavior = behavior
    .with_tagger(|_event| vec![Tag::new("counter")])
    .with_state_tagger(|state, _event| {
        vec![Tag::new("counter"), Tag::new(format!("bucket-{}", state / 100))]
    });

Calling both methods replaces the first tagger with the second. See the read-side guide for query modes, offsets, and decoding envelopes into Datum streams.

Confirmed events as a Topic

Event publishing is enabled by default. After spawn, published_events() returns a Topic<PublishedEvent<E>>. Subscribe before sending the command: like every Datum Topic, it is a live broadcast and does not replay earlier values.

let topic = handle
    .published_events()
    .expect("event publishing is enabled")
    .clone();
let published_source = topic.subscribe();

Each item carries the persistence id, sequence number, domain event, timestamp, and tags, and is published only after the journal confirms the append. Recovery never republishes. Disable the topic with with_event_publishing(false); handle.stop() gracefully closes it.

See also

• Persistence overview
• Durable state
• Schema evolution
• Event-sourced counter recipe