Download the PHP package kraz/messenger-workflow without Composer

MessengerWorkflow

[!WARNING] This Symfony bundle is a proof of concept. Using it in production is not recommended!

A Symfony bundle implementing Enterprise Integration Patterns on top of Symfony Messenger: CQRS message buses, transactional outbox/inbox, tracked async tasks and RabbitMQ integration. It is the messaging backbone for a modular monolith whose bounded contexts (modules) communicate only asynchronously — so they can later be split into independently deployed applications without touching application code.

Upgrading from 0.2.x? Read UPGRADE-0.3.md.

Message model

• Commands — change state, fire-and-forget, exactly one handler. Optionally tracked: dispatching with a by-reference task-id argument makes the result awaitable/pollable by UUID.
• Queries — return a result, no side effects, exactly one handler. Always tracked (the result storage is their return channel).
• Domain events — immutable notifications about domain changes. Zero or more handlers, fanned out per bounded context via topic bindings.

Every message goes through the message broker — there is no synchronous in-process handling, so behavior is identical whether the handler lives in the same deployment or in another one.

Flows

Segments in [...] are optional per configuration — see Reducing the flow. ⇄ marks the at-least-once hand-offs: a failed hand-off keeps the message on the safe side.

Every flow variant is traced end to end — configuration, derived workers, the exact worker commands and what happens at each hop — in MESSAGE_FLOWS.md.

Features

