TgwOutboxDispatcher

Reliable, backpressure‑aware background dispatcher for the Transactional Outbox pattern in .NET 8.

This library reads events from a durable outbox (your storage) and publishes them to an event bus (your broker), using leases, bounded channels, controlled concurrency, retries, and graceful shutdown.

Why use it

• Prevents lost messages by decoupling write and publish via a transactional outbox
• Respects backpressure via bounded channels and configurable parallelism
• Safe multi‑instance operation with leasing to avoid duplicate work
• Retries with attempt tracking and failure callbacks for dead‑lettering
• Clean shutdown with in‑flight draining

For background and detailed requirements, see docs/Requirements.md.

---

Installation

The project targets .NET 8 (C# 12). Until a NuGet package is published, include the project in your solution and reference it directly:

1. Add the project to your solution:

dotnet sln add src/TgwOutboxDispatcher/TgwOutboxDispatcher.csproj

2. Reference it from your application:

dotnet add <YourProject>.csproj reference src/TgwOutboxDispatcher/TgwOutboxDispatcher.csproj

---

Core abstractions

You integrate by implementing two small interfaces and handing them to the dispatcher.

1. IOutboxStore

public interface IOutboxStore
{
    Task<IReadOnlyList<OutboxMessage>> LeaseAsync(
        int maxCount,
        TimeSpan leaseTime,
        CancellationToken ct);

    Task MarkDispatchedAsync(Guid id, CancellationToken ct);

    Task OnDispatchFailedAsync(
        Guid id,
        Exception ex,
        int nextAttemptInSeconds,
        CancellationToken ct);
}

2. IEventBus

public interface IEventBus
{
    Task PublishAsync(
        string topic,
        ReadOnlyMemory<byte> payload,
        string? dedupKey,
        CancellationToken ct);
}

3. OutboxMessage

public record OutboxMessage(
    Guid Id,
    string Topic,
    byte[] Payload,
    DateTimeOffset CreatedAt,
    int Attempt = 0,
    string? DedupKey = null);

4. OutboxDispatcherOptions

public sealed class OutboxDispatcherOptions
{
    // See Configuration
}

---

How it works (high level)

• The dispatcher periodically leases up to MaxBatchSize messages from your outbox for LeaseTime.
• Messages are pushed into a bounded channel (ChannelCapacity), applying backpressure if consumers are slow.
• Up to DegreeOfParallelism workers call your IEventBus.PublishAsync.
• On success, MarkDispatchedAsync is called so you can delete or archive the message.
• On failure, OnDispatchFailedAsync is called. You decide how/when to reschedule or dead‑letter. The library passes a suggested delay via nextAttemptInSeconds and increments Attempt on the message you return later.
• After MaxAttempts, you should stop retrying and dead‑letter (policy is yours; the interface gives you the hook).
• On shutdown, the dispatcher stops leasing, drains the channel for up to ShutdownDrainTimeout, and exits gracefully.

---

Quick start

1. Implement IOutboxStore for your persistence (e.g., SQL table with an outbox and a leasing column/lock).
2. Implement IEventBus for your broker (e.g., Kafka, RabbitMQ, Azure Service Bus, SNS/SQS, or a test stub).
3. Configure OutboxDispatcherOptions as needed.
4. Register and run the dispatcher as a hosted background service in your app.

Example sketch (simplified):

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using TgwOutboxDispatcher;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddSingleton<IOutboxStore, SqlOutboxStore>();
builder.Services.AddSingleton<IEventBus, KafkaEventBus>();
builder.Services.Configure<OutboxDispatcherOptions>();

builder.Services.AddHostedService<OutboxDispatcher>();

var app = builder.Build();
await app.RunAsync();

Note: This repo currently provides the contracts and options. The concrete dispatcher service may live in this package or a companion one depending on how the repository evolves. See docs/Requirements.md for the intended behavior.

---

Configuration

This section lists all OutboxDispatcherOptions with their purpose and default values, and shows how to configure the dispatcher via DI.

Options (defaults in parentheses):

• DegreeOfParallelism (4)
  • Number of concurrent consumer workers publishing to your IEventBus.
• ChannelCapacity (512)
  • Size of the in-memory bounded channel used between the producer loop and consumers; controls backpressure.
• MaxBatchSize (128)
  • Maximum number of messages to lease from the outbox per producer iteration.
• MaxAttempts (10)
  • Maximum total publish attempts before you should dead-letter in your store (signaled via nextAttemptInSeconds: -1).
• LeaseTime (00:02:00)
  • Visibility timeout for leased outbox rows. While the lease is active, other dispatcher instances must not see them.
• PollInterval (00:00:00.250)
  • Minimum time between two leasing attempts. The producer adapts around execution time; this is the baseline cadence when idle.
• ShutdownDrainTimeout (00:00:30)
  • Maximum time to drain in-flight work on application shutdown before forcing cancellation.
• MinRetryDelay (00:00:01)
  • Lower bound for computed retry delays (exponential backoff with full jitter never goes below this).
• MaxRetryDelay (00:05:00)
  • Upper bound for computed retry delays; caps exponential growth.
• ExponentialDelayBaseInSeconds (5)
  • Base used when calculating the exponential backoff curve;

Registering in DI:

using Microsoft.Extensions.DependencyInjection;
using TgwOutboxDispatcher;

var services = new ServiceCollection();

// Your implementations
services.AddSingleton<IOutboxStore, SqlOutboxStore>();
services.AddSingleton<IEventBus, KafkaEventBus>();

// Add the dispatcher (registers hosted service and options)
services.AddOutboxDispatcher();

// Configure options (override defaults)
services.Configure<OutboxDispatcherOptions>(o =>{ ... });

Leasing and clock skew guidance:

• Multiple instances can run concurrently and rely on LeaseTime for exclusive visibility. To avoid accidental overlap due to time differences between machines, keep cross-instance clock skew significantly smaller than LeaseTime / 2.
• Practical recommendation: enable NTP time sync and target skew well under a few seconds. For example, with LeaseTime = 2 minutes, keep skew ≪ 60 seconds (ideally < 1–2 seconds) across hosts.

---

Delivery semantics and idempotency

• The dispatcher provides at‑least‑once delivery semantics. Duplicates can occur (e.g., process restarts, timeouts, multi‑instance leasing, or broker acks raced with persistence updates).
• Downstream must ensure idempotency using the message key. Use OutboxMessage.DedupKey as the idempotency key at the broker and in consumers.
• Recommended:
  • Set DedupKey to a unique, stable identifier per logical event.
  • Forward the key in IEventBus.PublishAsync to the broker if supported (e.g., Kafka key, SQS FIFO dedup). If the broker lacks native dedup, make your consumer idempotent by tracking processed keys.

---

Outbox storage guidance

• Use the same DB transaction that performs domain changes to insert OutboxMessage rows.
• Implement leasing so that multiple app instances can safely run the dispatcher concurrently.
• Persist and increment Attempt on each retry.
• Implement exponential backoff or a similar policy in OnDispatchFailedAsync when scheduling the next attempt.

Event bus guidance

• Forward DedupKey to the broker if it supports deduplication (e.g., Kafka keys, SQS FIFO dedup).
• Ensure PublishAsync is idempotent or safe under retries.

---

Development

Prerequisites:

• .NET 10 SDK (see global.json)

Build:

dotnet build

Run tests:

dotnet test

Project layout:

• src/TgwOutboxDispatcher — contracts and options
• tests/TgwOutboxDispatcher.Tests — test project scaffold
• docs/Requirements.md — full requirements and design notes

---

Third-Party Licenses

This project uses Shouldly and Moq for unit testing. Shouldly is licensed under the BSD-3-Clause License. See LICENSE-Shouldly.txt and LICENSE-Moq.txt for details.