Evento

A collection of libraries and tools that help you build DDD, CQRS, and event sourcing applications in Rust.

• Event sourcing — state changes stored as immutable events, with optimistic concurrency and a complete audit trail
• CQRS — commands, projections (read models), and continuous subscriptions
• Macros — #[evento::aggregate], #[evento::command], #[evento::projection], #[evento::snapshot], #[evento::handler], #[evento::subscription]
• Compact storage — fast binary serialization with bitcode

One Executor trait, many backends:

Installation

Evento 2.x is currently in alpha; pin the exact pre-release version (a bare "2" does not resolve pre-releases):

[dependencies]
evento = { version = "2.0.0-alpha.27", features = ["sqlite"] }
bitcode = "0.6"
anyhow = "1"

Swap sqlite for postgres, mysql, fjall, or remote as needed (see Feature flags).

Quick start

1. Define events with an aggregate enum

Each variant becomes an event struct with all required traits (bitcode serialization, Aggregate, AggregateEvent):

#[evento::aggregate]
pub enum BankAccount {
    AccountOpened { owner: String, initial_balance: i64 },
    MoneyDeposited { amount: i64 },
}

The aggregate type defaults to "{package_name}/{enum_name}". Pin it (and event names) so refactors never orphan stored events:

#[evento::aggregate(name = "bank/BankAccount")]
pub enum BankAccount {
    #[evento(name = "AccountOpened")]
    AccountOpened { owner: String },
}

2. Write events

create() starts a new aggregate; append(id) continues an existing one with optimistic concurrency:

# #[evento::aggregate]
# pub enum BankAccount {
#     AccountOpened { owner: String, initial_balance: i64 },
#     MoneyDeposited { amount: i64 },
# }
# async fn run<E: evento::Executor>(executor: &E) -> anyhow::Result<()> {
let id = evento::create()
    .event(&AccountOpened { owner: "Alice".into(), initial_balance: 1000 })
    .routing_key("accounts")
    .commit(executor)
    .await?;

// Fails with WriteError::InvalidOriginalVersion if another writer raced ahead.
evento::append(&id)
    .original_version(1)
    .event(&MoneyDeposited { amount: 100 })
    .commit(executor)
    .await?;
# Ok(())
# }

3. Build read models with projections

Handlers are pure (event, &mut view) functions; load(id) replays the aggregate's events through them:

use evento::{metadata::Event, projection::Projection};

# #[evento::aggregate]
# pub enum BankAccount {
#     AccountOpened { owner: String, initial_balance: i64 },
#     MoneyDeposited { amount: i64 },
# }
// bitcode derives make the view snapshottable through the executor.
#[evento::projection(bitcode::Encode, bitcode::Decode)]
pub struct AccountView {
    pub owner: String,
    pub balance: i64,
}

#[evento::handler]
async fn on_opened(event: Event<AccountOpened>, view: &mut AccountView) -> anyhow::Result<()> {
    view.owner = event.data.owner.clone();
    view.balance = event.data.initial_balance;
    Ok(())
}

#[evento::handler]
async fn on_deposited(event: Event<MoneyDeposited>, view: &mut AccountView) -> anyhow::Result<()> {
    view.balance += event.data.amount;
    Ok(())
}

# async fn run<E: evento::Executor>(executor: &E, id: &str) -> anyhow::Result<()> {
let view: Option<AccountView> = Projection::<_, AccountView>::new::<BankAccount>()
    .handler(on_opened())
    .handler(on_deposited())
    .load(id)
    .execute(executor)
    .await?;
# Ok(())
# }

4. Commands and the write gateway

The write side loads current state, guards invariants, then emits events through the loaded projection's write() gateway — which continues the stream at the version the load observed, so concurrent commands conflict instead of clobbering each other. #[evento::command] generates routing-key variants from one method body:

use evento::{metadata::Event, Executor, Projection, ProjectionAggregate};

# #[evento::aggregate]
# pub enum BankAccount {
#     AccountOpened { initial_balance: i64 },
#     MoneyDeposited { amount: i64 },
# }
// The write model: `id = id` implements ProjectionAggregate (enables `write()`),
// `snapshot(memory)` keeps a process-local materialized row per aggregate.
#[evento::projection(id = id)]
#[evento::snapshot(memory)]
pub struct Account {
    pub id: String,
    pub balance: i64,
}

# #[evento::handler]
# async fn on_opened(event: Event<AccountOpened>, row: &mut Account) -> anyhow::Result<()> {
#     row.id = event.aggregate_id.to_owned();
#     row.balance = event.data.initial_balance;
#     Ok(())
# }
# #[evento::handler]
# async fn on_deposited(event: Event<MoneyDeposited>, row: &mut Account) -> anyhow::Result<()> {
#     row.balance += event.data.amount;
#     Ok(())
# }
fn account_projection<E: Executor>() -> Projection<E, Account> {
    Projection::new::<BankAccount>()
        .handler(on_opened())
        .handler(on_deposited())
        .strict() // fail on events nobody handles
}

pub struct Command<E: Executor>(pub E);

#[evento::command]
impl<E: Executor> Command<E> {
    /// Written once; the trailing `routing_key` parameter makes the macro
    /// generate `deposit_money(id, amount)`, `deposit_money_with_routing(id,
    /// amount, key)`, and `deposit_money_opt(id, amount, Option<String>)`.
    pub async fn deposit_money(
        &self,
        id: impl Into<String>,
        amount: i64,
        routing_key: Option<String>,
    ) -> anyhow::Result<()> {
        let Some(account) = account_projection().load(id).execute(&self.0).await? else {
            anyhow::bail!("account not found");
        };
        if amount <= 0 {
            anyhow::bail!("invalid amount");
        }

        account
            .write()?
            .routing_key_opt(routing_key)
            .event(&MoneyDeposited { amount })
            .commit(&self.0)
            .await?;
        Ok(())
    }
}

