README

Русская версия

Transactional outbox pattern implementation for Yii3. Provides a stateless core for reliably publishing messages with configurable retry policies.

Using an AI coding assistant? llms.txt has a compact API reference you can use. Projects using the llm/skills Composer plugin also get this package's agent skill synced into .agents/skills/ automatically on install.

Requirements

PHP 8.3+
psr/clock ^1.0
psr/log ^3.0

Installation

composer require rasuvaeff/yii3-outbox

Usage

Recording a message

use DateTimeImmutable;
use Psr\Clock\ClockInterface;
use Rasuvaeff\Yii3Outbox\InMemoryStorage;
use Rasuvaeff\Yii3Outbox\Outbox;

$clock = new class implements ClockInterface {
    public function now(): DateTimeImmutable { return new DateTimeImmutable(); }
};

$outbox = new Outbox(storage: $storage, clock: $clock);

$message = $outbox->record(
    type: 'order.created',
    payload: json_encode(['orderId' => 42]),
    aggregateId: 'order-42',
);

The transactional guarantee

The pattern is only worth its name when the outbox write commits atomically with the business write it describes. The core cannot enforce this: it opens no transaction and knows nothing about your connection. Two obligations are therefore yours:

Call record() inside the same database transaction as the business write.
Use a storage that writes through the same connection as your business tables — rasuvaeff/yii3-outbox-db takes a ConnectionInterface for exactly this reason.

$db->transaction(static function () use ($orders, $outbox, $order, $json): void {
    $orders->insert($order);

    $outbox->record(
        type: 'order.created',
        payload: $json,
        aggregateId: $order->id,
    );
});

Break either obligation and the guarantee is void: commit the order without the message and the event is lost forever; commit the message without the order and consumers observe an event that never happened. A storage backed by a different database — or by a message broker — cannot provide this guarantee at all, and InMemoryStorage is a test double, not a durable one.

Message ids

The message id is the primary key of the outbox table and, when messages are exported to ClickHouse, the deduplication key of the ReplacingMergeTree. Two ways to control it:

Pass the domain event's id — the right choice whenever the message mirrors an event that already has an identifier. Republishing the same event then cannot mint a second id, so the consumer has something stable to deduplicate on:

$outbox->record(
    type: 'order.created',
    payload: $json,
    aggregateId: 'order-42',
    id: $domainEvent->getId(),
);

Bind a generator for messages that have no domain id. The default RandomHexIdGenerator keeps the historical format (32 random hex characters); a time-ordered id makes inserts append instead of scattering across InnoDB pages and gives batches a stable order:

use Rasuvaeff\Yii3Outbox\MessageIdGeneratorInterface;

// symfony/uid
final readonly class Uuid7IdGenerator implements MessageIdGeneratorInterface
{
    public function generate(): string
    {
        return \Symfony\Component\Uid\Uuid::v7()->toRfc4122();
    }
}

// ramsey/uuid — equally monotonic within the same millisecond
final readonly class RamseyUuid7IdGenerator implements MessageIdGeneratorInterface
{
    public function generate(): string
    {
        return \Ramsey\Uuid\Uuid::uuid7()->toString();
    }
}

$outbox = new Outbox(storage: $storage, clock: $clock, idGenerator: new Uuid7IdGenerator());

The package ships no UUID implementation and depends on no UUID library — id is VARCHAR(255) in rasuvaeff/yii3-outbox-db, so any format fits and the choice stays yours.

Implementing storage

claim() is the primitive the whole polling loop rests on — Processor calls it, never findPending(). It must atomically move messages to Processing and return them, so that two workers polling the same table never receive the same message. findPending() is the read-only counterpart: safe for dashboards and diagnostics, unsafe as a worker's fetch.

use Rasuvaeff\Yii3Outbox\StorageInterface;
use Rasuvaeff\Yii3Outbox\OutboxMessage;

