Crate architecturekit 

architecturekit

Building blocks for DDD-based applications with CQRS and event sourcing in Rust, on top of EventSourcingDB – a purpose-built database for event sourcing.

architecturekit takes care of what such an application needs besides its domain: events whose schemas are derived from their types, states that are replayed from events, commands that are decided on them with optimistic concurrency, projections that keep views up to date, answers to commands and queries over HTTP, and tests for deciders and projections. The API is documented on docs.rs.

Installation

Add architecturekit, together with serde to write and read the data of events, and Tokio to run it all:

cargo add architecturekit
cargo add serde --features derive
cargo add tokio --features macros,rt-multi-thread

architecturekit brings the client SDK for EventSourcingDB along as architecturekit::eventsourcingdb, and ed25519_dalek for verifying signatures as architecturekit::ed25519_dalek, so that their versions always fit. A new major version of the client SDK comes with a new major version of architecturekit.

architecturekit needs Rust 1.88 or later. Its features are:

• derive, which is on by default, derives events, their enums, and their schemas.
• http answers commands and queries over HTTP, with the types of the http crate.
• axum adds extractors and handlers for axum 0.8, and turns on http.
• testing adds helpers to test deciders and projections.
• testcontainers adds an EventSourcingDB in a container of a test’s own, and turns on testing.

Defining Events

An event describes what has happened. Derive Event for its data, together with serde’s Serialize and Deserialize, and name its event type:

use architecturekit::Event;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Event)]
#[event(type = "io.eventsourcingdb.library.book-acquired")]
struct BookAcquired {
    title: String,
    author: String,
}

#[derive(Serialize, Deserialize, Event)]
#[event(type = "io.eventsourcingdb.library.book-borrowed")]
struct BookBorrowed {
    reader: String,
}

The derive also derives the JSON schema of the event from the fields that serde writes. EventSourcingDB rejects events whose data does not match the registered schema, and a registered schema can not change, so the rules of the derive are fixed: the same fields always give the same schema.

For a type within the data of an event, such as a nested struct, derive Schema. A type whose schema can not be derived, such as an enum, implements Schema by hand, and the derive refuses to compile otherwise. For the rules, see the documentation of Schema on docs.rs.

Replaying States

A state holds what a command is decided on, such as whether a book has been borrowed. It is replayed from the events of its subject, which an enum lists, with one variant per event type, so that the compiler checks that the state handles every event it may see:

use architecturekit::{Events, State};

#[derive(Events)]
enum BookEvent {
    Acquired(BookAcquired),
    Borrowed(BookBorrowed),
}

#[derive(Clone, Default)]
struct Book {
    is_acquired: bool,
    is_borrowed: bool,
}

impl State for Book {
    type Event = BookEvent;

    fn evolve(self, event: &BookEvent) -> Self {
        match event {
            BookEvent::Acquired(_) => Book { is_acquired: true, ..self },
            BookEvent::Borrowed(_) => Book { is_borrowed: true, ..self },
        }
    }
}

A state starts from its Default. An event of a type that the enum does not list is left out, and upcasters turn events of older types into current ones (see State::upcasters).

Deciding on Commands

A command names the subject whose state it is decided on, and the preconditions that its events are written with. Precondition::on_state_read writes them only if no other event has been written to the subject since its state was read, which is optimistic concurrency. A Subject is a path that EventSourcingDB accepts, such as /books/42, and a SubjectScheme builds subjects from values, and reads values from subjects:

use std::sync::LazyLock;

use architecturekit::{Command, Precondition, Subject, SubjectScheme};

static BOOKS: LazyLock<SubjectScheme> = LazyLock::new(|| SubjectScheme::new("/books/{book_id}"));

#[derive(Clone)]
struct BorrowBook {
    book: Subject,
    reader: String,
}

impl Command for BorrowBook {
    fn subject(&self) -> Subject {
        self.book.clone()
    }

    fn preconditions(&self) -> Vec<Precondition> {
        vec![Precondition::on_state_read()]
    }
}

A decider takes a command and the state, and either returns the events that the command causes, or refuses it:

use architecturekit::{Decider, DomainError};

