README

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

Database-backed storage for rasuvaeff/yii3-outbox. Durably persists outbox messages in a yiisoft/db table so a worker can publish or export them asynchronously — surviving process restarts and downstream outages.

Using an AI coding assistant? llms.txt has a compact API reference you can use.

Requirements

• PHP 8.3+
• rasuvaeff/yii3-outbox ^1.7
• yiisoft/db ^2.0, yiisoft/db-migration ^2.1 (2.1.0 is where setSourceNamespaces() finds a vendor migration at all)
• symfony/console ^6.4 || ^7.0 — for the three housekeeping commands

Installation

composer require rasuvaeff/yii3-outbox-db

Usage

Migration

Register the bundled migration by namespace — no vendor paths:

// config/common/di/migration.php
use Yiisoft\Db\Migration\Service\MigrationService;

return [
    MigrationService::class => [
        'setSourceNamespaces()' => [[
            'App\\Migration',
            'Rasuvaeff\\Yii3OutboxDb\\Migration',
        ]],
    ],
];

./yii migrate:up

yiisoft/db-migration resolves the migration through Injector::make(), so it picks up the table-name value object from the container the same way the storage does — no manual wiring needed beyond setSourceNamespaces() above.

Set the table name in params — the same value reaches the migration and DbOutboxStorage:

// config/common/params.php
'rasuvaeff/yii3-outbox-db' => [
    'table' => 'my_outbox',
    'table_prefix' => '',   // prepended to `table`; e.g. 'rsv_' → rsv_my_outbox
],

Index names follow the table name (idx_my_outbox_pending, idx_my_outbox_processing), so two installations can share one PostgreSQL schema — index names are unique per schema there, not per table.

Migrations, in order:

M260820000000AddOutboxClaimedAt::down() works on MySQL and PostgreSQL only — yiisoft/db-sqlite cannot drop a column.

Payload size

payload is TEXT. PostgreSQL and SQLite treat that as unbounded; MySQL caps it at 65,535 bytes. That ceiling is generous for a domain event — an outbox payload should carry a reference, not a blob — so the schema does not force a table rebuild on every MySQL installation to raise it. If your events genuinely need more, widen the column yourself once:

-- Substitute your configured table: `table_prefix` + `table` from params,
-- `outbox` by default.
ALTER TABLE outbox MODIFY payload MEDIUMTEXT NOT NULL;

Know what the limit does if you hit it: in MySQL's strict mode (the default since 5.7) the insert fails, and because Outbox::record() runs inside your business transaction, that failure rolls back the business write too. In a permissive mode the payload is silently truncated instead, and the message is published with whatever survived the cut — a JSON payload will almost always be left unparseable, and one that does parse is worse, because the consumer accepts a corrupted event without noticing.

The DI entry point is MigrationService, not the migration class. Registering the namespace on MigrationService::setSourceNamespaces(), as above, is the supported recipe. A definition keyed by the migration itself — M...::class => ['__construct()' => ['table' => ...]] — has no effect: the migration is built by Injector::make(), which resolves constructor arguments by type from the container and never reads a container definition keyed by the class being made. Set the table in params instead; the OutboxTableName built from them is what Injector resolves by type, for the migration and the storage alike.

Recording and processing

use Rasuvaeff\Yii3Outbox\Outbox;
use Rasuvaeff\Yii3OutboxDb\DbOutboxStorage;

$storage = new DbOutboxStorage(db: $connection);          // ConnectionInterface
$outbox = new Outbox(storage: $storage, clock: $clock);

// request path — durable, no network call to the sink
$outbox->record(type: 'ab.exposure', payload: '{"experiment":"checkout"}');

// worker — atomically claim a batch of one consumer's types and process them
$claimed = $storage->claim(types: ['ab.exposure', 'ab.conversion'], limit: 1000);

Storage API

The backoff never reaches PHP

DbOutboxStorage implements Rasuvaeff\Yii3Outbox\RetryAwareStorageInterface, so Processor claims through claimReady() and a message whose retry delay has not elapsed is never taken from the table. Before, every Pending row was claimed and the core wrote the not-yet-due ones straight back — two writes per backing-off message per iteration, each occupying a slot in batchSize that a ready message could have used.

