Propulsion

Propulsion provides a granular suite of .NET NuGet packages for building Reactive event processing pipelines. It caters for:

• Event Sourcing Reactions: Handling projections and reactions based on event feeds from stores such as EventStoreDB and MessageDB, and the Equinox Stores (DynamoStore, CosmosStore, MemoryStore).
• Unit and Integration testing support: The AwaitCompletion mechanisms in MemoryStore and FeedSource provide a clean way to structure test suites in a manner that achieves high test coverage without flaky tests or slow tests.
• Generic Ingestion and Publishing pipelines: The same abstractions can also be used for consuming and/or publishing to any target.
• Serverless event pipelines: The core components do not assume a long-lived process.
  • The DynamoStore-related components implement support for running an end-to-end event sourced system using only Amazon DynamoDB and Lambda without requiring a long-lived host process.
• Strong metrics support: Feed Sources and Projectors provide comprehensive logging and metrics. (At present, the primary integration is with Prometheus, but the mechanism is exposed in a pluggable manner).

If you're looking for a good discussion forum on these kinds of topics, look no further than the DDD-CQRS-ES Discord's #equinox channel (invite link).

Core Components

• Propulsion Implements core functionality in a channel-independent fashion. Depends on FSharp.Control.TaskSeq, MathNet.Numerics, Serilog:

  1. StreamsSink: High performance pipeline that handles parallelized event processing. Ingestion of events, and checkpointing of progress are handled asynchronously. Each aspect of the pipeline is decoupled such that it can be customized as desired.
  2. Streams.Prometheus: Helper that exposes per-scheduler metrics for Prometheus scraping.
  3. ParallelProjector: Scaled down variant of StreamsSink that does not preserve stream level ordering semantics
  4. FeedSource: Handles continual reading and checkpointing of events from a set of feeds ('tranches') of a 'source' that collectively represent a change data capture source for a given system (roughly analogous to how a CosmosDB Container presents a changefeed). A readTranches function is used to identify the Tranches (sub-feeds) on startup. The Feed Source then operates a logical reader thread per Tranche. Tranches represent content as an incrementally retrievable change feed consisting of batches of FsCodec.ITimelineEvent records. Each batch has an optional associated checkpointing callback that's triggered only when the Sink has handled all events within it.
  5. Monitor.AwaitCompletion: Enables efficient waiting for completion of reaction processing within an integration test.
  6. PeriodicSource: Handles regular crawling of an external datasource (such as a SQL database) where there is no way to save progress and then resume from that saved token (based on either the intrinsic properties of the data, or of the store itself). The source is expected to present its content as an IAsyncEnumerable of FsCodec.StreamName * FsCodec.IEventData * context. Checkpointing occurs only when all events have been deemed handled by the Sink.
  7. SinglePassFeedSource: Handles single pass loading of large datasets (such as a SQL database), completing when the full data has been ingested.
  8. JsonSource: Simple source that feeds items from a File containing JSON (such a file can be generated via eqx query -o JSONFILE from cosmos etc)

  NOTE Propulsion.Feed is a namespace within the main Propulsion package that provides helpers for checkpointed consumption of a feed of stream-based inputs.

  • Supported inputs include custom bindings (e.g. a third-party Feed API) or various other input configurations (e.g. periodically correlating with inputs from a non-streamed source such as a SQL Database).
  • Provides a generic API for checkpoint storage, with diverse implementations hosted in the sibling packages associated with each concrete store (supported stores include DynamoStore, CosmosStore, SQL Server, Postgres).
  • Using a feed normally requires a checkpoint store that inmplements IFeedCheckpointStore from e.g., Propulsion.CosmosStore|DynamoStore|MessageDb|SqlStreamStore

• Propulsion.Prometheus Provides helpers for checkpointed consumption of a feed of stream-based inputs. Provides for custom bindings (e.g. a third-party Feed API) or various other input configurations (e.g. periodically correlating with inputs from a non-streamed source such as a SQL Database). Provides a generic API for checkpoint storage, with diverse implementations hosted in the sibling packages associated with each concrete store (supported stores include DynamoStore, CosmosStore, SQL Server, Postgres). Depends on Propulsion, a IFeedCheckpointStore implementation (from e.g., Propulsion.CosmosStore|DynamoStore|MessageDb|SqlStreamStore)

  1. Propulsion.Prometheus: Exposes processing throughput statistics to Prometheus.
  2. Propulsion.Feed.Prometheus: Exposes reading statistics to Prometheus (including metrics from DynamoStore.DynamoStoreSource, EventStoreDb.EventStoreSource, MessageDb.MessageDbSource and SqlStreamStore.SqlStreamStoreSource).

• Propulsion.MemoryStore . Provides bindings to Equinox.MemoryStore. Depends on Equinox.MemoryStore v 4.0.0, FsCodec.Box, Propulsion

  1. MemoryStoreSource: Presents a Source that adapts an Equinox.MemoryStore to feed into a Propulsion.Sink. Typically used as part of an overall test suite to enable efficient and deterministic testing where reactions are relevant to a given scenario.
  2. Monitor.AwaitCompletion: Enables efficient deterministic waits for Reaction processing within integration or unit tests.
  3. ReaderCheckpoint: ephemeral checkpoint storage for Propulsion.DynamoStore/EventStoreDb/Feed/MessageDb/SqlStreamSteamStore in test contexts.

Store-specific Components

• Propulsion.CosmosStore Provides bindings to Azure CosmosDB. Depends on Equinox.CosmosStore v 4.0.0

  1. CosmosStoreSource: reading from CosmosDb's ChangeFeed using Microsoft.Azure.Cosmos
  2. CosmosStoreSink: writing to Equinox.CosmosStore v 4.0.0.
  3. CosmosStorePruner: pruning from Equinox.CosmosStore v 4.0.0.
  4. ReaderCheckpoint: checkpoint storage for Propulsion.DynamoStore/EventStoreDb/Feed/MessageDb/SqlStreamSteamStore using Equinox.CosmosStore v 4.0.0.

  (Reading and position metrics are exposed via Propulsion.CosmosStore.Prometheus)

• Propulsion.DynamoStore Provides bindings to Equinox.DynamoStore. Depends on Equinox.DynamoStore v 4.0.0

  1. AppendsIndex/AppendsEpoch: Equinox.DynamoStore aggregates that together form the DynamoStore Index
  2. DynamoStoreIndexer: writes to AppendsIndex/AppendsEpoch (used by Propulsion.DynamoStore.Indexer, Propulsion.Tool)
  3. DynamoStoreSource: reads from AppendsIndex/AppendsEpoch (see DynamoStoreIndexer)
  4. ReaderCheckpoint: checkpoint storage for Propulsion.DynamoStore/EventStoreDb/Feed/MessageDb/SqlStreamSteamStore using Equinox.DynamoStore v 4.0.0.

  (Reading and position metrics are exposed via Propulsion.Prometheus)

• Propulsion.DynamoStore.Indexer AWS Lambda to index appends into an Index Table. Depends on Propulsion.DynamoStore, Amazon.Lambda.Core, Amazon.Lambda.DynamoDBEvents, Amazon.Lambda.Serialization.SystemTextJson

  1. Handler: parses Dynamo DB Streams Source Mapping input, feeds into Propulsion.DynamoStore.DynamoStoreIndexer
  2. Connector: Store / environment variables wiring to connect DynamoStoreIndexer to the Equinox.DynamoStore Index Event Store
  3. Function: AWS Lambda Function that can be fed via a DynamoDB Streams Event Source Mapping; passes