Laravel Outbox

Transactional outbox and inbox for Laravel. An event is written in the same database transaction as the business change, a relay delivers it to RabbitMQ, Redis Streams or a Laravel queue at least once, and the inbox on the consumer side makes the handler run once per message.

What sets it apart: messages of one aggregate are delivered in commit order even with several relay workers running in parallel, and that is proven by tests with real concurrent processes on MySQL and PostgreSQL.

The problem

An order service saves order.paid to its database and then publishes it to a broker. If the process dies between the two steps, you get an order nobody hears about, or, if you publish first, an event about an order that was never saved. That's the dual write problem.

flowchart LR
    A[Mark order paid] -->|commit| DB[(Database)]
    A -->|publish| MQ[[Broker]]
    X{{crash between the two}} -.-> A

The transactional outbox fixes it: the event goes into an outbox_messages table in the same transaction as the order. Either both are committed or neither is. A separate relay then reads the table and publishes.

flowchart LR
    subgraph app [Your app, one transaction]
        B[Business change] --> O[(outbox_messages)]
    end
    O --> R[outbox:relay workers]
    R -->|CloudEvents| MQ[[RabbitMQ / Redis Streams / queue]]
    MQ --> C[Consumer]
    C --> I[(inbox_messages + business change,<br/>one transaction)]

Installation

Requires PHP 8.2+ and Laravel 12 or 13, and MySQL 8 or PostgreSQL for production. SQLite works for tests with one relay worker.

composer require azhimov/laravel-outbox
php artisan vendor:publish --tag=outbox-config
php artisan vendor:publish --tag=outbox-migrations
php artisan migrate

Pick a transport (see Transports): RabbitMQ needs composer require php-amqplib/php-amqplib; Redis Streams works with phpredis or predis/predis.

Quick start

In a Laravel app with the package installed and migrated, start RabbitMQ and give the orders exchange a queue:

composer require php-amqplib/php-amqplib
docker run -d --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:4-management

# once it is up: a durable topic exchange "orders" and a queue that gets everything from it
H='content-type: application/json'
curl -u guest:guest -X PUT  -H "$H" localhost:15672/api/exchanges/%2F/orders -d '{"type":"topic","durable":true}'
curl -u guest:guest -X PUT  -H "$H" localhost:15672/api/queues/%2F/orders.all -d '{"durable":true}'
curl -u guest:guest -X POST -H "$H" localhost:15672/api/bindings/%2F/e/orders/q/orders.all -d '{"routing_key":"#"}'

Set OUTBOX_TRANSPORT=rabbitmq in .env, publish in a transaction and run one relay batch:

php artisan tinker --execute="DB::transaction(fn () => Azhimov\Outbox\Facades\Outbox::publish(topic: 'orders', type: 'order.paid', payload: ['order_id' => 1001, 'amount' => '25.00'], aggregateId: 'order-1001'));"
php artisan outbox:relay --once

outbox: claimed 1, published 1, failed 0, dead-lettered 0, lease lost 0

The message is in the queue: open http://localhost:15672 → Queues → orders.all → Get messages, or:

curl -u guest:guest -X POST -H "$H" localhost:15672/api/queues/%2F/orders.all/get -d '{"count":1,"ackmode":"ack_requeue_true","encoding":"auto"}'

{
  "routing_key": "order.paid",
  "payload": "{\"order_id\":1001,\"amount\":\"25.00\"}",
  "properties": {
    "message_id": "01a1cad7-bdd3-70b3-8c75-789d0e758274",
    "type": "order.paid",
    "headers": {
      "ce_id": "01a1cad7-bdd3-70b3-8c75-789d0e758274",
      "ce_source": "shop",
      "ce_specversion": "1.0",
      "ce_subject": "order-1001",
      "ce_time": "2026-11-11T13:41:38.901Z",
      "ce_type": "order.paid"
    }
  }
}

(Output trimmed to the interesting fields.) A full end-to-end demo with a payment API, the relay and a consumer with the inbox is coming soon in payments-outbox.

Usage

Publish inside the transaction of the change:

use Azhimov\Outbox\Facades\Outbox;
use Illuminate\Support\Facades\DB;

DB::transaction(function () use ($order) {
    $order->markPaid();

    Outbox::publish(
        topic: 'orders',
        type: 'order.paid',
        payload: ['order_id' => $order->id, 'amount' => $order->amount],
        aggregateId: "order-{$order->id}",   // keeps this order's events in order
        headers: ['tenant' => 'demo'],       // optional
    );
});

Outside a transaction, publish() throws NotInTransaction. That's on purpose: a message published that way would no longer commit together with the change. You can turn the check off with outbox.strict_transaction set to false.

