Reliability & governance helpers

Last updated: October 13, 2026

Beyond produce/consume, the PHP core ships a set of optional, opt-in helpers that mechanise the reliability and governance contracts the wire spec defines. None of them touch the frozen envelope (schema_version: 1); each is a tooling layer over what the codec already carries. They are the PHP face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

BabelQueue\Idempotency\Idempotent::wrap($store, $handler) makes a handler run at most once per meta.id, even on at-least-once redelivery — skip if already seen, run then remember on success, fail-open on a missing id.

use BabelQueue\Idempotency\Idempotent;
use BabelQueue\Idempotency\InMemoryStore;

$store = new InMemoryStore(); // tests / single process
$dispatch->on('urn:babel:orders:created', Idempotent::wrap($store,
    fn (\BabelQueue\Contracts\ConsumedMessage $m) => chargeOnce($m)));

For a fleet, swap the in-memory store for a shipped persistent one — both implement the ClaimingStore interface (which extends IdempotencyStore with an atomic claim / release so two workers can’t both run an in-flight id):

use BabelQueue\Idempotency\PdoStore;
use BabelQueue\Idempotency\RedisStore;

$pdo->exec(PdoStore::ddl());          // portable CREATE TABLE (Postgres/MySQL/SQLite)
$store = new PdoStore($pdo);          // or: new RedisStore($predisClient)

See Idempotency and the store deep-dive.

Transactional outbox

BabelQueue\Outbox 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.

use BabelQueue\Outbox\Outbox;
use BabelQueue\Outbox\OutboxRelay;
use BabelQueue\Outbox\InMemoryOutboxStore;

$store  = new InMemoryOutboxStore();   // production: a PDO/ORM-backed OutboxStore
$outbox = new Outbox($store);

// inside YOUR $db->transaction(...), beside the business write — no commit of its own:
$id = $outbox->write($envelope);       // encodes via EnvelopeCodec, calls $store->save()

// later, a separate process drains the durable rows to the broker:
$result = (new OutboxRelay($transport, $store))->drain(); // ->published / ->failed

OutboxStore is the four-method contract (save / fetchUnpublished / markPublished / markFailed); the transaction boundary is yours. The relay publishes the stored bytes verbatim, so trace_id is preserved. See Transactional Outbox.

DLQ redrive & replay-bypass

BabelQueue\Redrive\Redrive::run($io, $dlq, $options) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with dryRun, select, toQueue (sandbox) and bypass options. PHP’s Transport is publish-only, so you bind a RedriveIO (reserve / ack / publish) to your broker.

use BabelQueue\Redrive\Redrive;
use BabelQueue\Redrive\RedriveOptions;
use BabelQueue\Redrive\ReplayBypass;

$result = Redrive::run($io, 'orders.dlq', new RedriveOptions(
    toQueue: 'orders.sandbox',
    bypass: true, // stamp bq-replay-bypass (needs a HeaderRedriveIO)
));

// the handler skips effects that already fired on a deliberate replay:
$handler = function (\BabelQueue\Transport\KafkaMessage $m): void {
    saveOrder($m);                                                   // idempotent core
    ReplayBypass::bypassExternalEffects($m, fn () => sendEmail($m)); // skipped on replay
};

To keep a Kafka record key through the redrive, have your RedriveIO also implement BabelQueue\Redrive\KeyedRedriveIO (publishKeyed($queue, $body, ?string $key)) and return the DLQ record’s key from pop() in an optional key entry. Redrive::run() then hands the key back on the write to the work topic and on every restore write back to the DLQ: a null key stays null, an empty string stays an empty key. An IO that does not implement it republishes keyless, as before. The bypass path (HeaderRedriveIO) carries no key.

For Kafka there is a ready-made IO, BabelQueue\Redrive\KafkaRedriveIO. It is built from a KafkaRedriveClient (the seam you implement around RdKafka\KafkaConsumer, with poll, commitOffset and seek, auto-commit off) and a keyed producer: new KafkaRedriveIO($client, $producer, topicPrefix: ''). It implements KeyedRedriveIO, so the key survives. A restore writes the consumed record back verbatim (original value bytes, headers and key) with a fresh timestamp. An ack commits only the contiguous prefix of acknowledged offsets of a partition, so a record that is still unsettled is never committed past.

Kafka commits are cumulative, so a run that stops on a failed message must not leave the group past it. An IO that implements BabelQueue\Redrive\RewindingRedriveIO (rewind($handle)) is rewound by Redrive::run(): KafkaRedriveIO seeks every partition back to its lowest unacknowledged offset. If the rewind fails too, Redrive::run() throws BabelQueue\Redrive\RedriveRecoveryFailed, whose getErrors() returns the original error and the rewind error together.

bypassExternalEffects() and isReplay() take a HasHeaders message, and the handler that wrap() returns expects one; today only KafkaMessage exposes headers (PulsarMessage does not). Where you only hold a raw header map, use ReplayBypass::isReplayHeaders($headers). ReplayBypass::wrap($effectHandler) is the decorator form for a handler that is entirely an external effect: on a replay it is skipped as a whole.

See DLQ redrive & replay-bypass.

GDPR field encryption

BabelQueue\Gdpr\Gdpr::protect() / ::unprotect() encrypt only the data leaves a schema marks x-gdpr-sensitive, in place, leaving data pure JSON and the envelope frozen. The Cipher is caller-bound (so the core pulls no crypto dependency); OpenSslCipher is the reference over ext-openssl (a suggest, not a require). Validate cleartext — protect after validation on produce, unprotect before validation on consume.

use BabelQueue\Gdpr\Gdpr;
use BabelQueue\Gdpr\OpenSslCipher;

$cipher = new OpenSslCipher($key); // or implement BabelQueue\Gdpr\Cipher over a KMS

Gdpr::protect($data, $schema, $cipher);   // producer: encrypt marked leaves
// ... encode + publish ... decode ...
try {
    Gdpr::unprotect($data, $schema, $cipher); // consumer: inverse
} catch (\BabelQueue\Gdpr\DecryptException $e) {
    // wrong key / tampered → retry / dead-letter
}

See GDPR field encryption.

OpenTelemetry (traceparent)

BabelQueue\Otel\Tracing emits publish <urn> / process <urn> spans 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. The in-repo Redis/AMQP/SQS transports carry the header; Kafka/Pulsar/STOMP producer wiring is a documented follow-up. open-telemetry/api is an optional suggest, so the core stays ext-json. See Observability.

Per-URN schema validation

BabelQueue\Schema\SchemaValidated::assert() (producer guard) / ::wrap() (consumer safety net) validate a message’s data against the JSON Schema registered for its URN, bridged from a babelqueue-registry manifest via DirProvider. See Per-URN schema validation.