[Proposal] Event Sourcing V2

[willg1983]

Motivation

There have been various discussions about improvements to Orleans event sourcing support (see footer). Orleans is naturally a good fit for event-sourced aggregates because grains provide a strong consistency boundary, single activation guarantees, in-memory state, sequential message processing and single-threaded execution that makes safe state mutation simpler.

Event sourcing is often combined with a CQRS architectural pattern, event sourced aggregates provide the 'Command' portion and projections provide the 'Query' capability through event sourced read-models.

The current Orleans event sourcing support is quite limited, lacks credible in-the-box storage providers and does not support event sourcing's super power: projections.

This proposal adds

• a new event sourcing provider interface which is simple to implement.
• An approach for grain event sourced state(s) that can be constructor injected into an event sourced grain and extension methods for writing events.
• A new component for grain-based projections and side effects, allowing grains to aggregate events in an eventually consistent way.

Event Storage Provider Interface

A new interface for event storage providers that supports Read/Write/Truncate. Reads and writes include new standard event properties such as EventId, EventNumber, GlobalLogPosition and a Metadata dictionary alongside the event data payload. Truncation supports requirements around compaction and removal of event streams. This replaces the need to implement ICustomStorageInterface or ILogViewAdaptor.

It loses the concept of the provider monitoring and notifying of changes directly, instead relying on optimistic concurrency to detect changes on write. This suits the use case of a grain having its own dedicated event stream, and effectively being the single writer (transient duplicate grain activation handled by op concurrency). If there are multiple writers then a projection that notifies the grain of externally written changes is a better approach in my view.

There is no support for writing events unconditionally (without optimistic concurrency check), support could be added via an additional method on the interface.

public interface IGrainEventStorageProvider
{
    /// <summary>
    /// Reads all events for the given grain starting from the given event number, in ascending order.
    /// The event number is used as the starting point for reads, but does not guarantee the first event
    /// will be contigous as previously truncated events may mean the first event is greater than the given event number.
    /// The storage provider should return events with event numbers greater than the given event number.
    /// </summary>
    /// <param name="grainId">grain identity</param>
    /// <param name="exclusiveFrom">the event number from which to start reading (exclusive)</param>
    /// <param name="cancellationToken">token to cancel the read</param>
    public IAsyncEnumerable<EventRead> ReadEvents(GrainId grainId, GrainEventNumber exclusiveFrom, CancellationToken cancellationToken);

    /// <summary>
    /// Write one or more events to the grain's event stream atomically with optimistic concurrency control based on <paramref name="expectedEventNumber"/>.
    /// The write will only succeed if the current event number in storage matches the given event number, which represents the expected state of the stream at the time of write.
    /// If the write is successful, the storage provider will return the global position of the last written event, which can be used for reading from the stream or for projections.
    /// If the event number doesn't match, the storage provider should throw an InconsistentStateException to indicate a concurrency conflict.
    /// </summary>
    /// <param name="grainId">grain identity</param>
    /// <param name="expectedEventNumber">the expected event number of the stream at the time of write</param>
    /// <param name="events">the events to write</param>
    /// <param name="cancellationToken">token to cancel the write</param>
    /// <returns>The written position of the last event</returns>
    /// <exception cref="InconsistentStateException">if the event number doesn't match the expected value in storage</exception>
    public ValueTask<GlobalEventLogPosition> WriteEvents(GrainId grainId, GrainEventNumber expectedEventNumber, ReadOnlyMemory<EventWrite> events, CancellationToken cancellationToken);

    /// <summary>
    /// Removes all events prior to the given event number for the specified grain.
    /// This can be used for log compaction or truncation.
    /// If the given event number is greater than the current highest event number in storage, the storage provider should throw an ArgumentOutOfRangeException. 
    /// </summary>
    /// <param name="grainId">grain identity</param>
    /// <param name="inclusiveUpTo">the event number up to which all events will be removed (inclusive)</param>
    /// <param name="cancellationToken">token to cancel the delete</param>
    /// <returns></returns>
    /// <exception cref="ArgumentOutOfRangeException">if the given event number is greater than the current highest event number in storage</exception>"
    public ValueTask DeleteEvents(GrainId grainId, GrainEventNumber inclusiveUpTo, CancellationToken cancellationToken);
}

public readonly record struct EventRead(string EventType, // for deserialization
                                        BinaryData Data,
                                        GrainEventNumber EventNumber,
                                        GlobalEventLogPosition Sequence,
                                      //  DateTimeOffset WrittenTimestamp, // Bad idea: really just an infrastructure concern. If date is relevent to the event it should be captured in the event data
                                        Guid EventId,
                                        IReadOnlyDictionary<string, object> Metadata
                                        // Could include correlationId / causationId, or leave that to metadata
                                        );

public readonly record struct EventWrite(IReadOnlyDictionary<string, object> Metadata,
                                         string EventType, // for deserialization
                                         BinaryData Data,
                                         Guid EventId);

To enable translation from TEventBase to EventWrite, which is used to persist events and EventRead which is used to read events, a new interface will be added:

public interface IEventConverter 
    public DecodedEvent<TEventBase> DecodeEvent<TEventBase>(EventRead read) 
    public EventWrite EncodeEvent<TEventBase>(TEventBase @event) 
}

public readonly record struct DecodedEvent<TEventBase>(TEventBase Event, GrainEventNumber EventNumber) where TEventBase : class;

A default implementation will use IGrainStorageSerializer for serialization, it will populate the metadata with activity tracing information (if available, which can then be used to correlate the activity in projections/side-effects) we may place an interface constraint on TEvent to implement a Guid EventId property.

Constructor injected Grain Event Sourcing State(s)

Event sourced state read-models will be injected into event sourced grain's constructor.

public sealed class MyEventSourcedGrain(IEventSourcedState<MyReadModel> eventSourcedState) : Grain,
    IMyEventSourcedAggrgateGrain
{

    // Example of command pattern 
    public async ValueTask<EventSourcedCommandResult<TResult>> ExecuteCommand<TResult>(IEventSourcedGrainCommand<EventBase, TResult> command, CancellationToken token)
    {
        var result = command.Execute(eventSourcedState, this.GetPendingEventsWriter());
        var (globalLogPosition, eventNumber) = await this.CommitPendingEvents(token);
        return new(result, globalLogPosition, eventNumber);
    }

      // Query method etc
}

Each IEventSourcedState injected into the grain will be synchronously updated as events are written in the grain.

The IEventSourcedState interface will implement IAsyncEnumerable to provide observability of state changes.

This approach also allows a grain to host multiple state models (which could be useful for separating concerns, e.g. one model stores 'decider state' for command validation, one stores an audit history for the UI, one provides status). A built-in event capturing state model will, if used by the grain author, be able to dynamically create any state model on-demand and keep it updated.

public sealed class MyEventSourcedGrain(
IEventSourcedState<MyReadModel> eventSourcedState, 
IEventSourcedState<MyAuditHistory> auditHistory, 
IEventSourcedState<MyOrderStatusReadModel> customerOrderStatus) : Grain,
    IMyEventSourcedAggrgateGrain

Writing pending events and atomically committing all pending events to storage and refreshing state from storage will be extension methods of IGrainBase

It is expected IEventSourcedState models will be updated when a pending event is written, before the async persistence operation, to ensure state consistency for re-entrant operations. This risks phantom reads by concurrent operations on event write failure instead of stale reads all the time. The IAsyncEnumerable state model update yielding will happen after the write to ensure phantom state does not escape non-rentrant grains.

The grainId will be used for the event sourcing stream name, we could provide a grain attribute to configure the formatting.

We may want a mechanism for event type names that decouple them from their .net type names, perhaps adding a new EventAlias attribute.

These new event sourcing components will opt-in to grain migration if all the injected event sourced state models are serializable allowing them and pending events to be live migrated with the grain.

Event Sourced Projections and side-effects

A new interface that can be used to read grain events, allowing grains to be created that receive all matching historical catch-up events and live events from other grains.

The expectation is this will be implemented using the mechanism in the underlying event persistence provider, this ensures projections receive events as written in global event log order.

This proposal lays the ground work upon which more sophisticated abstractions can be built such as aprojection grain, that can host any event sourced read model in-memory or write to another storage mechanism.

This interface notifies of checkpoint opportunities.

 ///<summary>
/// Provides a mechanism to subscribe to events emitted by grains
/// </summary>
public interface IGrainEventProvider 
{
    /// <summary>
    /// Subscribes to a stream of events for a given subscription, with the event payload included in <see cref="GrainEvent{TEventBase}"/>
    /// </summary>
    /// <typeparam name="TEventBase">The type of events to subscribe to.</typeparam>
    /// <param name="subscriber">The subscriber for observability purposes.</param>
    /// <param name="startingPosition">The position in the event stream to start from.</param>
    /// <param name="eventFilter">An array of event types to filter the subscription.</param>
    /// <param name="cancellationToken">A token to cancel the subscription.</param>
    /// <returns>An asynchronous stream of <see cref="GrainEvent{TEventBase}"/>, <see cref="CaughtUp"/>, <see cref="FallenBehind"/> or <see cref="Checkpoint"/> objects.</returns>
    IAsyncEnumerable<EventStreamUpdate> SubscribeToGrainEvents<TEventBase>(GrainId subscriber, GlobalEventLogPosition startingPosition, Type[] eventFilter, CancellationToken cancellationToken) where TEventBase : notnull;

