NLnetLabs-krill/doc/development/04_es_krill.md

Event Sourcing in Krill

If you search for it you will find that there are many blog posts about how people made Event Sourcing and CQRS work for their project. There are also many libraries, for many languages, that provide frameworks. Some of these libraries will have you believe that their implementation is the right way of doing this... but we believe that there is more than one way to happiness. So, in Krill we have chosen to use our own local implementation. This lets us use the concepts that we need, but it allows us the flexibility to adapt it to our specific needs, and avoids the complication from concepts that we do not need.

Krill uses its own event sourcing library code which can be found under src/commons/eventsourcing. We will describe the concepts and code here, and include code (snippets). But, obviously, the real code is leading - so be sure to check it and fix this documentation if the two should disagree!

KeyValueStore and JSON

Before we delve in to the Krill eventsourcing code, we should talk a bit about storage. Krill stores all values in a KeyValueStore.

This is implemented in the kvx library: https://github.com/nlnetlabs/kvx

There is a PR to port the kvx implementation back into the core Krill code (Krill is the only user after all), and in the process make it support async. To support this, the code was updated to rely on an enum rather than a trait for KeyValueStore. This PR can be found here: https://github.com/NLnetLabs/krill/pull/1152

Currently, only disk and memory (for testing) implementations are supported. Database options may be added in future, but note that that will require async support. The kvx library claims to support postgresql but it can't be used in Krill because while the library is sync, it uses a runtime under the hood for the database connection and this conflicts with Krill because it already uses hyper and tokio.

Opt-in locking on disk relies on fd-lock in kvx and plain rustix in the PR (only supports UNIX). This locking is used by Krill to ensure that updates to CAs, the Publication Server Access (which Publishers have access) and Publication Server Content are always applied sequentially.

In principle, since the locking leverages flock which is supposedly NFS safe, this should mean that as of 0.14.4 it is safe to run multiple active Krill instances using the same shared NFS data directory. But.. this needs proper testing!

Aggregate

As said, the aggregate is a central concept in DDD that is used in combination with event sourcing and CQRS. Officially the root entity should be called 'aggregate root', but because this a bit wordy Krill just calls it 'Aggregate' instead. There is trait defined for this that is implemented by CertAuth and RepositoryAccess (more on these later).

See src/commons/eventsourcing/agg.rs for the following code:

/// This trait defines an Aggregate for use with the event sourcing framework.
///
/// An aggregate is term coming from DDD (Domain Driven Design) and is used to
/// describe an abstraction where a cluster of structs (the aggregate) provides
/// a 'bounded context' for functionality that is exposed only by a single top-level
/// struct: the aggregate root. Here we name this aggregate root simply 'Aggregate'
/// for brevity.
///
/// The aggregate root is responsible for guarding its own consistency. In the
/// context of the event sourcing framework this means that it can be sent a command,
/// through the [`process_command`] method. A command represents an intent to
/// achieve something sent by the used of the aggregate. The Aggregate will then take
/// this intent and decide whether it can be executed. If successful a number of
/// 'events' are returned that contain state changes to the aggregate. These events
/// still need to be applied to become persisted.
pub trait Aggregate: Storable + Send + Sync + 'static {
    type Command: Command<StorableDetails = Self::StorableCommandDetails>;
    type StorableCommandDetails: WithStorableDetails;
    type Event: Event;
    type InitEvent: Event;
    type Error: std::error::Error + Send + Sync;

    /// Creates a new instance. Expects an event with data needed to
    /// initialize the instance. Typically this means that a specific
    /// 'create' event is passed, with all the needed data, or just an empty
    /// marker if no data is needed. Implementations must return an error in
    /// case the instance cannot be created.
    fn init(event: Self::InitEvent) -> Result<Self, Self::Error>;

    /// Returns the current version of the aggregate.
    fn version(&self) -> u64;

    /// Applies the event to this. This MUST not result in any errors, and
    /// this MUST be side-effect free. Applying the event just updates the
    /// internal data of the aggregate.
    ///
    /// Note the event is moved. This is done because we want to avoid
    /// doing additional allocations where we can.
    fn apply(&mut self, event: Self::Event);

    /// Applies all events. Assumes that:
    /// - the list is contiguous (nothing missing) and ordered from old to new
    /// - the events are all applicable to this aggregate
    /// - the version of the aggregate matches that of the first (oldest) event
    fn apply_all(&mut self, events: Vec<Self::Event>) {
        for event in events {
            self.apply(event);
        }
    }

    /// Processes a command. I.e. validate the command, and return a list of
    /// events that will result in the desired new state, but do not apply
    /// these events here.
    ///
    /// The command is moved, because we want to enable moving its data
    /// without reallocating.
    fn process_command(&self, command: Self::Command) -> Result<Vec<Self::Event>, Self::Error>;
}

As we will see further down, when we get to describe the AggregateStore<A: Aggregate> which ties of all this together, this allows us to focus on business logic in our aggregates.

Just make sure that you define your Command and Event type, and implement the process_command method. The result should be either an error, or Vec<Event>. In case the command is a no-op, return an empty Vec. Furthermore implement apply and let your aggregate state be updated with event data.

Commands and StorableDetails

As can be seen in the trait above, Krill expects that an Aggregate defines a type that it will use for commands. The command MUST implement the Command trait which is defined in src/commons/eventsourcing/cmd.rs:

//// Commands are used to send an intent to change an aggregate.
///
/// Think of this as the data container for your update API, plus some
/// meta-data to ensure that the command is sent to the right instance of an
/// Aggregate, and that concurrency issues are handled.
pub trait Command: fmt::Display + Send + Sync {
    /// Identify the type of storable component for this command. Commands
    /// may contain short-lived things (e.g. an Arc<Signer>) or even secrets
    /// which should not be persisted.
    type StorableDetails: WithStorableDetails;

    /// Identifies the aggregate, useful when storing and retrieving the event.
    fn handle(&self) -> &Handle;

    /// The version of the aggregate that this command updates. If this
    /// command should update whatever the latest version happens to be, then
    /// use None here.
    fn version(&self) -> Option<u64>;

    /// The actor who sent the command. There is no default so as to avoid
    /// accidentally attributing a command by a user instead as if it were an
    /// internal command by Krill itself.
    fn actor(&self) -> &str;

    /// Get the storable information for this command
    fn store(&self) -> Self::StorableDetails;
}

We define the WithStorableDetails trait in the same file. As mentioned in the code comments above, we use this when we save commands to the audit log, because the actual command may contain a secret - or more likely an ephemeral thing that cannot and should not be saved - such as a KrillSigner which an Aggregate can use as part of the executing the command.

/// Must be implemented for all 'StorableDetails' used in Commands.
///
/// In addition to implementing Storable so that the details can be stored
/// *and* retrieved, the details also need to be able to present a generic
/// CommandSummary for use in history.
pub trait WithStorableDetails: Storable + Send + Sync {
    fn summary(&self) -> CommandSummary;
}

Skipping some details now (it's in the code), this is ultimately used when we store a command using the following struct:

/// A description of a command that was processed, and the events / or error
/// that followed. Commands that turn out to be no-ops (no events, no errors)
/// should not be stored.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StoredCommand<S: WithStorableDetails> {
    actor: String,
    time: Time,
    handle: Handle,
    version: u64,  // version of aggregate this was applied to (successful or not)
    sequence: u64, // command sequence (i.e. also incremented for failed commands)
    #[serde(deserialize_with = "S::deserialize")]
    details: S,
    effect: StoredEffect,
}

Finally StoredEffect here tells us whether the command was successful, or not:

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case", tag = "result")]
pub enum StoredEffect {
    Error { msg: String },
    Success { events: Vec<u64> },
}

Events

As mentioned above an Aggregate must specify which type of event it uses. Events MUST implement the Event trait (src/commons/eventsourcing/evt.rs):

pub trait Event: fmt::Display + Eq + PartialEq + Send + Sync + Storable + 'static {
    /// Identifies the aggregate, useful when storing and retrieving the event.
    fn handle(&self) -> &Handle;

    /// The version of the aggregate that this event updates. An aggregate that
    /// is currently at version x, will get version x + 1, when the event for
    /// version x is applied.
    fn version(&self) -> u64;
}

Events are stored using the StoredEvent struct:

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StoredEvent<E: fmt::Display + Eq + PartialEq + Storable + 'static> {
    id: Handle,
    version: u64,
    #[serde(deserialize_with = "E::deserialize")]
    details: E,
}

So, note that StoredEvent is not generic over Event. But, in practice the defined Event for our two main aggregates are of a type of StoredEvent:

pub type CaEvt = StoredEvent<CaEvtDet>;

IMPORTANT: We can change the internal implementation of an Aggregate pretty freely. As long as we keep backwards compatibility with regards to past events, we can just delete the snapshots and rebuild the current state based on past events. Well, if you look closely at the migration code in `src/upgrades/v0_9_0' you will find that it is of course possible to add code to parse all past commands and/or events, and reformat them. But, it's a major amount of work to do so, so it's better to avoid this if we can. Also note, that we can never add new information to past events. They already happened, so we have to deal with that.

Hybrid Model

So, time for a confession... and a digression that will make sense when we talk about event listeners, and things triggered by events.

Before Krill version 0.9 ALL CA and Publication Server functions were implemented in aggregates and used event sourcing. However, as you may know CAs in the RPKI have to re-publish CRLs and Manifests very, very often. Even when there are no changes to ROAs these objects must be refreshed before they go stale. By default Krill re-publishes every 16 hours.

This resulted in an enormous amount of commands and events to be generated for CAs, for a change that is done automatically - and which does not reflect any semantic change for a CA. Keeping all this stuff in the history led to excessive disk usage, and an audit trail that is hard to follow. It also has a serious impact on rebuilding state based on events from scratch i.e. without using snapshots.

Therefore we decided to implement a hybrid model in Krill. The Krill CertAuth is still in charge of almost all changes, and in particular all semantic changes that users made. But, the generation of Manifests and CRLs is offloaded to associated component CaObjects that just keeps the latest Manifest and CRL. It can re-sign these because it has access to a KrillSigner and it can get the public key identifier needed for signing from the CertAuth.

This is relevant here, because under the hood we use a PreSaveEventListener to ensure that a new Manifest and CRL are written when there is a change in ROAs or issued certificates, observed in events.

Event Listeners

The Krill eventsourcing stack defines two different EventListener traits which are called by the AggregateStore (see below) when an Aggregate is successfully updated. The first is called before updates are saved, and it can fail, the second is called after all changes have been applied, and cannot fail.

/// This trait defines a listener for events which is designed to receive
/// the events *before* the Aggregate is saved. Thus, they are allowed
/// to return an error in case of issues, which will then roll back the
/// intended change to an aggregate.
pub trait PreSaveEventListener<A: Aggregate>: Send + Sync + 'static {
    fn listen(&self, agg: &A, event: &[A::Event]) -> Result<(), A::Error>;
}

/// This trait defines a listener for events which is designed to receive
/// them *after* the updated Aggregate is saved. Because the updates already
/// happened EventListeners of this type are not allowed to fail.
pub trait PostSaveEventListener<A: Aggregate>: Send + Sync + 'static {
    fn listen(&self, agg: &A, event: &A::Event);
}

In a nutshell, we use the event listeners for two things:

• Trigger that the CaObjects for a CA gets an updated Manifest and CRL (pre-save).
• Trigger that follow-up tasks are put on the Scheduler, based on events.

As discussed in issue: https://github.com/NLnetLabs/krill/issues/1182 it would be best to remove the PreSaveEventListener trait and do everything through (idempotent) triggered tasks on the queue in the Scheduler. Note that Krill will add any missing tasks on this queue at startup, so this means that even if an aggregate, like a CA, is saved and then the follow-up task scheduling fails because of an outage, then the task will simply be re-added when Krill restarts.

AggregateStore

When Krill interacts with an Aggregate, it does so through an AggregateStore<A: Aggregate>. The full code for this type can be found in src/commons/eventsourcing/store.rs. Here we will stick to the highlights of how this type works!

Essentially, the AggregateStore provides a convenient access layer to dealing with any Aggregate. Using this let's you focus on pure business logic in your aggregates. Essentially we just need to implement the Aggregate, Command and Event traits and then the framework will deal with storage concerns.

The most important pub function signatures (implementation omitted):

impl<A: Aggregate> AggregateStore<A>
where
    A::Error: From<AggregateStoreError>,

{
    /// Creates an AggregateStore using a disk based KeyValueStore 
    pub fn disk(work_dir: &PathBuf, name_space: &str) -> StoreResult<Self> { ... }

    /// Warms up the cache, to be used after startup. Will fail if any aggregates fail to load.
    pub fn warm(&self) -> StoreResult<()> { ... }

    /// Adds a listener that will receive all events before they are stored.
    pub fn add_pre_save_listener<L: PreSaveEventListener<A>>(&mut self, sync_listener: Arc<L>) { ... }

    /// Adds a listener that will receive a reference to all events after they are stored.
    pub fn add_post_save_listener<L: PostSaveEventListener<A>>(&mut self, listener: Arc<L>) { ... }

    /// Adds a new aggregate instance based on the init event.
    pub fn add(&self, init: A::InitEvent) -> StoreResult<Arc<A>> { ... }

    /// Send a command to the latest aggregate referenced by the handle in the command.
    ///
    /// This will:
    /// - Wait for a lock for the latest aggregate for this command.
    /// - Call the A::process_command function
    /// on success:
    ///   - call pre-save listeners with events
    ///   - save command and events
    ///   - call post-save listeners with events
    ///   - return aggregate
    /// on no-op (empty event list):
    ///   - do not save anything, return aggregate
    /// on error:
    ///   - save command and error, return error
    pub fn command(&self, cmd: A::Command) -> Result<Arc<A>, A::Error> { ... }

    /// Returns true if an instance exists for the id
    pub fn has(&self, id: &Handle) -> Result<bool, AggregateStoreError> { ... }

    // Lists all known ids.
    pub fn list(&self) -> Result<Vec<Handle>, AggregateStoreError> { ... }
}

As said, users of AggregateStore do not need to worry about the actual storage, as this is done inside of it using private functions. Still, it's good to talk a bit more about how this works. And looking at how the struct if built up will help to explain its inner workings:

pub struct AggregateStore<A: Aggregate> {
    kv: KeyValueStore,
    cache: RwLock<HashMap<Handle, Arc<A>>>,
    history_cache: Option<Mutex<HashMap<MyHandle, Vec<CommandHistoryRecord>>>>,
    pre_save_listeners: Vec<Arc<dyn PreSaveEventListener<A>>>,
    post_save_listeners: Vec<Arc<dyn PostSaveEventListener<A>>>,
}

• Applying changes

ALL changes to Aggregates, like CertAuth are done through the command function. This function waits for a lock (using flock mentioned earlier) to ensure that all changes to the Aggregate are applied sequentially.

First the Aggregate is retrieved from the in memory cache if present. If there is no cached instance then latest snapshot is retrieved from storage instead. If there is also no snapshot, then the INIT command (i.e. with version 0) is retrieved and applied instead. Then the key value store is queried for follow-up StoredCommand values (for the version of the Aggregate) which are then applied. Note that this would mean, in a possible cluster set up that guarantees locking, that even if a cluster node is behind the other, it will simply find the missing updates.

Once the latest Aggregate has been retrieved, the Command is sent to it. Note that this Command is not a StoredCommand. Command contains the intent for a change, while StoredCommand contains the result of such a change. The Aggregate is responsible for verifying the command and it can return with: an error, no effect, or a vec of change events.

If a command has no effect, i.e. there is no error and no change, then it is simply forgotten.

If a command has an effect then a StoredCommand is created that contains a storable representation of the command (e.g. certain values in a command, like an Arc<Signer>, or Arc<Config> are not included) and either the error or a Vec of Events. This StoredCommand gets a unique key name based on the version of the Aggregate that it affects.

If there should be an existing key-value pair for this StoredCommand, then this indicates that the locking mechanism failed somehow. This should not happen, but if it did, then the command is NOT saved. Instead Krill exits with an error message.

But if all is well (as expected), then the command with events is applied to the aggregate. Note that this just updates its version in case of a command that resulted in an error. The in-memory cached aggregate is updated to help performance when retrieving it later. Then the command is saved.

• Retrieving

When retrieving an Aggregate for read access Krill actually follows the same code path that is used for applying a Command, except that in this case the underlying function that takes care of locking and retrieving the latest Aggregate is called with None instead of Some(command), so it simply returns without trying to apply any changes. This ensures however, that the same locking rules are observed in both cases and allows us to avoid code duplication (thus increasing the loci for bugs).

• Snapshots

Note that we do not update the full snapshot on every change because this could create a performance issue in cases where aggregates are big (e.g. a CA with many children or objects) and serialization takes a long time. Instead, updating the snapshots is done daily through a different code path from the Scheduler. See Task::UpdateSnapshots. This code, again, actually uses the same underlying function as above to retrieve the snapshot, this time setting the save_snapshot parameter to true to ensure that the snapshot is saved.