Tyto.Transports.Postgres 0.0.1-alpha.97

Tyto.Transports.Postgres

PostgreSQL-backed transport for Tyto. Designed for teams that want a real queue (ack/nack, retry, dead-letter, competing consumers) without operating Kafka or RabbitMQ — when you already run Postgres, this reuses it as the broker.

This is not a Kafka replacement. It targets thousands-of-messages/second workloads where operational simplicity and transactional integrity matter more than raw broker throughput.

How it works

A single table acts as the queue:

• Send → INSERT a Ready row, then pg_notify to wake consumers.
• Receive → SELECT ... FOR UPDATE SKIP LOCKED leases a batch and marks it InFlight with a locked_until lease. Consumers also LISTEN for instant wake-ups and fall back to polling (PollingInterval).
• Settle → Complete deletes the row; Abandon/retry bumps attempt and delays visible_at; exceeding MaxRetryCount sets state = DeadLetter.
• Crash safety → an InFlight row whose locked_until has passed is re-leased automatically (at-least-once delivery).

The table, schema and index are created idempotently at startup by PostgresTopologyManager (set AutoCreateSchema = false to run DDL-free under least-privilege credentials).

Priority queues

Each message carries a priority (default PostgresOptions.DefaultPriority, or a per-message Tyto-Priority header). Higher values are leased first — the lease order is priority DESC, visible_at, id, backed by a matching index.

// per-message
envelope.Headers["Tyto-Priority"] = "10";
// or a transport-wide default
pg.Configure(o => o.DefaultPriority = 0);

Retries, backoff & poison messages

A failed message is settled by policy, keyed off the DB attempt column (crash-safe — no reliance on headers):

• Backoff: RetryStrategy is Exponential by default — the delay is RetryDelayMilliseconds * 2^(attempt-1), capped at MaxRetryDelayMilliseconds, with ±RetryJitterFactor random jitter to avoid retry stampedes. Set RetryStrategy = Fixed for a constant delay.
• Give up: once attempt >= MaxRetryCount the message is dead-lettered.
• Poison messages: exceptions that can never succeed on retry (validation, deserialization) can be dead-lettered immediately via NonRetryableExceptions or a custom ShouldRetry predicate — no wasted retries, no queue head-of-line blocking. The dead-letter reason records whether it was exhaustion or a non-retryable exception.

pg.Consumers(c => {
    c.MaxRetryCount = 5;
    c.RetryDelayMilliseconds = 1000;      // base delay
    c.RetryStrategy = PostgresRetryStrategy.Exponential;
    c.MaxRetryDelayMilliseconds = 30_000; // cap
    c.RetryJitterFactor = 0.2;            // ±20%

    // Poison-message protection: never retry these.
    c.NonRetryableExceptions.Add(typeof(ValidationException));
    c.NonRetryableExceptions.Add(typeof(JsonException));
    // or a predicate:
    c.ShouldRetry = ex => ex is not ArgumentException;
});

With the defaults above and base 1s: retries land at ~1s, ~2s, ~4s, ~8s … (each ±20%), until MaxRetryCount, then dead-letter.

TTL / expiry & retention

• Messages with ExpiresAt in the past are never leased and are reclaimed by the maintenance worker.
• Completed messages are deleted on Complete, so they never accumulate.
• Dead-lettered rows are purged after Maintenance.DeadLetterRetention (default 7 days; TimeSpan.Zero keeps them forever for inspection).

PostgresMaintenanceWorker runs these sweeps on Maintenance.Interval (default 5 min), keeping table bloat — and the resulting autovacuum pressure — in check.

pg.Configure(o => {
    o.Maintenance.Interval = TimeSpan.FromMinutes(2);
    o.Maintenance.DeadLetterRetention = TimeSpan.FromDays(3);
    o.Maintenance.DeleteExpiredMessages = true;
});

Observability (OpenTelemetry)

Meter and ActivitySource are both named Tyto.Transports.Postgres.

• Traces: postgres.send (producer), postgres.receive (consumer).
• Metrics: enqueued, leased, completed, abandoned, deadlettered, expired, retention.deleted counters and a processing.duration histogram (all under tyto.transports.postgres.*).

tracing.AddSource("Tyto.Transports.Postgres");
metrics.AddMeter("Tyto.Transports.Postgres");

Usage

services.AddTyto(tyto => {
    tyto.Transports(t => {
        t.AddPostgres("Postgres_Main", pg => {
            pg.Connection("Host=localhost;Database=app;Username=app;Password=secret");
            pg.Schema("public");        // optional, default "public"
            pg.Table("tyto_messages");  // optional
            pg.Consumers(c => {
                c.DefaultConcurrencyLimit = 4;
                c.MaxRetryCount = 5;
                c.RetryDelayMilliseconds = 2000;
                c.PerQueueSettings["orders"] = new() { ConcurrencyLimit = 8 };
            });
        });
    });
    // ... endpoints route to "Postgres_Main" like any other transport
});

Transactional outbox seam

IPostgresMessageStore.EnqueueAsync accepts an optional caller-owned NpgsqlConnection / NpgsqlTransaction. When supplied, the message is inserted in the same transaction as the caller's business data — the foundation for a true transactional outbox (message and state commit atomically, or not at all).

The current "publish now" path (PostgresSendingTransport) passes null, opening its own connection. When the outbox module is wired into Tyto's send pipeline, it can call the store directly with its transaction — no transport changes required.