    /// <summary>
    /// Subscribes to all events from <typeparamref name="TGrain"/> with the event payload included in <see cref="GrainEvent{TEventBase}"/>
    /// </summary>
    /// <typeparam name="TGrain">The type of the grain whose events to subscribe to.</typeparam>
    /// <typeparam name="TEventBase">The base type of events to subscribe to.</typeparam>
    /// <param name="subscriber">The subscriber for observability purposes.</param>
    /// <param name="startingPosition">The position in the event stream to start from.</param>
    /// <param name="cancellationToken">A token to cancel the subscription.</param>
    /// <returns>An asynchronous stream of <see cref="GrainEvent{TEventBase}"/>, <see cref="CaughtUp"/>, <see cref="FallenBehind"/> or <see cref="Checkpoint"/> objects.</returns>
    IAsyncEnumerable<EventStreamUpdate> SubscribeToGrainEvents<TGrain, TEventBase>(GrainId subscriber, GlobalEventLogPosition startingPosition, CancellationToken cancellationToken) where TGrain : IGrain where TEventBase : notnull;
}



/// <summary>
/// Represents an update in the event stream, such as a <see cref="GrainEvent{T}"/> <see cref="Checkpoint"/>, <see cref="CaughtUp"/>, or <see cref="FallenBehind"/> notification.
/// This is the base type for all event stream update notifications returned by IGrainEventProvider 
/// </summary>
public abstract record EventStreamUpdate
{

    /// <summary>
    /// Indicates that the event stream subscription has caught up to the latest available event.
    /// Used to notify subscribers that no further historical events are pending.
    /// </summary>
    public sealed record CaughtUp : EventStreamUpdate
    {
        private CaughtUp() { }

        public static CaughtUp Instance { get; } = new();
    }

    /// <summary>
    /// Indicates that the event stream subscription has fallen behind and is no longer up-to-date.
    /// Used to notify subscribers that the stream is lagging.
    /// </summary>
    public sealed record FallenBehind : EventStreamUpdate
    {
        private FallenBehind() { }

        public static FallenBehind Instance { get; } = new();
    }

    /// <summary>
    /// Provides a regular checkpoint in the event stream when processing through events not relevent to the subscriber
    /// </summary>
    /// <param name="Position"></param>
    public record Checkpoint(GlobalEventLogPosition Position) : EventStreamUpdate;

    /// <summary>
    /// Provides a notification that a grain event occurred (without the payload)
    /// </summary>
    /// <param name="Position"></param>
    /// <param name="EventGrainId"></param>
    /// <param name="GrainEventNumber"></param>
    public record GrainEventNotification(GlobalEventLogPosition Position, GrainId EventGrainId, GrainEventNumber GrainEventNumber)
        : Checkpoint(Position);


    /// <summary>
    /// Represents an event emitted by a grain in the event stream.
    /// This record encapsulates the event data, its global position in the event log, the originating grain's identifier,
    /// and the version of the grain at the time the event was produced.
    /// </summary>
    /// <typeparam name="TEventBase">The base type of the event payload.</typeparam>
    /// <param name="Position">The global position of the event in the event log stream.</param>
    /// <param name="Event">The event payload emitted by the grain.</param>
    /// <param name="EventGrainId">The unique identifier of the grain that produced the event.</param>
    /// <param name="GrainEventNumber">The version of the grain at the time the event was generated.</param>
    public sealed record GrainEvent<TEventBase>(
        GlobalEventLogPosition Position,
        TEventBase Event,
        GrainId EventGrainId,
        GrainEventNumber GrainEventNumber) : GrainEventNotification(Position, EventGrainId, EventGrainVersion);
}

Improved developer experience

Polymorphic immutable state

Currently with JournaledGrain it's a pain to have a read-model that is immutable (each event applied produces a new read model) or state that is polymorphic (where an event can transition the state type) allowing a state machine to represent an aggregate's lifecycle e.g. Basket -> Processing-> Shipped -> Delivered-> Archived, these new components will enable both scenarios - but they are not mandatory.

Read-your-writes
Pending event write operations will return a GrainEventNumber and Commit operations will return a GlobalEventLogPosition which can be returned by the grain so callers can ensure 'read-your-write consistency' for projections shown in the UI.

Observability
The the code will be instrumented, support logging and activity tracing using the latest .net best practices.

POCO Grains
will be supported, no requirement to derive from JournaledGrain

Built-in event sourcing event storage providers
It would be useful to have some supported event storage providers, in addition to an in-memory provider for testing.

Initially a provider for Kurrent would be implemented as a reference
Ideally another provider would be available to ensure the abstractions are flexible enough - community suggestions welcomed.

Relationship to other parts of Orleans

Orleans.EventSourcing
Given the need to make breaking changes, backward compatibility with the existing event sourcing support in Orleans will be limited to developers being able to reuse their existing TLogView and TEventBase types (that they would have used with JournaledGrain) on the new event sourcing stack, there is no backward compatibility planned for ILogViewAdapter implementations. This proposal would add the new approach side-by-side with the existing event sourcing support in the same package.

Orleans.Journaling
Orleans.Journaling is optimised for high performance large-state storage involving multiple state machines with incremental atomic writes.

