Experiment in sharding to enable competing consumers pattern across multiple instances

[quezlatch]

One of the things i'd really like to be able to do is to run a subcription across mutiple instances.
This would enable things like load balancing, fault tolerance and all those other good things.
Unfortunately, doing this naively doesn't work with event sourcing as the events must be strictly
processed in order. This cannot be guaranteed if there is a free for all across multiple instances.

However, a key insight here is that in general individual streams can be treated as independent.
So long as the events in an individual stream are processed in order that should be good enough
for most cases.

We therefore have the concept of a shard. Each stream is assigned to a shard and each instance has
responsiblity for one or more shards. Shard responsibilty will be rebalanced in the cluster as instances
come and go, but essentially for each stream everything stays the same. A single instance processes
the event stream in order.

This is a pretty standard pattern. This is an attempt to put together a potential implementation in
Eventuous.

Note. It's quite possible to do a lot of this in the current architecture by using more fine
grained services.

• HTTP api service can be load balanced. turns commands into events
• Singe instance event handlers can subscribe to events
• One instance per checkpoint if that makes sense
• HTTP API service can also return projected read models

But this does mean a more complex architecture, as well as more coupling between services. Possible
means mono repos, sharing packages for data contracts, etc. And we don't always get to choose our architecture.

First some terminology:

• cluster a collection instances. Probably in kubernetes these days
• instance a member of a cluster. They may come and they may go. Especially
  if the cluster has elastic scaling
• shard a way of partitioning streams, each stream is assigned to a shard.
  The number of shards should be a multiple of the maximum number of instances in
  a cluster to ensure they are balanced across instances.
• lease shards are leased to an instance. This is to ensure that if rebalancing
  occurs there is not contention between instances for the same shard.

Implementation

This experiment will use SQL as that's what i'm most familiar with but there is nothing
that inheritently ties it to this.

Cluster membership

A new hosted service, this will:

• Register itself as a member of the cluster
• Crucially the membership has an expiry time
• Its membership is constantly renewed before it can expire
• This way we can tell if an instance has terminated
• We also regularly interrogate the membership list and notify the application if it has changed

Shard leases

A new hosted service, this will:

• Decide which shards it's servicing using Rendezvous hashing
• Revise the shards list in response to a cluster membership change
• Notify the application if the shards list has changed
• Also maintain an expiration time and constantly renew it. This protects us when a shard changes nodes
• if the shard is already ours update the expiry. If it's not only grab it when it's expired.

Event writer

• when an event is written, it should now also contain the ShardId and ShardPosition
• Makes sense to go on Messages table for SQL. Maybe ShardId goes on metadata for KurrentDB?

Subscriptions

When using SQL there are two options as i see it:

One subscription that reads the all the relevant shards in one SQL query

• offloads a lot of the complexity onto SQL

A subscription manager that maintains a list of subscriptions per shard:

• application does more
• hits sql with more queries
• though the queries are simpler
• can maybe reuse more of the existing code
• can process multiple shards in parallel
• i think it might be easier on my brain
• Note. handlers should be re-entrant if they are being reused across shards

And there's always a halfway house, where the a single query can service all the multiple
subscriptions though that sounds a bit tricky...

An added complication is the existing gap processing. This is complex and subtle so prioritise using
existing code.

SQL

Modify current schema or have new one? Tempted to modify current schema, but that does mean there is a schema migration step.
But I think it can be done it such a way that existing subscription types continue to work without migration.

• Calculate the ShardId from the StreamId using the multiplicative (Knuth) hash expression
• In SQL (CAST(StreamId AS BIGINT) * 2654435761 % 4294967296) % <number of shards>. Hopefully StreamId won't be so big as to overflow.
• Can maybe get added to check_stream SP and Streams table, even if that's denormalised.
• define a sequence for each shard
• in the append_events SP have a big CASE block to get the appropriate shard sequence. This is the shard position.
• we will use this for gap detection per shard.

Question

• Does this sketch seem like reasonable first stab?
• Is this proof of concept something that would be useful?

[linear]

EVE-249