• Transactional outbox (*-outbox:// transports): messages are stored with your domain changes in one database transaction and relayed to RabbitMQ by a publisher worker — at-least-once delivery even when the broker is down. FIFO by auto-increment id; PostgreSQL LISTEN/NOTIFY wakes idle relays instantly. Awaiting a task whose message still sits in your own uncommitted transaction throws PendingOutboxMessageException instead of deadlocking.
• Inbox with deduplication (*-inbox:// transports): a receiver worker moves broker messages into the handling context's database; a message UUID is handled at most once — broker redeliveries of already-processed UUIDs are dropped, even when the processing had failed permanently. Handlers run inside the transaction that deletes the inbox row, so application writes on the same connection commit atomically with the message consumption. Competing consumers via FOR UPDATE SKIP LOCKED (commands default) or single-consumer FIFO (events default); long-running handlers stay protected from redelivery via messenger:consume --keepalive.
• Tracked tasks: tracked command results and all query results land in the result storage (Redis in production, in-memory for tests/dev) under rs:[<ns>:]<uuid>. Optional task services add ownership records, a status provider (pending|completed|failed, plus unknown for a result that exists but cannot be decoded — never misreported as success) and a non-blocking result provider.
• Opinionated retry policies per flow (commands / queries / events) with transient-error detection, exception-class lists, custom decider services, exponential backoff + jitter and bounded total budgets; failure transports (DLQ) for commands and events.
• Worker auto-derivation: the required workers (publisher, receiver, handler, notifier, query handler) are derived from the transport topology; messenger:supervisor-config renders the supervisord configuration (stdout or --output-dir). Manual worker declarations override the derived set.
• RabbitMQ integration on the stock jwage/phpamqplib-messenger transport (no subclassing): direct exchanges for commands/queries, topic exchange for events; routing keys derived from the message FQCN (Contracts\* = public contract, everything else internal.<Context>); queue binding keys derived from the queue owner and the #[AsEventHandler(fromTransport: ...)] handler declarations.
• Multi-database support: one database per bounded context; the DSN host of every outbox/inbox transport is the Doctrine DBAL connection name, so the owning database is always inferable from the transport.

Installation

Requires PHP ≥ 8.4, Symfony ^8.1, ext-redis, PostgreSQL, RabbitMQ and Redis. Register the bundles:

Configuration

The bundle prepends the three buses (command.bus, query.bus, event.bus), the broker transports (commands, queries, events — you supply the DSNs), the marker-interface routing and the serializer defaults. A typical bounded context ("book_store", own database) adds:

Provision the broker topology and the database tables once per deploy:

The full configuration shape (all keys, defaults and descriptions) is generated into config/reference.php.

Full default configuration

All keys of the messenger_workflow extension with their default values and inlined comments (cross-check anytime with bin/console config:dump-reference messenger_workflow):

Transport DSN options

Usage

Messages are plain classes implementing the marker interfaces Application\CommandInterface, Application\QueryInterface or Domain\DomainEventInterface. Handlers:

The attributes also work on methods, and a handler method may declare extra parameters after the message — they are resolved from the container with full autowiring semantics (#[Autowire]/#[Target] included; nullable parameters degrade to null when no service matches, defaults are kept). This lets a controller class host its feature's handler without constructor injection:

(Batch handlers keep Symfony's native ($message, Acknowledger) signature and are never wrapped.)

Domain events are published through the context's outbox bus (atomic with the domain transaction) or directly to the event bus (immediate, less resilient):

Task status and results (e.g. for a polling HTTP endpoint), with tasks.enabled: true:

Workers

Every flow segment is executed by messenger:consume workers under supervisord. The worker set is derived automatically from the transport topology — for the configuration above: an events publisher (relay), receiver + handler pairs for the commands and events queues, a commands notifier and a query handler. Inspect/override under messenger_workflow.messenger.workflow:

Manual entries merge with the derived set by name or type|source|queue identity; enabled: false removes a worker; labels carries free-form metadata. cmd_extra_options.keepalive renders messenger:consume --keepalive=<s>, protecting a long-running handler on a competing-consumer source from mid-flight redelivery (keep it below the transport's redeliver_timeout; single-consumer sources ignore the in-flight marker entirely). instances > 1 requires a competing-consumer source — on a single-consumer transport the container fails at compile time, because every process would handle the same messages.

cmd_extra_options.sleep renders messenger:consume --sleep=<s>, the idle pause between polls. The derived defaults are already tuned per transport kind and rarely need overriding: broker-queue workers (receivers, query handlers, no-inbox handlers) get sleep: 0 because the AMQP consumer blocks on the broker socket — no polling loop exists to pace. PostgreSQL-backed workers (publishers, notifiers, inbox handlers) keep the messenger default (1 s), but their idle pacing really comes from LISTEN/NOTIFY: the worker blocks query-free until a notification or the get_notify_timeout wake-up, so lowering sleep there buys nothing. One combination to avoid: sleep: 0 on a worker that consumes a PostgreSQL transport together with another transport in one messenger:consume — with mixed transports the NOTIFY wait is capped to the worker's sleep to keep the sibling transport polled, and a zero cap disables the wait entirely, leaving an unpaced loop of empty polls against the database. Keep the derived one-transport-per-worker shape and sleep: 0 stays where it belongs — on AMQP-only workers.

Generate the supervisord config:

Reducing the flow

The optional segments can be removed per context — each removal is an informed trade-off:

All reductions are exercised by the integration suite (tests/Integration/Flow/ReducedFlowTest.php) and walked through hop by hop in MESSAGE_FLOWS.md.

Ordered event delivery

Events are FIFO end-to-end when every segment has a single consumer: the outbox relay is single-consumer by design, and the events inbox defaults to single-consumer FIFO. A retrying (poison) message blocks its ordered queue only for its bounded retry budget, then drains to the failure transport and the queue resumes. Declaring strict_order=true on a transport documents the intent and makes conflicting multiple_consumers configuration fail at boot (container compile time).

Retry backoff delays apply on the inbox hop too: a retry redelivery stamps the row with an available_at. In single-consumer FIFO mode a message in backoff blocks its successors (ordering is preserved — the queue waits, bounded by the flow's total retry budget); in competing-consumer mode the row is simply skipped until due. On PostgreSQL a worker sleeping on LISTEN/NOTIFY picks a due retry up at the next idle wake-up — governed by get_notify_timeout (default 60 s) with the idle listener active (the normal bundle mode), or by the check_delayed_interval re-poll (default 60 s) when the transport is wired standalone without the listener. Lower the matching interval on transports where precise backoff timing matters.

The unit of work at the transaction boundary

WorkflowTransactionMiddleware opens the transaction around a received message — and closes the unit of work inside it. Before committing, every open ORM entity manager running on that transaction's connection is flushed, so an ORM application gets a real unit of work per message:

A handler that flushes itself is unaffected — the boundary flush then finds nothing to do. The write happens before the inbox row is removed, so a failing flush leaves the message in the inbox and the whole transaction rolls back: the message is retried, never silently lost.

Managers are selected by connection name, resolved at container compile time: the inbox transport and the entity managers both name their connection in configuration and resolve the same doctrine.dbal.<name>_connection service, so the compiled connection name → entity managers map is exactly the set of managers taking part in this message's transaction. Per message that is a single hash lookup — with dozens of bounded contexts, the other contexts' managers are never instantiated or scanned. Other bounded contexts, and managers running their own transactions (a projection writer, say), are untouched.

Two guards catch the misconfiguration this map cannot repair — handlers writing through an entity manager on a different connection than the message transaction (such writes could never be atomic with the inbox-row removal):

• Compile time: a transactional inbox transport whose connection has no entity manager is flagged in the container's compiler log (fine for a DBAL-only context, suspicious if that context uses the ORM).
• Debug mode (kernel.debug): a handler that leaves scheduled (persisted/removed, never flushed) entities in a manager outside the transaction fails the message with a descriptive LogicException instead of silently dropping the writes on commit.

Only received messages are wrapped, which is also what makes the flush safe to do here: a sender-side dispatch() carries no ReceivedStamp and returns before reaching the flush — including the outbox's own dispatch, which happens from inside a flush and would otherwise re-enter one.

Turn it off to leave writing entirely to the application:

Failure transports (DLQ) and replays

Inbox deduplication covers broker redelivery only: a message UUID already recorded as processed is dropped when RabbitMQ delivers it again. An operator replay (messenger:failed:retry) deliberately bypasses that record and re-executes the handler — that is the escape hatch for the crash window where a message was marked processed but its side effects were lost (e.g. a non-transactional handler crashing between the application write and the inbox ack). The operational consequences:

• Handlers behind an inbox must stay idempotent under operator replay — replaying an already-applied event otherwise re-applies it (a projector may resurrect a deleted row).
• transactional_handler=true commits the application writes atomically with the inbox-row removal (same DBAL connection), closing the crash window that makes replays necessary — and with it most of the double-execution risk.
• Replay order is the failure transport's, not the original queue order — relevant for strictly-ordered event streams.

Why replays do not consult the dedup index (decision, 0.4): the index marks a message UUID processed on both terminal outcomes — successful ack and permanent rejection — so every message sitting in a failure transport is already marked processed, and a dedup check on replay would block all replays. More fundamentally, no marker can distinguish "failed before applying its side effects" from "failed after": whether a replay is safe is a property of the handler, not of the message's delivery history. Idempotent handlers and transactional_handler=true are the supported mechanisms; the bypass is deliberate and will stay.

Testing

The PHPUnit suite (unit + functional + integration) expects live local infrastructure for the integration groups:

• RabbitMQ v3.13 on localhost:5672 (management console on :15672), user guest / password guest
• Redis v8.4 on localhost:6379, password xxx (tests use db 15)
• PostgreSQL v18.4 on localhost:5432, user test / password test (tests skip without pdo_pgsql)

Connection overrides: MWF_TEST_PG_*, MWF_TEST_REDIS_*, MWF_TEST_AMQP_DSN (defaults in phpunit.xml.dist).

License

This library is licensed under the MIT License. See the LICENSE file for details.