Projections often involve aggregating many events into large models, which if based on Orleans.Journaling would provide efficient storage of projection and side-effect state.

Streams
Streams (at least v1) do not provide a complete replay of all events from the beginning of time, they operate with the concept of a replay window, nor do they guarantee global ordering of received events, making them unsuitable for projections. It's not yet clear v2 addresses the above.

Related Issues

#8408
#8408
#486
#9734 (comment)
#9734 (comment)

[flensrocker]

Just a first "brainstorm" kind of note:

1. How close do you think should the (metadata) structure of the events be in regards to cloud events?
2. I have mixed feelings about IEventSourcedReadModel, because esp. in CQRS the term "read model" already has a meaning. And it's different from the "aggregate" you usually have for the commands as the consistency boundary. I would prefer IEventSourcedState (like IPersistedState) and the generic argument actually defines by its name if it's a CQRS read model, an aggregate etc.
3. What about versioning/rebuilding projections? The logic of projections can change so that they have to be rebuilt (on the fly maybe). Should this be as a provided functionality or some "execise for the reader"? 🙂 Of course we can do that with "just" another ReadModelV2 thing and deprecating the ReadModelV1. I don't have a lot experience with rolling live upgrades where this can be important.
4. Having multiple read models on one grain should be possible (sometimes I wish I can modify two or more aggregates in the same transaction), but it also feels like a code smell, because they all get populated but may not be needed on every method call. So maybe they should be different grains.
5. I like "implicit projections" (like implicit stream subscriptions), because that's the only way how to build "queryable" content somewhere in an async fashion. On the other hand for "detail state" I like having non-persistend projections which are directly populated from the events (so the logic can change in an easier way) and having explicit event subscriptions to be kept up-to-date while they are active. That's where Orleans can shine as a real good in-memory cache for active entities.

I will re-read your content on the weekend to be able to make a better comment and disect the code examples.

Thanks, that looks promising!

[willg1983]

Thanks for the initial feedback.

1. Cloud Events I am not super familiar with, what would your expectations be on how they would be used? Would you expect the grain author to use CloudEvent as their TEvent type? Or would you expect their TEvent type to be serialized into the Data property of a CloudEvent?
2. I agree. I've changed the name to IEventSourcedState as you suggested 👍
3. I think the story here should be to deploy your new projection as a new grain side-by-side with your old projection grain, and then switch over consumers when appropriate, or if the projection is quick to rebuild, simply discard existing state and let it be rebuilt (state contains the checkpoint position, so this would also reset the projection to start replaying from scratch). I have considered encoding a hash of the event types in the checkpoint data, so any change in the events the projection uses would invalidate the checkpoint, but that does not protect against interpreting the same events differently in the projection so I don't think it's worth doing. A projection will be aware of whether or not it is still catching up, or live, so it can decline to serve queries until it is caught up, or it can forward them to the old projection for example.
4. The proposal allows for multiple state models within the same event sourced grain, on the assumption those models are coupled to a single aggregate's event, but not multiple aggregates - that would be a projection. Are there specific use cases you are thinking of here? Usually if you need to modify multiple aggregates (or external systems) you use a side-effect possibly with a saga (this might be a good fit with durable tasks/jobs in Orleans)
5. I like this idea 👍 Need to have a think about how this would work? Something like put a [Projection] attribute on the grain and a source generator will find all the IEventSourcingProjection<TEvent> interfaces the grain implements and automatically subscribe it and feed it events? The Grain will also need to implement an interface to allow the system to query the checkpoint GlobalLogPosition to resume from on startup, perhaps IEventSourcingProjection<TEvent> could derive from IProjectionGrain which has a GetCheckpoint() method? This feels like one of those higher-level abstractions we could build on top of the basic IGrainEventProvider interface.

[flensrocker]

1. CloudEvents

Like you I'm not very familiar with CloudEvents, but this is some kind of standard on how to store persistent events and then they may be consumable by other tools. They also include versioning etc. and since Event Sourcing is about (very) persisted events, that might be at least something to think about.

It's more about what kind of metadata should be stored (something like Source, DataSchema etc. can be interesting). At least the required bits may be interesting so that some kind of CloudEvent compatible wrapper can be provided to be able to export in that format.
The TEvent base class should not derive from CloudEvent, because that's the payload in the Data property.

Homework for me: research if this spec is actually good for event sourcing.

2. Renaming IEventSourcedState<T>

✅♥️

3. Rebuilding/Re-interpreting projections

I think the story here should be to deploy your new projection as a new grain side-by-side with your old projection grain, and then switch over consumers when appropriate,

Yes, that sounds good. Until now I haven't had to deal with "live updating" systems while they run, but in Orleans context this is sure a thing. I have to get used to this kind of multi step deployment: first deploy the changes to the data (like new projections, DB schemas etc.) and backfill them while also running the old projections, and if they are up-to-date, deploy the next version to switch the consumers of those projections to the new one and finally remove the old ones.

