Authors tests for the transactional outbox pattern: atomic DB-write-plus-event-insert in one transaction, relay/poller publishing with at-least-once delivery and consumer deduplication, insertion-order preservation, idempotent consumers, and relay failure/retry. Use when adding outbox infrastructure, changing the relay or poller, or auditing whether dual-write atomicity and at-least-once delivery guarantees hold under failure.
74
93%
Does it follow best practices?
Run evals on this skill
Adds up to 20 points to the overall score
View guide
Passed
No findings from the security scan
The transactional outbox is the canonical solution to the dual-write problem: a service cannot atomically update its database AND publish a message to a broker in one step without 2PC. Per microservices.io/transactional-outbox, "the service that sends the message first stores the message in the database as part of the transaction that updates the business entities. A separate process then sends the messages to the message broker." This gives the guarantee that "messages are guaranteed to be sent if and only if the database transaction commits."
saga-transaction-tests names outbox atomicity in its Step 6 but contains no
relay, deduplication, ordering, or retry recipes. This skill covers all five
test dimensions end to end.
Before writing any test, enumerate which dimensions apply:
| Dimension | What to verify |
|---|---|
| Atomicity | DB write + outbox insert commit together; partial failure rolls both back |
| Relay publishing | Poller picks up pending rows and publishes to the broker |
| At-least-once + dedup | Relay may re-publish; consumer deduplicates on message ID |
| Ordering | Events published in insertion order per aggregate |
| Relay retry | Relay crash/restart re-publishes unpublished rows; no loss |
Each dimension maps to one test group in the sections below.
Per microservices.io/transactional-outbox, the outbox insert must be inside the same DB transaction as the business entity update. Test both the happy path and the mid-transaction failure path:
def test_outbox_insert_commits_with_business_entity(db):
with db.transaction():
db.execute("INSERT INTO orders (id, status) VALUES ('o1', 'PLACED')")
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-1', 'OrderPlaced', '{\"order_id\":\"o1\"}', 'PENDING')"
)
assert db.scalar("SELECT COUNT(*) FROM orders WHERE id='o1'") == 1
assert db.scalar("SELECT COUNT(*) FROM outbox WHERE id='evt-1'") == 1
def test_mid_transaction_failure_rolls_back_both(db):
with pytest.raises(SimulatedDBError):
with db.transaction():
db.execute("INSERT INTO orders (id, status) VALUES ('o2', 'PLACED')")
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-2', 'OrderPlaced', '{\"order_id\":\"o2\"}', 'PENDING')"
)
raise SimulatedDBError()
# Both must roll back - this is the guarantee from [tx-outbox]
assert db.scalar("SELECT COUNT(*) FROM orders WHERE id='o2'") == 0
assert db.scalar("SELECT COUNT(*) FROM outbox WHERE id='evt-2'") == 0The polling publisher queries the outbox for rows with status='PENDING'
and publishes them to the broker. Per
microservices.io/polling-publisher, the relay works with "any
SQL database" and is the simpler alternative to transaction log tailing.
def test_relay_publishes_pending_events(db, broker, relay):
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-3', 'OrderPlaced', '{\"order_id\":\"o3\"}', 'PENDING')"
)
relay.run_once()
assert broker.published_ids() == ["evt-3"]
assert db.scalar("SELECT status FROM outbox WHERE id='evt-3'") == "PUBLISHED"
def test_relay_skips_already_published_events(db, broker, relay):
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-4', 'OrderPlaced', '{}', 'PUBLISHED')"
)
relay.run_once()
assert broker.published_ids() == [] # nothing re-sentPer microservices.io/transactional-outbox, "the message relay might publish a message more than once. It might, for example, crash after publishing a message but before recording the fact that it has done so." Consumers must therefore be idempotent by tracking processed message IDs.
def test_relay_may_publish_same_event_twice(db, broker, relay):
# Simulate: relay published but crashed before marking PUBLISHED
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-5', 'OrderPlaced', '{\"order_id\":\"o5\"}', 'PENDING')"
)
broker.inject_publish_then_crash_before_ack()
with contextlib.suppress(RelayRestartedError):
relay.run_once()
relay.run_once() # second pass after restart
# Broker received it at least once - possibly twice
assert len(broker.received("evt-5")) >= 1
def test_consumer_deduplicates_duplicate_delivery(consumer, broker):
event = {"id": "evt-5", "event_type": "OrderPlaced", "payload": {"order_id": "o5"}}
consumer.handle(event)
consumer.handle(event) # duplicate delivery
# Business effect applied exactly once
assert consumer.processed_order_count("o5") == 1
# Both invocations recorded (dedup table has the ID)
assert consumer.seen_message_id("evt-5") is TrueThe consumer's dedup check must compare the outbox row's id column (a
stable, relay-assigned UUID) - not content hashing, which breaks for
non-deterministic payloads.
Per microservices.io/transactional-outbox, "messages must be
sent to the message broker in the order they were sent by the service. This
ordering must be preserved across multiple service instances."
microservices.io/polling-publisher notes ordering is "tricky"
for the polling implementation - verify the relay uses ORDER BY inserted_at ASC (or equivalent sequence column) explicitly.
def test_relay_publishes_events_in_insertion_order(db, broker, relay):
for i in range(1, 4):
db.execute(
"INSERT INTO outbox (id, event_type, payload, status, inserted_at) "
"VALUES (?, 'ItemAdded', '{}', 'PENDING', ?)",
f"evt-order-{i}",
datetime.utcnow() + timedelta(milliseconds=i),
)
relay.run_once()
published = broker.published_ids()
assert published == ["evt-order-1", "evt-order-2", "evt-order-3"]If the relay does not enforce ordering, this test will catch non-deterministic sequences on high-concurrency inserts.
The relay must re-publish any rows still PENDING after a crash. This
validates the recovery contract that gives at-least-once semantics:
def test_relay_retries_pending_rows_after_crash(db, broker, relay):
db.execute(
"INSERT INTO outbox (id, event_type, payload, status) "
"VALUES ('evt-retry', 'PaymentCaptured', '{}', 'PENDING')"
)
relay.crash_after_n_publishes = 0 # crash before any publish
with contextlib.suppress(RelayCrashError):
relay.run_once()
# Row is still PENDING after crash
assert db.scalar("SELECT status FROM outbox WHERE id='evt-retry'") == "PENDING"
relay.crash_after_n_publishes = None # recover
relay.run_once()
assert broker.published_ids() == ["evt-retry"]
assert db.scalar("SELECT status FROM outbox WHERE id='evt-retry'") == "PUBLISHED"Pair this with the consumer deduplication test (Step 4) - a relay retry produces a duplicate delivery that the consumer must absorb.
| Anti-pattern | Consequence | Fix |
|---|---|---|
| Outbox insert outside the transaction | Business entity commits; event never published (silent data loss) | Step 2 atomicity test |
| Relay marks PUBLISHED before broker ACK | Crash window loses the event | Step 6 retry test |
| Consumer lacks dedup table | Duplicate delivery causes double-charge, double-refund | Step 4 dedup test |
Relay polls without ORDER BY | Non-deterministic publish order; downstream ordering bugs | Step 5 ordering test |
| Reuse saga-transaction-tests Step 6 only | No relay, dedup, ordering, or retry coverage | All steps above |
saga-transaction-tests - saga
orchestration and choreography; pairs with this skill when sagas use the
outbox for step-event publishingeventual-consistency-tests -
assertion of the consistency window between outbox publish and consumer
processing