AssegaiPHP Queues: Typed Jobs and Reliable Settlement

A queue processor should receive the work the application asked it to perform. It should not need to understand the broker object that carried that work, reproduce a decoder, or decide how to settle a delivery.

The first-party queue drivers did not consistently enforce that boundary.

RabbitMQ passed a raw AMQPMessage into the processor and defaulted to automatic acknowledgement. A message could therefore be considered handled before application processing had succeeded.

Beanstalkd passed raw JSON to the processor. Its failure handling did not cover every PHP Error, and settlement did not provide one reliable success-and-retry contract across decoding and processor failures.

Those were two transport symptoms of one architectural problem. Fixing only a driver would have left Console unable to preserve the processor's type and every transport free to invent its own result semantics. The repair therefore spans Common 0.10.1, Console 0.10.3, RabbitMQ 1.1.1, and Beanstalkd 1.1.1.

One job contract in Common

Common owns the shared queue boundary.

JsonQueueJobCodec encodes a domain object into a versioned JSON envelope. The envelope records its version and top-level job class separately from the payload. On delivery, QueueJobTypeResolver inspects the processor callback's first parameter and gives the declared type to the codec.

That lets a processor state its real contract:

final readonly class NotificationJob
{
  public function __construct(
    public string $recipient,
    public string $message,
  ) {
  }
}

final class NotificationsProcessor
{
  public function process(NotificationJob $job): void
  {
    // Perform the application work.
  }
}

Hydration supports readonly constructor DTOs, inherited private state, nested and intersection-typed objects, enums, dates, arrays, and nullable values. Envelope class metadata must be compatible with the class or interface declared by the processor; an incompatible job fails before application code runs.

Legacy JSON objects remain consumable. If the processor declares a concrete class, the codec hydrates that class from the old payload. If the processor declares only object, it receives stdClass; the envelope does not turn an untyped callback into arbitrary class instantiation.

Common also supplies the canonical process result for one delivery attempt. It retains callback data, captured errors, and the hydrated job when decoding succeeded. Drivers own that result. Application processors do not create or return transport-specific result objects.

Every decoding, processing, and settlement path catches Throwable. Both Exception and PHP Error implementations, including TypeError, therefore enter the same controlled failure path.

Console preserves the processor type

Typed hydration only works if the callable passed to the driver retains its concrete parameter.

Console 0.10.3 resolves the processor provider and passes its method directly to QueueInterface::process(). It no longer hides the declaration behind a generic callback. Processor discovery accepts a public process, handle, configured, or invokable method that takes one job and requires no additional arguments.

The worker reports errors before deciding that a result is empty. That distinction matters when decoding fails and no hydrated job exists. Failed deliveries also observe the configured worker sleep interval before the next poll, preventing an immediate failure loop.

One call to a queue driver's process() handles at most one delivery. That rule gives the worker options consistent meaning:

• --once performs one poll and exits
• --max-jobs counts successful jobs without a driver draining several messages inside one call
• an empty poll can sleep or satisfy --stop-when-empty
• a failed delivery can report the error and apply worker backoff before polling again

RabbitMQ settles after processing

RabbitMQ 1.1.1 uses manual acknowledgement by default through no_acknowledgement: false.

The driver retrieves one message, decodes and hydrates the domain job, calls the processor, and acknowledges only after the callback succeeds. A decoding or processor failure is nacked. With requeue_on_failure: true, RabbitMQ returns that delivery to the queue; with false, the broker can discard it or apply a configured dead-letter policy.

Exchange configuration is also effective end to end. When exchange_name is set, the driver declares the exchange using exchange_type, exchange_durable, and exchange_auto_delete, then binds the queue with routing_key. Publishing uses that exchange and routing key instead of silently ignoring them.

The package communicates through PhpAmqpLib. The unused ext-amqp Composer requirement has been removed; applications do not need the PECL AMQP extension for this driver.

Beanstalkd settles and retries consistently

Beanstalkd 1.1.1 watches its configured tube and reserves at most one job per poll. A named worker removes the implicit default tube from its watch list, preventing it from taking work intended for another queue.

reserve_timeout controls how many seconds that poll waits for a job. A successful callback deletes the reserved job. A decoding or processor failure releases it with retry_priority and retry_delay.

Production workers should use a non-zero retry delay. Without one, a persistent failure can cycle through reserve, fail, and release without useful backoff.

The application boundary is transport-free

Producing and consuming code now share one domain type:

use Assegai\Common\Interfaces\Queues\QueueInterface;
use Assegai\Core\Attributes\Injectable;
use Assegai\Core\Queues\Attributes\InjectQueue;
use Assegai\Core\Queues\Attributes\QueueProcessor;

#[Injectable]
final readonly class NotificationsService
{
  public function __construct(
    #[InjectQueue('rabbitmq.notifications')]
    private QueueInterface $queue,
  ) {
  }

  public function send(NotificationJob $job): void
  {
    $this->queue->add($job);
  }
}

#[Injectable]
#[QueueProcessor('rabbitmq.notifications')]
final class NotificationsProcessor
{
  public function process(NotificationJob $job): void
  {
    // No JSON decoding or broker settlement here.
  }
}

Register the processor as a module provider so Console can discover it, then inspect and run the connection:

assegai queue:list
assegai queue:work rabbitmq.notifications
assegai queue:work rabbitmq.notifications --once

The Queues and Background Jobs guide contains complete configuration for both transports, worker stopping options, typed hydration behavior, and idempotency guidance.

Migrating existing processors

Applications with transport-shaped processors should move the boundary once:

• replace an AMQPMessage parameter with the concrete domain job class
• replace a raw JSON string parameter with that same class
• replace generic transport payloads with a concrete job type when the contract is known
• keep object only when receiving stdClass is deliberate
• remove application-owned JSON decoding, acknowledgement, nacking, deletion, and release calls
• confirm legacy JSON contains every constructor value required by the typed job

Retrying transports provide at-least-once delivery. A callback may complete an external side effect and then lose settlement because of a connection failure. Use stable job identifiers, unique constraints, or an idempotency record to make repeated delivery safe. Do not design processors around an exactly-once assumption.

Coordinated release and upgrade order

The dependency order is:

1. Common 0.10.1 establishes the shared codec, hydration, and result contracts.
2. Console 0.10.3 preserves the processor callable and applies worker semantics around that result.
3. RabbitMQ 1.1.1 and/or Beanstalkd 1.1.1 implement typed delivery and safe settlement.

Upgrade Common first:

composer require assegaiphp/common:^0.10.1 --with-all-dependencies

Upgrade the global Console:

composer global require assegaiphp/console:^0.10.3 --with-all-dependencies

Then upgrade the driver or drivers used by the application:

composer require assegaiphp/rabbitmq:^1.1.1 --with-all-dependencies

composer require assegaiphp/beanstalkd:^1.1.1 --with-all-dependencies

Composer can resolve Common and an application driver in one project operation when preferred. The important constraint is that the driver must resolve with Common 0.10.1, and the worker should run through Console 0.10.3.

Release references

• Common 0.10.1 release notes
• Console 0.10.3 release notes
• RabbitMQ 1.1.1 release notes
• Beanstalkd 1.1.1 release notes