fn borrow_book() -> Decider<BorrowBook, Book, DomainError> {
    Decider::new(|command: BorrowBook, book: Book| async move {
        if !book.is_acquired {
            return Err(DomainError::new(format!("{} does not exist", command.book)));
        }
        if book.is_borrowed {
            return Err(DomainError::new(format!("{} is already borrowed", command.book)));
        }

        Ok(vec![BookEvent::Borrowed(BookBorrowed { reader: command.reader })])
    })
}

A decider is asynchronous, so that it can load further states from the store. DomainError refuses a command with a message. An error type of your own works as well, and implements Categorized to say whether it refuses the command or reports a failure.

Executing Commands

The store connects to EventSourcingDB through the client SDK, and names the source of the events that it writes. Register the schemas of the events once at the start, and then execute commands:

use architecturekit::{Store, eventsourcingdb::Client};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new("http://localhost:3000".parse()?, "secret");
    let store = Store::builder(client, "https://library.eventsourcingdb.io").build();
    store.register_schemas(BookEvent::schemas()).await?;

    let command = BorrowBook { book: BOOKS.build(&["42"]), reader: "Jane".to_owned() };
    let written = store.execute(&borrow_book(), command).await?;
    println!("borrowed with {} event(s)", written.len());

    Ok(())
}

execute loads the state, decides, and writes the events with their preconditions. If another command has written to the subject in the meantime, it fails with a conflict, unless StoreBuilder::conflict_retries lets it decide again on the new state. Every error tells by its category what to do about it: a refusal of the decider is final, a transient failure, such as a conflict, may pass when trying again, and a permanent one will not. A write whose outcome is unknown has a variant of its own, since its events may have been stored.

Projecting Events into Views

A projection applies events to views, which queries read. An InMemoryView keeps its items in memory, together with the revision of the event that last changed each of them, so that an event that is applied twice changes nothing:

use architecturekit::{Envelope, InMemoryView, Projection, Revision, ViewError};
use serde::Serialize;

#[derive(Clone, Default, Serialize)]
struct ListedBook {
    id: String,
    title: String,
    is_borrowed: bool,
}

struct BookList {
    books: InMemoryView<String, ListedBook>,
}

impl Projection for BookList {
    type Event = BookEvent;
    type Error = ViewError;

    async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), ViewError> {
        let Some(book_id) = BOOKS.value(&event.subject, "book_id") else {
            return Ok(());
        };

        match &event.data {
            BookEvent::Acquired(acquired) => {
                let book = ListedBook {
                    id: book_id.to_owned(),
                    title: acquired.title.clone(),
                    is_borrowed: false,
                };
                self.books.insert(&event, book)?;
            }
            BookEvent::Borrowed(_) => {
                self.books.update(book_id, &event, |book| book.is_borrowed = true)?;
            }
        }

        Ok(())
    }

    fn seen(&self, event_id: Revision) {
        self.books.seen(event_id);
    }
}

seen tells the view about every event the projection has handled, including those it had nothing to do with, so that a query can wait for a revision. The store starts a run of the projection, which reads the events stored so far, and then follows the events as they are written:

use std::sync::Arc;

use architecturekit::{CancellationToken, ProjectionOptions, Subjects, wait_caught_up};

let book_list = Arc::new(BookList {
    books: InMemoryView::new(|book: &ListedBook| book.id.clone()),
});
let token = CancellationToken::new();

let run = store.start_projection(
    Subjects::tree(BOOKS.root()),
    Arc::clone(&book_list),
    token.clone(),
    ProjectionOptions::named("book list"),
);
wait_caught_up(&[&run]).await?;

If reading fails in a way that may pass, the run reconnects and goes on from the last event it applied, and a failure that will not pass ends it, as does cancelling the token. ProjectionRun::status tells where it stands. A view in a database of its own implements ResumableProjection, which goes on from a checkpoint after a restart, or TransactionalProjection, which applies the events and the checkpoint within one transaction.

Serving Commands and Queries

The feature axum adds what a web server built with axum needs to answer commands and queries:

cargo add architecturekit --features axum
cargo add axum
cargo add tower-http --features catch-panic

