Event-sourced actors

An event-sourced device

Persist readings as events and recover the last one in a new actor system.

An event-sourced actor stores what happened, not what its state is. For a device, what happens is that a reading is recorded. In this step the device persists every reading as an event, and a new actor system over the same store recovers the last reading by replaying those events.

Add the persistence-in-memory feature, which brings an in-memory store for trying things out and for tests:

cargo add tellus --features persistence-in-memory
cargo add serde --features derive

Commands and events

tellus/examples/tutorials/event_sourced/01_device.rs

enum Command {
    Record(f64),
    Read(ReplyTo<Option<f64>>),
}

#[derive(Serialize, Deserialize)]
struct Recorded(f64);

impl Versioned for Recorded {
    const MANIFEST: &'static str = "recorded";
    const VERSION: SchemaVersion = SchemaVersion::new(1);
}

The device still receives two messages, now called commands: Record and Read. What it persists is an event, Recorded. Events are stored, so they implement Serialize, Deserialize and Versioned. The manifest is a stable name for the event type, independent of the Rust type, and the version allows the event to evolve later.

The device

tellus/examples/tutorials/event_sourced/01_device.rs

impl EventSourced for Device {
    type Command = Command;
    type Event = Recorded;
    type State = Option<f64>;
    type Snapshot = Nothing;
    type Error = Infallible;

    fn persistence_id(&self) -> PersistenceId {
        PersistenceId::new("device", "left").expect("the segments are valid")
    }

    fn init(&self) -> Result<Self::State, Self::Error> {
        Ok(None)
    }

    fn init_from_snapshot(&self, snapshot: Self::Snapshot) -> Result<Self::State, Self::Error> {
        match snapshot {}
    }

    fn handle(
        &self,
        _: &ActorContext<Self::Command>,
        incoming: Incoming<Self::Command>,
        _: &Self::State,
    ) -> Result<Effect<Self>, Self::Error> {
        let Incoming::Message(command) = incoming else {
            unreachable!("the device watches no actor, hence never gets a terminated signal")
        };

        match command {
            Command::Record(degrees) => Ok(Effect::persist(Recorded(degrees))),

            Command::Read(reply_to) => {
                Ok(Effect::none().then(move |last_reading| reply_to.reply(*last_reading)))
            }
        }
    }

    fn apply(&self, _: Self::State, Recorded(degrees): Self::Event) -> Self::State {
        Some(degrees)
    }
}

An event-sourced actor implements EventSourced instead of Actor. The work of receive is now split in two:

handle decides. It gets the current state by reference and returns an Effect. For Record, the effect persists a Recorded event. For Read, it persists nothing, and the reply runs in a continuation, then, which receives the state.
apply changes the state. It folds one event into the state and is the only state transition.

tellus appends the events to the store, and only then applies them. On recovery, it replays the stored events through apply, the same way, so the recovered state is exactly the state the device had.

persistence_id names the device's stream of events. Every incarnation of this device, today or next year, reads and writes the same stream. init is the state before any event, and init_from_snapshot is unreachable here, because Nothing as snapshot type means no snapshots; step 3 adds them.

Run it

tellus/examples/tutorials/event_sourced/01_device.rs

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let store = InMemoryStore::default();

    let system = ActorSystem::event_sourced(Device, Persistence::new(store.clone()));
    system.root().tell(Command::Record(21.5));
    system.root().tell(Command::Record(22.0));
    let last_reading = system
        .root()
        .ask(Duration::from_secs(1), Command::Read)
        .await
        .context("asking for the last reading")?;
    if let Some(degrees) = last_reading {
        println!("Last reading before stopping: {degrees:.1}");
    }
    system.stop();
    system
        .terminated()
        .await
        .context("awaiting the first actor system's termination")?;

    let system = ActorSystem::event_sourced(Device, Persistence::new(store));
    let last_reading = system
        .root()
        .ask(Duration::from_secs(1), Command::Read)
        .await
        .context("asking for the last reading")?;
    match last_reading {
        Some(degrees) => println!("Last reading after recovery: {degrees:.1}"),
        None => println!("No reading after recovery"),
    }

    system.stop();
    system
        .terminated()
        .await
        .context("awaiting the second actor system's termination")
}

ActorSystem::event_sourced spawns the root as an event-sourced actor, with a Persistence that names the store. The first system records two readings and stops. The second system, a new incarnation of the same device over the same store, starts without any reading in memory, replays the two events and answers with the last one.

Notice the ask before the first stop. stop takes effect between two commands, so without waiting for an answer, the queued readings might never be handled.

cargo run --quiet -p tellus --features persistence-in-memory --example event_sourced_01_device

Last reading before stopping: 22.0
Last reading after recovery: 22.0

What you learned

An event-sourced actor decides in handle and changes its state only in apply. Its events are stored before they are applied, and replaying them recovers the state. Next, the device fails.