All posts

Eventfold: Building an Event Sourcing System from Scratch in Rust

14 min readrust, event-sourcing, architecture, infrastructure

When we built a bespoke CRM for a client, the core requirement wasn't "make Salesforce but simpler." It was "build something that fits how we already work — and never lose a record of what happened." That second part — a complete, immutable audit trail of every interaction — led us to event sourcing. And event sourcing led us to building Eventfold.

This post covers the two pieces: eventfold-db, the event store itself, and eventfold-es, the Rust framework that sits on top of it.

Why Build Our Own

The existing options for event sourcing infrastructure fall into two camps: distributed databases that carry massive operational overhead (EventStoreDB clusters, Kafka with schema registries), or lightweight libraries that punt on persistence and make you wire up your own storage.

We needed something in between. A single-node event store with real durability guarantees, optimistic concurrency, push-based subscriptions, and a gRPC API — but without the operational complexity of running a distributed system for a workload that would comfortably fit on one machine. Nothing off the shelf hit that spot, so we built it.

eventfold-db: The Event Store

Eventfold-db is an append-only event store. No tables, no indexes, no query language. Events go in, events come out, and nothing ever gets mutated or deleted.

The Log File

The entire persistence layer is a single binary file. No WAL, no B-tree, no LSM tree. Events are written sequentially and read sequentially. The on-disk format uses length-prefixed records grouped into batches, each checksummed with CRC32:

[FILE HEADER: 8 bytes]
  MAGIC: "EFDB" (4 bytes)
  FORMAT_VERSION: u32 (currently 3)

[BATCH]
  BATCH_HEADER: magic + record_count + first_global_position
  [RECORD 1]
    LENGTH: u32
    BODY: global_pos | recorded_at | stream_id | stream_version
          | event_id | event_type | metadata | payload
    CRC32: u32 (over body)
  [RECORD 2...]
  BATCH_FOOTER: magic + batch_crc (CRC32 over header + all record bodies)

Every record carries its own CRC32 checksum, and every batch carries a checksum over the entire batch. This means we can detect corruption at both the record and batch level — and distinguish between a partial write from a crash (truncate and recover) and mid-file corruption (stop and report data loss).

In-Memory Index

At startup, the entire log is read into memory and two structures are built:

  1. A Vec<RecordedEvent> — the global log. Index i is the event at global position i. Direct array access, no seeking.
  2. A HashMap<Uuid, Vec<u64>> — the stream index. Maps each stream ID to its list of global positions. Index j in the vec is stream version j.

Both structures live behind an Arc<RwLock<EventLog>>. Readers take the read lock — no contention among concurrent reads. The writer takes the write lock only after fsync, and only long enough to append to the vec and update the hashmap.

This means every read operation is a slice into an in-memory array. Zero disk I/O on the read path.

The Write Path

All writes are serialized through a single tokio task that owns the file handle. The flow:

  1. gRPC handler validates the request and sends an AppendRequest through a bounded mpsc channel.
  2. The writer task wakes up, drains all pending requests from the channel (batching), and stamps them with the same recorded_at millisecond.
  3. For each request: check the dedup cache (an LRU of recent event IDs), validate the expected version against the in-memory stream index, encode the records to binary.
  4. Write the batch to the file and call sync_all() — blocking until the data is on the storage device.
  5. Acquire the write lock briefly to extend the in-memory index.
  6. Publish the new events to the broadcast channel for live subscribers.
  7. Return the result through a oneshot channel back to the waiting gRPC handler.

The key property: fsync happens before the index update. If the process crashes between fsync and the index update, the next startup replays the log and reconstructs the index. No data loss, no inconsistency.

// Simplified write path — encode, fsync, then update index
let batch_bytes = codec::encode_batch(&records, first_global_pos);
self.file.write_all(&batch_bytes)?;
self.file.sync_all()?; 
 
// Only after durable write: update in-memory state
let mut log = self.event_log.write().expect("poisoned"); 
for record in &records {
    log.streams
        .entry(record.stream_id)
        .or_default()
        .push(record.global_position);
    log.events.push(record.clone());
}