The extra condition is one clause:

AND (attempts >= :maxAttempts
     OR last_attempt_at IS NULL
     OR last_attempt_at <= :readyThreshold)

attempts >= :maxAttempts is not an optimisation. A message out of attempts can only ever be marked Failed, and Processor can only fail a message the storage handed it — filter it out and nothing terminates it: it stays Pending forever, invisible to an alert watching Failed.

No migration and no new index come with this. idx_<table>_pending (status, type, created_at) still narrows the scan and serves the ordering; the added disjunction is an OR across two columns, which no index can satisfy as a whole, and it is evaluated on rows the existing index already selected.

Two things change for an operator:

• ProcessingResult::$skipped reads 0 — the messages it used to count are no longer claimed. Count Pending rows whose last_attempt_at is recent if you want to know how many are backing off.
• A message that has spent its attempts is marked Failed up to delaySeconds later than before, since it waits for a batch that includes it.

claim() vs findPending()

claim() is the primitive a worker must use, and the one Processor calls. It runs inside a transaction: it selects the pending ids, stamps them Processing with a random claimed_by token, then re-reads exactly the rows carrying that token. Two workers polling concurrently therefore never receive the same message.

findPending() is a plain read. Nothing is locked or marked, so two workers polling it both get the same rows and publish the same message twice. Use it for dashboards, admin screens and diagnostics — never as a worker's fetch.

Every claimed message must reach a terminal state: markPublished(), markFailed(), or save($message->withStatus(OutboxStatus::Pending)) to release it.

Several workers on MySQL or PostgreSQL: skipLocked: true. The token scheme is correct with any number of workers, but their claims serialise on row locks: a worker whose candidate rows overlap with another's waits for that transaction to finish (on MySQL up to innodb_lock_wait_timeout, 50 s by default) before it learns the rows are gone. With the flag the id select runs FOR UPDATE SKIP LOCKED, so a claim takes whatever is free at once and never waits on a sibling:

new DbOutboxStorage(db: $connection, skipLocked: true);
// or params: 'skip_locked' => true

MySQL 8+ and PostgreSQL 9.5+ only. SQLite has no FOR clause and rejects the claim with NotSupportedException at query time — keep the flag off in a SQLite-backed test environment.

Recovering stale claims

A worker killed between claim() and the finalising write — SIGKILL under supervisor or k8s, an OOM, a daemon timeout — leaves its rows in Processing, and no amount of retry logic brings them back on its own. claim() stamps claimed_at, so an abandoned claim is distinguishable from a fresh one:

$threshold = $clock->now()->modify('-15 minutes');

// Look first — this is also what a monitoring endpoint should report.
$stuck = $storage->findStaleClaims($threshold);

// Then put them back; they return to Pending without spending an attempt.
$released = $storage->releaseStaleClaims($threshold);

Run the release from a cron or a supervisor hook, with a threshold comfortably longer than the slowest batch: releasing a claim a live worker still holds means the message is delivered twice, which the at-least-once contract permits but nobody enjoys. A Processing row with no timestamp counts as stale, whether it was left by a version predating the column or written by save() — which always clears claimed_by along with claimed_at. Neither row is held by a live claim, which is exactly what the missing claimed_by says.

A growing Processing count still deserves an alert; now it also has a cure. stats() is how to read it without SQL:

$stats = $storage->stats();           // one GROUP BY query
$stats->processing;                   // what the alert watches
$stats->failed;                       // what the other alert watches
$stats->oldestPendingAgeSeconds($clock->now());   // how far behind the worker is

Console commands

Three housekeeping commands — Console\PurgeOutboxCommand, Console\ReleaseStaleOutboxClaimsCommand, Console\RequeueFailedOutboxCommand — are registered for yiisoft/yii-console (they work in any Symfony Console application; the container needs a StorageInterface bound to DbOutboxStorage, which config/di.php does):