final class DbStorage implements StorageInterface
{
    public function save(OutboxMessage $message): void
    {
        // INSERT INTO outbox ... ON CONFLICT(id) DO UPDATE ...
        // Must run on the caller's connection so it commits with the business write.
    }

    public function claim(array $types = [], int $limit = 1000): array
    {
        // Atomically: SELECT ids of status = 'pending' [AND type IN (:types)]
        //   LIMIT :limit FOR UPDATE SKIP LOCKED
        // then UPDATE outbox SET status = 'processing', claimed_by = :worker
        //   WHERE id IN (...) — and return the claimed rows.
        // Every claimed message must end up markPublished(), markFailed(),
        // or save($msg->withStatus(Pending)); none may stay Processing.
    }

    public function findPending(array $types = [], int $limit = 1000): array
    {
        // SELECT * FROM outbox WHERE status = 'pending'
        //   [AND type IN (:types)] LIMIT :limit  -- empty $types = all types
        // Read-only: no atomicity, so two workers would both get the same rows.
        // For retry support, also return status = 'pending' with attempts > 0
    }

    public function markPublished(OutboxMessage $message): void
    {
        // UPDATE outbox SET status = 'published' WHERE id = ?
    }

    public function markFailed(OutboxMessage $message): void
    {
        // UPDATE outbox SET status = 'failed' WHERE id = ?
    }

    public function getById(string $id): ?OutboxMessage
    {
        // SELECT * FROM outbox WHERE id = ?
    }
}

Letting the storage apply the retry policy

claim() returns every Pending message, so Processor receives the ones still waiting out their backoff and writes each of them straight back as Pending. That is two writes per backing-off message per iteration, and each one occupies a slot in batchSize that a message ready to go could have used — with a large retry queue, fresh messages wait behind it.

A storage that can express the predicate in its own query language implements RetryAwareStorageInterface, and Processor claims through it automatically:

use Rasuvaeff\Yii3Outbox\RetryAwareStorageInterface;

final class DbStorage implements RetryAwareStorageInterface
{
    public function claimReady(
        DateTimeImmutable $readyThreshold,
        int $maxAttempts,
        array $types = [],
        int $limit = 1000,
    ): array {
        // Same atomic claim as claim(), with one more condition:
        //   AND (attempts >= :maxAttempts
        //        OR last_attempt_at IS NULL
        //        OR last_attempt_at <= :readyThreshold)
    }

    // ... the rest of StorageInterface unchanged
}

The attempts >= :maxAttempts clause is not an optimisation and must not be dropped. A message out of attempts cannot be retried, and the only thing left to do with it is mark it Failed — which Processor can only do to a message the storage handed it. Filter it out and nothing ever terminates it: it stays Pending, invisible to an alert watching Failed, forever.

$readyThreshold comes from RetryPolicy::readyThreshold($now) — the delay is the core's business, and an implementation must not reconstruct it. The interface exists separately from StorageInterface because adding the parameter to claim() itself would break every third-party implementation.

rasuvaeff/yii3-outbox-db implements it. A storage that does not is still correct: Processor falls back to claim() and filters in PHP, exactly as before.

Two consequences worth knowing before you alert on them:

ProcessingResult::$skipped counts messages the batch claimed and discarded. Against a retry-aware storage there are none, so it reads 0 — the work it used to count is what this interface removes.
A message whose attempts are spent is terminated up to delaySeconds later than before, since it now waits for a batch that includes it.

Implementing a publisher

use Rasuvaeff\Yii3Outbox\PublisherInterface;
use Rasuvaeff\Yii3Outbox\OutboxMessage;
use Rasuvaeff\Yii3Outbox\PublishException;

final class RabbitPublisher implements PublisherInterface
{
    public function publish(OutboxMessage $message): void
    {
        try {
            // publish to RabbitMQ, Kafka, etc.
        } catch (\Throwable $e) {
            throw new PublishException(
                message: $e->getMessage(),
                outboxMessage: $message,
                previous: $e,
            );
        }
    }
}

