Tyto.DeadLettering 0.0.1-alpha.94

Tyto.DeadLettering

Application-level, transport-agnostic dead-letter store for Tyto. When a transport gives up on a message — after its own retries are exhausted, or the failure is non-retryable — the message is captured into a durable, queryable store with full failure context (reason, attempts, endpoint, and the original envelope) so it can be inspected and replayed.

Model: retry-then-dead-letter. The transport owns retry/backoff (e.g. the Postgres transport's exponential backoff + poison-message rules). Only the final failure is reported here — this is not a per-attempt log.

Why

Every transport already has some notion of a failed message (RabbitMQ dead-letter exchanges, a Postgres state = DeadLetter row, a Kafka offset that never commits). Those are transport-specific, thin, and hard to query uniformly. Tyto.DeadLettering gives you one place — independent of transport — where every finally-failed message lands with:

• the reason it died and how many attempts were made,
• the exception (type / message / stack) when available,
• the full original envelope (payload Base64-encoded) so it can be replayed,
• the endpoint and transport address it came from.

Install

dotnet add package Tyto.DeadLettering
# for the EF Core store:
dotnet add package Tyto.DeadLettering.EntityFrameworkCore

Quick start

In-memory store (tests / local dev)

builder.AddTyto(tyto => {
    tyto.AddInMemoryDeadLettering(out InMemoryDeadLetterStore dlq);

    tyto.Transports(t => t.AddInMemory("memory"));
    // ... message definitions, endpoints, handlers ...
});

// Later, inspect what was dead-lettered:
foreach (DeadLetterMessage m in dlq.Messages)
    Console.WriteLine($"{m.MessageType} from {m.TransportName}:{m.Address} — {m.Reason}");

EF Core store (production)

1. Reference Tyto.DeadLettering.EntityFrameworkCore and make your DbContext implement IDeadLetterDbContext:

using Microsoft.EntityFrameworkCore;
using Tyto.DeadLettering.EntityFrameworkCore;

public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
    : DbContext(options), IDeadLetterDbContext {
    public DbSet<DeadLetterEntity> DeadLetters => Set<DeadLetterEntity>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
        => modelBuilder.ApplyDeadLetterConfiguration();
}

2. Register it (your DbContext must be registered via AddDbContext):

builder.Services.AddDbContext<AppDbContext>(o => o.UseNpgsql(connectionString));
builder.AddTyto(tyto => tyto.AddEfCoreDeadLettering<AppDbContext>());

3. Add a migration for the tyto_dead_letters table:

dotnet ef migrations add AddDeadLetters
dotnet ef database update

Custom store

Implement IDeadLetterStore and register it:

public sealed class BlobDeadLetterStore : IDeadLetterStore {
    public Task StoreAsync(DeadLetterMessage message, CancellationToken ct) {
        // write message (e.g. as JSON) to blob storage, a log, an alerting system, ...
    }
}

builder.AddTyto(tyto => tyto.AddDeadLettering<BlobDeadLetterStore>());
// AddDeadLettering<TStore>(ServiceLifetime.Scoped) — lifetime is configurable

How it fits together

Transport (retries exhausted / poison)
        │  IDeadLetterSink.DeadLetterAsync(DeadLetterContext)
        ▼
   DeadLetterSink ──► IDeadLetterStore   (InMemory | EF Core | your own)

• IDeadLetterSink (in Tyto.Abstractions) — the contract transports call at their final dead-letter point. Resolved optionally: if no store is registered it's a no-op and the transport keeps its native behaviour.
• DeadLetterSink — maps the DeadLetterContext to a DeadLetterMessage (serializing the envelope) and writes it to the configured IDeadLetterStore.
• IDeadLetterStore — pluggable persistence.

Per-transport behaviour

All four transports are wired. Configure the limit via each transport's ConsumerSettings.MaxRetryCount (a per-queue/topic override is also available).

All four capture the full failure context — reason, attempts, and the exception (type / message / stack). The failing exception flows to the settlement point through the IMessageActions.AbandonAsync(Exception?, ...) / DeadLetterAsync(Exception?, ...) overloads.

Ordering caveat: the RabbitMQ/Kafka retry re-publishes to the same destination, which trades strict ordering for transport-agnostic retry-then-dead-letter.

The stored record

public sealed record DeadLetterMessage {
    public Guid Id { get; init; }
    public string MessageId { get; init; }
    public string EndpointName { get; init; }   // logical endpoint
    public string TransportName { get; init; }  // e.g. "postgres"
    public string Address { get; init; }        // queue/topic — used for replay
    public string MessageType { get; init; }
    public string Reason { get; init; }         // why it died
    public int Attempts { get; init; }
    public string? ExceptionType { get; init; }
    public string? ExceptionMessage { get; init; }
    public string? StackTrace { get; init; }
    public string EnvelopeJson { get; init; }   // original envelope (payload Base64)
    public DateTimeOffset FailedAt { get; init; }
}

Replaying

Every record carries the full original envelope, so it can be re-delivered to the transport and address it came from.

Load the records from wherever you stored them, then hand each to IDeadLetterReplayer (registered automatically by AddDeadLettering / AddInMemoryDeadLettering):

// e.g. from the EF Core store
public sealed class ReplayService(IDeadLetterReplayer replayer, AppDbContext db) {
    public async Task ReplayAsync(Guid id, CancellationToken ct) {
        DeadLetterEntity row = await db.DeadLetters.FindAsync([id], ct)
            ?? throw new InvalidOperationException("Not found.");

        DeadLetterMessage record = new() {
            Id = row.Id, MessageId = row.MessageId, EndpointName = row.EndpointName,
            TransportName = row.TransportName, Address = row.Address, MessageType = row.MessageType,
            Reason = row.Reason, EnvelopeJson = row.EnvelopeJson,
        };

        await replayer.ReplayAsync(record, ct); // reconstructs the envelope and re-dispatches
        db.DeadLetters.Remove(row);
        await db.SaveChangesAsync(ct);
    }
}

If you need just the envelope (e.g. to inspect or transform it before resending), use record.ToEnvelope().

Projects