adimiko / transactionalbox

Outbox and Inbox Pattern in .NET (scalability & fault tolerance). Ensures eventual consistency between services. (microservices, event-driven architecture, message streaming)

⭐ - The star motivates me a lot!

Transactional box is an implementation of the outbox and inbox pattern in .NET.
Ensures eventual consistency when modules need to communicate with each other over the network.

Examples of problems that occur during network communication:

• lost messages
• the same messages were processed again
• unavailable services

✨ Features

Actions

• Add a message to send to the outbox
• Add a message to publish to the outbox
• Get messages from outbox and add them to transport
• Get a message from transport and add them to the inbox
• Get a message from inbox and process it
• Get messages from inbox and process them

Storage

• InMemory (Default)
• Entity Framework (Relational)
  • Migrations
  • Distributed Lock (Based on atomic write operation, Standalone Package)
• MongoDB

Transport

• InMemory (Default)
• Apache Kafka
• RabbitMQ
• Iggy

Scalability & Fault Tolerance

• Support for multiple outbox worker instances
  • Multiple instances of the same service
  • Multiple processes in the same service
• Support for multiple inbox instances
  • Multiple instances of the same service
  • Multiple processes in the same service
• Standalone outbox worker
• Standalone inbox
• Error handling in background services
• Task throttling in background services (Long Running Jobs and Max Degree of Parallelism for Jobs)
• Dead messages
• Hook from service to outbox worker (when service add new message to outbox)
  • InMemory
  • SignalR
• When the inbox adds messages to the storage, it should notify the service that the messages should be processed

Observability

• Support for OpenTelemetry
• Outbox size
• Inbox size
• Message failure rate
• Message delivery latency
• Number of duplicated messages
• Message duplication rate

Maintenecne

• Remove processed messages from the outbox
• Remove processed messages from the inbox
• Remove expired idempotency keys
• Archive processed messages from the outbox
• Archive processed messages from the inbox
• Correlation ID

Other

• Modular package architecture
• Support for TimeProvider
• Unordered messages
• Internal high-performance logging (Source Generators)
• Execution context in Inbox
• High-performance invoking of inbox message handlers from assemblies (Compiled Lambda Expressions)
• Grouping of messages
  • Group by topic outbox messages to a single transport message from batch (better compression)
  • Adjusting optimal transport message size
• Messages serialization and deserialization
  • System.Text.Json (default)
• Messages compression and decompression
  • No compression (default)
  • Brotli
  • GZip
• Message streaming
• Package configuration using appsetings.json
• Own transport message serialier and deseralizer (with StringBuilder)
• Idempotent messages
• Keyed in memory lock (based on SemaphoreSlim and ConcurrentDictionary)
• Transport discriminator (one outbox many transport, tagged message)

🎬 Run Sample

Note

Docker is required.

Clone this repo and open TransactionalBox.sln via Visual Studio 2022. Set the TransactionalBox.Sample.WebApi as startup and then run. You should see the following view.

Have fun 😃!

🏰 Architecture

The transactional box consists of four basic components. The following diagrams show the basic flow (omits details).

Outbox

Outbox is responsible for adding messages to the storage.

Outbox Worker

Outbox worker is responsible for getting the messages from storage and adding them to the transport.

Inbox Worker

Inbox worker is responsible for getting messages from transport and adding them to the storage.

Inbox

Inbox is responsible for processing messages from the storage.

🏅 Competition '100commitow'

The project is part of the competition 100 commitow.

Topics

• Distributed lock (prevent race condition)
• Hook processing
• Concurrency control
• Scaling and parallel processing (distributed processing)
• Synchronization primitives
• Idempotency
• Retry Pattern

transactionalbox's Issues

Entity Framework Outbox Model

Shared IEntityTypeConfiguration for InboxMessage (EntityFramework)

Inbox Deserializer Configuration

|POC| OutboxWorker: EntityFramework - Scaling and parallel processing

Get message from inbox

using for DistributedLock

DistributedLock should be used with using, because when will be exception Dispose method will be executed and lock will be released.

InMemory Transport

IInboxHandler - compile handle method using LabdaExpression

Envelope Pattern

Transport discriminator

One outbox with messages for multiple transports

ITransactionalBoxLogger (Wrapper Pattern)

Create wrapper for Microsoft ILogger.

ServiceNameProvider

App service, which will contain the name of the service

Add logo to all packages

Entity Framework Inbox Model

Sequential GUID

Shared IEntityTypeConfiguration for OutboxMessage (EntityFramework)

Message expiration time

Get message from Transport

TTL for IdempotentInboxKeys (concrete message type)

Multiple messages in payload (Transport)

Custom compression & decompression

TransportMessageFactory - divides into multiple TransportMessages when the payload will be greater than the transport limit per message

• maximum expected bytes of transport message => X
• transport message bytes limit (retrun Transport package) => Y

Round_up(X/Y) = number of transport messages

Duplicate detection (Idempotent messages)

Outbox Serializer Configuration

TimeProvider abstraction

InMemory InboxStorage

InMemory OutboxStorage

BackgroundServiceBase (Problem): when there are a lot of fired tasks, not all are processed

Add message to Outbox (Storage)

Gzip (compression & decompression)

Inbox Idempotency

Prototype refactoring

Settings Validator

HostName Service

Wrap Dns.GetHostName() to app service

Place where can be used

• TransactionalBox.OutboxWorker.Kafka

Remove processed messages from outbox

Split one big message into multiple small

InboxExecutionContext

InMemory lock SemaphoreSlim(1,1)

Cancellation Tokens for Workers

Model for Transport

Outbox worker: processing in packages

OutobxWorker should get the package of messages and add them to the transport.
It should then mark as processed the messages that have been added to the transport.

ConfigureAwait(false)

Decorate BackgroundService (errors, retry, logging etc.)

Create a single logic that can be reused by background services

InboxWorker: Listening to multiple topic

Push message to Transport

Distributed lock to standalone package

New tasks when the prototype is ready

Create new tasks based on problems encountered while creating the prototype

Messages compression and decompression

Database structure optimized for performance

System.Text.Json Serializer

for JsonSerializer etc.