Add an opt-in transactional queue outbox

Merged
BI
2d ago
+10659-328

Summary

This adds an opt-in transactional outbox to the queue package. Jobs are stored alongside the business write in the same database transaction, then published to their existing queue by a separate relay process.

afterCommit() prevents a worker from seeing uncommitted data, but it cannot close the gap between committing the database transaction and publishing the job. If the process stops in that gap, the business write survives without its job. The outbox closes that gap without replacing the existing queue drivers, dispatch APIs, or workers.

DB::transaction(function () use ($attributes) {
    $order = Order::create($attributes);

    SendOrderReceipt::dispatch($order)->viaOutbox();
});

Usage

Applications can select the outbox per dispatch with viaOutbox(), per job class with ShouldQueueViaOutbox, or per queue connection with outbox => true. withoutOutbox() overrides either default. onOutboxDatabase() selects the owning connection when more than one transaction is active.

The existing bulk and batch APIs use set-based inserts. Batch defaults also apply to jobs added later. Queued listeners, mailables, notifications, and broadcasts pass their selection to their existing wrapper jobs. Jobs that do not select the feature keep their normal dispatch path, without outbox database access or outbox service resolution.

Setup consists of an outbox table and a supervised relay alongside the existing workers:

php artisan make:queue-outbox-table
php artisan migrate
php artisan queue:relay --batch=100 --concurrency=4
php artisan queue:work redis

Horizon remains optional and continues managing Redis jobs after publication. The detailed guide is src/docs/queue-outbox.md; the queue and producer guides link to it.

Delivery and recovery

  • Staging uses the exact database session that owns the transaction. Rollback removes the rows and releases dispatch-owned locks. With several active connections, the caller must choose an owner rather than relying on a guess.
  • The relay claims bounded batches, publishes through the existing drivers, and deletes only confirmed deliveries. PostgreSQL uses a single claim statement; MySQL and MariaDB use locking reads with SKIP LOCKED; SQLite has an atomic-update fallback.
  • Database connections and locks are released before broker calls. Publication and settlement have separate, short resource lifetimes. Multiple relay processes and bounded concurrent batches are supported.
  • Redis and database publication are batched. SQS uses its request limits and partial-success responses. Other drivers use their existing raw-payload API, with one pooled connection per publication batch.
  • Failed and uncertain deliveries remain retryable with jittered backoff. Failed sources and destinations are throttled without blocking healthy sources. Ownership tokens prevent a stale relay from deleting another relay's claim.
  • Payload UUIDs, intended delays, SQS FIFO options, and Horizon classification survive staging. Separate SQS overflow publication attempts have separate cleanup ownership; existing overflow retention settings are unchanged.

The guarantee is at least once, not exactly once. A crash after broker acceptance but before settlement can publish the same job again. Handlers must be idempotent where repeated effects would be harmful. Publication order is best effort, and the outbox cannot make writes to separate databases or external services atomic.

Driver compatibility and operations

Selecting the feature does not require changing job code when switching drivers. A database queue already participating in the owning transaction inserts directly. Sync, deferred, background, and failover connections preserve their execution behavior through an after-commit fallback, without claiming crash durability. Null and fake queues do not create outbox rows. Direct pushRaw() remains a bypass.

The relay supports restart signals, graceful shutdown, bounded runtime and job limits, and one-shot execution. Operator commands provide pending-row inspection, monitoring, immediate retry, and deliberate removal with owner-matched unique-lock cleanup:

php artisan queue:outbox
php artisan queue:outbox-monitor --max=10000 --age=60
php artisan queue:outbox-retry JOB_UUID
php artisan queue:outbox-forget JOB_UUID

Delivery events and OpenTelemetry metrics are recorded at batch boundaries. There is no backlog polling on the application dispatch path and no duplicate producer span at relay time.

Related correctness changes

The integration also fixes behavior shared with ordinary dispatch:

  • Transaction records and callbacks identify the resolved base/read/write session, preventing separate sessions from consuming each other's callbacks.
  • Dispatch-lock cleanup captures the context in which the lock was acquired. Transactional storage keeps an independent rollback snapshot after acceptance.
  • Batch notifications preserve confirmed acceptance even when an observer fails. Horizon's post-publication work cannot turn an accepted job into a failed dispatch or release its lock.
  • Generic lifecycle events retain the finalized payload used by existing tracing integrations, separate from transport decoration.
  • Uninitialized public broadcast properties are skipped, and custom or memoized cache repositories retain their store identity for unique-lock cleanup.