BTW: How do we discard state associate with a type of grain that doesn't exist anymore? With good old relational database I would just drop the tables on the third deployment step. But I wonder how this works e.g. with Blob-Storage or something similar.

4. Multiple Aggregates

The proposal allows for multiple state models within the same event sourced grain

Ok, I would expect that all needed informations are aggregated in the same projection logic (like an order with its order-lines etc). But you may have different usecases and experience why this can be useful.

Speaking of multiple aggregates: invoices (at least in Germany) have to be sequentially numbered without gaps. So to create a new invoice with the next number, I think the number group for these invoices (per year) is an aggregate on its own. The only thing I can come up with is some kind of reservation pattern, where the invoice is e.g. stored with some InvoiceRegistered event that kicks off some "processor" (that's how they are called in event modelling) that creates a new command for the number group like NumberInvoice with the year as id and the invoice-id as payload, that then reads the current state of that number group (which should obviously be snapshotted because it's just a number [which leads me to another idea of having some kind of "last event is my snapshot" kind of functionality for lightweight snapshots without actually doing snapshots]) and then emits a NumberIncreased event on the number group, which then gets picked up by the next processor to file a NumberInvoice command on the invoice.
Does this makes sense?
Having the possibility to raise two events at the same time like the NumberIncreased and InvoiceRegistered (then with the correct number already included) seems nice.

I don't know if there are other usecases, but having consecutive numbering on aggregates is a common scenario in my business. That's why relational database have some kind of sequence generator built in (but using them gapless is another topic).

5. Implicit/Explicit Event Subscription

Implicit subscriptions to (domain) events from the Event Souring can almost be like the current stream implementation. Having an attribute like e.g. ImplicitEventSubscription and settings up something like the OnNextAsync in streaming (which can be named ApplyAsync, because we are "applying" events to a projection) is pretty straight forward - assuming we will have some kind of stream of the events.

Explicit subscriptions can be handled with a base class for the grain (or similiar), which subscribes on activate and unsubscribes on deactivate. But I like the idea of a source generator - but no experience here.
I posted a link to my experiments on implementing Event Sourcing in Orleans on top of PostgreSQL multiple times on the Discord server, but for reference:
https://codeberg.org/flensrocker/OrleansPlayground/src/commit/c9cc2642b72489d3b6a09639741e67e06bfd2a11/src/EventSourcing.Orleans.Server/ExplicitDomainEventProjectionGrain.cs

The grain is just "hooking things up":
https://codeberg.org/flensrocker/OrleansPlayground/src/commit/c9cc2642b72489d3b6a09639741e67e06bfd2a11/samples/ToDo/src/Orleans.Server/ToDoDetailsViewGrain.cs

And the projection is "in the domain":
https://codeberg.org/flensrocker/OrleansPlayground/src/commit/c9cc2642b72489d3b6a09639741e67e06bfd2a11/samples/ToDo/src/Abstractions/ToDoView/ToDoDetails.cs

[flensrocker]

Typo in the example code (missing "d" on IEventSourcedState):

