Provide docs for Log Consistency Providers and split test

[galvesribeiro]

First of all, thank you @sebastianburckhardt for the great work!

After #1854 got merged I tried play with it but was unable to get it to work. I was unable to find among the tons of tests what are the files related to OrleansEventSourcing project. The configuration and requirements is what is hardest to find.

I would suggest a simple page with the minimum required configuration to get it to work.

Also, would be good if that project has its own test project. I can do it myself if you point out what exactly is related to this project inside the test projects.

Thank you Sebastian! :)

[sebastianburckhardt]

No doubt: the very next step is to add documentation and samples, both for the geo-distribution use case and for the simpler, non-geodistributed event sourcing. I will create a new PR shortly.

The only things already there right now are

• a couple unit tests in Tester/EventSourcingTests. These demonstrate a few aspects of the API, but not much about the configuration.
• Several tests for log-consistency providers in TesterInternal/GeoClusterTests. These are unlikely to be of much interest unless you are planning to write your own log-consistency provider, which is probably not the first thing you want to try.

In a separate branch, I already have three geo-distributed samples, including information about how to configure and run them under Windows Azure. I will add them to the current Orleans samples.

[galvesribeiro]

Thank you for promptly reply @sebastianburckhardt

Yes, I think the first thing you can do, is just a very simple page that allow people that are already familiar with ES (including the old code) to configure Orleans to use your new provider mode.

After that, with time, you can elaborate more samples and a more complete doc page. That will unlock right now people (like me) to use it and provide feedback.

Also, I don't see in that code where we can replay events. For example, someone introduced a new grain type or an external system that behave as a projection. How can the events be replayed? That would be good to have in the "Hello world ES".

Again, thanks for your great work!

[sebastianburckhardt]

Maybe I can give you enough information right here, to unblock you now before I get to writing the documentation.

To get the equivalent of the previous ES support in Orleans, for a class like in the JournaledPerson example at TestGrains/EventSourcing/JournaledPersonGrain.cs, you would use something like the following configuration

    [StorageProvider(ProviderName = "WhateverStorageProviderYouWantToUse")]
    [LogConsistencyProvider(ProviderName = "LogStorage")]
    public class JournaledPersonGrain : JournaledGrain<PersonState,IPersonEvent>, IJournaledPersonGrain

with corresponding provider configuration

    <LogConsistencyProviders>
      <Provider Type="Orleans.EventSourcing.LogStorage.LogConsistencyProvider" Name="LogStorage" />
    </LogConsistencyProviders>

(the storage provider is optional, if you leave it out the default storage provider is used).

The Orleans.EventSourcing.LogStorage.LogConsistencyProvider implements a simple form of event sourcing where the storage contains the full log of events, and each read/write from/to storage reads this entire log (so you would probably not want to use this provider for grains that accumulate a very long log of events because then reads and writes become slow).

All events in the log are replayed automatically right after the grain is activated. From then on, events are applied incrementally.

Finally, there is a little pitfall here. A grain can start executing even before the log of events is loaded and replayed (this is so a grain can be available even if the connection to storage is temporarily broken, or if it takes a long time to load the state). In most cases, you probably want to avoid this behavior; to make sure the grain is fully caught up during activation, you can call RefreshNow explicitly as follows:

        public override async Task OnActivateAsync()
        {
            await RefreshNow();
        }

[galvesribeiro]

Thanks for get back.

I tried it but I'm getting the same error from #2597 :(
I though it was a configuration problem but it isn't.

