Reliability & governance helpers

Beyond produce/consume, BabelQueue.Core ships optional, zero-dependency helpers that mechanise the reliability and governance contracts the wire spec defines. None touch the frozen envelope (schema_version: 1); each is a tooling layer over the codec. They are the .NET face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

Idempotency.Wrap(store, handler) (namespace BabelQueue) makes a handler run at most once per meta.id, even on at-least-once redelivery. The store contract is synchronous:

using BabelQueue;

IIdempotencyStore store = new InMemoryStore(); // tests / single process
Handler guarded = Idempotency.Wrap(store, handler);

The in-memory InMemoryStore is the reference; for a fleet, implement IIdempotencyStore (Seen / Remember / Forget) over a shared backend (no persistent store ships in core yet — bring your own). See Idempotency and the store deep-dive.

Transactional outbox

The BabelQueue.Outbox namespace removes the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows verbatim. It is async throughout (CancellationToken-threaded):

using BabelQueue.Outbox;

IOutboxStore store = new InMemoryOutboxStore(); // production: an ADO.NET-backed IOutboxStore
var outbox = new Outbox(store);

// inside YOUR DB transaction, beside the business write — no commit of its own:
string id = await outbox.WriteAsync(envelope, ct); // encodes via EnvelopeCodec, calls SaveAsync

// later, a relay drains the durable rows to the broker:
OutboxRelayResult res = await new OutboxRelay(publisher, store).DrainAsync(0, ct);
// res.Published / res.Failed

IOutboxStore is the four-method async contract (SaveAsync / FetchUnpublishedAsync / MarkPublishedAsync / MarkFailedAsync); the relay forwards through the OutboxPublisher delegate ((body, queue, ct)). The transaction boundary is yours. (Under using BabelQueue.Outbox; the writer type Outbox is disambiguated from the same-named namespace with a using alias — the standard C# pattern.) See Transactional Outbox.

DLQ redrive & replay-bypass

Redrive.RedriveAsync(transport, dlq, options) (namespace BabelQueue) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with ToQueue (sandbox), Max, DryRun, Select and Bypass. The Replay guard takes the delivered headers explicitly and skips effects that already fired:

using BabelQueue;

public async Task Handle(IDictionary<string, object?> data, IReadOnlyDictionary<string, string> headers)
{
    SaveOrder(data);                                                  // idempotent core
    await Replay.BypassExternalEffectsAsync(headers, () => SendEmailAsync(data)); // skipped on replay
}

Redrive with Bypass = true stamps the bq-replay-bypass marker through an IHeaderPublisher transport. See DLQ redrive & replay-bypass.

GDPR field encryption

Gdpr.Protect() / Gdpr.Unprotect() (namespace BabelQueue.Gdpr) encrypt only the data leaves a schema marks x-gdpr-sensitive, in place. ICipher is caller-bound; AesGcmCipher (on the in-box System.Security.Cryptography.AesGcm) is the reference. Validate cleartext — protect after validation on produce, unprotect before validation on consume.

using BabelQueue.Gdpr;

ICipher cipher = new AesGcmCipher(key); // or implement ICipher over a KMS

Gdpr.Protect(data, schema, cipher);     // producer: encrypt marked leaves
try
{
    Gdpr.Unprotect(data, schema, cipher); // consumer: inverse
}
catch (ProtectedFieldException)
{
    // wrong key / tampered → retry / dead-letter
}

BabelQueue.Schema.SchemaSensitivity.SensitivePaths(schema) exposes the marked leaves directly. See GDPR field encryption.

OpenTelemetry (traceparent)

BabelQueue.Tracing.Telemetry emits publish <urn> / process <urn> activities and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. Telemetry.Wrap(handler, headers) and the header-aware Telemetry.PublishAsync(…, headers, …) overloads are the entry points; Traceparent.Inject / Traceparent.RemoteParentFromHeaders do the W3C work. It is built only on the in-box System.Diagnostics.Activity, so the core stays zero-dependency. The SQS, Redis and MassTransit transports carry the header. See Observability.

Per-URN schema validation

The core validator (producer guard + consumer wrap, via MapProvider) validates a message’s data against the JSON Schema registered for its URN. See Per-URN schema validation.