Event sourcing

Deep dive

How it works

In: command + entity id · Out: appended immutable events, rebuilt state, read models · State: per-entity event stream, snapshots, projection checkpoints · Cost: append one multi-row insert; load = snapshot + events since it

command ─► load stream (snapshot + events after it) ─► state ─► decide(cmd, state) ─► new events
                                                                           │
                                       append(stream, expected_version=n, events) ─► conflict? reload, retry
                                                                           │
                                                   global log ─► projections / outbox consumers (checkpointed)

1. Store: rows (global_pos, stream_id, version, type, data, metadata) with a unique key on (stream_id, version).
2. Load: read the latest snapshot (state, version) for the stream, then events with version > snapshot.version, applying each.
3. Decide: a pure function from (state, command) to events or a rejection. No I/O, no clocks (pass time in the command).
4. Append: insert events with versions n+1, n+2, … in one transaction; a unique violation means someone else appended first: reload and retry (bounded).
5. Project: consumers read the global log in order, update read models, and store their checkpoint in the same transaction as the read-model write.
6. Evolve: never rewrite stored events; add V2 types and an upcaster that converts old payloads when reading.

Code

use std::collections::HashMap;

#[derive(Clone, Debug, PartialEq)]
pub enum Event { Opened { owner: u64 }, Deposited { amount: i64 }, Withdrawn { amount: i64 } }

pub enum Cmd { Open { owner: u64 }, Deposit { amount: i64 }, Withdraw { amount: i64 } }

#[derive(Default, Clone, Debug)]
pub struct Account { pub open: bool, pub balance: i64 }

impl Account {
    pub fn apply(&mut self, e: &Event) {                       // must be total and deterministic
        match e {
            Event::Opened { .. } => self.open = true,
            Event::Deposited { amount } => self.balance += amount,
            Event::Withdrawn { amount } => self.balance -= amount,
        }
    }
    pub fn decide(&self, c: &Cmd) -> Result<Vec<Event>, &'static str> {
        match *c {
            Cmd::Open { owner } if !self.open => Ok(vec![Event::Opened { owner }]),
            Cmd::Open { .. } => Err("already open"),
            _ if !self.open => Err("not open"),
            Cmd::Deposit { amount } if amount > 0 => Ok(vec![Event::Deposited { amount }]),
            Cmd::Withdraw { amount } if amount > 0 && amount <= self.balance => Ok(vec![Event::Withdrawn { amount }]),
            _ => Err("invalid amount or insufficient funds"),
        }
    }
}

#[derive(Debug, PartialEq)]
pub struct Conflict { pub actual: u64 }

#[derive(Default)]
pub struct Store {
    log: Vec<(String, u64, Event)>,                            // global order: (stream, version, event)
    streams: HashMap<String, Vec<usize>>,                      // stream -> positions in log
    snapshots: HashMap<String, (u64, Account)>,
}

impl Store {
    pub fn version(&self, s: &str) -> u64 { self.streams.get(s).map_or(0, |v| v.len() as u64) }

    /// Atomic conditional append: the whole batch or nothing.
    pub fn append(&mut self, s: &str, expected: u64, evs: Vec<Event>) -> Result<u64, Conflict> {
        let actual = self.version(s);
        if actual != expected { return Err(Conflict { actual }); }
        let idx = self.streams.entry(s.to_string()).or_default();
        for (i, e) in evs.into_iter().enumerate() {
            idx.push(self.log.len());
            self.log.push((s.to_string(), expected + 1 + i as u64, e));
        }
        Ok(idx.len() as u64)
    }

    pub fn load(&self, s: &str) -> (Account, u64) {
        let (from, mut st) = self.snapshots.get(s).map_or((0, Account::default()), |(v, a)| (*v, a.clone()));
        let pos = self.streams.get(s).map_or(&[][..], |v| &v[..]);
        for &p in &pos[from as usize..] { st.apply(&self.log[p].2); }
        (st, pos.len() as u64)
    }

    pub fn snapshot(&mut self, s: &str) { let (a, v) = self.load(s); self.snapshots.insert(s.to_string(), (v, a)); }

    /// Projection / consumer: everything after `checkpoint` in global order.
    pub fn read_all(&self, checkpoint: usize) -> impl Iterator<Item = (usize, &(String, u64, Event))> {
        self.log.iter().enumerate().skip(checkpoint)
    }
}