public sealed class MyEventSourcedGrain(IEventSourcedState<MyReadModel>

[flensrocker]

I'm not sure about this part:

public sealed class MyEventSourcedGrain(IEventSourceState<MyReadModel> eventSourcedState) : Grain,
    IMyEventSourcedAggrgateGrain
{

    // Example of command pattern 
    public async ValueTask<EventSourcedCommandResult<TResult>> ExecuteCommand<TResult>(IEventSourcedGrainCommand<EventBase, TResult> command, CancellationToken token)
    {
        var result = command.Execute(eventSourcedState, this.GetPendingEventsWriter());
        var (globalLogPosition, eventNumber) = await this.CommitPendingEvents(token);
        return new(result, globalLogPosition, eventNumber);
    }

      // Query method etc
}

For me a command is "just" some kind of DTO like object, which gets fed into a function on the aggregate and returns a list of events (or EventWrites). I'm a fanboy of the Result monad thing, but error handling of the command can also be done with an exception. But I'm not sure if this should be part of the framework or be a decision of the developer depending on their needs.
Usually my command handler will only return a "success" (with an optional response like the id of the created entity) or "error", but sometimes I return the list of events for optimistic UI updates. It depends...

[willg1983]

1. CloudEvents

Taking a step back, I think this is a more general class of problem: how does one provide a general programming abstraction over disparate specific implementations? In this case the specification for an event in an event sourced system.

This proposal tackles this by making EventWrite and EventRead the contracts between IEventConverter and IGrainEventStorageProvider with minimum required properties for event sourcing systems. The Metadata dictionary is a catch-all for everything else.

We could instead make IEventWrite and IEventRead interfaces, allowing freedom to pass any type that implements those interface between IEventConverter and IGrainEventStorageProvider, you could then have a vendor specific IEventConverter and IGrainEventStorageProvider that pass their own type between them. Inevitably if you start using a proprietary implementation of those, you can no longer change to a different implementation and expect things to just work.

On CloudEvents specifically, let's assume we wanted to implement them for a IGrainEventStorageProvider

The REQUIRED attributes of a CloudEvent are: id (String), source (URI-reference), specversion (String), type (String)

It seems me the implementation of WriteEvents would have everything it needs to produce CloudEvents with the minimum required properties:
CloudEvent. Id = @event.EventId
CloudEvent.Source = grainId
CloudEvent.specVersion = 1.0
CloudEvent.Type = @event.EvenType

In terms of OPTIONAL attributes, datacontenttype would be the @event.Data.MediaType.

The metadata dictionary could be used to set everything else - dataschema, subject and time.

The only that might be worth thinking about promoting is the concept of a timestamp? - see my comment in EventRead about writtenTimestamp - we need to be careful not to confuse the time of a write (which is an infrastructure concern), with the time stamp of an event (which is a business concern) - let me know your thoughts on this?

3. Rebuilding/Re-interpreting projections

BTW: How do we discard state associate with a type of grain that doesn't exist anymore? With good old relational database I would just drop the tables on the third deployment step. But I wonder how this works e.g. with Blob-Storage or something similar.

Often in event sourcing you write a tombstone event and then truncate all events prior, this ensures a record exists of the deletion containing who+why+when and allows projections to read the deletion event and clean-up any associated state.

The grain would write a tombstone event (which may be marked as such for example) and that would be written and call IGrainEventStorage.DeleteEvents to delete all prior events in the stream.

If you wanted to remove all events, calling IGrainEventStorage.DeleteEvents passing the last known GrainEventNumber would remove all events for that grain.

4. Multiple Aggregates

Ok, I would expect that all needed informations are aggregated in the same projection logic (like an order with its order-lines etc). But you may have different usecases and experience why this can be useful.

You can implement it all in one event sourced state type. It might be easier to split them out, for example: if you want your event sourced Order grain return the line items, current customer order status, internal order status and the history for the order for example (setting aside the argument all non-command decider state should be separate projections for a moment, perhaps because of practical concerns around eventually consistency here) then a single state implementation's event handler might look like:

Apply(OrderCreatedEvent @event)
{
    this.OrderHistory.Add(new OrderCreatedHistoryEvent(@event));
    this.CustoimerOrderStatus = CustomerOrderStatus.Processing;
    this.InternalOrderStatus = InternalOrderStatus.AwaitingPaymentGateway;
    this.LineItems = @event.ShoppingBasket.ToLineItems();
}

The challenge is that code is doing four different things to your state, it might be cleaner to split them into different state implementations each of which has one responsibility. These can be tested and evolved independently. The event sourced grain still has current 'state' its just decomposed into smaller state implementations that collectively provide it.

Speaking of multiple aggregates: invoices (at least in Germany) have to be sequentially numbered without gaps.

I think I would centralising invoice number generation to a single aggregate, as part of the process of placing an order it would assign an invoice number and issue it as part of the command to create the order. If you needed more scalability then invoice number assignment could be a side-effect that looks for OrderPlaced events and then requests an invoice number and assigns it via a command to the order (orders would have a UUID internally, an Invoice number would just be a property of the order that is added after creation). I'm sure there are other approaches.

Generally that's how transactions (multiple aggregate updates) are avoided and that's often seen as a net benefit to scalability and robustness - the trade-off being eventual consistency.

5. Implicit/Explicit Event Subscription
   Looks interesting, I think we need to look into this further and refine the proposal.

[m3nax]

Hi @willg1983 and @flensrocker I have read the proposal and would like to add a couple of points for consideration about the state snapshot support.

First of all without snapshots we can't truncate the event log of an active grain because it would be impossible to recalculate its state and it would only be usable for “thombstoned” grains, where only the event representing the deletion remains.

In my company, we have some long-lived event-sourced grains with more than 3000 events each. With our actual implementation of ICustomStorageInterface we append events in the event log and create a materialized view of the state (aka snapshot) every x events, when the grain is activated we read the last snapshot and than apply the missing events to the state.

So, based on the implementation made for the IStateMachineStorage interface, we can and a bool to indicate when we want to create a snapshot and a method to read the lastest available snapshot.

public interface IGrainEventStorageProvider
{
    /// [..] Other interface methods

    /// <summary>
    /// Reads, if available, the last persisted snapshot of the state.
    /// </summary>
    /// <param name="grainId">grain identity</param>
    /// <param name="cancellationToken">token to cancel the read</param>
    /// <returns>Returns the most recent state snapshot, null if no snapshots are available.</returns>
    ValueTask<Snapshot?> ReadLatestSnapshot(GrainId grainId, CancellationToken cancellationToken);

    /// <summary>
    /// Writes the generated snapshot to storage.
    /// </summary>
    /// <param name="grainId">grain identity</param>
    /// <param name="snapshot">snapshot value</param>
    /// <param name="cancellationToken">token to cancel the read</param>
    ValueTask<GlobalEventLogPosition> WriteSnapshot(GrainId grainId, Snapshot snapshot, CancellationToken cancellationToken);

    /// <summary>
    /// Gets a value indicating whether the storage provider has requested a snapshot.
    /// The default value is false and disables the creation/saving of snapshots.
    /// </summary>
    bool IsSnapshotRequested { get; }
}

public readonly record struct Snapshot(
                                        string StateType, // for deserialization
                                        BinaryData Data,
                                        GrainEventNumber EventNumber,
                                        GlobalEventLogPosition Sequence
                                        );

Potential uses:

1. For storage providers that do not support snapshots, the IsSnapshotRequested value will be set to false, and the ReadLatestSnapshot method called during grain activation will always return null, forcing the system to read the entire event log directly.
2. For storage providers that natively support snapshot creation and do not require Orleans to generate it, the IsSnapshotRequested value will be set to false, and ReadLatestSnapshot will return the latest snapshot if available.
3. For storage providers that require Orleans to generate the snapshot the IsSnapshotRequested value will be set to true based on the provider-specific rules, and ReadLatestSnapshot will return the latest snapshot if available.

[willg1983]

Thanks for the feedback @m3nax

There are two types of snapshot in my experience, let me set out how I see it and then others can chime in:

Read-optimization snapshot

Scenario: A projection or aggregate has a large number of events and replaying them each time to build current state is expensive.

Solution: To speed up initial loading a snapshot of state is created at regular intervals. On load the snapshot plus events emitted after the snapshot are applied to get back to current state quickly.

This is purely a technical read optimization of state. A key differentiator of this scenario is the snapshot could be discarded and state rebuilt from the complete event stream - this may have an RTO impact but will not affect RPO.

This optimization is especially impactful on stateless middle tiers where a command handler would typically load a complete event stream for an aggregate, execute a command and then persist emitted events. The process would be repeated for every command.

As Grains are stateful they avoid the need to reload the aggregate event stream for every command. Support for grain live migration in the proposal and the configuration of grain idle collection lifetime maximize this benefit.

There may be cases where the initial penalty for a grain to load an aggregate's event stream to build a state model is still too high and a previously written snapshot of the state model plus delta of events would accelerate initial loading.

For projections their state models may already be persistent, for example, if a projection maintains a read model in a database. In other cases the projected state model may not be persisted.

---

Business Snapshot

Scenario: The business requires very detailed events initially but the value of discrete events diminishes after certain business conditions are met and retaining that level of detail over time has negative value.

Solution: At a natural business boundary an event summarizing the aggregate's current state is written and prior business events are discarded.

Examples: Data Retention Privacy Laws / deletion tombstoning / aggregate archival / end of trading summary

This is a business operation, and whilst superficially similar to the Read-optimization snapshot scenario there is a critical difference: The business snapshot is not throw-away as the event stream discards all events prior to the snapshot. If the snapshot were lost, the aggregate's state could not be retrieved from events in the event stream. This has an RPO impact and RTO impact.

Deriving the aggregate's current state from this event does not depend on any prior events. The business event may also contain summarized (lower fidelity) information from the original events and necessitate projections also discarding or replacing information they contain about the entity based on this event.

---

They are independent of one another and complimentary. An aggregate can have both a read-optimisation snapshot and also write business snapshots.

I've purposely not proposed technical solutions to them as I think it's important to ensure we agree those are the scenarios we are dealing with first?

[ReubenBond]

Thank you for the proposal, @willg1983 and for the comments, @flensrocker & @m3nax! I am supportive of the idea. I like the concept of being able to observe grains externally and building that in as a feature. I see a lot of similarities between this and Orleans.Journaling and I wonder whether we can have one library that supports both patterns well. The IDurableStateMachine types in Orleans.Journaling are already event sourced in some sense (they are more like replicated state machines on a shared log, but the lines blur if you take a non-dogmatic view of what Event Sourcing is). They already have snapshots and the storage provider can indicate whether a snapshot is requested (like @m3nax requests), etc.

Orleans.Journaling is still a pre-release library, so we have a chance to change it to suit the needs of event sourcing if that would be appropriate.

My question is what would it take to accomplish that? In Orleans.Journaling, each IDurableStateMachine effectively gets its own log, even if changes to multiple state machines are guaranteed to be committed atomically (i.e, remove from a queue + add to a dictionary, then call WriteStateAsync = 1 atomic write). It's not clear to me that each of the IEventSourcedState<T> instances in your proposal are isolated from each other in the same way. If they are, how do we determine which state machine an event is destined for? I might have missed the detail - but I feel I have an understanding gap.

I have been experimenting with making those state machine implementations more open so instead of hard-coding a serialization format internally, they have objects that represent their state transitions - you might even call them events. The main driver for that has been to support JSON-native storage and Protocol Buffers native storage as alternatives to the current hand-crafted encodings.

[flensrocker]

I'm basically fine with the current Orleans.EventSourcing setup. And I guess the IEventSourcedState is somehow similar to the state of the "old school" JournaledGrain<TState, TEventBase> - it must have some kind of Apply method to transition to the next state depending on the raised events.

What I really miss is an easy way to implement (async) projections.

If I'm right (trying to read the source) I cannot use the same TEventBase on another grain with a different TState, because the storage of the events are tied to the type of the originating grain (ofc I can use that event class, but it will not read the same persisted event data than the other grain - please correct me if I'm wrong). So I don't know how to provide a different read-model of an event stream (assuming that the originating grain is "the aggregate" or "command handler" for handling the write side of an event sourced entity - CQRS is not a must, but a natural companion to event sourcing).
But this may be a limitation of the provided LogConsistencyProvider. As mentioned I try to implement a custom provider based on PostgreSQL, so this is not a limitation per se.