Batching amortizes the fsync cost — the most expensive operation — across multiple appends that happen to arrive in the same window. Under load, this means one disk sync per batch of requests rather than one per request.

Optimistic Concurrency

Every append specifies an expected version for the target stream:

  • Any — no check, always succeeds
  • NoStream — stream must not exist yet
  • Exact(n) — stream must be at exactly version n

If the check fails, the server returns FAILED_PRECONDITION and the client retries. No locks on the client side, no transactions — just a version check on the hot path and a retry loop in the framework. For the CRM workload (single-digit concurrent users, hundreds of events per day), conflicts are rare enough that this is simpler and faster than pessimistic locking.

Crash Recovery

On startup, the store reads the log file sequentially. If it encounters a partial trailing record — a length prefix that promises more bytes than the file contains, or a CRC mismatch on the last record — it truncates back to the last complete batch and continues. This handles the common crash scenario: the process died mid-write.

If it finds corruption followed by valid batches (corruption in the middle of the file), it errors out. This would indicate hardware failure or filesystem bugs, not an application crash, and silent recovery would risk data loss.

Subscriptions

Live subscriptions use a tokio broadcast channel. The critical ordering guarantee: the subscriber registers on the broadcast channel before starting the catch-up read. This prevents the race where events appended between the end of catch-up and the start of live listening are missed.

1. Register broadcast receiver (events buffer from now on)
2. Catch-up: read_all(cursor, 500) in a loop until batch < 500
3. Emit CaughtUp marker to client
4. Live: recv from broadcast, dedup against catch-up cursor

The CaughtUp marker tells the client that everything from this point forward is real-time. Projections use this to know when they've finished replaying history.

What It Doesn't Do

Intentionally absent: clustering, replication, server-side projections, snapshots, and schema validation. The store treats event payloads as opaque bytes. It assigns global positions and stream versions, enforces concurrency, and guarantees durability. Everything else — projection logic, schema evolution, read model construction — is the client's responsibility.

For a single-machine CRM serving a 30-person company, this is the right set of trade-offs. We get a storage engine we fully understand, with crash recovery we can reason about, in under 2,000 lines of Rust.

eventfold-es: The Framework

Eventfold-es is the client-side framework that talks to eventfold-db over gRPC. It provides the abstractions you'd expect from an event sourcing toolkit: aggregates, projections, process managers, and a runtime to wire them together.

Aggregates

An aggregate is a domain model whose state is derived entirely from its event history. The Aggregate trait has four pieces:

pub trait Aggregate: Default + Clone + Send + Sync + 'static {
    /// Identifies the aggregate type in the event store (e.g., "contact").
    const AGGREGATE_TYPE: &'static str;
 
    type Command: Send + 'static;
    type DomainEvent: Serialize + DeserializeOwned + Send + Sync + Clone + 'static;
    type Error: std::error::Error + Send + Sync + 'static;
 
    /// Pure decision function: validate a command against current state,
    /// return zero or more events, or reject with an error.
    fn handle(&self, cmd: Self::Command) -> Result<Vec<Self::DomainEvent>, Self::Error>;
 
    /// Pure state transition: fold a single event into the aggregate state.
    /// Must be total — unknown event variants are ignored for forward
    /// compatibility.
    fn apply(self, event: &Self::DomainEvent) -> Self;
}

The implementing type is the state. handle is the decision function — given the current state and a command, produce events or reject. apply is the state transition — given an event, produce a new state. Both are pure functions with no I/O. This makes them trivially testable:

#[test]
fn handle_then_apply_roundtrip() {
    let contact = Contact::default();
    let events = contact.handle(CreateContact {
        name: "Jane Doe".into(),
        email: "jane@example.com".into(),
    }).unwrap();
 
    let state = events.iter().fold(Contact::default(), |s, e| s.apply(e));
    assert_eq!(state.name, "Jane Doe");
    assert_eq!(state.email, "jane@example.com");
}

No database, no gRPC connection, no async runtime. Just a struct, a command, and an assertion.

Event Encoding

Domain events are Rust enums serialized with serde's adjacently-tagged format:

#[derive(Serialize, Deserialize, Clone)]
#[serde(tag = "type", content = "data")]
enum ContactEvent {
    Created { name: String, email: String, phone: Option<String> },
    EmailUpdated { email: String },
    NoteAdded { content: String },
    Archived,
}

This serializes to {"type": "Created", "data": {"name": "Jane", ...}}. The framework extracts the type field as the event type stored in eventfold-db, and the data field becomes the payload. The split matters — it means projections can filter by event type without deserializing the full payload.

Actor-per-Instance Runtime

Each aggregate instance (e.g., "contact/abc-123") runs as a dedicated tokio task. The task owns the in-memory state and the current stream version. Commands arrive through an mpsc channel and are processed sequentially — no concurrent command handling per instance, which eliminates an entire class of race conditions.

The execution loop for a command:

  1. Deserialize the command from the channel.
  2. Call aggregate.handle(cmd) to produce events (or an error).
  3. Encode the events and append to the event store with ExpectedVersion::Exact(current_version).
  4. If the store returns FAILED_PRECONDITION (version conflict), re-read the stream, re-fold the state, and retry. Up to 3 attempts.
  5. If the append succeeds, apply the events to the local state and advance the version.
for attempt in 0..MAX_RETRIES {
    let events = state.handle(cmd.clone())?;
    match store.append(stream_id, ExpectedVersion::Exact(version), &events).await {
        Ok(resp) => {
            // Apply events locally, advance version
            for event in &events {
                state = state.apply(event);
            }
            version = resp.last_stream_version;
            return Ok(events);
        }
        Err(e) if e.code() == Code::FailedPrecondition => { 
            // Conflict — re-read full stream, rebuild state, retry
            let recorded = store.read_stream(stream_id, 0, u64::MAX).await?; 
            state = recorded.iter().fold(A::default(), |s, e| s.apply(e)); 
            version = recorded.len() as u64 - 1; 
        }
        Err(e) => return Err(e.into()),
    }
}

Actors idle-evict after a configurable timeout (default 5 minutes). When a new command arrives for an evicted aggregate, the runtime spawns a fresh task, replays the stream from the store, and rebuilds the state. No snapshots — full replay every time. For aggregates with hundreds of events this takes single-digit milliseconds; for aggregates with thousands, it's still under 50ms. Snapshots are a complexity we haven't needed yet.

Projections

Projections consume the global event log to build read models. They subscribe to all events across all streams, not individual streams:

pub trait Projection: Default + Clone + Send + Sync + 'static {
    const NAME: &'static str;
 
    fn apply(&mut self, event: &StoredEvent);
}

A projection's apply receives a StoredEvent with the aggregate type, instance ID, event type, and JSON payload. It can filter and transform however it needs:

impl Projection for ActiveContactsList {
    const NAME: &'static str = "active_contacts";
 
    fn apply(&mut self, event: &StoredEvent) {
        if event.aggregate_type != "contact" { return; }
        match event.event_type.as_str() {
            "Created" => {
                // Parse payload, add to list
            }
            "Archived" => {
                // Remove from list
            }
            _ => {} // Unknown events ignored — forward compatible
        }
    }
}

Projections are eventually consistent. The runtime catches them up from the global log on demand, then keeps them current through the live subscription loop. The CaughtUp marker from the store tells the projection when it's transitioned from replaying history to receiving live events.

Process Managers

Process managers react to events by issuing commands against other aggregates — cross-aggregate workflows:

pub trait ProcessManager: Default + Send + Sync + 'static {
    const NAME: &'static str;
 
    fn react(&mut self, event: &StoredEvent) -> Vec<CommandEnvelope>;
}

When a process manager returns a CommandEnvelope, the runtime dispatches it to the target aggregate. If the command fails, it's logged to a dead-letter file (JSONL format) rather than retried indefinitely. This keeps the system from getting stuck on a permanently failing command while preserving a record of what went wrong.

Command Context

Commands are pure domain objects — they carry no cross-cutting concerns. Audit trail, correlation IDs, and device tracking travel alongside the command in a CommandContext:

pub struct CommandContext {
    pub actor: Option<String>,
    pub correlation_id: Option<String>,
    pub metadata: Option<Value>,
    pub source_device: Option<String>,
}

The framework burns this context into event metadata at write time. Every event in the store carries the identity of who issued the command, the correlation ID for request tracing, and optional device information. This gives the CRM a complete audit trail — not just "what happened" but "who did it and from where" — without polluting the domain model with infrastructure concerns.

Wiring It Together

The AggregateStoreBuilder sets up the entire runtime:

let store = AggregateStoreBuilder::new()
    .endpoint("http://127.0.0.1:2113")
    .base_dir("/data/crm")
    .aggregate_type::<Contact>()
    .aggregate_type::<Deal>()
    .projection::<ActiveContactsList>()
    .projection::<DealPipeline>()
    .process_manager::<FollowUpScheduler>()
    .idle_timeout(Duration::from_secs(300))
    .open()
    .await?;
 
// Execute a command
store.execute::<Contact>(
    "contact-123",
    ContactCommand::UpdateEmail { email: "new@example.com".into() },
    CommandContext { actor: Some("user-456".into()), ..Default::default() },
).await?;
 
// Read a projection
let contacts = store.projection::<ActiveContactsList>().await?;

Multiple aggregate types, multiple projections, multiple process managers — all registered on a single store instance. The store handles actor lifecycle, event routing, projection catch-up, and subscription management. The application code defines domain logic and reads projections. That's the entire integration surface.

The Trade-offs (For Now)

The initial release deliberately left several features on the cutting room floor to keep the system simple and shippable. Each was a conscious decision for the CRM workload — not a permanent limitation. All of these are on the roadmap.

No snapshots (yet). Aggregate state is rebuilt from the full event stream on every actor spawn. This is fast enough for aggregates with hundreds or even thousands of events, but it won't scale to tens of thousands. Snapshot support — both for aggregate state and projection cursors — is planned.

Single-node (for now). One node means one point of failure. We mitigate this with filesystem-level backups of the log file — it's a single append-only file, so incremental backups are trivial. Clustering and replication are on the roadmap for workloads that need higher availability.

No server-side projections. Projections run in the application process, not in the event store. The framework makes this transparent, but it's a different operational model than something like EventStoreDB's persistent subscriptions. Subscription checkpoint messages — explicit signals from the store telling projection services when to persist their progress — are planned to make this more robust.

Full event log in memory. The in-memory index means memory usage grows linearly with event count. For the CRM's workload (thousands of events per month), this is negligible. For higher-throughput systems, a disk-backed index would be the natural evolution.

These trade-offs let us ship a system simple enough to fully understand from top to bottom. When something goes wrong at 2am — and something always goes wrong at 2am — we can reason about the entire storage engine from first principles, not debug a distributed consensus protocol.

What's Next

Eventfold is under active development. Beyond snapshots and clustering, the roadmap includes:

  • Event schema evolution — upcasting hooks and versioned event types, so domain models can evolve without breaking the log
  • Stream compaction and archival — lifecycle management for long-lived aggregates
  • Shared proto crate — a single eventfold-proto crate as the source of truth for the gRPC schema, eliminating drift between the store and framework
  • Interactive console TUI — a terminal interface for browsing streams, reading events, and live-tailing the global log during development (already in progress)
  • Derive macros — reducing the boilerplate of implementing Aggregate, Projection, and ProcessManager traits
  • Multi-aggregate transactions — best-effort atomic appends across multiple streams for workflows that can't be expressed cleanly as sagas
  • Open-source publishing — both crates are being prepared for public release on crates.io and GitHub

The guiding principle remains the same: add complexity only when the workload demands it, not before.

The Result

The CRM runs on eventfold-db and eventfold-es in production. The event store is under 2,000 lines of Rust. The framework is another few thousand. Together they give us:

  • A complete, immutable audit trail of every action ever taken in the system
  • Optimistic concurrency without distributed locks
  • Eventually consistent read models that catch up automatically
  • Sub-millisecond reads from the in-memory index
  • Durable writes with CRC32 integrity verification
  • A crash recovery model we can explain in one paragraph

For a bespoke CRM that needed to fit an existing workflow rather than force a new one, event sourcing turned out to be the right foundation. And building the infrastructure ourselves meant we could make it exactly as complex as the problem required — and no more.