A command is an ordinary handler of axum:

use std::sync::Arc;

use architecturekit::httpapi::{CommandAnswer, HttpError, Strict};
use axum::extract::{Path, State};
use serde::Deserialize;

#[derive(Clone)]
struct App {
    store: Arc<Store>,
    borrow_book: Arc<Decider<BorrowBook, Book, DomainError>>,
    book_list: Arc<BookList>,
}

#[derive(Deserialize)]
struct BorrowRequest {
    reader: String,
}

async fn borrow(
    State(app): State<App>,
    Path(book_id): Path<String>,
    Strict(request): Strict<BorrowRequest>,
) -> Result<CommandAnswer, HttpError> {
    let command = BorrowBook { book: BOOKS.try_build(&[&book_id])?, reader: request.reader };
    let written = app.store.execute(&app.borrow_book, command).await?;

    Ok(CommandAnswer::new(&written))
}

Strict reads the body as JSON, and refuses a body that does not fit, also because of a field that the type does not have, or a list where an object belongs, with 400 and what is wrong, such as {"message":"malformed request: unknown field \"raeder\""}. serde decodes an enum with #[serde(untagged)] or #[serde(tag = "…")], and a field with #[serde(flatten)], from a copy of its own that no strict reading reaches, so within them neither of these two is refused. SubjectScheme::try_build refuses an ID that can not be part of a subject, such as a.b, which ? answers with 400 as well, where build would panic. A command that has succeeded is answered with the revision it wrote, such as {"revision":"42"}. Every error of architecturekit turns into an HttpError with ?, which answers it by its category:

Failure	Status	Message
an ID that can not be part of a subject	400	its own
a refusal of the decider	422	its own
a conflict	409	conflict: the data has changed since it was read
another transient failure	503	internal server error
a write whose outcome is unknown	500	outcome unknown: the request may have succeeded
any other failure	500	internal server error

A fixed message keeps internals from the caller, and its cause is logged with the log crate. The one error that does not turn into an HttpError with ? is the one of wait_for_written, which waits until a view has seen what a command wrote: the command has succeeded, so it is answered as a success, with CommandAnswer::with_result, which logs that the view lags behind. Who may send a command is up to the application: it determines the user with an extractor of its own, and refuses with HttpError::unauthorized or HttpError::forbidden.

A query is a handler of axum as well, which reads a view:

use architecturekit::KeyedView;
use architecturekit::httpapi::{DEFAULT_WAIT, QueryAnswer, Reading};
use serde::Serialize;

#[derive(Deserialize, Serialize)]
struct BookById {
    id: String,
}

async fn book_by_id(
    State(app): State<App>,
    reading: Reading,
    Strict(query): Strict<BookById>,
) -> Result<QueryAnswer<ListedBook>, HttpError> {
    let books = &app.book_list.books;

    reading
        .revisioned(books, &query, DEFAULT_WAIT, || async {
            let Ok(book) = books.get(&query.id).await;
            Ok(QueryAnswer::found(book))
        })
        .await
}

A query is sent with the HTTP method QUERY, with the query as its body, and answered with what the view holds, or with 404 if it holds nothing. A caller that has just sent a command passes its revision in the header Wait-For-Revision, and the query waits until the view has seen it, for at most DEFAULT_WAIT, so that the caller reads what it has written. The answer names the revision of the view, and carries a tag, so that a caller that holds the answer already gets 304.

The router puts it all together, with the probes of an orchestrator such as Kubernetes, and an answer to a handler that panics:

use architecturekit::httpapi::{Health, answer_panic, query};
use axum::{Router, routing::{get, post}};
use tower_http::catch_panic::CatchPanicLayer;

let app = App {
    store: Arc::new(store),
    borrow_book: Arc::new(borrow_book()),
    book_list,
};

let router: Router = Router::new()
    .route("/books/{book_id}/borrow", post(borrow))
    .route("/books/by-id", query(book_by_id))
    .route("/health/ready", get(Health::readiness([run.clone()])))
    .route("/health/alive", get(Health::liveness([run])))
    .layer(CatchPanicLayer::custom(answer_panic))
    .with_state(app);

