Architecture Note: Symfony Messenger for Asynchronous Workloads

The Problem: Decoupling Work Without Losing Control

\n

Applications that process user uploads, send notifications, or integrate with third‑party APIs often block HTTP responses while doing heavy lifting. Symfony Messenger solves this by moving work into background consumers, but the component introduces new operational boundaries: message durability, retry semantics, serialization trust, and failure visibility. This note captures the smallest viable design for a team adopting Messenger, the trust assumptions each transport demands, and the checks that keep the system honest in production.

\n\n

Requirements That Shape the Design

\n

\n
• HTTP handlers must return in <100 ms while work continues asynchronously.
\n
• Messages must survive process restarts and broker failures.
\n
• Retry behavior must be predictable and observable.
\n
• Failed messages must be inspectable and replayable without code changes.
\n
• No external broker (RabbitMQ, Kafka) for the first iteration; the team runs PostgreSQL.
\n

\n

These requirements point to the Doctrine transport as the starting point, with a clear migration path to AMQP when throughput or HA demands exceed what a database table can provide.

\n\n

Smallest Suitable Design

\n

Message and Handler

\n

Messages are plain PHP objects—no base class, no interface. Handlers are invokable services tagged messenger.message_handler. Routing is zero‑configuration: Messenger matches the message class to the handler via PHP's type system.

\n

// src/Message/ProcessUpload.php\nfinal class ProcessUpload {\n    public function __construct(\n        public readonly string $uploadId,\n        public readonly int $userId,\n        public readonly \DateTimeImmutable $receivedAt\n    ) {}\n}

\n

// src/MessageHandler/ProcessUploadHandler.php\n#[AsMessageHandler]\nfinal class ProcessUploadHandler {\n    public function __construct(\n        private UploadProcessor $processor,\n        private LoggerInterface $logger\n    ) {}\n\n    public function __invoke(ProcessUpload $message): void {\n        $this->processor->handle($message->uploadId);\n        $this->logger->info('Upload processed', ['upload_id' => $message->uploadId]);\n    }\n}

\n

The #[AsMessageHandler] attribute (Symfony 6.1+) replaces YAML/XML tagging. The handler receives the exact message instance; no envelope unwrapping is required.

\n\n

Transport: Doctrine with Explicit Serialization

\n

Default PHP serialization is unsafe for any message that might cross trust boundaries (e.g., messages produced by a separate admin panel or scheduled command). Use the JSON serializer with an explicit allow‑list.

\n

# config/packages/messenger.yaml\nframework:\n    messenger:\n        serializer:\n            default_serializer: messenger.transport.symfony_serializer\n            symfony_serializer:\n                format: json\n                context:\n                    allowed_classes:\n                        - 'App\\Message\\ProcessUpload'\n                        - 'DateTimeImmutable'\n        transports:\n            async: 'doctrine://default?queue_name=async&auto_setup=true'\n            failed: 'doctrine://default?queue_name=failed&auto_setup=true'\n        routing:\n            'App\\Message\\ProcessUpload': async\n        failure_transport: failed\n        retry_strategy:\n            max_retries: 3\n            multiplier: 2\n            max_delay: 300000 # 5 minutes\n            service: null

\n

Key settings:\n

\n
• auto_setup=true creates the messenger_messages table on first consume; disable in production migrations and run bin/console messenger:setup-transports during deploy.
\n
• queue_name isolates workloads in the same table.
\n
• retry_strategy uses exponential backoff: 1s, 2s, 4s (capped at 5 min).
\n
• failure_transport moves messages after max_retries exhausted.
\n

\n\n

Consumer Command

\n

Run in the application container or VM with the same PHP environment as the web workers.

\n

# Foreground for systemd/Supervisor\nphp bin/console messenger:consume async -vv --memory-limit=128M --time-limit=3600

\n

\n
• -vv logs each received, handled, and failed message.
\n
• --memory-limit restarts the worker before PHP OOM; tune after load testing.
\n
• --time-limit recycles the process hourly to release DB connections and clear opcode cache drift.
\n

\n\n

Trust and Data Boundaries

\n\n\n\n\n\n\n\n\n\n\n

Boundary	Assumption	Mitigation
Message producer → transport	Producer code is trusted; no external input directly creates messages.	Validate and sanitize in the controller before dispatching.
Transport → consumer	JSON payload only contains allowed classes.	Symfony Serializer throws on unexpected classes; log and alert on deserialization failures.
Consumer → handler	Handler is idempotent for retries.	Design handlers with natural keys (e.g., uploadId) and use INSERT ... ON CONFLICT or state machines.
Failed queue → ops	Failed messages retain original stamps (bus name, sent timestamp, retry count).	Replay via messenger:failed:retry only after fixing the root cause; audit replay logs.

\n

