Projections

Defining Projections#

A projection turns events into a view. Define it as a struct that holds the view, and implement the Projection trait. Its Event names the enum of the events it depends on, with #[derive(Events)], and its apply method receives every event as an Envelope, which holds the metadata of the event, such as its id, time, and subject, and its data, decoded into the enum.

Which parts of the subject a view needs is up to the application. In this example, the SubjectScheme called BOOKS takes the ID of the book out of the subject (see Composing Subjects):

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

struct CatalogProjection {
  catalog: InMemoryView<String, BookItem>,
}

impl Projection for CatalogProjection {
  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) => {
        self.catalog.insert(
          &event,
          BookItem {
            id: book_id.to_owned(),
            title: acquired.title.clone(),
            author: acquired.author.clone(),
            ..BookItem::default()
          },
        )?;
      }
      BookEvent::Borrowed(borrowed) => {
        self.catalog.update(book_id, &event, |item| {
          item.is_borrowed = true;
          item.borrowed_until = borrowed.borrowed_until.clone();
        })?;
      }
      BookEvent::Returned(_) => {
        self.catalog.update(book_id, &event, |item| {
          item.is_borrowed = false;
          item.borrowed_until = String::new();
        })?;
      }
    }

    Ok(())
  }

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

The event type of every variant is taken from the event it holds, so it does not have to be repeated, and the type of an event and the type its data is decoded into can not drift apart. An event whose subject does not follow the scheme is skipped. The seen method records every event in the view, so that a query can wait until the view has seen a write (see Tracking Revisions).

The projection shares its view with the queries that read it, so a run takes it in an Arc, and the queries read the view through the same Arc:

let catalog_projection = Arc::new(CatalogProjection { catalog: new_catalog() });

apply takes &self, since the projection shares what it builds with those who read it. Events of types the enum has no variant for are skipped, since a projection usually reads more events than it depends on, so a projection may have an enum of its own, with only the event types it needs, or share the enum of the state, as here. If the data of an event can not be decoded, the run fails with ProjectionError::Permanent. An error returned by apply is passed on together with the category its type names (see Running Projections).

If the view keeps the revisions of its items for a precondition, a skipped event makes the revision of its item fall behind the subject (see Defining Views). In that case, give every event type of the subject a variant, also one that does not change the item, such as BookInspected, whose arm in evolve returns the state unchanged. Its arm calls update with a change that does nothing, which only moves the revision of the item on. A projection that shares the enum of the state gets the variant along, and the compiler insists on an arm for it:

impl Projection for CatalogProjection {
  // ...

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), ViewError> {
    // ...

    match &event.data {
      // ...
      BookEvent::Inspected(_) => {
        self.catalog.update(book_id, &event, |_| {})?;
      }
    }

    Ok(())
  }

  // ...
}

To have the projection see the same events as the state, return the same upcasters from its upcasters method as the state does (see Versioning Events):

impl Projection for CatalogProjection {
  // ...

  fn upcasters() -> Option<&'static Upcasters> {
    Some(&LIBRARY_UPCASTERS)
  }
}

Handling Every Event#

A projection that has to see every event, for example to log it, names eventsourcingdb::Event, the event of the client SDK, as its Event. Its apply method then receives every event as it is stored, with its hash, without running any upcasters:

struct LogProjection;

impl Projection for LogProjection {
  type Event = eventsourcingdb::Event;
  type Error = Infallible;

  async fn apply(&self, event: Envelope<eventsourcingdb::Event>) -> Result<(), Infallible> {
    println!("{} {}", event.subject, event.event_type);

    Ok(())
  }
}

For a projection that needs no type of its own, use the projection_fn function, which turns a closure into a projection. The closure gets every event as an Envelope, typed by an enum or as it is stored, and returns a future of the result:

let log_projection = projection_fn(|event: Envelope<eventsourcingdb::Event>| async move {
  println!("{} {}", event.subject, event.event_type);

  Ok::<(), Infallible>(())
});

Note that a closure brings along neither upcasters nor a seen method, so a view that it fills reaches no revision by itself, and waiting for one runs into the timeout. A projection that needs upcasters or a reliable revision has a type of its own.

Running Projections#

To run a projection, call the start_projection method of the store with the subjects to read, the projection in an Arc, a CancellationToken, and the options. Say which subjects with Subjects::tree, which stands for the given subject together with every subject below it, or with Subjects::exact, which stands for the given subject alone. A projection usually reads a tree, such as every book below /books. There is no default, so that every projection says which one it means, since the client SDK reads a single subject unless told otherwise.

The method runs the projection on a task of its own and returns a ProjectionRun at once, which can be cloned, every clone watching the same run. The run first applies all events that are already stored, then follows new events until the token is canceled. An application usually answers queries only once its views have caught up, since a half-built view answers wrongly rather than slowly, so wait for that with the wait_caught_up function, which takes the runs:

let token = CancellationToken::new();

let run = store.start_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::clone(&catalog_projection),
  token.clone(),
  ProjectionOptions::named("catalog"),
);

wait_caught_up(&[&run]).await?;

// Start to answer queries.

ProjectionOptions::named gives the projection a name, by which its status, its failures, the observer of reconnects, and the health checks report it (see Checking Health over HTTP). ProjectionOptions::default() leaves it without a name, and the subject it reads tells it apart instead. The name method of the run returns it.

Note that an empty name makes start_projection panic, and so does Subject::new for a subject that is not one, such as one that does not start with a slash. start_projection needs a Tokio runtime with the time driver, as the client SDK does.

wait_caught_up waits until every run has applied the events that were stored when it started, and returns Ok(()) then, and only then, so that Ok(()) always means that the views are complete. To wait for several projections, hand over all of their runs, as in wait_caught_up(&[&catalog_run, &readers_run]). It waits for all of them at once, not one after the other, and fails with a CaughtUpError as soon as one of them ends before it has caught up:

• If a run ends with a failure before it catches up, it fails with CaughtUpError::Failed, which holds the failure of the run. Its text names the run by its name, as in projection "catalog": …, or, for a run without a name, by the subject it reads, as in projection of "/books": …, and it keeps the category of the failure.
• If a run ends without a failure before it catches up, because its token was canceled, it fails with CaughtUpError::Stopped, which belongs to no category, and whose text names the run the same way, as in the projection "catalog" stopped before it caught up or the projection of "/books" stopped before it caught up.

A run that has caught up counts as caught up, even if it has ended since. If several runs have ended before they caught up by the time wait_caught_up looks, the failure comes first, since it tells why, that of the first such run among the ones handed over, and a run that stopped comes next. Dropping the future stops waiting, so to wait for a limited time only, wrap it in tokio::time::timeout.

Note that wait_caught_up returns Ok(()) at once without any run, since none of them is behind.

For finer control, the run offers what wait_caught_up waits for. caught_up waits until the run has applied the events that were stored when it started, and counts once only, also while the run reconnects later on. If the run ends before it catches up, caught_up never returns, so wait for done as well. done waits until the run has ended, which happens when the token is canceled, or on a failure that trying again will not fix. error returns the failure the run ended with, as a ProjectionError in an Arc. It returns None as long as the run has not ended, and if it ended because its token was canceled, since canceling the token is how a projection is stopped. If apply fails with an error that trying again will not fix, the run ends, and error returns it.

The failure names the event that apply failed at by its ID, as ProjectionError::Projection, whose text reads applying event 42: …, so that the log tells which event to look at. It keeps the error of the projection, which a caller can downcast to its type, together with its category. A failure outside of apply, such as one of checkpoint, or of beginning or committing a transaction, names no event (see Resuming Projections).

If the projection panics, for example because apply unwraps a value that is not there, the run ends as well, rather than the task. error then returns ProjectionError::Panicked, which belongs to Category::Permanent, since a panic is a mistake in the code that trying again will not fix. It holds the value of the panic, if it is text, and for a panic in apply, it names the event as well, as in permanent failure: applying event 42: the projection panicked: …, while the panic hook of Rust reports where the panic happened, as for any other panic. This holds for every method of the projection that the run calls, and for the observer of reconnects (see below). Once the run has ended, Health::liveness answers 503, so that the orchestrator restarts the application (see Checking Health over HTTP).

Note that only canceling the token stops the projection. Dropping the run, or the token, does not, so keep a clone of the token for as long as the application runs, and cancel it when it shuts down (see Putting It Together). Canceling it stops waiting for the next event at once, while an event that is being applied is applied to the end.

To wait until the run ends, as a process does that runs nothing else, wait for done:

run.done().await;

match run.error() {
  Some(error) => Err(error.into()),
  None => Ok(()),
}

Note that if the database can not be reached at the start, the run keeps trying, and caught_up keeps waiting. To wait for a limited time only, wrap wait_caught_up in tokio::time::timeout. Its end stops the waiting, but not the run (see Putting It Together).

If reading fails with a transient failure, the run waits and continues after the last event it has applied, until the token is canceled. That includes a stream that ends, for example because the database restarts, or because a load balancer ends long-lived connections: the database never ends a stream on its own, so the run takes it for a stream that broke off. It also includes a failure of the projection whose error belongs to Category::Transient. The delay starts at one second, doubles with every attempt in a row, and never exceeds one minute. It starts over once the projection has applied an event again, or has caught up and followed the stream for longer than the initial delay, even if no event arrived. That way, a load balancer that ends long-lived connections regularly does not hold back a quiet projection, while a database that fails before the projection has caught up, or within the initial delay after, is given ever more time.

Note that how long the projection followed the stream is compared with the initial delay, not with the delay it has grown to. Otherwise, once an outage had let the delay grow to one minute, a load balancer that ends connections every 30 seconds would keep a quiet projection waiting for a minute after each of them, and a caller who wants to read their own write in the meantime would not see it (see Reading Your Own Writes). In return, a database that ends every stream a few seconds after the projection has caught up is tried again every few seconds rather than once a minute. That is not a loop without a pause, and catching up, the expensive part, succeeds each time.

A connection can also stall without being closed, for example behind a proxy that keeps it open but no longer passes anything on. While there is nothing else to send, the database sends a heartbeat every second, so the client ends a stream on which neither an event nor a heartbeat has arrived for 30 seconds, its stall timeout, and the run reconnects as after any other transient failure. Without that, the run would wait for the next event forever, while its status kept reporting Phase::Live.

To use other delays, or to learn about every attempt, for example to log it, call the reconnect_delays and reconnect_observer methods of the builder when creating the store:

let store = Store::builder(client, "https://library.eventsourcingdb.io")
  .reconnect_delays(Duration::from_millis(500), Duration::from_secs(30))
  .reconnect_observer(|reconnect| {
    println!(
      "observing again {:?} {} {} {:?}",
      reconnect.projection, reconnect.attempt, reconnect.error, reconnect.delay
    );
  })
  .build();

Note that an initial delay of zero panics, since the projections would then read again without any pause, and so does a maximum delay below the initial one.

The observer receives a Reconnect with these fields:

• projection is the name the projection was given with ProjectionOptions::named, or None if it has none.
• subject is the subject the projection reads.
• error is the reason, such as a stream that broke off, as a ProjectionError in an Arc.
• delay is how long the projection waits before the next attempt.
• attempt counts the attempts in a row, starting at one. It starts over together with the delay once the projection has applied an event, or has followed the stream for longer than the initial delay. So it stays at 1 while a load balancer ends long-lived connections regularly.

Note that a database that can not be reached is retried as well, since that is usually transient. The observer is how to notice a database that stays unreachable. A failure that trying again will not fix, for example a rejected API token, ends the run (see Handling Errors).

To find out where a run stands, call the status method. It returns a ProjectionStatus with these fields:

• phase is Phase::CatchingUp, Phase::Live, Phase::Reconnecting, or Phase::Stopped.
• since is when the phase began. For Phase::Reconnecting, that is when the disruption began, not when the latest attempt did.
• error is why the run is reconnecting or has stopped, such as a stream that broke off. It is None if the run stopped because its token was canceled.
• attempts counts the attempts to read again within the current disruption. Unlike attempt of Reconnect, it starts over whenever the run has caught up, even if the stream ends again within the initial delay.
• revision is the ID of the last event the run has applied and committed, or None if there is none.
• has_caught_up tells whether the run has caught up at least once. Like caught_up, it stays true while the run reconnects later on.

let status = run.status();

let is_stale = SystemTime::from(status.since)
  .elapsed()
  .is_ok_and(|elapsed| elapsed > Duration::from_secs(5 * 60));
if status.phase == Phase::Reconnecting && is_stale {
  // The view has not been up to date for more than five minutes.
}

To only apply the events that are already stored, for example for a batch job or in a test, call the catch_up_projection method instead. It takes the subjects and a reference to the projection, and returns once all stored events have been applied. Dropping its future stops it, so that a read model that is only partly built never looks complete. If the projection panics, it fails with the same ProjectionError::Panicked a run ends with, rather than panicking:

store
  .catch_up_projection(&Subjects::tree(Subject::new("/books")), catalog_projection.as_ref())
  .await?;

For a resumable or a transactional projection, call the methods for its kind instead (see Resuming Projections).

Resuming Projections#

By default, a projection starts from the first event every time it runs, which fits a view held in memory. For a view that keeps its data, the projection can resume where it stopped instead.

If the view can store a checkpoint, but not together with the data, additionally implement the ResumableProjection trait. checkpoint returns the ID of the last event saved, or None if there is none, and save_checkpoint saves it:

struct BookTableProjection {
  // ...
}

impl Projection for BookTableProjection {
  type Event = BookEvent;
  type Error = BookTableError;

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), BookTableError> {
    todo!()
  }
}

impl ResumableProjection for BookTableProjection {
  async fn checkpoint(&self) -> Result<Option<Revision>, BookTableError> {
    todo!()
  }

  async fn save_checkpoint(&self, event_id: Revision) -> Result<(), BookTableError> {
    todo!()
  }
}

BookTableError is the error of the view in a database (see Indexing Items).

Note that the checkpoint is saved after the events have been applied. After a crash, events may therefore be applied a second time, so apply must be idempotent.

Which way a projection is run is decided by the method that starts it, not by the traits it implements. start_projection and catch_up_projection run every projection from the first event, a resumable one as well, without reading or saving its checkpoint. To resume a projection, call start_resumable_projection or catch_up_resumable_projection instead, which take the same arguments, and which the compiler accepts only for a projection that implements ResumableProjection:

let run = store.start_resumable_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(BookTableProjection {}),
  token.clone(),
  ProjectionOptions::named("book-table"),
);

If the view can store the data and the checkpoint together, as a relational database can, implement the TransactionalProjection trait instead of Projection. A transactional projection applies events only within a transaction, so it has no apply method of its own. begin starts a transaction and returns a Transaction, which applies the events, and commits them together with the ID of the last event, or rolls them back:

struct TransactionalBookTableProjection {
  // ...
}

impl TransactionalProjection for TransactionalBookTableProjection {
  type Event = BookEvent;
  type Error = BookTableError;
  type Transaction = BookTableTransaction;

  async fn checkpoint(&self) -> Result<Option<Revision>, BookTableError> {
    todo!()
  }

  async fn begin(&self) -> Result<BookTableTransaction, BookTableError> {
    todo!()
  }
}

struct BookTableTransaction {
  // ...
}

impl Transaction for BookTableTransaction {
  type Event = BookEvent;
  type Error = BookTableError;

  async fn apply(&mut self, event: Envelope<BookEvent>) -> Result<(), BookTableError> {
    todo!()
  }

  async fn commit(self, last_event_id: Revision) -> Result<(), BookTableError> {
    todo!()
  }

  async fn rollback(self) -> Result<(), BookTableError> {
    todo!()
  }
}