./yii outbox:release-stale --claimed-before=15m        # the cron above; default 15m, --limit=1000
./yii outbox:purge --older-than=7d                     # Published rows created more than 7 days ago
./yii outbox:purge --status=failed --older-than=30d    # or Failed ones; Pending/Processing are never purged
./yii outbox:requeue --type=order.created --limit=500  # Failed -> Pending with attempts reset; all types by default

Ages are <count><unit> with s, m, h or d; a malformed age or limit exits Command::INVALID without touching a row. Without --older-than, outbox:purge deletes every row in the status — the pre-2.4 behaviour.

Guarding the transaction in development

The outbox pattern only holds when record() runs inside the business transaction, and nothing can enforce that in production without a cost. In development and CI the cost is fine:

new DbOutboxStorage(db: $connection, requireTransaction: true);
// or params: 'require_transaction' => true

A save() that would create a new row — what record() does — then throws Exception\OutboxWriteOutsideTransactionException unless a transaction is open on the connection; saveBatch() (behind recordMany()) likewise. A worker updating an existing row between attempts is not affected: the guard checks whether the row exists, which is the one extra query the mode costs.

The $types filter lets several consumers — a generic Processor and a specialized exporter — share one outbox. Because claim() hands each message to exactly one caller, their type sets must not overlap: a message matching both is delivered only to whichever worker claimed it first.

Acknowledging a batch, and whether to keep what was sent

markPublished() is one upsert per message. A consumer that delivers a batch as a unit — yii3-outbox-clickhouse writes one bulk insert per group — used to acknowledge a thousand-message group with a thousand statements in the OLTP database. DbOutboxStorage implements BatchAcknowledgingStorageInterface from the core, and such a consumer acknowledges the whole group through markPublishedBatch(): one UPDATE … WHERE id IN (…) per distinct attempt stamp, which for a group acknowledged together is one statement. Each row ends up exactly as markPublished() would have left it.

What happens to an acknowledged row is a constructor flag:

new DbOutboxStorage(db: $connection, deletePublished: true);
// or, through the config-plugin:
'rasuvaeff/yii3-outbox-db' => ['delete_published' => true],

The flag governs markPublished() too, so Processor and a batching exporter sharing one table agree on what it holds. The price of true is the audit trail of what was sent — observe it at the sink instead (the ClickHouse event id is the outbox id). At-least-once is unchanged either way: a crash between the sink write and the acknowledgement leaves the row Processing, the stale-claim release returns it to Pending, and it is delivered again and deduplicated downstream.

The acknowledgement carries no status = 'processing' guard on purpose. It runs after a successful delivery and a row is immutable apart from its status, so acknowledging it is right whatever state it is in: a row the stale-claim cron released and a second worker re-claimed is acknowledged by the first worker's batch all the same, and the second worker's own acknowledgement then finds nothing to do. A guard would only leave the released row in place — a guaranteed second delivery.

Yii3 DI

The config-plugin binds StorageInterface to DbOutboxStorage from config/di.php. Core yii3-outbox binds nothing, so this backend (or the application) is the single source of StorageInterface. Set the table name in params:

// config/common/params.php
'rasuvaeff/yii3-outbox-db' => [
    'table' => 'outbox',
    'delete_published' => false,    // true: acknowledged rows are deleted, no purge cron needed
    'require_transaction' => false, // true in dev/CI: record() outside a transaction throws
    'skip_locked' => false,         // true on MySQL 8+/PostgreSQL 9.5+ with several workers
],

The same file registers outbox:purge, outbox:release-stale and outbox:requeue under yiisoft/yii-console.

Security

• All values are written through yiisoft/db parameterized commands.
• OutboxRowMapper validates every column and rejects corrupt rows with InvalidOutboxRowException — no silent coercion.
• Payloads may contain PII; retention/purging is the application's responsibility (deleteByStatus helps, deletePublished: true removes a row the moment it is acknowledged).

Examples

Runnable scripts live in examples/.

Development

make build        # full gate: validate + normalize + require-checker + cs + psalm + test
make cs-fix
make psalm
make test
make test-coverage
make mutation

Core yii3-outbox is consumed via a path repository while unpublished — see AGENTS.md for the monorepo-root Docker invocation.

License

BSD-3-Clause. See LICENSE.md.