Support for global log (e.g. $all stream in EventStore)

[eulerfx]

This would support the approach nowadays referred to as bottled water.

One way would be to do an additional insert, into a row space ordered by a global timestamp. Then at read time, a merge sort across partitions.

[yevhen]

Hi Lev! Cool to see you here.

Nice idea. That could be implemented but will require few additional things:

1. Stream directory. At the moment SS doesn't maintain directory of all streams. There samples for how to implement that, which is really trivial. Anyway, that feature is on roadmap and will be baked into core soon.
2. Maintaining global timestamp. I'm not sure about how best to proceed with that. Currently, SS is just simple serveless in-process library (a component). So if there multiple hosts each will have it's own view of the global timestamp. So having a global timestamp will require some synchronization protocol to be implemented. Perhaps, you can advice a simple solution.

If we disregard 2). and assume single-host deployment, this can be done very easily by anyone, since out-of-the-box SS supports including additional table operations along with event, so you can do additional insert, and implementing a stream directory is very simple as well.

I'll give that some more thinking.

[yevhen]

BTW, one of the drivers behind birth of SS was to have an ultra-scalable cheap PaaS event store by avoiding global ordering of events.

Jay Kreps talks about log partitions in Building a Scalable Log section here. What he describes are actually DDD aggregates each mapped to a distinct stream. In that sense global order of events doesn't make sense, really.

Why do you need global ordering? Are you're dealing with an aggregate with a long lifecycle? Busy aggregate? Perhaps you're looking for a sharded stream?

I can think of distributing events of a single stream into multiple partitions (shards) to get ultimate write/read speeds. That could be done by combining stateful processing (actor per stream, Akka.Net/Orleans) and consistent hashing algorithms. I'm thinking about implementing something like that just for fun, as an example of using two of my OSS projects (Streamstone/Orleankka).

Unfortunately, such stream won't guarantee a perfect order of events, could have gaps due to write failures and retries. This might be ok for something like scalable logging. Thus it would be probably better partitioned by natural key such as date.

[jkonecki]

I think the link to Jay's talk is broken.

[yevhen]

@jkonecki thanks. fixed!

[yevhen]

AFAIK, Greg is able to provide _all feature and be perfectly ACID only due to EventStore being a single partition.

[eulerfx]

Why do you need global ordering?

This global stream only needs to maintain order with respect to events within individual streams, but not necessarily across streams. One approach is to do a transactional dual write in AT - one row corresponding to the event in an individual stream and another entry in a global log. This way, you split a partition into two halves - one containing all global log . This global log would be ordered by a timestamp plus some value to ensure uniqueness (such as the id of individual stream). If operations on individual streams are forced to specify explicit versions (OCC), this ensures consistency and ordering within individual streams.

Re Kafka and event sourcing, I'm not quite sure how to actually do it given that Kafka wasn't designed to support arbitrarily large numbers of topics. The log compaction feature can help, but only for integration - you still have no way to read events corresponding to an individual entity. Kafka is great as a bus, but I'm not sure it fits as a source of truth store.

Stream directory.

Would this be implemented using AT entity group transactions? Also, I'm not sure this and the global timestamp would suffice. I suppose you could scan the stream directory and pick events greater than a certain global timestamp, but this seems prohibitive.

Maintaining global timestamp

As alluded to above, I think it should be possible to do this using a simple timestamp plus a unique connection id or event the stream id + event number of the event being written. Since OCC ensures consistency within an individual stream and we don't require global timestamp to fully respect ordering across streams.

[yevhen]

This global log would be ordered by a timestamp plus some value to ensure uniqueness (such as the id of individual stream).

Now I'm confused. What did you mean by global timestamp previously? My understanding was that it's some monotonically increasing counter like in GY EventStore or some form of the Lamport Timestamp. But seems it's not.

If operations on individual streams are forced to specify explicit versions (OCC), this ensures consistency and ordering within individual streams.

Yes, SS guarantees that for individual streams. It's fully ACID.

Stream directory. Would this be implemented using AT entity group transactions?

Nope, since AT transaction is supported only within boundaries of a single partition. Basically, stream directory is just a registry of stream addresses (partition ids), which could be read in efficient way (instead of inefficient full table scan). It could be a single partition or a sharded directory with multiple partitions.

The stream directory is required so that you can enumerate all available partitions, something you will need to have to be able to implement an aforementioned merge sort across all partitions.

I think it should be possible to do this using a simple timestamp plus a unique connection id or event the stream id + event number of the event being written

Ok, as I understand by simple timestamp you really meant a current datetime? Proposed addition of stream id + event number will fix the precision problem and make that event id globally unique, right? Seems legit, but ...