Public dispatch method signatures remain unchanged. Two extension behaviors are documented: role-qualified transaction records now use the exact session name, and Horizon's normal push() uses the shared Redis storage helper instead of routing through its public pushRaw() override. Direct pushRaw() remains supported.

Performance

Measurements compare the pre-feature baseline with this branch, using independent installs with identical dependencies. The environment was a Linux VM exposing six CPUs, PHP 8.4.26, Swoole 6.3.0RC1, CLI OPcache enabled, JIT disabled, and local PostgreSQL 17.11, MySQL 9.5.0, MariaDB 10.11.19, and Redis 8.2.1. PostgreSQL durability settings stayed enabled. Job contents were 256 bytes; stored outbox envelopes were 1,246 bytes.

These measurements include the branch's related correctness changes. They measure publication, not execution of application jobs. Tables show median jobs/second with the observed range.

Ordinary dispatch, without the outbox

Ten alternating baseline/current pairs per case. Single runs dispatched 20,000 jobs; bulk runs dispatched 100,000 jobs in batches of 1,000. Warmup was 1,000 jobs.

DispatchBaseline jobs/sThis branch jobs/s
Single5,074 (4,992–5,210)5,073 (4,981–5,180)
Bulk, 1,000 per call40,074 (39,107–41,351)43,491 (40,829–44,366)

Single dispatch was effectively unchanged. Bulk improved by 8.5% using the ratio of medians; every matched pair improved. Median PHP CPU for bulk fell from 1.990 to 1.809 seconds. Ending heap growth was identical between revisions in both cases; the peak heap difference was about 24 KB.

Transactional staging

Three samples per engine and batch size. Single runs staged 2,000 jobs; bulk runs staged 10,000. Each call had its own transaction, so bulk amortizes commit cost.

DatabaseSingle jobs/s1,000 per call jobs/s10,000 per call jobs/s
PostgreSQL266 (260–266)10,964 (9,425–11,241)11,312 (11,007–12,124)
MySQL86 (82–89)1,046 (378–4,581)7,504 (7,426–7,508)
MariaDB283 (276–284)11,281 (8,455–11,428)11,237 (9,021–11,657)

Database wall times varied substantially while PHP CPU stayed steadier. The ranges retain those outliers; these results are not an engine ranking.

Relay

PostgreSQL source, 30,000 jobs per sample, 1,000 warmup jobs, and three samples per configuration. The table below uses four concurrent tasks and four pool slots. Claim, publication, settlement, and worker lifecycle are timed; seeding is not. Restart checks use an in-memory cache.

TargetClaim batchJobs/s
Redis1008,211 (7,770–8,421)
Redis1,00022,387 (14,396–22,700)
Database10010,963 (9,450–11,153)
Database1,00011,754 (6,867–14,457)

The full report also includes concurrency-one and concurrency-eight runs. Those retained four pool slots, so the concurrency-eight results include pool contention rather than isolating scaling with matched capacity.

Claim design

A PostgreSQL comparison tested the current indexed eligibility deadline against a separate, unindexed reservation deadline. Five alternating pairs per workload used eight clients, eight pool slots, batches of 100, and a mixed backlog under continuous claim/settle/refill activity.

Backlog / live leasesCurrent claim statement, median msReservation alternative, median msEligibility-index buffer hits, current / alternative
10,000 / 8002.0862.505158 / 657
13,800 / 8,0002.0664.948166 / 2,219
100,000 / 8001.8572.309158 / 642
100,000 / 8,0001.7324.581158 / 2,109

The alternative scanned live leases and showed no consistent statement-WAL saving. The production design is unchanged, and the temporary comparison code is removed. An earlier oversubscribed run hit checkout timeouts; the report retains that failure and the follow-up investigation rather than hiding it or changing timeout settings. No connection-lifetime defect was confirmed.

The retained harness and full results, including commands, ranges, runtime settings, resource checks, and diagnostic limitations, are in tests/Benchmarks/QueueOutbox/.

Verification

