Messenger, and the queue that is part of the framework

The job class knew which queue it went on, which broker to use and how many times to retry, because there was nowhere else to put any of it. Moving a job to a different transport meant editing the class, and testing the handler meant a broker. 4.3 in May made Messenger stable and the interesting part is not the features — it is that dispatch and transport stopped being one decision.

The symptom

final class SendReceipt
{
    const QUEUE = 'emails';          // a deployment decision, in the class
    const RETRIES = 3;
    const TIMEOUT = 30;

    public function handle(): void
    {
        // and the business logic, underneath all of that
    }
}

// which means the test needs a broker, or a mock of one, or both

Three infrastructure constants above the only two lines that are about the domain. The class cannot be tested without deciding what to do about the transport, and moving it to a different queue is a code change with a deploy attached.

Why it happens

Most queue libraries model a job as a thing that knows how to run itself, which merges three concerns: the data describing what should happen, the code that makes it happen, and the arrangements for getting it from one process to another. That merge is convenient at first and is the reason the class above looks the way it does.

Separating them is not a new idea — it is the command bus pattern, and PHP has had several implementations for years. What changed is that one of them is now in the framework, which means packages can rely on it and the abstraction stops being a project-specific choice.

The fix

A message, a handler, and nothing else

// the message: data. no framework dependency at all.
final class SendReceipt
{
    public $orderId;

    public function __construct(int $orderId) { $this->orderId = $orderId; }
}

// the handler: the type hint is the registration
final class SendReceiptHandler implements MessageHandlerInterface
{
    public function __invoke(SendReceipt $message): void
    {
        $this->mailer->send($this->build($message->orderId));
    }
}

The message carries no framework dependency, so it can be shared as a plain data object with a consumer in another application. Several handlers for one message are allowed and all of them run, which is the difference between a command bus and an event bus expressed as configuration rather than as two libraries.

The type hint being the registration has exactly one failure mode: a typo in the class name produces a handler that is never called and never complains. debug:messenger is the first thing to run when a message appears to vanish.

Routing, which is where the transport decision lives now

framework:
  messenger:
    transports:
      async:  '%env(MESSENGER_TRANSPORT_DSN)%'
      failed: 'doctrine://default?queue_name=failed'

    routing:
      'AppMessageSendReceipt':   async
      'AppMessageRebuildReport': async
      # anything unrouted is handled SYNCHRONOUSLY, immediately

Unrouted messages being handled inline is what makes local development pleasant — no worker to run — and is a trap in production, because a message nobody remembered to route runs during the request and the endpoint gets slower for no visible reason. A test asserting that every message class appears in the routing table is worth the ten lines.

The DSN in an environment variable means the same code runs against Doctrine locally, Redis on staging and AMQP in production with no conditional anywhere. Routing by interface rather than by class is also supported, which is how a whole category of message moves at once.

Middleware on the bus, which is where the transaction belongs

buses:
  command.bus:
    middleware: [validation, doctrine_transaction, AppMessengerCorrelationIdMiddleware]
  event.bus:
    default_middleware: allow_no_handlers
    middleware: [AppMessengerCorrelationIdMiddleware]

Ordering is load-bearing and not obvious. Validation before the transaction means an invalid message never opens one. The correlation middleware belongs outside both, so that a rollback is still logged with its id — which is exactly the case where the log line matters most.

public function handle(Envelope $envelope, StackInterface $stack): Envelope
{
    if ($envelope->last(CorrelationStamp::class) === null) {
        $envelope = $envelope->with(new CorrelationStamp($this->current()));
    }

    $this->logger->pushProcessor(/* ... attach the id ... */);

    try {
        return $stack->next()->handle($envelope, $stack);
    } finally {
        $this->logger->popProcessor();
    }
}

A stamp is metadata travelling with the message, and the correlation id is the canonical use — it survives serialisation, so the worker on another machine logs with the same id as the request that dispatched it. That single stamp is what makes a distributed trace possible without any tracing infrastructure.

The allow_no_handlers setting on the event bus is what stops an event with no subscribers throwing, which is correct for events and wrong for commands. Having two buses with different middleware is the arrangement that makes both behaviours available without a runtime flag.

Retries, delays and the failure transport

transports:
  async:
    dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
    retry_strategy:
      max_retries: 3
      delay: 1000
      multiplier: 3        # 1s, 3s, 9s
      max_delay: 0
    failure_transport: failed

Exponential backoff by default is the right choice and the multiplier is worth setting deliberately — three retries at one second apart is not a backoff, it is three attempts during the same outage. A message can also declare itself unrecoverable by throwing an exception implementing UnrecoverableExceptionInterface, which skips the retries entirely and is the correct response to a validation failure.

$ php bin/console messenger:failed:show
  1  AppMessageSendReceipt   2019-06-14 09:12:04   SMTP timeout
  2  AppMessageRebuildReport 2019-06-14 09:40:11   Deadlock

$ php bin/console messenger:failed:show 1 -vv     # the full stack trace
$ php bin/console messenger:failed:retry 1 2

Reading the exception that killed a message without opening a log aggregator is the practical win, and it is what makes an operator willing to look at the queue at all. The transport still needs a depth alert, because these commands only help somebody who already knows there is something to look at.

Verifying it worked

$ php bin/console debug:messenger
AppMessageSendReceipt
  handled by AppMessageHandlerSendReceiptHandler
  routed to async

$ vendor/bin/phpunit --filter SendReceiptHandler    # no broker, no kernel
OK (6 tests, 14 assertions)

$ php bin/console messenger:consume async 
    --time-limit=3600 --memory-limit=128M --limit=1000

The three limits on the consumer are not optional in a supervised setup: a PHP process that runs forever accumulates whatever the framework touched, and bounding it by time, memory and message count means the supervisor restarts it before that matters. It is a seatbelt rather than a fix, and it is the difference between a leak that is invisible and one that takes the machine down.

Testing the handler by constructing it and calling __invoke with a message is the assertion that says the migration achieved something — the handler is now a class with dependencies and no framework involvement, and that was the whole point.

What this costs

Another abstraction between the code and the queue, and a configuration file that can silently send a message nowhere. A routing table with a typo, a handler with the wrong type hint, a bus whose middleware order is subtly wrong — none of these produce an error, and all of them produce a message that appears to be dispatched and is never handled. debug:messenger answers all three and has to be a habit rather than a thing looked up during an incident.

The honest comparison with a simpler library is that Messenger is more machinery for a project with four background jobs. What it buys is that the machinery is the framework’s rather than yours: the retry strategy, the failure transport and the operator commands all exist and are maintained by somebody else, and that stops mattering only if the four jobs never grow into forty.