Health::readiness answers with 200 once every projection has caught up, and with 503 before, and once one of them has stopped. Health::liveness answers with 503 only once a projection has stopped. answer_panic answers a handler that panics with 500, and logs the panic. For another web framework, the feature http offers the same with the types of the http crate, such as body_of and Reading::of.

Testing Deciders

The feature testing adds given-when-then for deciders in architecturekit::testing, and the feature testcontainers adds an EventSourcingDB in a container of a test’s own, which needs Docker. Turn them on for the tests only:

cargo add architecturekit --dev --features testcontainers

A decider decides asynchronously, so a test runs on Tokio, and waits for the decision:

use architecturekit::testing::given;

// In a #[tokio::test]:
given(
    &borrow_book(),
    [
        BookEvent::Acquired(BookAcquired { title: "2001".to_owned(), author: "Arthur C. Clarke".to_owned() }),
        BookEvent::Borrowed(BookBorrowed { reader: "Jane".to_owned() }),
    ],
)
.when(BorrowBook { book: BOOKS.build(&["42"]), reader: "John".to_owned() })
.await
.then_failed(DomainError::new("/books/42 is already borrowed"));

then_failed compares the error of the decider, which therefore implements PartialEq and Debug. then_events compares events by their type and the JSON of their data, which is what would be stored, so they need no PartialEq, and then_preconditions compares the preconditions they are written with.

Testing Projections

The feature testing also drives a projection with events in the shape that the database stores them in, without a database, and the same way a run does: the upcasters run, an event of a type that the projection has no variant for is left out, an event that an upcaster splits arrives part by part, and the projection learns of every event in seen. A test then reads what the projection built from its views:

use architecturekit::{KeyedView, testing::{project, stored_events}};

// In a #[tokio::test]:
let book_list = BookList { books: InMemoryView::new(|book: &ListedBook| book.id.clone()) };

project(
    &book_list,
    &stored_events(
        &BOOKS.build(&["42"]),
        [BookEvent::Acquired(BookAcquired { title: "2001".to_owned(), author: "Arthur C. Clarke".to_owned() })],
    ),
)
.await
.expect("the events are applied");

let Ok(book) = book_list.books.get("42").await;
assert_eq!(book.map(|book| book.title), Some("2001".to_owned()));

project returns the failure of the projection, so that a test can check that a projection refuses an event. The fields of a StoredEvent are public, so that a test can give an event a type that the projection does not know, data that it can not read, or a time of its own. project_transactional drives a transactional projection within one transaction.

License

architecturekit is licensed under the MIT License.

Re-exports

pub use eventsourcingdb;

pub use ed25519_dalek;

Modules

httpapi

Serving commands and queries over HTTP. The module needs the feature http, and its parts for axum the feature axum, which turns on http as well.

testing

Given-when-then for deciders, driving projections with stored events, and a database in a container for the tests that need a real one. The module needs the feature testing, and Database the feature testcontainers, which turns on testing as well, so that a test that only checks a decider builds without Docker.

Structs

CancellationToken

Stops a run of a projection from outside (see Store::start_projection). A token which can be used to signal a cancellation request to one or more tasks.

Decider

A decision on a command, given the state of its subject: the events to write, or an error if the command is refused.

DomainError

A business rule that refused a command, with a message for the caller.

EmptyRange

The bounds of a read leave no room for an event, such as one before 0, one after 0 and before 1, or one after the largest revision, 2^63-1. No event can ever lie in such a range, so it is a mistake of whoever chose the bounds, which usually come from a request, and it belongs to no category. A range that is only empty for now, such as the one after the last event written so far, is no such mistake, since events can still come there, so reading hands out no events for it, without an error.

Envelope

An event as a projection gets it: the metadata of the event as the database recorded it, and its data, as Projection::Event says.

EventOn

An event together with the subject it is written to, for Store::write.

EventSchema

The schema of an event type, for registration with the database.

Failure

What failed, as in reading "/books/42", and the error that caused it, if there is one, such as the error of the client SDK.

InMemoryIndex

A secondary index of an InMemoryView, created with InMemoryView::index, which finds the items by a value taken from them, such as the shelf they stand on. Several items may share a value, so a change through the index reports how many items it changed rather than an Outcome.