At that point I want to ask you about the use-cases you have at hands for this _all stream? If you want to use it as an integration point between bounded contexts a.k.a feed of all changes you can quickly get into a trouble due to a clock skew on different nodes (stream writers). If for something else - I'm curious what for.

[jkonecki]

@eulerfx It seems to me as well that you're not looking for a global timestamp but just for a timestamp which is already supported. Global ordering between streams could be achieved by implementing vector clock - as @yevhen mentioned synchronisation in distributed system is very difficult.

[eulerfx]

What did you mean by global timestamp previously?

I was referring to a possible implementation. It could also be a monotonic integer, but this would require a central authority to generate it. Using a wall-clock timestamp is one way to go. (I can already hear a co-worker calling out true time...) One challenge here is with commitment ordering.

Global ordering between streams could be achieved by implementing vector clock

Not sure how a vector clock would apply. I'm proposing that you don't need global ordering across different streams, but you still need a global log of all events. This will impose some order on events. This order needs to respect the ordering within individual streams, but not across streams. Suppose you've streams of the form Sn (S1, S2, S3, ...) and then events within them (S1E0, S1E1, S2E0, ...). Each event would also have a global timestamp Gn (G0, G1, G2, ...). Let G(SnEm) be the global timestamp of event m in stream n. We must have SnEi < SnEj => G(SnEi) < G(SnEj) but otherwise unconstrained. One challenge with this model of global timestamps is commitment ordering - the generation of global timestamps can be monotonic, but their commitment may fall slightly out of order. This places the burden on the consumer of the global log to allow some degree of skew, however it should be possible to bound it.

as @yevhen mentioned synchronisation in distributed system is very difficult.

Yes indeed :) we're trying to avoid it as much as possible.

Nope, since AT transaction is supported only within boundaries of a single partition.

How will the directory be written to? If not in the same partition, won't you need a non-transactional dual-write? It seems the directory write needs to be atomic, but not linearized, so something like RAMP could be used, but would require some coordination.

The stream directory is required so that you can enumerate all available partitions

Yes, although in the simplest case, the set of partitions can be fixed and hardcoded at configuration time.

If you want to use it as an integration point between bounded contexts a.k.a feed of all changes you can quickly get into a trouble due to a clock skew on different nodes (stream writers).

Yes, this is the use-case - this is the "bottled water" approach I referred to above. Yes, the skew could be an issue, but as described above, it can be bounded. (Its possible to commit G2 before G1, but this disorder should only exist for a limited amount of time, after which it can be ignored).

[jkonecki]

We must have SnEi < SnEj => G(SnEi) < G(SnEj)

I think you have the implication the wrong way round.
Also the above is not a global timestamp as it's local to the stream. Streamstone does support ordering within a single stream.

It is unclear what are you asking for: on one hand you're mentioning a global timestamp but on the other you're providing examples that do not involve a global timestamp.

Global timestamp would be described as:
G(SnEi) < G(SmEj) => SnEi < SmEj

[eulerfx]

I think you have the implication the wrong way round.

For the purpose of this discussion, I'm defining G as the global timestamp (or alternatively version), with global meaning that it is defined for all events, in all streams (within some namespace, such as all shopping cart events). The reason I wrote the implication that way I did is, we already have total ordering within individual streams and we're specifying the meaning of G based on that.

Also the above is not a global timestamp as it's local to the stream.

Not sure what you're referring to. In SnEi the i is the i-th event within stream n and yes, this is local to that stream. On the other hand, I'm defining G(SnEi) as the global timestamp of the i-th event in stream n making it the G(SnEi)-th event within the global log.

It is unclear what are you asking for: on one hand you're mentioning a global timestamp but on the other you're providing examples that do not involve a global timestamp.

What I'm asking for is an API that allows consuming all events across all streams. This API should be akin to the EventStore $all stream (but some of the guarantees can be relaxed) or the transaction log of an RDBMS. In particular, it should be possible to request a stream of events that have been committed after a specific global timestamp. This is where having a stream directory alone may be prohibitive. Global timestamp as defined here corresponds to Position in EventStore (but again, with some guarantees relaxed). How global version is actually implemented is a separate discussion, although I did sketch one guarantee it has to provide.

[Scooletz]

EventStore internally uses the physical log position for ordering all the events. When using a sharded, non open source version it correlates logs in a given time window, AFAIK, still using the log positions of singular shards. A good summary of different strategies for providing data to queries can be found at ReaderStrategy which defines how $all is handled. You can see that depending on the criteria, different strategies can be used like:

• reading through the transaction log itself
• using system stream created by build in projection

