Critter Stack Event Sourcing Game Plan for 2023

[jeremydmiller]

Daemon & Event Store Improvements

Warning folks upfront that some of this will be part of the forthcoming "Critter Stack" paid product on top of Marten/Wolverine

Problems

• Too many DB connections, especially when both Marten & Wolverine are polling the database for their respective async daemon & durability agent
• Azure Postgresql hosting puts a cap on maximum concurrent database connections to the same server. That kills folks using a database per tenant strategy on one server
• Blue/green deployment. What if new event types are introduced?
• How to handler inline projection rebuilds?
• Can we do automatic projection rebuilds with zero downtime?

Potential Remedies

• MemoryPack or other alternative, higher performance serialization alternatives for only the event store. Would mean that Linq querying within event bodies is not longer supported
• In the daemon, share pages of events across multiple projection shard agents. This could dramatically shave off serialization overhead, but does limit us a bit for locking the different projections together.
• Collect/debounce/accumulate database changes in the projections. Try to find ways to make fewer database round trips to the database
• Try to batch database lookups -- including aggregated document lookup -- between projections to again reduce the amount of database calls
• Batch "event enrichment" in async projections. Got an N+1 query issue right now if folks use IQuerySession within projections to look up additional data
• Introduce 2nd level caching in aggregation projections with some kind of most recently used cache so that Marten does not have to make so many database round trips and serialization
• Npgsql Prepare() command. Not sure we do this today
• Introduce database sharding through Weasel and on through Marten. Shard the event store by Achived = 0 or 1. Encourage folks to be aggressive with archiving for cold v. hot storage

Questions

• Is the new NpgsqlDataSource worth anything for us? Can it be used across databases maybe?
• For the issue with the hard cap by server, would it be possible or advantageous to use foreign data wrappers within one "master" db to parallelize some queries? Or just to reuse connections?

Daemon Changes

• The existing Daemon only supports pure "hot/cold" leadership where all processing happens on one single node
• The HighWaterAgent that detects new event activity runs consistently
• Each "shard agent" which today is a single agent per async projection, is responsible for its own
  event fetching. This might still be valuable if there's predominantly projections that only use a subset of the projections being captured or if there's very little sharing of events between
  projections. If multiple projections are using the same events in different ways, this may be introducing a lot of overhead
• In the future, opt into a mode where the there's only one event fetcher & async daemon for all the projections running for a single tenant database.
• Try to batch up database updates from all projections in one batch
• Maybe try to find a way to batch up event fetching and the high water detection query to avoid round trips and soften the usage of database connections.

Inline to Snapshotting

Today, the FetchForWriting() functionality is built for aggregating streams, either for a "write" or "read" model. FetchForWriting() assumes that the projection is either "Inline" or "Live". Instead, what if folks use the Async lifecycle as a default, and the FetchForWriting() works by efficiently starting from the currently saved version and applying any events with newer versions than the current snapshot. This would go a long way toward effecting "zero downtime" rebuilds at least for aggregate by stream scenarios.

Tasks:

1. Add a EventVersion number to the document storage table for the aggregate projection
2. Set the EventVersion value to the latest stream version when updating
3. Strongly recommend folks use FetchForWriting() when working w/ per stream aggregates
4. Async/snapshotted version of FetchForWriting() that fetches both any snapshotted version of an aggregate and any events for the stream applied after the snapshotted version and aggregates live from that baseline

Concept: Combine Marten Async Daemon and Wolverine Durability Agent

Both the Wolverine "Durability Agent" and the Marten "Async Daemon" today do a lot of polling
on every single application node. Marten would be independently polling on each tenant database while
wolverine is not yet able to handle multiple databases.

• Combine durability agent operations like trying to recover dormant nodes or deleting expired envelopes into a batched call to the database. Maybe use Marten's IStorageOperation model to batch up
  commands and the response to them. Might move that interface and some generic handling code for that to Weasel so that it could also be used w/ Sql Server some day
• Only allow the durability agent & the async daemon high water agent to run on one single node. Maybe that's modeled as "system controller" / "database controller" / "worker node"
• Might be best to be "sticky" for database operations to certain nodes to try to reduce the connection count. Talking specifically about having multiple databases here
• Need to have a way to communicate through the network of application nodes. Re-introduce the idea of tracked nodes from FubuTransportation where a new node inserts a record into a shared DB table announcing itself, and also some way (http endpoint? Rabbit MQ queue?) to message that specific node.
• Try to use Wolverine's error handling model instead of the one-off error handling in the current async daemon
• Might want a Postgresql backed queue mechanism for Wolverine just so you don't absolutely have to introduce any other infrastructure to communicate
• Try to move some of the projection building work inside of Wolverine handlers for the Wolverine diagnostics, instrumentation, and error handling
• Expose metrics about daemon throughput piggy-backing off of Wolverine's metrics
• Try to reduce polling in non-leader durability/daemon agents

Leadership Election

• Possible use a database sequence for every new application node on start up, then always the leader
  is the node with the lowest number -- but the deployed application version matters here, such that a new node coming up with a higher app version takes precedence. That might give us a blue/green deployment strategy

• If there's a way to do request/reply messaging with each individual node, use something off the
  shelf to do a bully algorithm or Raft algorithm. But we are not writing our own leadership protocol unless it's dirt simple

Improved Event Projection Rebuilds & Blue/green deployment

Projection Versioning

1. Add a Revision : int property to both IProjection & IProjectionSource
2. New table (or Marten document) should store each permutation of active node (see Wolverine enhancements below), projection name, and projection version
3. Use the projection version as part of the table/function naming for a projected document such that older and newer versions could run
   simultaneously
4. If a new, async projection version is detected, spin up that projection in a new "catch up" model
5. Also store history of what projections and projection versions have ever existed such that when no active nodes are running with that version, that projection version is permanently deleted. Assume some grace period rather than doing this on node closing.

"Catch Up" Projection Mode

When a projection (like a new projection version) is known to be way behind the current high water mark, some types of projections could
be in a "catch up" mode. For any projection that is aggregated by stream, what we do in "catch up" mode is to:

• Work stream by stream so that you can make fewer database calls and fetch all the events for a singular stream (or handful of streams) rather than naively working event by event across streams
• Try to start with the streams that were most recently updated first under the assumption that these streams are being worked on by the system

Introduce formal concept of "time-sliced" projections

It would clearly help to have some concept of "time-sliced" projections across streams that would lead to:

• Helpers for event slicing based on some date within the event, but this would work WAY better by using the Marten Timestamp

Wolverine Enhancements

• Node storage & "control" endpoints for each Node. We had this years ago w/ FubuTransportation. Easy with Rabbit MQ. Could be done w/ HTTP too. Need to figure out how to have transient queues in other transports

• Maybe a Postgresql backed queue? One is already in flight at least.

• HTTP message transport? Had it early in Jasper, could be done now w/ Wolverine.HTTP

• The Node storage including application versioning

• Generalized leadership election

• Durable inbox/outbox needs to be able to span multiple databases (somewhat big)