Download the PHP package rasuvaeff/yii3-outbox without Composer

rasuvaeff/yii3-outbox

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

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

Usage

Recording a message

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:

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

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:

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:

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.

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:

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

Processing the outbox

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.

Using InMemoryStorage for tests

API reference

Outbox

StorageInterface

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.

OutboxMessage

MessageIdGeneratorInterface

OutboxStatus

RetryPolicy

Processor

ProcessingResult

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.

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 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.