Modeling
SynthEvents
Remembering
EventSourcingDB
Building
ArchitectureKit
Language
GoRust
Docs / ArchitectureKit / Rust / Fundamentals / Reading Your Own Writes

Reading Your Own Writes

A view lags behind the events that have been written, by however long its projection takes. To read your own writes, wait until the view has seen the events you have written.

The ID of the last event a view has seen is its revision. Since the database assigns event IDs in ascending order across all subjects, revisions can be compared.

A view sees only the events of the subjects its projection reads. So wait only for a write to one of those subjects: if the write went to another subject, the view does not reach its revision until a later event lands in one of its own subjects, and waiting for it lasts the full time, with wait_for, with Reading, and with the Wait-For-Revision header alike. wait_for_written even fails then, with NotCaughtUp::TimedOut, so hand it only the events written to the subjects the view reads.

Tracking Revisions#

To track the revision of a view, the projection records every event that reaches it in the view, including the ones it skips. The run calls the seen method of the projection with the ID of every stored event, once the projection has applied it, and the projection hands it on to the seen method of the view:

impl Projection for CatalogProjection {
  // ...

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

If the projection writes to several views, such as a LibraryProjection that keeps both the catalog and the readers up to date (see Sharing Items with Readers), record the event in all of them, so that each one knows how far it has come:

impl Projection for LibraryProjection {
  // ...

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

The run calls seen for every stored event that reaches the projection, also for one of a type the projection has no variant for, and that is the point: a projection skips what does not concern it, but a reader may be waiting for exactly that event. For an event that an upcaster splits, it calls seen once the projection has applied every part. It does not call it for an event that failed, and it calls it before the checkpoint of a batch is saved. Events may arrive more than once after a restart, so seen of InMemoryView does not move the revision backwards for an event the view has already passed.

seen does nothing unless the projection implements it, so a projection that leaves it out, and one made with projection_fn, which has none, never moves the revision of its view, and waiting for a revision runs into the timeout. The method is meant for a view that keeps no transaction, since the run calls it once the projection has applied the event, apart from its data. Where data and revision have to become durable together, as in a view in a database, the revision belongs inside the transaction, so a TransactionalProjection has no seen method, and the projection writes the revision itself.

To get the revision your own write has produced, call the revision_of function with the written events. It returns the highest event ID, as a Revision, or None if no events were written:

let revision = revision_of(&written);

Waiting for Revisions#

To wait until a view has reached a revision, call the wait_for method of the Revisioned trait with the revision. It returns once the view has reached the revision, at once if it has already. It waits for as long as it takes, so limit it with tokio::time::timeout, which fails once the time has run out:

if let Some(revision) = revision {
  match tokio::time::timeout(Duration::from_secs(5), catalog.wait_for(revision)).await {
    Ok(Ok(())) => {
      // The view has reached the revision.
    }
    Err(_) => {
      // The view has not reached it within five seconds.
    }
  }
}

The view in memory can always be asked, so the error of wait_for is Infallible, and the match needs no arm for it. A view of your own, which keeps its revision in a database, may fail to be asked, and then wait_for returns its error.

Note that running out of time is a failure here. The awaiting method of Reading in the httpapi module, which waits for the revision an HTTP request asks for, takes it for none instead, and returns Ok(()), so that the handler answers with what the view holds (see Waiting for a Revision in a Handler of Your Own).

To get the current revision of a view, call the revision method. It returns None as long as the view has not seen any event, and an error if it can not read the revision, which may happen to a view of your own that keeps its revision in a database. InMemoryView keeps its revision in memory, so it never fails:

let Ok(current) = catalog.revision().await;

Both methods form the Revisioned trait, which InMemoryView implements. To wait for revisions of a view of your own, implement it as well.

Comparing Revisions#

A revision is a Revision, which sorts as a number, so that 10 comes after 9. Compare two of them with < and >, or with cmp, which returns an Ordering. A view that has not seen any event has no revision, None, which Option orders before every revision:

let ordering = Revision::from(9).cmp(&Revision::from(10));
// ordering == Ordering::Less

A revision is a decimal number from 0 to 2^63-1, the range of the event IDs of the database, so a value that is no event ID can not be a Revision at all. To read a revision that comes from outside, such as the one a caller hands over, call parse. It accepts digits only, and refuses everything else, an empty text as well, since that is the ID of no event, so it is the way to check a revision before it becomes part of a command (see Handling Commands over HTTP):

let revision: Revision = "42".parse()?;
// revision.get() == 42

let refused = "".parse::<Revision>();
// refused.is_err() == true

For a value that is not a revision, parse fails with NotARevision, which names the value, as in not a revision: "4 2". A number becomes a revision with try_from, which refuses one beyond 2^63-1 the same way, and a u32, which always fits, with from. serde writes and reads a revision as a string of its decimal, as the database writes an event ID.

Waiting for Written Events#

Sometimes a step on the server builds on what was just written, for example a handler that executes a command and then answers with what a view holds after it. To wait until a view has seen the events that execute or write returned, call the wait_for_written function with the view, the written events, and how long to wait at most:

let written = store.execute(&borrow_book(), command).await?;

match wait_for_written(&catalog, &written, Duration::from_secs(5)).await {
  Ok(()) => {
    // Read what the command has changed.
  }
  Err(NotCaughtUp::TimedOut(_)) => {
    // The command has succeeded, so answer with success all the same, and
    // with the revision, for which the caller can wait with
    // Wait-For-Revision.
  }
}

It waits with wait_for for the revision that revision_of returns for the events, and returns Ok(()) once the view has reached it. If no events were written, it returns Ok(()) at once, since there is nothing to wait for. Otherwise it fails with NotCaughtUp, since the events were written in any case, which says why the view has not caught up:

  • If the time runs out first, it fails with NotCaughtUp::TimedOut, whose text says so, as in not caught up: the events were written, but the view did not catch up within 5s.
  • If the view fails otherwise, for example because a view in a database can not read its revision, it fails with NotCaughtUp::View, which holds the error of the view, with its category, and whose text reads not caught up: the events were written, but the view could not be asked. The view in memory never fails, so for it, the match needs no arm for NotCaughtUp::View.

Dropping the future stops waiting, as for every other future.

NotCaughtUp belongs to no category, since the write has succeeded, and only the view lags behind: trying the write again would store its events twice, and the view may still catch up (see Handling Errors). So a handler that waits after a command answers with success, as in the example above, and with the revision of the written events. If the caller needs to see what was written, it can wait for the revision with the Wait-For-Revision header of a query (see Reading Your Own Writes over HTTP). That is why NotCaughtUp does not turn into an HttpError with ?, which would answer a command that has succeeded as a failure: CommandAnswer::with_result answers it as the success it is, and logs that the view lags behind.

Note that the view sees only the events of the subjects its projection reads, so hand over only the events written to those subjects. For a write to several subjects, such as one with write, the highest ID of all events may lie in a subject the view does not read, and then the view does not reach it until a later event lands in one of its own subjects (see Waiting for Revisions).

Note that running out of time is a failure here, since the step needs what was written. The awaiting method of Reading, on the other hand, waits for the revision a caller asks for, and takes running out of time for none, so that the handler answers with what the view holds (see Waiting for a Revision in a Handler of Your Own).

Note that wait_for_written panics for a timeout of zero, since a view that catches up with a write that has just happened needs some time, and waiting no time at all would fail almost always.