InMemoryView

Holds the items of a read model in memory, each under a key of its own. It is the one view the kit ships, and because it keeps no checkpoint, its projection is rebuilt on every start.

NotARevision

A value that was meant to be a revision, but is not one.

NotASubject

A value that was meant to be a subject, but is not one, such as books/42 or /books//42.

Precondition

A condition that must hold for the events of a command to be written.

ProjectionOptions

How a projection is started. Build it with named or default, which leaves the projection without a name:

ProjectionRun

A run of a projection, which follows the events of its subjects until its token is cancelled, or until a failure that trying again will not fix. Start one with Store::start_projection or its counterparts. It can be cloned, and every clone watches the same run.

ProjectionStatus

The status of a run of a projection, as ProjectionRun::status returns it. It may gain fields, so it can not be built outside the kit.

ReadOptions

Narrows down which events Store::read hands out, or turns around the order it hands them out in. A read has one lower bound, one upper bound, and one order, so options that contradict each other can not be written.

Reconnect

An attempt of a projection to read again, after reading failed, which includes a stream of events that ended, as the observer set with StoreBuilder::reconnect_observer receives it. It may gain fields, so it can not be built outside the kit.

Revision

A revision, which is the ID of an event of EventSourcingDB: a number from 0 to 2^63-1, since the database numbers its events with signed 64-bit integers, as one ascending sequence across all subjects. Revisions sort as numbers, so that 10 comes after 9, and a value that is no event ID can not be a revision at all.

Store

Reads from and writes to EventSourcingDB, through the client SDK.

StoreBuilder

Builds a Store, which Store::builder creates.

Subject

A subject of EventSourcingDB, which is a path such as /books/42: it starts with a slash, / alone is the root, and its segments are separated by single slashes, none of them empty, each made only of the characters A-Z, a-z, 0-9, underscores, and hyphens. A single slash at the end is taken, and left out, as the database does, so /books/42/ is /books/42. A value that is no such path can not be a subject at all.

SubjectScheme

How a subject is composed: literal segments and placeholders in braces, as in /tenant/{tenant}/workshop/{workshop}.

SubjectValuesError

The values that SubjectScheme::check refuses, with a message that may reach the caller of an API.

Subjects

What a read or a projection reads: a subject, and whether the subjects below it belong to it. Create it with tree or exact, so that every read says which of the two it means, rather than relying on a default.

UpcastEvent

A stored event as upcasters see it: its type and its data, which an upcaster changes, and its subject, ID, time and source, which it reads.

Upcasters

The set of upcasters for the event types of an application, which translate stored events of an older type into a newer shape, before any state or projection sees them.

Enums

Category

What a failure means for its caller, on two axes: whether it is a matter of the domain or of the machinery, and whether trying again unchanged could help.

CaughtUpError

Why wait_caught_up failed. It may gain variants and fields, so match it with a catch-all arm, or ask for its category.

Error

Any failure of architecturekit, where D is the error of a decider, so that a function that calls several operations can pass their failures on with ?.

ExecuteError

Why executing a command failed, where D is the error of its decider.

IfLatestEventIsMissing

What LowerBound::FromLatestEvent reads if the subject has no event of the type yet.

LoadError

Why loading a state failed.

LowerBound

Where the events of a read start.

NotCaughtUp

Why a view did not catch up with written events (see wait_for_written). The events were written, so the write has succeeded, and only the view lags behind.

Order

The order in which a read hands out the events.

Outcome

What a change of an InMemoryView did to the one item it concerns. A change of several items, such as InMemoryView::update_where or InMemoryIndex::update, reports how many it changed instead, since each of them may fare differently.

Permanent

A failure of the category Category::Permanent: trying again will not help.

Phase

Where a run of a projection stands.

PreconditionsError

Why a command declares its preconditions in a way they can not be honored (see check_preconditions).

ProjectionError

Why a projection failed. It may gain variants and fields, so match it with a catch-all arm, or ask for its category.

ReadError

Why reading events failed. It has exactly the cases reading can end in, as the error of every operation does, so that a match handles all of them.