Run the relay as a long-lived process, for example under Supervisor or as a Kubernetes Deployment:

php artisan outbox:relay

outbox: relaying (pid 4182)
outbox: claimed 3, published 3, failed 0, dead-lettered 0, lease lost 0

Run as many relay processes as you like. They share the work without delivering a message twice at the same time, and they keep each aggregate in order.

Event classes can describe themselves:

final class OrderPaid implements \Azhimov\Outbox\Contracts\ShouldBeOutboxed
{
    public function __construct(private Order $order) {}

    public function topic(): string { return 'orders'; }
    public function type(): string { return 'order.paid'; }
    public function payload(): array { return ['order_id' => $this->order->id]; }
    public function aggregateId(): ?string { return "order-{$this->order->id}"; }
}

DB::transaction(fn () => Outbox::publishEvent(new OrderPaid($order)));

On the consumer side: reading from the broker

The transports can turn what they published back into a ReceivedMessage (id, type, source, subject, time, topic, headers, body, payload), so a consumer does not need to know where the CloudEvents attributes live:

use Azhimov\Outbox\Facades\Inbox;
use Azhimov\Outbox\Transports\RabbitMqTransport;
use PhpAmqpLib\Message\AMQPMessage;

$channel->basic_consume('ledger', callback: function (AMQPMessage $amqp): void {
    $message = RabbitMqTransport::receive($amqp);   // UnexpectedValueException if it is not ours

    Inbox::handle('ledger', $message->id, fn () => Ledger::post($message->payload));

    $amqp->ack();                                   // only after the inbox transaction committed
});

Redis Streams: app(TransportManager::class)->driver('redis')->receive($stream, $fields) for an entry from XREAD / XREADGROUP. The topic is the stream name without stream_prefix.

On the consumer side: inbox

Delivery is at-least-once, so a consumer can see a message twice. The inbox stores a "processed" mark in the same transaction as your handler's changes:

php artisan vendor:publish --tag=outbox-inbox-migrations && php artisan migrate

use Azhimov\Outbox\Facades\Inbox;
use Azhimov\Outbox\Inbox\InboxOutcome;

$outcome = Inbox::handle('billing', $messageId, function () use ($payload) {
    Ledger::credit($payload['account'], $payload['amount']);
});

// InboxOutcome::Processed, or InboxOutcome::AlreadyProcessed (the handler did not run)

The scope is consumer + message id, so two services consuming the same message don't interfere. If the handler throws, the mark is rolled back with the change, and the redelivery gets processed. Queue jobs can use the DeduplicateWithInbox middleware.

Testing your app

With RefreshDatabase every test runs inside a transaction, so the strict check can't catch a publish() you forgot to wrap. Use the fake to assert what would be published:

use Azhimov\Outbox\Facades\Outbox;

Outbox::fake();

$this->postJson('/api/orders/1/pay')->assertOk();

Outbox::assertPublished('order.paid', fn ($m) => $m->aggregateId === 'order-1');
Outbox::assertPublishedTimes('order.paid', 1);
Outbox::assertNotPublished('order.refunded');
// Outbox::assertNothingPublished();

The fake validates input the same way the real outbox does.

How it works / why it's built this way

Claiming with SKIP LOCKED and leases

Each relay worker has a random owner token. Each batch goes through three steps:

1. In a short transaction it runs SELECT … FOR UPDATE SKIP LOCKED LIMIT n on due messages. Rows another worker is claiming are skipped, not waited for. The worker then leases the rows: status = processing, lease_owner = token, lease_until = now + lease_seconds. Then it commits.
2. It publishes outside any transaction, so the broker call doesn't hold database locks.
3. It marks a message published with a compare-and-set on lease_owner. A worker whose lease expired can't overwrite a message another worker took over.

A worker that dies doesn't keep messages forever: once lease_until passes, any worker claims them again. That's exactly why delivery is at-least-once, and why the inbox exists.

sequenceDiagram
    participant A as Relay A
    participant DB as outbox_messages
    participant B as Relay B
    participant MQ as Broker
    A->>DB: SELECT … FOR UPDATE SKIP LOCKED, lease to A
    B->>DB: SELECT … FOR UPDATE SKIP LOCKED
    Note over B,DB: A's rows are skipped, B leases others
    A->>MQ: publish (no transaction open)
    MQ-->>A: confirm
    A->>DB: UPDATE … SET published WHERE lease_owner = A
    Note over A,DB: 0 rows if A lost its lease: nothing overwritten

The test suite starts four relay processes on 300 messages and checks that every message is delivered exactly once. With SKIP LOCKED removed, the same test sees 420–510 deliveries.

Order within an aggregate, with parallel workers

Ordering by id isn't enough. Ids are generated before commit, so transaction A can take id 10, transaction B id 11, and B commits first. A relay would then send 11 before 10.