The real power of event sourced system is to be able to define various projections of the event streams, even combining events of different TEventBase in "multi stream projections". And to make them not interfere with the creation of the events there should be some kind of messaging system like Orleans Streams, so "projection grains" (or services) can consume the events - either through some kind of implicit subscription or even an explicit one (for live updates that are not necessary after deactivation).
Usually one needs some kind of outbox thing to decouple the storage of the events from their publishing. And the outbox may depend on the actual storage implementation.

I'm curious how the new Orleans Journaling can help us here! Because "inbox" and "outbox" are mentioned in that PR...

[ReubenBond]

The Orleans.Journaling PR mentions inbox & outbox as potential things that can be built on top, using the same underlying log.

[willg1983]

It's not clear to me that each of the IEventSourcedState<T> instances in your proposal are isolated from each other in the same way. If they are, how do we determine which state machine an event is destined for? I might have missed the detail - but I feel I have an understanding gap.

As @flensrocker suggested each IEventSourcedState<T> implementation would be able to receive one or more events, a grain component would multiplex events to every IEventSourcedState instance for the grain synchronously, on load and before each event write. We would probably provide some higher-level abstractions so developers would implement strongly typed Apply() methods and the componentry would dispatch only events they required.

My question is what would it take to accomplish that?

We implemented an early version of Orleans Journaling based on your original proposal and I see them as different but complimentary. Event sourcing is about recording business facts as a series of immutable events with business-defined retention and Orleans.Journaling is really about state storage, the log is an implementation detail to achieve efficient writes with large-state.