I think that your requirement of requesting a stream of event after a specific global timestamp can't be handled without an external indexing structure. One easily imagine a WorkerRole that was about to commit a batch of events but suffered from long GC/network problem. The data, marked with old timestamps would be added after sometime breaking the queries that should have read this events.

As far as I see it now, the only approach would be to have an indexing role that would scan every stream head and atomically, in the same partition write an indexing entry and bump up the indexed stream version. Detection of new streams would require full scans though.
It's not a perfect solution but provided linearizablity over all the events. What's your opinion?

[Scooletz]

One more proposal of providing such an indexer. When one creates a new stream, before storing the head and the first event, the name of the stream is stored in the indexer partition. Then, the normal Streamstone storage operations are applied. This ensures a property, that indexer does not have to scan other tables to find streams, but rather queries stream after stream directly.
I'm aware of Azure Storage Account limitations and that one account can take no more than 20kIOPS. This would mean that indexer can't update more than 20.000 per second which isn't the best performance ever.
Still, this approach clearly suffers from the need of having a separate entity, an indexer, that updates the index scanning the original storage.

[eulerfx]

I think that your requirement of requesting a stream of event after a specific global timestamp can't be handled without an external indexing structure.

It can be using AT, since it provides transactional writes of up to 100 rows in a partition. This library already makes use of this approach.

I'm aware of Azure Storage Account limitations and that one account can take no more than 20kIOPS

These limitations are being lifted in the next version. However, it is always possible to distribute across storage accounts much like you distribute across partitions.

When using a sharded, non open source version it correlates logs in a given time window, AFAIK, still using the log positions of singular shards

EventStore doesn't directly support sharding. The OSS and commercial versions ere merged a few years ago. We can use AT to mimic the functionality of the EventStore index as described above.

[yevhen]

When one creates a new stream, before storing the head and the first event, the name of the stream is stored in the indexer partition.

That's what I call stream directory.

How will the directory be written to? If not in the same partition, won't you need a non-transactional dual-write? It seems the directory write needs to be atomic, but not linearized, so something like RAMP could be used, but would require some coordination.

I don't intend to write/update any of event sequence information to the stream directory but merely store an address (table, partitionId) of a stream, so I can later do an efficient for (var stream in directory) from a single partition. Stream directory maintenance could be done via explicit provisioning as @Scooletz rightly pointed.

These limitations are being lifted in the next version.

Interesting. Source?

However, it is always possible to distribute across storage accounts much like you distribute across partitions.

Indeed, this is simple. SS includes reusable sharding function just for that.

One challenge with this model of global timestamps is commitment ordering - the generation of global timestamps can be monotonic, but their commitment may fall slightly out of order. This places the burden on the consumer of the global log to allow some degree of skew, however it should be possible to bound it.

Ye, I understand. As a consumer I might read _all stream at 11:20PM and got latest position 1120_S101_001. Then, host with skewed timestamp (-1 minute) can write an event with global timestamp 1119_S202_005. If I then ask _all stream to give me (compute) new events since 1120_S101_001 I will miss it.

Yes, the skew could be an issue, but as described above, it can be bounded. (Its possible to commit G2 before G1, but this disorder should only exist for a limited amount of time, after which it can be ignored)

We do something similar with 3rd party aggregators, reading from past cursor position and then dealing with duplicates. Which is a good workaround for indexing lags. Depending on how big is you window and how many events are generated per this timeframe, that could have severe performance impact and processing lags. But it's doable.

What I'm asking for is an API that allows consuming all events across all streams. This API should be akin to the EventStore $all stream (but some of the guarantees can be relaxed) or the transaction log of an RDBMS. In particular, it should be possible to request a stream of events that have been committed after a specific global timestamp.

I can think of 2 possible implementations, each with its own set of trade-offs:

1. Computed (virtual) stream. Just-in-time enumeration of all streams. Could be very prohibitive in a system with millions of streams to check each and every stream every time you poll _all stream.

   Possible optimization is to store the global timestamp of the first/last event of each stream in some easily queryable medium, could be an Index Table (on in the stream directory itself) and do a range query. That could greatly reduce a number of partitions which need to be read (for a merge sort).

   The downside is a possible inconsistency between index table and actual stream due to non-transactional write. But that could be easily tolerated. Another downside, is WATS partition IOPS limits. That could be mitigated by sharding of index table.

2. Pre-computed (durable) stream. This is similar (including downsides) to the Index Table approach above but will store an actual event. It's basically a global log. It could be heavily partitioned to allow parallel reads.

How global version is actually implemented is a separate discussion, although I did sketch one guarantee it has to provide.

Right, that should be tackled separately. Wall clock might be sufficient if you can control the drift window and deal with it at read time.

Anyway, this is a massive feature and I don't think I would be able to make it alone. But I do accept pull requests! 😄