If a message must cross a trust boundary (e.g., webhook payload from a third party), add a dedicated InboundWebhook message with scalar‑only properties, validate with a dedicated validator, and map to internal messages inside a trusted handler.

\n\n

Operational Checks

\n

Daily: Queue Health

\n

# Run from app container; requires DB read access\nphp bin/console messenger:stats async failed\n# Expected: async queue near 0; failed queue 0 or documented backlog

\n

Alert if async > 1000 messages for > 5 minutes or failed > 0 without an active incident.

\n\n

Weekly: Failure Inspection

\n

# Show last 20 failed messages with exception class\nphp bin/console messenger:failed:show --limit=20 --transport=failed\n# Replay a specific message by ID after fix\nphp bin/console messenger:failed:retry 142 --transport=failed\n# Remove poison pill after root‑cause analysis\nphp bin/console messenger:failed:remove 142 --transport=failed

\n

Each command runs in the app container with the same Doctrine connection. No elevated permissions beyond doctrine:query:sql access.

\n\n

Deploy‑Time: Schema Sync

\n

# CI step before deploy\nphp bin/console messenger:setup-transports --no-interaction\n# Verifies table exists, indexes match, no pending migrations

\n\n

Failure Modes and Detection

\n\n\n\n\n\n\n\n\n\n\n\n

Mode	Symptom	Detection	Recovery
Handler throws transient exception	Message retries with backoff; appears in logs with retry_count stamp	Log metric on Messenger::RETRY stamp; alert on retry rate > 5/min	Auto‑recovers; investigate upstream dependency if sustained
Handler throws permanent exception	Message moves to failed transport after max_retries	Failed queue gauge > 0; messenger:failed:show reveals exception class	Fix code, deploy, replay via messenger:failed:retry
Consumer OOM or SIGKILL	Message stays in async with delivered_at set; redelivered after timeout	Consumer process count drops; delivered_at rows age	Restart consumer; message reprocessed automatically
Database outage	Producer dispatch() throws; HTTP 500	DB connection pool exhaustion alerts	Circuit‑breaker at producer level to return 503 fast
Duplicate processing (fan‑out)	Same message routed to multiple transports; each retries independently	Idempotency key collision in handler logs	Use single transport with multiple handlers, or deduplicate via Redis

\n\n

Conditions That Change the Design

\n

\n
• Throughput > 5k msg/min sustained: Migrate async transport to AMQP (RabbitMQ). Doctrine transport polling (default 1000 ms) and single‑table contention become bottlenecks.
\n
• Multi‑datacenter HA required: Doctrine transport has no built‑in replication. Move all transports to mirrored RabbitMQ quorum queues or Kafka.
\n
• Message ordering per aggregate: Add a ConversationIdStamp and configure a single consumer with prefetch=1.
\n
• External consumers (other languages): Switch serializer to JSON with a shared schema (OpenAPI/Protobuf). Remove PHP‑specific classes from allowed_classes.
\n
• Regulatory audit trail: Enable AuditStamp middleware that writes every envelope transition to an append‑only store.
\n

\n\n

Verification Checklist for the First Deploy

\n

\n
1. Dispatch a ProcessUpload from a controller; confirm messenger:stats async shows 1 queued.
\n
2. Run consumer with -vv; verify handler logs appear and queue drains to 0.
\n
3. Throw \\RuntimeException in handler; confirm message appears in failed after 3 retries.
\n
4. Run messenger:failed:show; verify exception class and stack trace are readable.
\n
5. Fix handler, run messenger:failed:retry <id>; confirm processing succeeds.
\n
6. Send a message with a DateTimeImmutable property; confirm JSON payload in messenger_messages.body is valid RFC3339.
\n
7. Load test: 10k messages via bin/console messenger:consume --limit=10000; monitor DB connections and table bloat.
\n

\n\n

Limitations to Accept Today

\n

\n
• Doctrine transport polling interval (1 s default) adds latency; reduce to 200 ms only after measuring DB CPU.
\n
• No built‑in priority queues; simulate with separate queue_name and dedicated consumers.
\n
• Delayed messages require a custom middleware or the symfony/messenger‑delayed‑message bridge.
\n
• Transactionality: Doctrine transport commits the dequeue and enqueue‑to‑failed in the same DB transaction only if wrapped in DoctrineTransactionMiddleware.
\n

\n\n

Practical Result Check

\n

After the first production deploy, the team should be able to answer:\n

\n
• \"How many messages are currently waiting?\" → messenger:stats in < 2 s.
\n
• \"Why did upload #442 fail?\" → messenger:failed:show 442 shows exception and context.
\n
• \"Can we replay last night's failures?\" → messenger:failed:retry --all.
\n
• \"Is the consumer alive?\" → Systemd/Supervisor status + delivered_at freshness query.
\n

\nIf any answer requires SSH + grep, the observability gap is the next architecture decision.