Instead, publish() with an aggregateId increments a counter row in outbox_aggregates with one upsert, in your transaction:

• PostgreSQL, SQLite: INSERT … ON CONFLICT … DO UPDATE … RETURNING
• MySQL: ON DUPLICATE KEY UPDATE last_seq = LAST_INSERT_ID(last_seq + 1)

The upsert locks that row until commit. A second writer of the same aggregate waits, so the aggregate_seq values are handed out in commit order.

The relay claims a message only if every earlier message of its aggregate is already published. That gives three properties:

• each aggregate has at most one message in flight;
• different aggregates are spread over all workers;
• a message that ends up failed blocks only its own aggregate (event AggregateBlocked) until you run outbox:retry. Skipping it would deliver order.shipped before order.paid.

The suite tests this with four relay processes, 20 aggregates × 15 messages. With ordering disabled, the same test reports aggregates delivered out of order.

Messages without aggregateId have no order and no counter. Set outbox.ordering to none to turn ordering off everywhere.

Retries and the dead letter

A failed delivery comes back after min(base × multiplier^(n−1), max) ± jitter. Defaults: 1 s, ×2, 5 min, ±20%. After max_attempts (default 10) the message becomes failed and fires MessageFailed; alert on it and fix the cause.

The relay also dead-letters a message whose lease expired on its last attempt, because the worker died while delivering. Claiming it again could take the next worker down the same way.

Message format: CloudEvents 1.0, binary mode

The body is your payload JSON byte for byte. It is stored as text, not as a json column, because MySQL reorders keys and jsonb reformats them, and consumers that hash or sign a body need the exact bytes.

The attributes travel beside the body:

Spec: CloudEvents 1.0.2. Header names starting with ce_, and the names data and content-type, are reserved.

Transports

For the queue transport, an unknown type fires UnhandledOutboxMessage and logs a warning instead of failing the job. Set deduplicate to true to run deliveries through the inbox.

To add your own transport, implement Transport::publish(OutboxMessage $m): void: return when the broker has accepted the message, throw otherwise. Register it with app(TransportManager::class)->extend('sqs', fn ($app) => new SqsTransport(...)).

CI tests RabbitMQ and Redis Streams against real servers. Redis Streams runs with both phpredis and predis.

Commands

Configuration

Everything lives in config/outbox.php; the main keys also read env variables.

Events

Compared with alternatives

Checked against README and source on 11.11.2026; corrections are welcome.

If you already use Ecotone, use its outbox. If one relay is enough, ianfoxdev/outbox is a well-engineered choice with a fast Go relay. This package is for Laravel apps that want outbox, relay and inbox in one Composer package, with ordering that survives horizontal scaling.

Limitations

• At-least-once, not exactly-once. A worker can die after the broker accepted a message and before it is marked published; another worker then sends it again. Make consumers idempotent; that's what the inbox is for.
• The outbox must live on the business connection. publish() checks the transaction on outbox.connection. A transaction on another connection doesn't count.
• Several aggregates in one transaction can deadlock. Each aggregate's counter row is locked until commit. If two transactions touch aggregates X and Y in opposite order, the database aborts one of them. Publish aggregates in a stable order, for example sorted by id.
• A failed message blocks its aggregate until outbox:retry. That's the price of strict order. Alert on AggregateBlocked, or turn ordering off.
• One message per aggregate per batch. Throughput per aggregate is one message per relay round trip, which is plenty for most aggregates and wrong for a single very hot one.
• MySQL claims lock briefly. Under REPEATABLE READ, the claim query also locks rows it scanned and filtered out until its transaction ends, a few milliseconds. Other workers skip them, so a worker may see an empty batch while work remains. That costs latency, not correctness.
• SQLite is for tests only. There are no row locks, so run one relay worker.
• RefreshDatabase hides a missing transaction. Use Outbox::fake() in application tests.
• No Kafka yet; it's planned for 0.3.

Roadmap

• 0.3: Kafka transport (ext-rdkafka, key = aggregate id).
• 0.3: a docker compose demo: relay + a consumer with the inbox (with payments-outbox).
• Prometheus metrics for the relay: lag, attempts, dead letters.

Testing

composer check          # Pint, Larastan (max), Pest on SQLite
docker compose up -d    # MySQL, PostgreSQL, RabbitMQ, Redis for the integration suites
composer test:mysql
composer test:pgsql

The concurrency tests start separate PHP processes against MySQL or PostgreSQL. The broker tests need RabbitMQ and Redis. CI runs all of them, plus PHP 8.2–8.5 × Laravel 12–13 and coverage of at least 90%.

Changelog

See CHANGELOG.md.

License

MIT. See LICENSE.