Processing the outbox

use Rasuvaeff\Yii3Outbox\Processor;
use Rasuvaeff\Yii3Outbox\RetryPolicy;

$processor = new Processor(
    storage: $storage,
    publisher: $publisher,
    retryPolicy: new RetryPolicy(maxAttempts: 3, delaySeconds: 60),
    clock: $clock,
    batchSize: 100,
);

$result = $processor->process();
// $result->published — successfully published
// $result->failed   — publish failures and messages that ran out of attempts
// $result->skipped  — claimed but not yet ready for retry; always 0 against
//                     a RetryAwareStorageInterface, which never claims them

Retry behaviour

When a publish fails:

If attempts < maxAttempts → message stays Pending, will be retried after delaySeconds
If attempts >= maxAttempts → message is marked Failed (terminal)

Every message a batch claims leaves Processing. A message that is claimed with its attempts already spent — restored from a backup, or left behind by a markFailed() that never reached the database — is marked Failed on sight rather than saved back as Pending, which would make it circle claim → skip → save forever with no alert on Failed ever firing.

A publisher that throws something other than PublishException is a bug, and process() rethrows it: the failure is not silently retried as if it were a delivery problem. Before the exception propagates, the current message is persisted per the retry policy and every message the batch had claimed but not yet attempted is released back to Pending (or failed, if it too was out of attempts), so nothing is left in Processing for a human to find with raw SQL.

The same applies when the storage — not the publisher — is what fails. If markPublished() throws after publish() succeeded, the message reached its consumer and only the record of that did not: it goes back to Pending and a later run publishes it again. This package delivers at least once, and the message id is what a consumer deduplicates on; leaving the row Processing instead would be a row nothing in this API can move. That branch logs Outbox message was published but could not be marked published, not the publisher warning — during a storage incident an operator should not be sent to debug the publisher.

The release is best-effort by construction: it reaches for the same storage that may be the reason the batch aborted, and a storage that is down cannot be told anything. What it does guarantee is that its own failures are logged (Failed to release a claimed outbox message) rather than thrown — the exception you catch is always the one that aborted the batch, never a symptom raised while reacting to it.

$policy = new RetryPolicy(maxAttempts: 3, delaySeconds: 60);

$policy->shouldRetry($message);           // bool — attempts remaining?
$policy->isReadyForRetry($message, $now); // bool — delay elapsed?
$policy->readyThreshold($now);            // DateTimeImmutable — the same question
                                          // as a boundary a storage can filter on

Using InMemoryStorage for tests

use Rasuvaeff\Yii3Outbox\InMemoryStorage;

$storage = new InMemoryStorage();
$storage->save($message);

$pending = $storage->findPending();
$storage->count();
$storage->clear();

API reference

Outbox

Method	Description
`__construct(storage, clock, idGenerator?)`	Main entry point; default generator = `RandomHexIdGenerator`
`record(type, payload, aggregateId?, id?)`	Create and persist message, returns `OutboxMessage`. `id` = the domain event's id; omitted → generator. Call inside the business transaction

StorageInterface

Method	Description
`save(message)`	Persist. Must commit with the business write — see The transactional guarantee
`claim(types = [], limit = 1000)`	Atomically moves up to `limit` `Pending` messages to `Processing` and returns them. What `Processor` uses; safe for concurrent workers
`findPending(types = [], limit = 1000)`	Read-only listing of `Pending` messages. No atomicity — for dashboards, not for workers
`markPublished(message)`	Terminal success
`markFailed(message)`	Terminal failure
`getById(id)`	`?OutboxMessage`

types filters by message type (empty = all), which is how several consumers share one outbox. Since claim() hands a message to exactly one caller, the type sets of independent consumers must not overlap — otherwise each message reaches only whichever worker claimed it first.

RetryAwareStorageInterface

Extends StorageInterface. Optional: implement it when the backend can apply the retry policy inside the claim itself — see Letting the storage apply the retry policy.

