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:

Backend	Feature	Crate	Notes
SQLite / PostgreSQL / MySQL	`sqlite` / `postgres` / `mysql`	evento-sql	via sqlx, with built-in migrations
Fjall (embedded LSM-tree)	`fjall`	evento-fjall	no external server
Remote (client/server TCP)	`remote`	evento-remote	serve any executor over framed TCP
Accord (consensus, alpha)	—	evento-accord	leaderless replicated store, strictly serializable (design)

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, keyed by (aggregate type, projection name, aggregate id) — so an aggregate can have any number of snapshotted views. The projection name defaults to "<module path>::<Struct>"; pin it with name = "..." so that renaming or moving the struct does not orphan its snapshots.
#[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);

// Executor-backed, with a name that survives refactors:
#[evento::projection(name = "myapp/BalanceView", bitcode::Encode, bitcode::Decode)]
pub struct BalanceView {
    pub balance: i64,
}

Changing the shape of an executor-backed projection needs a .revision(n) bump on its Projection, so snapshots taken with the old shape are dropped instead of mis-decoded. A stored snapshot that no longer decodes is treated as a miss and rebuilt from events.

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>.

7. Evolving events

A stored event never changes: bitcode is positional, so its layout is frozen once a database holds one. When an event needs a new shape, add a new variant and point the old one at it. Older stored events are then converted before handlers see them, so only the newest handler has to exist:

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

#[evento::aggregate(name = "myapp/Payment")]
pub enum Payment {
    // Still in old streams, no longer written. Keep it: it is the decode schema.
    #[evento(upcast_to = PaymentRefundedV2)]
    PaymentRefunded { amount: i64 },
    PaymentRefundedV2 { amount: i64, reason: String },
}

impl From<PaymentRefunded> for PaymentRefundedV2 {
    fn from(old: PaymentRefunded) -> Self {
        Self { amount: old.amount, reason: "unknown".to_owned() }
    }
}

#[evento::projection]
#[evento::snapshot(none)]
pub struct RefundsView {
    pub total: i64,
}

#[evento::handler]
async fn on_refunded(event: Event<PaymentRefundedV2>, view: &mut RefundsView) -> anyhow::Result<()> {
    view.total += event.data.amount;
    Ok(())
}

fn refunds<E: Executor>() -> Projection<E, RefundsView> {
    // The only handler: it receives `PaymentRefunded` events too, upcast.
    Projection::new::<Payment>().handler(on_refunded()).strict()
}

It is declared once, on the aggregate. Every Projection and SubscriptionBuilder that registers a handler — or a .skip::<New>() — for the newer event picks it up; tombstone::<New>() and has_event::<New>() follow the older names as well.
Chains work (V1 -> V2 -> V3) and are folded into one decode and one encode. With handlers for both V2 and V3, a V1 event goes to the nearest one.
A handler registered for the older event itself wins over the upcast, so consumers can migrate one at a time.
The handler sees the newer event: event.name and event.data are the newer ones, everything else (id, version, timestamp, metadata) is the stored event's. #[evento::subscription_all] handlers keep receiving stored events as they are.
Snapshots hold folded state, not events: if the conversion yields different state than the handler you removed did, bump the projection's .revision(n).

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

Concern	Entry point
Define events	`#[evento::aggregate] enum`
Start a new aggregate	`evento::create()` → `WriteBuilder`
Append to an aggregate	`evento::append(id)` → `WriteBuilder`
Command with routing variants	`#[evento::command] impl Command<E>`
Read model	`#[evento::projection]` + `#[evento::handler]` fns
Emit events from loaded state	`#[evento::projection(id = ...)]` → `view.write()?`
Snapshot strategy	bitcode derives / `#[evento::snapshot(memory)]` / `#[evento::snapshot(none)]`
Load a read model	`Projection::new::<A>().handler(..).load(id).execute(exec)`
Co-keyed secondary aggregate	`.load(id).aggregate::<Other>(other_id)`
Filter events when reading	`EventFilter::by_type / by_id / by_event / exact`
Continuous processing	`SubscriptionBuilder::new(key)...start(exec)`
One-shot processing	`SubscriptionBuilder::new(key)...run_once(exec)`
Keep a projection updated	`projection.subscription(key).start(exec)`
Fail on unhandled events	`.strict()`
Keep going after a handler error	`.continue_on_error()`

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

Crate	Purpose
evento	Facade: re-exports core + feature-gated backends
evento-core	`Executor` trait, write path, projections, subscriptions
evento-macro	Procedural macros
evento-sql	SQLite/MySQL/PostgreSQL executor (sqlx)
evento-sql-migrator	Schema migrations for the SQL backend
evento-fjall	Embedded LSM-tree executor
evento-remote	Client/server executor over framed TCP
evento-accord	Accord consensus replicated executor (alpha)

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.

Name		Name	Last commit message	Last commit date
.claude/skills/evento		.claude/skills/evento
.github		.github
evento-accord		evento-accord
evento-core		evento-core
evento-fjall		evento-fjall
evento-macro		evento-macro
evento-remote		evento-remote
evento-sql-migrator		evento-sql-migrator
evento-sql		evento-sql
evento-test		evento-test
evento		evento
examples		examples
.dockerignore		.dockerignore
.gitignore		.gitignore
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
compose.yml		compose.yml
flake.lock		flake.lock
flake.nix		flake.nix

Navigation Menu

Repository files navigation

Evento

Installation

Quick start

1. Define events with an aggregate enum

2. Write events

3. Build read models with projections

4. Commands and the write gateway

5. Snapshots

6. Subscriptions

7. Evolving events

Wiring a backend

Fjall (embedded, zero setup)

SQLite (or PostgreSQL/MySQL) with migrations

Remote (client/server split)

Accord (replicated, alpha)

Core API at a glance

Feature flags

Workspace crates

Examples

License

About

Releases

Packages

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

Evento

Installation

Quick start

1. Define events with an aggregate enum

2. Write events

3. Build read models with projections

4. Commands and the write gateway

5. Snapshots

6. Subscriptions

7. Evolving events

Wiring a backend

Fjall (embedded, zero setup)

SQLite (or PostgreSQL/MySQL) with migrations

Remote (client/server split)

Accord (replicated, alpha)

Core API at a glance

Feature flags

Workspace crates

Examples

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages