Skip to content

Commit 3d3c1fe

Browse files
muhammetsafakmuhammetsafak
committed
feat(outbox): transactional outbox helper (ADR-0029)
New dependency-free src/Outbox/: OutboxStore interface + Outbox writer + OutboxRelay (at-least-once, idempotent re-publish, bounded backoff) + InMemoryOutboxStore. The envelope is stored and re-published byte-identical (GR-1/GR-4/GR-5); the caller owns the DB transaction so the message row is written atomically with the business data. Core stays ext-json only (GR-7); DB binding is via the OutboxStore seam (InitORM adapter shipped as an example).
1 parent 25c428e commit 3d3c1fe

8 files changed

Lines changed: 748 additions & 0 deletions

File tree

‎README.md‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ composer require babelqueue/php-sdk
3232
| Validation | `BabelQueue\Validation\EnvelopeValidator` | Consumer-side validation **with a reason** — quarantine an unsupported `meta.schema_version` instead of dropping it. |
3333
| Transports | `BabelQueue\Transport\RedisTransport` / `AmqpTransport` | Optional framework-less reference `Transport` impls (Redis `RPUSH`; RabbitMQ durable + contract AMQP properties). |
3434
| Dead-letter | `BabelQueue\DeadLetter\DeadLetter` | Annotate an envelope with the additive `dead_letter` block (ADR-0009). |
35+
| Outbox | `BabelQueue\Outbox\Outbox` / `OutboxRelay` / `OutboxStore` | Transactional outbox (ADR-0029): persist the message **atomically with the business write**, relay it later. Dependency-free — `OutboxStore` is an interface you bind to your DB. |
3536
| Routing | `BabelQueue\Routing\UnknownUrnStrategy` | `fail` / `delete` / `release` / `dead_letter` constants. |
3637
| Support | `BabelQueue\Support\Uuid` | Dependency-free UUIDv4 (no ramsey/symfony-uid needed). |
3738
| Errors | `BabelQueue\Exceptions\BabelQueueException` / `UnknownUrnException` / `InvalidEnvelopeException` | Exception hierarchy; `InvalidEnvelopeException` carries the rejection reason + envelope. |
@@ -72,6 +73,43 @@ if ($reason = EnvelopeValidator::check($envelope)) {
7273
phpredis (`ext-redis`) users can implement the one-method `Transport` directly —
7374
it is just an `rpush`.
7475

76+
## Transactional outbox (ADR-0029)
77+
78+
A plain producer makes a **dual write** — commit the business row *and* publish to the
79+
broker — two systems that can disagree on a crash. The outbox removes it: the message is
80+
written into the **same database, in the same transaction** as the business data (so they
81+
commit or roll back atomically), and a separate **relay** publishes it afterwards. No
82+
distributed transaction; exactly-once *handoff* into the broker (then at-least-once on the
83+
wire, deduped on `meta.id` by the consumer-side `Idempotent::wrap`, ADR-0022).
84+
85+
The helper is dependency-free (GR-7): the core defines `OutboxStore` and you bind it to
86+
your DB. **The transaction boundary is yours** — `OutboxStore::save()` runs inside the
87+
transaction *you* opened around your business write. The envelope is stored **verbatim**
88+
(GR-1) and relayed byte-for-byte, so `trace_id` is preserved end-to-end (GR-4).
89+
90+
```php
91+
use BabelQueue\Codec\EnvelopeCodec;
92+
use BabelQueue\Outbox\Outbox;
93+
use BabelQueue\Outbox\OutboxRelay;
94+
95+
// WRITE side — one transaction for the business row AND the message (your tx boundary).
96+
$outbox = new Outbox($store); // $store implements OutboxStore (your DB)
97+
$db->transaction(function () use ($db, $outbox, $order): void {
98+
$db->insertOrder($order); // business write
99+
$outbox->write(EnvelopeCodec::make('urn:babel:orders:created', $order, 'orders'));
100+
}); // both commit, or neither
101+
102+
// READ side — a worker/cron drains the durable rows onto the broker.
103+
$relay = new OutboxRelay($transport, $store); // $transport is any BabelQueue Transport
104+
$relay->drain(); // publishes verbatim, marks published/failed
105+
```
106+
107+
`OutboxStore` is four methods — `save()`, `fetchUnpublished()`, `markPublished()`,
108+
`markFailed()`. A reference `InMemoryOutboxStore` ships for tests; a runnable
109+
**InitORM-backed** adapter + the outbox-table DDL live in
110+
[`babelqueue-examples/outbox-initorm/`](https://github.com/BabelQueue/babelqueue-examples/tree/main/outbox-initorm),
111+
keeping this core DB-free.
112+
75113
## Design
76114

77115
This core is the **contract runtime**, not a worker. It does not own a broker

‎src/Outbox/InMemoryOutboxStore.php‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace BabelQueue\Outbox;
6+
7+
/**
8+
* Process-local reference {@see OutboxStore} backed by arrays — for tests and single-process
9+
* demos. It has **no real transaction**: {@see save()} just appends, so it cannot deliver the
10+
* atomic-with-the-business-write guarantee a production store gives. Use a database-backed
11+
* adapter (e.g. the InitORM one in `babelqueue-examples/`) in production.
12+
*
13+
* It still faithfully models the relay contract: rows are pending until {@see markPublished()},
14+
* {@see fetchUnpublished()} returns them oldest-first, and {@see markFailed()} bumps the attempt
15+
* count and stores the last error while leaving the row pending for retry.
16+
*/
17+
final class InMemoryOutboxStore implements OutboxStore
18+
{
19+
/** @var array<string, array{body: string, queue: string, attempts: int, published: bool, error: string}> */
20+
private array $rows = [];
21+
22+
private int $sequence = 0;
23+
24+
public function save(string $encodedEnvelope, string $queue): string
25+
{
26+
// A non-numeric id keeps the array key a genuine string (PHP coerces numeric-string
27+
// keys to int), so the contract's string-keyed store stays honest.
28+
$id = 'ob-' . ++$this->sequence;
29+
$this->rows[$id] = [
30+
'body' => $encodedEnvelope,
31+
'queue' => $queue,
32+
'attempts' => 0,
33+
'published' => false,
34+
'error' => '',
35+
];
36+
37+
return $id;
38+
}
39+
40+
public function fetchUnpublished(int $limit): array
41+
{
42+
$records = [];
43+
foreach ($this->rows as $id => $row) {
44+
if ($row['published']) {
45+
continue;
46+
}
47+
$records[] = new OutboxRecord($id, $row['body'], $row['queue'], $row['attempts']);
48+
if (count($records) >= $limit) {
49+
break;
50+
}
51+
}
52+
53+
return $records;
54+
}
55+
56+
public function markPublished(array $ids): void
57+
{
58+
foreach ($ids as $id) {
59+
if (isset($this->rows[$id])) {
60+
$this->rows[$id]['published'] = true;
61+
}
62+
}
63+
}
64+
65+
public function markFailed(string $id, string $error): void
66+
{
67+
if (isset($this->rows[$id])) {
68+
$this->rows[$id]['attempts']++;
69+
$this->rows[$id]['error'] = $error;
70+
}
71+
}
72+
73+
/**
74+
* Test/inspection helper: the number of rows still pending publish.
75+
*/
76+
public function pendingCount(): int
77+
{
78+
$pending = 0;
79+
foreach ($this->rows as $row) {
80+
if (! $row['published']) {
81+
$pending++;
82+
}
83+
}
84+
85+
return $pending;
86+
}
87+
88+
/**
89+
* Test/inspection helper: the recorded attempt count for one row (0 if unknown).
90+
*/
91+
public function attemptsOf(string $id): int
92+
{
93+
return $this->rows[$id]['attempts'] ?? 0;
94+
}
95+
96+
/**
97+
* Test/inspection helper: the last recorded error for one row ('' if none).
98+
*/
99+
public function lastErrorOf(string $id): string
100+
{
101+
return $this->rows[$id]['error'] ?? '';
102+
}
103+
}

‎src/Outbox/Outbox.php‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace BabelQueue\Outbox;
6+
7+
use BabelQueue\Codec\EnvelopeCodec;
8+
9+
/**
10+
* The **write side** of the transactional outbox (ADR-0029): turn a BabelQueue envelope
11+
* into a stored outbox row, so the message is persisted *atomically with the business
12+
* data* and a separate {@see OutboxRelay} publishes it later.
13+
*
14+
* Usage — the caller owns the transaction boundary (this is the whole point):
15+
*
16+
* $db->begin();
17+
* try {
18+
* $db->insertOrder($order); // the business write
19+
* $envelope = EnvelopeCodec::fromJob($job, 'orders');
20+
* $outbox->write($envelope); // same connection, same tx
21+
* $db->commit(); // both, or neither
22+
* } catch (\Throwable $e) {
23+
* $db->rollBack();
24+
* throw $e;
25+
* }
26+
*
27+
* Because both writes share one transaction, a crash can never leave the business row
28+
* committed without its message (the classic dual-write bug) — they commit or roll back
29+
* together. The handoff to the broker becomes a *local* problem the relay solves.
30+
*
31+
* This helper is intentionally tiny and dependency-free: it only encodes via the frozen
32+
* {@see EnvelopeCodec} (GR-1 — the envelope bytes are stored unchanged; the outbox never
33+
* adds an envelope field) and delegates persistence to the injected {@see OutboxStore},
34+
* which the caller binds to their own DB (GR-7). It does **not** begin/commit anything.
35+
*/
36+
final class Outbox
37+
{
38+
public function __construct(
39+
private readonly OutboxStore $store,
40+
) {
41+
}
42+
43+
/**
44+
* Encode the envelope (frozen codec, bytes unchanged) and persist it via the store,
45+
* inside the transaction the caller has already opened. Returns the new outbox row id.
46+
*
47+
* @param array<string, mixed> $envelope A canonical envelope from
48+
* {@see EnvelopeCodec::make()} / {@see EnvelopeCodec::fromJob()}.
49+
* @return string The outbox row id (for the caller's own correlation, if wanted).
50+
*
51+
* @throws \JsonException When the envelope is not cleanly encodable (same as the codec).
52+
*/
53+
public function write(array $envelope): string
54+
{
55+
$queue = $this->queueOf($envelope);
56+
57+
return $this->store->save(EnvelopeCodec::encode($envelope), $queue);
58+
}
59+
60+
/**
61+
* The logical queue the message targets: its `meta.queue`, falling back to "default".
62+
* Captured at write time so the relay can publish to the right queue without decoding
63+
* the body.
64+
*
65+
* @param array<string, mixed> $envelope
66+
*/
67+
private function queueOf(array $envelope): string
68+
{
69+
$meta = $envelope['meta'] ?? null;
70+
if (is_array($meta) && isset($meta['queue']) && is_string($meta['queue']) && $meta['queue'] !== '') {
71+
return $meta['queue'];
72+
}
73+
74+
return 'default';
75+
}
76+
}

‎src/Outbox/OutboxRecord.php‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace BabelQueue\Outbox;
6+
7+
/**
8+
* One pending row read back from an {@see OutboxStore} for the {@see OutboxRelay} to
9+
* publish. It pairs the store's own bookkeeping (id, attempts) with the verbatim,
10+
* frozen wire envelope ({@see body}) and the queue it should go to.
11+
*
12+
* {@see body} is the exact {@see \BabelQueue\Codec\EnvelopeCodec}-encoded JSON that was
13+
* handed to {@see OutboxStore::save()} — the relay publishes these bytes unchanged
14+
* (GR-1/GR-5), so `trace_id` is preserved end-to-end (GR-4) without the relay ever
15+
* decoding or rebuilding the envelope.
16+
*/
17+
final class OutboxRecord
18+
{
19+
/**
20+
* @param string $id The outbox row id (the store's primary key, not `meta.id`).
21+
* @param string $body The frozen, encoded envelope JSON, byte-for-byte as stored.
22+
* @param string $queue The logical queue the relay should publish to.
23+
* @param int $attempts How many times the relay has already tried to publish this row.
24+
*/
25+
public function __construct(
26+
public readonly string $id,
27+
public readonly string $body,
28+
public readonly string $queue,
29+
public readonly int $attempts = 0,
30+
) {
31+
}
32+
}

0 commit comments

Comments
 (0)