Method	Description
`claimReady(readyThreshold, maxAttempts, types = [], limit = 1000)`	Like `claim()`, but skips messages still waiting for their next attempt. Takes a message when it has never been attempted, was last attempted at or before `readyThreshold`, or has already spent `maxAttempts` attempts

OutboxMessage

Method	Description
`create(type, payload, aggregateId?, createdAt?, id?)`	Factory; `id` omitted → 32-char hex
`getId()`	Message ID (32-char hex)
`getType()`	Message type
`getPayload()`	Raw payload string
`getStatus()`	`OutboxStatus` enum
`getCreatedAt()`	`DateTimeImmutable`
`getAttempts()`	Number of publish attempts
`getLastAttemptAt()`	`?DateTimeImmutable`
`getAggregateId()`	`?string`
`withStatus(status)`	Returns new instance with status
`withAttempt(at)`	Returns new instance with incremented attempts and timestamp

MessageIdGeneratorInterface

Implementation	Produces
`RandomHexIdGenerator` (default)	32 hex characters, 128 random bits
your own	anything non-empty; `id` is `VARCHAR(255)` in the DB adapter

OutboxStatus

Case	Value	Meaning
`Pending`	`'pending'`	Awaiting publication, including retries with `attempts > 0`
`Processing`	`'processing'`	Claimed by a worker; no other worker may take it
`Published`	`'published'`	Terminal success
`Failed`	`'failed'`	Terminal failure, retries exhausted

RetryPolicy

Method	Description
`__construct(maxAttempts, delaySeconds)`	Default: 3 attempts, 60s delay
`shouldRetry(message)`	Checks attempt count
`isReadyForRetry(message, now)`	Checks attempts + delay elapsed
`readyThreshold(now)`	`now - delaySeconds`: the same check as a boundary a storage can filter on. Feeds `RetryAwareStorageInterface::claimReady()`

Processor

Method	Description
`__construct(storage, publisher, retryPolicy, clock, batchSize, logger)`	Default batch: 100
`process()`	Returns `ProcessingResult`

ProcessingResult

Property/Method	Description
`$published`	Count of successfully published messages
`$failed`	Count of publish failures this run, plus messages claimed with no attempts left
`$skipped`	Count of claimed messages not ready for retry. `0` against a `RetryAwareStorageInterface`, which never claims them
`total()`	Sum of all counters

Serializer

Serializer implements SerializerInterface — the extension point a storage backend or transport uses to move a message across a boundary as a string. Swap in your own implementation for a different wire format; the interface is the contract, Serializer is the JSON default.

Method	Description
`serialize(message)`	Message to JSON
`deserialize(data)`	JSON to Message

deserialize() rejects every malformed input with InvalidArgumentException — missing fields, wrong types, an unknown status, an unparsable or empty datetime. A caller catching "bad input" never has to also catch ValueError or DateMalformedStringException from a field the parser forgot to guard.

Security

Storage implementations must use parameterized queries for all user values.
Message payload is stored as-is; validate before saving if needed.

Examples

See examples/ for complete usage examples.

Development

make install
make build
make cs-fix
make test
make test-coverage
make mutation
make release-check

make test-coverage and make mutation bootstrap pcov inside the composer:2 container because the base image has no coverage driver.

License

BSD-3-Clause. See LICENSE.md.

rasuvaeff / yii3-outbox

Maintainers

Package info

Statistics

Security

README

Requirements

Installation

Usage

Recording a message

The transactional guarantee

Message ids

Implementing storage

Letting the storage apply the retry policy

Implementing a publisher

Processing the outbox

Retry behaviour

Using InMemoryStorage for tests

API reference

Outbox

StorageInterface

RetryAwareStorageInterface

OutboxMessage

MessageIdGeneratorInterface

OutboxStatus

RetryPolicy

Processor

ProcessingResult

Serializer

Security

Examples

Development

License