The key differences are:

• They operate at different granularity: Durable state machine events are discrete component mutation operations for that one state machine only. Event sourcing events are business operations scoped to the business concept (e.g. 'add key and value to dictionary' vs 'customer placed an order')
• They have different contractual requirements: Durable state machine events are contractual only to the state machine that emitted them. Event sourcing events are contracts with other components of the system, e.g. zero or more other projections that may depend on them.
• Durable state machine events are temporary until a heuristic decides a snapshot is required to consolidate the state machine state. Event sourcing events must be retained indefinitely and removal is a business operation rather than a technical one.
• Event sourcing projections must be able to aggregate all events from the system, replay through historical ones and then observe live ones. Typically this is globally ordered - but being pragmatic it might be acceptable to just ensure consistent ordering per-aggregate. This usually requires the concept of a global log that is used to discover and replay all events that have occurred.

I guess you could make changes to Orleans.Journaling to optionally disable log truncation, format events using STJ and also write a pointer to the grain (or at least the first event, so the grain can be discovered) into some kind of global log for discovery by projections, that could then fetch and read the grain's logs?

Then we could implement an event sourcing component as an IDurableStateMachine perhaps on top of the existing infrastructure.

I'm not sure if mixing the two requirements would be a benefit for both or dilute both?

[ReubenBond]

The purpose of journaling is more than just making larger state more feasible; it's also about atomicity across multiple bits of state within a grain.

State machines aren't required to be snapshottable: they can 'snapshot' by emitting all of their events again. If we retained (eg archived) the log instead of truncating/replacing it at snapshot time, that should enable the property you seek, where the total history is maintained. The snapshot for event sourced state could be empty, a pointer, or a materialized view. For very large collections (large than you would want eagerly loaded into memory), I had this idea that the log is really just the latest events (it provides atomicity) and snapshotting involves taking the events from the log and flushing them down to another system (eg, a table, or a database, or even some other grain).

I agree that there's a conceptual difference between replicated state machines and event sourcing in that ES logs domain events whereas these durable collection RSMs log data manipulation operations. Mechanically, they are similar: they both involve replaying a log to materialize a view.

On granularity, I see these things as largely isomorphic. In Orleans.Journaling, we route events to the relevant state machine. If you route every event to every state machine, that is ok as long as the state machines have a way to filter them out.

Given that event sourcing is fundamentally about logging & replaying events, it still seems like a good fit to me even if Orleans.Journaling is just an implementation detail or an optional implementation detail (with another option being to implement the interface directly on Kurrent? Is it necessary/useful?)