Reliability & governance helpers

The Symfony adapter is a thin Messenger serializer on the framework-agnostic PHP core (babelqueue/php-sdk). It adds two Symfony-native pieces of wiring — an idempotency middleware and a trace-propagation middleware — and otherwise re-exposes the core’s optional reliability and governance helpers, which you apply at produce/consume time. None touch the frozen envelope (schema_version: 1); Messenger’s worker, retry and failure transport stay as they are. They are the Symfony face of the cross-SDK spec — follow each linked spec page for the full contract, and the php-sdk reference for the complete core API.

Idempotency (native middleware)

The adapter ships BabelQueue\Symfony\Messenger\IdempotencyMiddleware — the Messenger-idiomatic wrapper around the core’s Idempotent::wrap / ClaimingDispatch. It deduplicates a redelivered message on its canonical meta.id (carried on a BabelMessageIdStamp the serializer attaches on decode), so a handler runs once per logical message under at-least-once delivery. It acts only on the receive path; a fresh outbound dispatch passes straight through.

Enable it in the bundle config and register it on the consuming bus, before the handler middleware:

# config/packages/babelqueue.yaml
babelqueue:
    idempotency:
        enabled: true
        store: App\Babel\RedisIdempotencyStore   # a BabelQueue\Idempotency\IdempotencyStore service
        ttl: 3600                                  # in-flight claim TTL (ClaimingStore only)

framework:
    messenger:
        buses:
            messenger.bus.default:
                middleware:
                    - 'babelqueue.messenger.idempotency_middleware'

With a plain IdempotencyStore the middleware is post-success seen-set dedupe; with a ClaimingStore (a shared, persistent backend — the core’s PdoStore / RedisStore) it uses the stronger claim/commit/release lifecycle, so exactly one of N concurrent deliveries wins. A delivery that loses to an in-flight peer throws ClaimParkedException so Messenger does not ack it — the broker redelivers it later, by when the winner has committed. Left unset, store falls back to a bundled in-memory store (single-process / tests only). See Idempotency and the store deep-dive.

Transactional outbox

The adapter has no outbox of its own — use the core BabelQueue\Outbox helpers to remove the dual write between your Doctrine write and the publish. Persist the encoded envelope in the same DB transaction as your business row, then drain it with a relay (a messenger:consume worker on a dedicated bus, or a scheduled command):

use BabelQueue\Outbox\Outbox;

$em->wrapInTransaction(function () use ($order, $outbox, $envelope) {
    $em->persist($order);        // business row
    $outbox->write($envelope);   // the message, same transaction — no commit of its own
});

Outbox, OutboxRelay and the four-method OutboxStore (save / fetchUnpublished / markPublished / markFailed) live in BabelQueue\Outbox; back the store with a Doctrine DBAL adapter whose save() runs on the open transaction. See Transactional Outbox.

DLQ redrive & replay-bypass

BabelQueue\Redrive\Redrive::run($io, $dlq, $options) moves dead-lettered messages back onto a queue (reset, dry-run, sandbox toQueue, select, bypass) — typically from a console command bound to your broker. The core ReplayBypass guard lets a deliberate replay re-run the idempotent core while skipping effects that already fired:

use BabelQueue\Redrive\ReplayBypass;

ReplayBypass::bypassExternalEffects($message, fn () => $this->sendEmail($data));

See DLQ redrive & replay-bypass.

BabelQueue\Gdpr\Gdpr::protect() / ::unprotect() encrypt only the data leaves a schema marks x-gdpr-sensitive, leaving data pure JSON and the envelope frozen. OpenSslCipher (over ext-openssl) is the reference; bind your own Cipher to a KMS in production. Protect before the message is dispatched/serialized, unprotect at the top of the handler — and validate cleartext on both sides. See GDPR field encryption.

OpenTelemetry (`traceparent`)

The adapter’s TracePropagationMiddleware already forwards an inbound trace_id onto any message a handler dispatches, keeping a chain of work in one trace (the v0.1 floor). The core BabelQueue\Otel\Tracing adds publish <urn> / process <urn> spans and, on a header-carrying transport (Redis/AMQP/SQS), 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. 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 — run bqschema (including the gdpr --require PII audit) in CI. See Per-URN schema validation.