Local verification covered formatting, source and type-fixture static analysis, the framework tests, Testbench package contracts, and focused checks after corrections. Live database tests cover PostgreSQL, MySQL, MariaDB, and SQLite; Redis integration tests exercise actual publication and worker delivery.

Coverage includes transaction ownership and rollback, batch atomicity, producer propagation, partial and uncertain publication, lease expiry and stale settlement, cancellation, observer failures, operator commands, and tiny-pool resource lifetimes. Benchmark runs check completed counts and released connections, and clean up their own tables and queue keys. No timing thresholds were added to CI.

Summary by cubic

Adds an opt-in transactional outbox so queued jobs can be stored in the same database transaction as the business write and published by a separate relay. This closes the crash window between commit and publish without replacing existing queue drivers, dispatch APIs, or workers.

New Features

  • Select the outbox with viaOutbox(), the ShouldQueueViaOutbox marker, or per-connection outbox => true; withoutOutbox() overrides any default, and onOutboxDatabase() picks the owning connection.
  • Queued listeners, mailables, notifications, and broadcasts propagate the selection; unselected jobs keep their ordinary dispatch path without outbox access.
  • Setup is one migration plus a supervised queue:relay alongside existing workers; Horizon remains optional and continues managing jobs after publication.
  • The relay claims bounded batches, publishes through the existing drivers, and deletes only confirmed deliveries; failed and uncertain deliveries retry with jittered backoff and throttled sources.
  • Delivery is at-least-once, so handlers must be idempotent where repeated effects are harmful. Sync, background, deferred, and failover connections degrade to an after-commit fallback without crash durability; without a transaction, local dispatch keeps its immediate result.
  • Adds operator commands for listing, monitoring, retrying, and discarding pending messages, plus batch-boundary delivery events and OpenTelemetry metrics.
  • Includes related correctness fixes: resolved base/read/write transaction identity for records and callbacks, context-captured dispatch-lock cleanup, preserved confirmed acceptance on observer failures, finalized payload retention for tracing, and batch outbox owners validated against the batch repository before dispatch so a rolled-back business transaction cannot leave committed batch counters.
  • Ordinary single dispatch is effectively unchanged; bulk dispatch improved in paired measurements.

Migration

  • Run php artisan make:queue-outbox-table and php artisan migrate, then start queue:relay alongside queue:work.
  • Extensions registering callbacks directly with DatabaseTransactionsManager must use the resolved connection's getNameWithReadWriteType() instead of base getName().

[!NOTE]

Add opt-in transactional queue outbox with relay worker and console commands

  • Adds a database-backed transactional outbox. Jobs marked via viaOutbox(), onOutboxDatabase(), the ShouldQueueViaOutbox marker, or a queue connection default are staged in a job outbox table inside the dispatching transaction, then published by a relay worker after commit. See StagesOutboxJobs.php, OutboxDispatcher.php, and OutboxWorker.php.
  • Adds outbox publishers for Redis, SQS, database, and queue-pool transports, honoring delayed delivery, FIFO fields, and batch limits. SQS delayed jobs are staged no more than 900 seconds before delivery.
  • Adds console commands: queue:relay, queue:outbox (listing), queue:outbox-monitor, queue:outbox-retry, queue:outbox-forget, and make:queue-outbox-table, plus new OutboxBusy, OutboxBatchDelivered, and OutboxBatchDeliveryFailed events and OpenTelemetry delivery metrics.
  • Extends opt-in APIs to batches (PendingBatch::viaOutbox and friends), pending dispatches, and queued mailables, notifications, broadcasts, and listeners; unused outbox properties are not serialized.
  • Refactors transaction-owner identity so transaction records, callbacks, and applicability checks use the connection name with its read/write role (ManagesTransactions, DatabaseTransactionsManager, RequestContextSynchronizer). Also reworks Horizon RedisQueue payload preparation around prepareForQueueing / push-time hooks and removes the last-pushed coroutine context.
  • Behavioral Change: queue enqueue now raises queued events and accepts dispatch locks after transport acceptance; a post-acceptance listener failure is rethrown after all member notifications. Transaction callbacks keyed on base connection names no longer match — extensions comparing against getName() must switch to getNameWithReadWriteType(). RedisTagRepository returns early for empty tag lists. SyncQueue jobs selected for outbox are deferred until commit instead of stored.

Macroscope summarized 2f11e2e.