Appouse.Outbox.Core 0.1.0-alpha

Appouse.Outbox

Transactional outbox for .NET, on SQL Server. Business data and the message are written in one transaction, so a message is never lost and never appears without the work that produced it.

Ordering is per partition: within a PartitionKey, messages are delivered strictly in sequence, and a message that has not reached a terminal state holds the ones behind it. Delivery is claimed under a fencing token, so a second worker cannot acknowledge work that a stalled first worker still holds.

0.1.0-alpha. The storage engine is complete and covered by 20 integration tests against a real SQL Server. The hosting layer is not written yet — see What is not here before you take a dependency.

Install

dotnet add package Appouse.Outbox.SqlServer --prerelease

Appouse.Outbox.Abstractions, .Core and .Storage.Relational come with it.

Getting started

Three steps, in this order. Skipping the second is the most common mistake.

var connectionString = "Server=.;Database=Shop;Integrated Security=true;TrustServerCertificate=true";

// 1. Create or upgrade the schema. Idempotent and safe to run from every instance at startup.
await new SqlServerMigrator(connectionString).MigrateAsync();

var store = new RelationalMessageStore(
    new SqlServerConnectionFactory(connectionString),
    new SqlServerDialect());

// 2. Register the message types you will send. Also idempotent — this is an upsert, so it belongs
//    in startup next to the migration. Enqueueing an unregistered type is rejected.
await ((IMessageTypeAdmin)store).RegisterAsync(new[]
{
    new MessageTypeRegistration("ORDER_CREATED", MessageDirectionScope.Outbound),
    new MessageTypeRegistration("ORDER_CANCELLED", MessageDirectionScope.Outbound)
    {
        MaxAttempts = 3,
        FailurePolicy = MessageFailurePolicy.SkipAndContinue,
    },
});

Writing a message with your business data

Pass your own transaction and the message is committed with it, or not at all.

await using var connection = new SqlConnection(connectionString);
await connection.OpenAsync();
await using var transaction = await connection.BeginTransactionAsync();

await SaveOrderAsync(connection, transaction, order);          // your work

await store.AppendOutboundAsync(
    new IDocEnvelope
    {
        MessageId = Guid.NewGuid(),
        MessageType = "ORDER_CREATED",
        SchemaVersion = 1,
        CorrelationId = order.Id.ToString(),
        PartitionKey = $"customer:{order.CustomerId}",         // ordering scope
        SenderPartner = new PartnerRef("SHOP"),
        ReceiverPartner = new PartnerRef("ERP"),
        Direction = MessageDirection.Outbound,
        ContentType = "application/json",
        PayloadFormat = PayloadFormat.Typed,
        Payload = JsonSerializer.SerializeToUtf8Bytes(order),
    },
    (DbTransaction)transaction,
    CancellationToken.None);

await transaction.CommitAsync();

Delivering it

var batch = await store.ClaimMessageBatchAsync(
    MessageDirection.Outbound, instanceId: Environment.MachineName,
    batchSize: 32, leaseSeconds: 120, ct);

foreach (var message in batch)
{
    try
    {
        await SendToPartnerAsync(message, ct);
        await store.CompleteAsync(message.MessageId, message.LeaseToken, instanceId, ct);
    }
    catch (Exception ex)
    {
        await store.FailAsync(
            message.MessageId, message.LeaseToken, instanceId,
            ErrorClass.Transient, ex.Message, internalCategory: null, ct);
    }
}

ErrorClass.Transient reschedules with exponential backoff and jitter, up to MaxAttempts. ErrorClass.Permanent dead-letters immediately. What happens to the rest of the partition is the message type's FailurePolicy: BlockPartition stops it (nothing behind a failed message gets delivered out of order), SkipAndContinue cancels the message and lets the partition drain.

Run ReapExpiredLeasesAsync on a timer: it returns messages whose worker died to the retry queue without counting a failure against them.

Delaying a message

Set NotBefore on the envelope. The message keeps its place in the partition and becomes claimable when the time arrives.

NotBefore = DateTimeOffset.UtcNow.AddMinutes(30),

Business-key uniqueness

There is deliberately no unique index on (SenderPartnerId, ReceiverPartnerId, MessageType, CorrelationId). CorrelationId is a trace correlation and is normally the same across every message of one workflow, so such an index rejects the second message of any conversation.

If your domain does have a business key that must be unique, add it in your own migration with the columns that actually identify it:

CREATE UNIQUE NONCLUSTERED INDEX UX_Messages_MyBusinessKey
    ON comm.Messages (SenderPartnerId, ReceiverPartnerId, MessageType, CorrelationId)
    WHERE CorrelationId <> '';

Duplicate-key violations are classified as ErrorClass.Permanent, so an offending message dead-letters rather than retrying forever.

What is not here

0.1.0-alpha ships the storage engine and the SQL Server dialect. Not yet written:

• A dispatcher. There is no hosted service; the claim/complete loop above is yours to run.
• IIDocSender. The interface is published as the planned high-level API, but no implementation ships in this version. Use IMessageStore directly.
• PostgreSQL and Oracle. RelationalMessageStore is engine-agnostic and IDbDialect is the only seam, but only SqlServerDialect exists today.
• A DI integration package. Construct the store yourself, as shown above.

The public surface of IMessageStore is intended to be stable across those additions.

Requirements

SQL Server 2019 or later. The schema lives in the comm schema and the migrator needs rights to create it. sp_getapplock serialises concurrent startup, so every instance may safely migrate.

License

MIT.