Regarding the replay, I understand that all the event stream is applied to the state serially up to the HEAD version. My question was, once the events are applied, if there is a way to replay it again so a new projection can be generated from that. I guess the only way to do it today, is to loop into the event (btw, I don't see a this.Events in JournaledGrain<T>) and generate a projection based on it.

[galvesribeiro]

@sebastianburckhardt I got it to "work"

However this line here https://github.com/dotnet/orleans/blob/master/test/TestGrains/EventSourcing/JournaledPersonGrain.cs#L135 never is true. ConfirmedVersion == 1 so it fail.

Is there anything wrong?

[galvesribeiro]

Ok... I changed that delay to 200 and it worked... I'm not quite sure I followed why... The Confirm() should be atomic, right? Or is it writing to the storage somewhere else in parallel?

Also,

orleans/src/OrleansEventSourcing/LogStorage/LogConsistencyProvider.cs

Lines 16 to 22 in 0f4ba04

	/// <summary>
	/// A log-consistency provider that stores the latest view in primary storage, using any standard storage provider.
	/// Supports multiple clusters connecting to the same primary storage (doing optimistic concurrency control via e-tags)
	///<para>
	/// The log itself is transient, i.e. not actually saved to storage - only the latest view (snapshot) and some
	/// metadata (the log position, and write flags) are stored in the primary.
	/// </para>

mention that the events are transient... Is that correct?

The GrainState table is getting this:

{
  "$id": "1",
  "$type": "Orleans.EventSourcing.LogStorage.LogStateWithMetaData`1[[OrleansESTest.Grains.IPersonEvent, OrleansESTest.Grains]], OrleansEventSourcing",
  "Log": {
    "$type": "System.Collections.Generic.List`1[[OrleansESTest.Grains.IPersonEvent, OrleansESTest.Grains]], System.Private.CoreLib",
    "$values": [
      {
        "$id": "2",
        "$type": "OrleansESTest.Grains.PersonLastNameChanged, OrleansESTest.Grains",
        "LastName": "Organa"
      },
      {
        "$id": "3",
        "$type": "OrleansESTest.Grains.PersonLastNameChanged, OrleansESTest.Grains",
        "LastName": "Solo"
      }
    ]
  },
  "GlobalVersion": 2,
  "WriteVector": ""
}

[sebastianburckhardt]

My question was, once the events are applied, if there is a way to replay it again so a new projection can be generated from that

Currently, the API does not include a direct way to enforce a replaying of the events, nor a way to get access to the whole event history. This is because the API is trying to be as generic as possible (make as few assumptions as possible about the underlying provider). Not all providers even persist the events, and are in that sense not strictly event sourcing. For example, Orleans.EventSourcing.StateStorage.LogConsistencyProvider only persists the latest state, and never replays events.

However, if you are using Orleans.EventSourcing.LogStorage.LogConsistencyProvider, you can force a replaying of the events by deactivating and reactivating the grain. I am not sure why this would be desirable... I would think in most cases, the projection changes only if you are changing the code, in which case you would need to deactivate/reactivate the grain anyway.

If there is an important reason why the provider should grant access to the events, we can consider adding additional methods to the API, perhaps to retrieve whatever piece of the event sequence is still around.

[sebastianburckhardt]

Ok... I changed that delay to 200 and it worked... I'm not quite sure I followed why... The Confirm() should be atomic, right? Or is it writing to the storage somewhere else in parallel?

The event storage API is indeed somewhat "parallel" / "asynchronous". After the call to RaiseEvents(events), the events are written to storage asynchronously (atomically, but not instantly).

In this particular unit test, I am using an estimated time of 20ms for this provider to complete the storage operations, which is of course fragile - it definitely depends on the storage provider, here memory storage which was always fast on my machine.

Regular application code should never make such such timing assumptions. Rather, to ensure that all pending event storage operations have completed, call await ConfirmEvents() , which is meant for just that purpose.

By the way, all the included log consistency providers are reentrancy-safe: adding the [Reentrant] attribute on a JournaledGrain does not cause dangerous race conditions on storage. The provider ensures only one storage operation is pending at all times, and it batches multiple such operations together to improve performance.

[galvesribeiro]

However, if you are using Orleans.EventSourcing.LogStorage.LogConsistencyProvider, you can force a replaying of the events by deactivating and reactivating the grain. I am not sure why this would be desirable... I would think in most cases, the projection changes only if you are changing the code, in which case you would need to deactivate/reactivate the grain anyway.

Yes I'm using that. Hence why I'm seeing the full events in the storage. Regarding the projections, imagine a scenario that you have a long series of events. Whenever one event happen you publish to a stream so you can generate a projection in another grain. Later on, you introduce a new grain type that will represent a new kind of projection. That will require you to replay all the events from begining and process them on this new grain type to generate the new projection. That is something really common and one of the benefits of ES. To be able to leverage the application event history even on new systems. So, I think you should expose the list of commited events so you can based on that generate projections anytime (e.g. when someone call a grain method publish each one of the events a new stream). It is important that we can get all, or a filtered list of events, for exemple, "starting on version x and stop on y".

[sebastianburckhardt]

Ok, I think I see the point.

One way to support this scenario elegantly is to let several grains share the same event log. This is not yet supported (right now, each grain uses its own log), but it may be a good feature to add. I believe @jkonecki also proposed this feature, we have a prototype that does this for a specialized class of providers based on event storage.

Also, we can add support for log retrieval to the API, like you say. However, as mentioned earlier, log retrieval may not behave uniformly for all providers.

It would be helpful to have a clearer idea on what kind of storage back-end people would like to use in conjunction with this feature. We need to formulate an event storage API that works across a variety of such storage options.

[galvesribeiro]

One way to support this scenario elegantly is to let several grains share the same event log

That would be good, but I think will introduce a lot of complexity for now. I think projections are totally unrelated to the source of the events which in this case is the JournaledGrain<T> in the sense of how they generate it. The ES-capable grain, must deal with the events just like you did and @jkonecki did on his previous code. The grain should be able to publish his events as is or by adapting that event as a custom message to a stream so the consumers would be able to process it and generate their projections. There are 2 ways to generate projection IMHO. One is by push the event (as is or in whatever other form) - in that way the consumer will recalculate its projections based on that - or, a snapshot, which will push the current state snapshot as is, so others can have it (useful in CQRS scenarios).

Regarding the backends, I think we should provide a common interface to just read and write events to a store. The way it is persisted, is totally out from the JournaledGrain<T> or its Log adapter IMHO. Look at how @yevhen's very smart lib https://github.com/yevhen/Streamstone work. SS could definitively be one of the many Event Stores that we could add by implement the common ES store interface. The same for GetEventStore or many others. Just like we have a common grain state interface which we implement for regular grain state, have an API that focus on read and write events (really, two methods is what we need) would be perfect and deal with the majority of scenarios IMHO.

[galvesribeiro]

Hey @sebastianburckhardt do you think that will be too long before more complete docs are around or an ES API is in place?

[sebastianburckhardt]

not too long. I will resume work on this later this week, and hope to make some progress by next week.

The Event Sourcing is likely to keep developing over a period of time, also based on feedback. Note that the plan is to have two separate APIs; the programming API inside the grain, and the event-store API.

What's there so far is the programming API for use inside grains. Based on our discussion, I will add a function for accessing events in the log. This is a minor addition that can go into a small PR.

Adding the event store API is the big next step. We already have a prototype for this written by @jkonecki but I need to update it slightly to fit together with the rest of the story. This prototype works with GetEventStore, but the API is meant to be more general. I will create a PR for this with the work-in-progress flag.

[galvesribeiro]

Ok thanks, that sounds good. Looking forward for it. Thanks

[sebastianburckhardt]

As I am writing the docs I get the strong impression most of the time, users should work with the confirmed state, not the tentative state. Unless anyone has strong objections, I will change the default names so "State" refers to the confirmed state, not the tentative one, while "TentativeState" can still be used to access the tentative state.

[galvesribeiro]

I agree @sebastianburckhardt

I think that is the way the state is named on Reliable collections APIs on SF as well. Makes it clear.

[ReubenBond]

@sebastianburckhardt that sounds good to me.

[sebastianburckhardt]

We can close this issue, it is covered by #2651, #2667, #2711.