With sqlx, begin starts a transaction of the database, which the Transaction holds, and apply writes the events into it, decoded into the enum of the projection, as apply of a Projection gets them:

struct TransactionalBookTableProjection {
  pool: PgPool,
}

impl TransactionalProjection for TransactionalBookTableProjection {
  // ...

  async fn begin(&self) -> Result<BookTableTransaction, BookTableError> {
    let transaction = self.pool.begin().await.map_err(BookTableError)?;

    Ok(BookTableTransaction { transaction })
  }
}

struct BookTableTransaction {
  transaction: sqlx::Transaction<'static, Postgres>,
}

impl Transaction for BookTableTransaction {
  type Event = BookEvent;
  type Error = BookTableError;

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

    if let BookEvent::Acquired(acquired) = &event.data {
      sqlx::query("INSERT INTO books (id, title, author) VALUES ($1, $2, $3)")
        .bind(book_id)
        .bind(&acquired.title)
        .bind(&acquired.author)
        .execute(&mut *self.transaction)
        .await
        .map_err(BookTableError)?;
    }

    Ok(())
  }

  async fn commit(mut self, last_event_id: Revision) -> Result<(), BookTableError> {
    let saved = sqlx::query("UPDATE checkpoint SET event_id = $1")
      .bind(last_event_id.to_string())
      .execute(&mut *self.transaction)
      .await;
    if let Err(error) = saved {
      let _ = self.transaction.rollback().await;
      return Err(BookTableError(error));
    }

    self.transaction.commit().await.map_err(BookTableError)
  }

  async fn rollback(self) -> Result<(), BookTableError> {
    self.transaction.rollback().await.map_err(BookTableError)
  }
}

Note that rollback is not called after commit fails, so commit has to roll back itself if it can not finish, as above.

To run a transactional projection, call the start_transactional_projection or the catch_up_transactional_projection method. They take the same arguments as those for the other kinds:

let run = store.start_transactional_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(TransactionalBookTableProjection { pool }),
  token.clone(),
  ProjectionOptions::named("book-table"),
);

Note that a transactional projection has no seen method, since where data and revision have to become durable together, the revision belongs inside the transaction, and the projection writes it itself (see Tracking Revisions).

The place where a view keeps its data can fail as well, and such a failure counts like an error of apply, whether it comes from checkpoint, save_checkpoint, begin, commit, or the apply method of a Transaction. An error that does not belong to Category::Transient ends the run, so Health::liveness answers 503, and the orchestrator restarts the application (see Checking Health over HTTP). That fits a mistake, such as a statement that the database refuses, but not a failure that may pass, such as a lost connection or a deadlock. Report such a failure with an error whose category is Category::Transient, as BookTableError does for a lost connection. The run then waits and tries again from where it stopped, with the same growing delay as when reading from EventSourcingDB fails (see Running Projections). A panic in any of these methods, on the other hand, always ends the run, since it is a mistake in the code. A panic in the apply method of a Transaction rolls the transaction back first, as an error does, and if rolling back fails as well, the run reports that failure as ProjectionError::RollbackFailed, alongside the one that made it roll back, never instead of it.

Note that which failures may pass depends on the database and its driver, so mark only those. A failure that is marked as transient but never passes keeps the run trying forever, while Health::liveness keeps answering 200.

Batching Events#

By default, the checkpoint is saved, or the transaction is committed, after every event. While a projection catches up, for example when it starts for the first time, that takes a lot longer than needed. To apply several events at once while catching up, implement the catch_up_batch_size method of ResumableProjection or TransactionalProjection, and return how many:

impl ResumableProjection for BookTableProjection {
  // ...

  fn catch_up_batch_size(&self) -> usize {
    1000
  }
}

The default is 1, and 0 counts as 1. For a transactional projection, a batch is one transaction, so a failure rolls back all of its events, and the next attempt starts after the last commit.

Once the projection has caught up, it saves the checkpoint, or commits, after every event again. Events then arrive one at a time, and a batch that waited to fill up would hold back the ones that have already arrived.

Note that a resumable projection may apply up to that many events a second time after a crash.