ReadOptionsError

Why the options of a read are refused (see check_read_options). It has exactly the cases the options can be refused for, as the error of every operation does, so that a match handles all of them; a new one comes with a new option, and so with a new major version.

RegisterSchemasError

Why registering the schemas of events failed.

Transient

A failure of the category Category::Transient: trying again later, unchanged, may succeed.

UpperBound

Where the events of a read end.

ViewError

Why a change of an InMemoryView was refused. It leaves the view as it is, and belongs to the category Category::Permanent, since applying the same event again will fail again.

WriteError

Why writing events failed.

Traits

Categorized

An error that names its Category.

Command

A command, which knows the subject it acts on, and the conditions under which its events may be written.

Event

The data of an event, which describes what has happened.

Events

The events a state evolves by, as one enum with a variant per event type.

KeyedView

A view that finds a single item by its key, without reading all of them. InMemoryView is one, for the type of its keys, and for every type that a key can be borrowed as, such as str for a String.

Projected

What a projection can get as its events: an enum with #[derive(Events)], or eventsourcingdb::Event, the event as it is stored. The kit implements it for both, and for nothing else.

Projection

A projection that applies events, and is rebuilt from the first event on every start, which is right for a view kept in memory. A projection whose target can hold a checkpoint is a ResumableProjection as well, and one whose target can make data and checkpoint durable together is a TransactionalProjection instead.

ResumableProjection

A projection whose target can hold a checkpoint, but not together with the data, so that it goes on from there on the next start. It writes the checkpoint after the data, so an event may be applied twice after a crash, and apply has to be idempotent.

Revisioned

A view that knows how far its projection has come, which lets a reader wait for a revision instead of polling. InMemoryView is one.

Schema

The JSON schema of a type, which describes exactly what serde writes for it.

State

The state a command decides on, together with the rule that builds it from events.

Transaction

One unit of work of a TransactionalProjection. Commit has to make the applied events and the ID of the last one durable together, or neither. Rollback is not called after commit fails, so commit has to roll back itself if it can not finish. A panic in apply is rolled back like an error.

TransactionalProjection

A projection whose target can make data and checkpoint durable together, as a relational database can. It applies events only within a Transaction.

View

A collection of items that a query reads, wherever they live, in memory or in a database. InMemoryView is one.

Functions

check_preconditions

Checks that a command declares its preconditions in a way they can be honored, before anything is read, as executing a command does.

check_read_options

Checks the options of a read as Store::read checks them before it asks the database. Use it to check bounds that come from a request before a side effect that must not happen for a read that is refused, such as recording who reads. The database may still refuse a read that passes, such as one from a latest event that comes after the upper bound.

replay

Folds a sequence of events into a state, starting from Default. It is meant for tests, where the history is at hand as typed events.

replay_stored

Folds stored events into a state, running the upcasters on the way, exactly as reading from the database would. It is meant for tests of upcasters, which need the shape of an older version.

revision_of

Returns the highest event ID among written events, which is the revision a caller has to see before its own write shows up in a read model, or None if nothing was written. The order of the events does not matter, and an ID that is not a revision is skipped.

step

Takes a state a single event further, by the same rules as replay. It starts from the given state instead of Default, and leaves that as it is, so that the state before and after the event are both at hand, for example for a history that tells what an event changed.

step_stored

Takes a state a single stored event further, running the upcasters on the way, exactly as reading from the database would. If the upcasters turn the event into several, the state evolves by all of them. Otherwise, it behaves like step.

wait_caught_up

Waits until every given run has caught up, as a service does before it takes requests. A run that has caught up once counts, even if it has ended since. Dropping the future stops waiting.

wait_for_written

Waits until the view has seen the events that Store::execute or Store::write returned, for a step on the server that builds on what was just written, such as one that reads from the view what a command has changed:

Derive Macros

Event

Derives architecturekit::Event and architecturekit::Schema for the data of an event. See architecturekit::Event.

Events

Derives architecturekit::Events for the enum of the events a state evolves by. See architecturekit::Events.

Schema

Derives architecturekit::Schema for a type in the data of an event. See architecturekit::Schema.