# async fn run<E: Executor>(cmd: Command<E>) -> anyhow::Result<()> {
cmd.deposit_money("account-1", 50).await?;
cmd.deposit_money_with_routing("account-1", 50, "eu-west").await?;
# Ok(())
# }

See examples/bank for the full pattern with domain errors and ten commands.

5. Snapshots

Loading replays an aggregate's events; snapshots cut that short. Three modes:

• Executor-backed (default): derive bitcode::Encode/bitcode::Decode on the projection (#[evento::projection(bitcode::Encode, bitcode::Decode)]) and the snapshot is persisted in the event store.
• #[evento::snapshot(memory)]: a process-local table keyed by aggregate id, with a snapshot_rows() accessor for reading materialized rows.
• #[evento::snapshot(none)]: opt out — always replay from scratch.

#[evento::projection]
#[evento::snapshot(memory)]
pub struct MemView {
    pub id: String,
    pub balance: i64,
}

#[evento::projection]
#[evento::snapshot(none)]
pub struct StatusView {
    pub frozen: bool,
}

let rows = MemView::snapshot_rows().read().unwrap();
# drop(rows);

6. Subscriptions

Process events continuously (side effects allowed), with cursor tracking, retries, and graceful shutdown:

use evento::{metadata::Event, subscription::{Context, SubscriptionBuilder}, Executor};

# #[evento::aggregate]
# pub enum BankAccount {
#     MoneyDeposited { amount: i64 },
# }
#[evento::subscription]
async fn notify<E: Executor>(
    _ctx: &Context<'_, E>,
    event: Event<MoneyDeposited>,
) -> anyhow::Result<()> {
    println!("deposited {}", event.data.amount);
    Ok(())
}

# async fn run<E: Executor + Clone>(executor: &E) -> anyhow::Result<()> {
let subscription = SubscriptionBuilder::new("deposit-notifier")
    .handler(notify())
    .routing_key("accounts")
    .chunk_size(100)
    .retry(5)
    .start(executor)
    .await?;

// On application shutdown
subscription.shutdown().await?;
# Ok(())
# }

To drain currently-pending events once instead of running a background loop, use run_once(&executor) (optionally after no_retry()). To keep a projection auto-updated, use projection.subscription("key").start(&executor). Handlers for all events of an aggregate without deserializing go through #[evento::subscription_all] with RawEvent<A>.

Wiring a backend

Fjall (embedded, zero setup)

# fn run() -> anyhow::Result<()> {
let executor = evento::Fjall::open("./data")?;
# let _ = executor;
# Ok(())
# }

SQLite (or PostgreSQL/MySQL) with migrations

use evento::migrator::{Migrate, Plan};
use sqlx::sqlite::SqlitePoolOptions;

# async fn run() -> anyhow::Result<()> {
let pool = SqlitePoolOptions::new().connect("sqlite:events.db").await?;

// Run migrations (generic over the database type)
let mut conn = pool.acquire().await?;
evento::sql_migrator::new::<sqlx::Sqlite>()?
    .run(&mut *conn, &Plan::apply_all())
    .await?;
drop(conn);

let executor: evento::Sqlite = pool.into();
# let _ = executor;
# Ok(())
# }

Remote (client/server split)

Serve any executor over framed TCP; the client implements Executor, so commands, projections, and subscriptions work unchanged across the network:

# async fn run() -> anyhow::Result<()> {
// Server process
let executor = evento::Fjall::open("./data")?;
let listener = tokio::net::TcpListener::bind("0.0.0.0:4321").await?;
let handle = evento::remote::serve(listener, executor);

// Client process
let client = evento::RemoteClient::connect("127.0.0.1:4321".parse()?).await?;
# let _ = (handle, client);
# Ok(())
# }

Accord (replicated, alpha)

evento-accord replicates writes through the Accord consensus protocol (Cassandra CEP-15): leaderless, strictly serializable, highly available, with any local backend (Fjall/SQL) serving reads. See its README, DESIGN.md, and OPERATIONS.md, plus the bank-axum-accord 3-node demo.

Core API at a glance

Full macro reference: evento-macro/README.md.

Feature flags

• macro (default) - Procedural macros for aggregates and handlers
• sql - Enable all SQL database backends
• sqlite / postgres / mysql - Individual SQL backends with migrations
• fjall - Embedded key-value storage with Fjall
• remote - Client/server executor over framed TCP
• group - Multi-executor support for querying across databases
• rw - Read-write split executor for CQRS patterns

Workspace crates

Examples

Complete working examples in examples/:

• quickstart - Smallest end-to-end run (Fjall): cargo run -p quickstart
• bank - Bank domain: aggregates, ten commands, projections, snapshots
• bank-axum-sqlite - Axum + SQLite + migrations: cargo run -p bank-axum-sqlite
• bank-axum-fjall - Axum + embedded Fjall: cargo run -p bank-axum-fjall
• bank-axum-remote - Two-process client/server split: cargo run -p bank-axum-remote -- store then cargo run -p bank-axum-remote
• bank-axum-accord - 1- or 3-node Accord cluster: make accord or make accord.cluster

License

Licensed under the Apache License, Version 2.0.