/// Load, decide, append; retry on a concurrent writer.
pub fn handle(store: &mut Store, s: &str, c: &Cmd) -> Result<u64, &'static str> {
    for _ in 0..3 {
        let (state, v) = store.load(s);
        let evs = state.decide(c)?;
        match store.append(s, v, evs) { Ok(nv) => return Ok(nv), Err(_) => continue }
    }
    Err("too much contention")
}

The durable version is one table:

CREATE TABLE events (
  global_pos bigint GENERATED ALWAYS AS IDENTITY,
  stream_id  text        NOT NULL,
  version    bigint      NOT NULL,
  type       text        NOT NULL,
  data       jsonb       NOT NULL,
  metadata   jsonb       NOT NULL DEFAULT '{}',   -- causation id, correlation id, actor
  tx_id      xid8        NOT NULL DEFAULT pg_current_xact_id(),
  PRIMARY KEY (stream_id, version)                -- this is the optimistic concurrency check
);
-- append: INSERT … VALUES ($s, $n+1, …), ($s, $n+2, …);  unique_violation (23505) = conflict

Complexity and constants

• Append: one multi-row insert; load: one indexed range scan. With snapshots every N events, load reads at most N events; N of a few hundred keeps loads to a single-digit number of index pages.
• Storage grows forever: order of 100 B-1 KB per event as JSON; binary encodings shrink it several-fold.

Efficiency

Cost: the bound is durable-write latency (one fsync per commit) on the write side and index range scans on the read side; CPU for apply is negligible. With snapshots every N events a load reads at most N events; events are order of 100 B-1 KB each as JSON.

Levers

• Batch appends — several events of one command in one multi-row insert and commit; group commit (write-ahead log) gives many commits per second — any write path.
• Snapshots — cap load at N events (a cache, rebuild lazily) — long streams.
• Batched projections — checkpoint once per batch, in the same transaction as the read-model writes — heavy read models.
• Binary payloads — events several-fold smaller than JSON (order of) — storage and replay I/O bound.

Naive → standard → state of the art

• Naive: replay the whole stream on every command · O(stream length) per load
• Standard: expected-version append + snapshot every few hundred events + checkpointed projections · bounded load
• Cheaper alternative: a mutable row plus an append-only change-log table in one transaction, when audit is needed but not replay or new projections

Implementation notes

• Global ordering trap: identity/sequence values are assigned at insert, but transactions commit in a different order. A projection reading WHERE global_pos > checkpoint can see 101 commit before 100 and skip 100 forever. Fixes: a single writer per log; or record pg_current_xact_id() per row and only read rows whose tx_id < pg_snapshot_xmin(pg_current_snapshot()) (PostgreSQL 13+), i.e. rows of transactions that can no longer be in flight.
• Outbox for free: other services consume the event log instead of receiving a dual write; publishing is just another checkpointed consumer.
• Idempotent consumers: store the checkpoint (or processed event ids) in the same transaction as the side effect.
• Metadata: causation id (command that produced it), correlation id (business flow), actor, and the idempotency key of the request. They make replays and audits explainable.
• Snapshots are a cache: versioned by state schema, deletable, rebuilt from events.
• Decide must be deterministic: inject time and ids through the command so replays and retries produce the same events.
• Ledgers: a double-entry ledger is an event-sourced system whose events are balanced transactions; the balance table is its projection.

Pitfalls

• Kafka (or any log without conditional append per key) as the event store: no expected-version check, so concurrent writers break invariants.
• Events that describe intent or carry derived state (BalanceChangedTo(70)) instead of facts: replay then depends on old logic.
• Changing apply semantics for old event types: historic state silently changes. Version the event type instead.
• Long streams without snapshots (an account with 10⁶ events): every command replays them all.
• Read-your-writes expectations against projections; return the new state from the write path or wait on the checkpoint.
• Storing personal data in events you can never delete; encrypt per subject and delete the key (crypto-shredding).

Variants and alternatives

In production

EventStoreDB (now KurrentDB), Axon Server, Marten (event store on PostgreSQL). Kurrent publishes case studies in healthcare (NHS Wales, a pharmaceutical company using it for audit trails) and warehouse management (SkuVault); banking cores and payment platforms keep ledgers as event streams; LMAX journals inputs in the same spirit (trading). Microsoft documents the pattern in its Azure Architecture Center.

Implementations

Sources

• Martin Fowler, Event Sourcing, martinfowler.com, 2005. article
• Greg Young, CQRS Documents, 2010. PDF
• Martin Kleppmann, Designing Data-Intensive Applications, O'Reilly, 2017, chapter 11. book
• PostgreSQL documentation, System Information Functions (pg_current_xact_id, pg_current_snapshot, pg_snapshot_xmin). docs