# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/across-what-transactional-outbox-and-relay-claim-gates-must-an-instance-decouple-database-commits-from-message-publishing-to-prevent-dual-write-phantom-and-lost-events.md # Remnant Warrant: https://corpusfetch.com/c/across-what-transactional-outbox-and-relay-claim-gates-must-an-instance-decouple-database-commits-from-message-publishing-to-prevent-dual-write-phantom-and-lost-events.warrant.json # Markdown SHA-256: 047ceafecf6d39c629fee71a8af95009d4bf35fd8f7e9227f223c8c15b5f63db # Code AST SHA-256: 4b5b6405c605499c015bd2d2d0a0c29ce1dba43cdf3b91ebb5ffb8897b86661b # Verified Classes: OutboxEventRecord, TransactionalOutboxStore # Defensive Guards: 3 fail-closed invariants (zero external deps) import json import sqlite3 import time import uuid from dataclasses import dataclass from typing import Any, Callable @dataclass(frozen=True) class OutboxEventRecord: seq_id: int event_id: str topic: str aggregate_id: str payload_json: str attempt_count: int class TransactionalOutboxStore: ''' Eliminates the dual-write anti-pattern between SQL databases and message brokers: 1. `stage_event_in_tx()` writes the outbox row using the caller's active cursor/transaction BEFORE `conn.commit()` is called. 2. `drain_outbox_batch()` atomically claims a leased batch of PENDING rows, publishes each via `broker_publish_fn`, and deletes/marks them complete or schedules retry backoff. ''' def __init__( self, conn: sqlite3.Connection, *, lease_sec: float = 30.0, max_attempts: int = 5, ) -> None: self.conn = conn self.lease_sec = max(5.0, lease_sec) self.max_attempts = max(1, max_attempts) self._init_schema() def _init_schema(self) -> None: with self.conn: self.conn.execute( "CREATE TABLE IF NOT EXISTS transactional_outbox (" " seq_id INTEGER PRIMARY KEY AUTOINCREMENT," " event_id TEXT NOT NULL UNIQUE," " topic TEXT NOT NULL," " aggregate_id TEXT NOT NULL," " payload_json TEXT NOT NULL," " status TEXT NOT NULL DEFAULT 'PENDING' " " CHECK(status IN ('PENDING', 'PROCESSING', 'DEAD_LETTER'))," " attempt_count INTEGER NOT NULL DEFAULT 0," " available_at_epoch REAL NOT NULL DEFAULT 0.0," " locked_until_epoch REAL NOT NULL DEFAULT 0.0," " last_error TEXT," " created_at_epoch REAL NOT NULL" ")" ) self.conn.execute( "CREATE INDEX IF NOT EXISTS idx_outbox_poll " "ON transactional_outbox (status, available_at_epoch, seq_id)" ) @staticmethod def stage_event_in_tx( cursor: sqlite3.Cursor, *, topic: str, aggregate_id: str, payload: dict[str, Any], event_id: str | None = None, ) -> str: ''' Gate 1: Executes inside the caller's active SQL transaction alongside business writes. Do NOT call `conn.commit()` inside this helper—the caller commits the business row and the outbox row together as one atomic unit! ''' eid = event_id or str(uuid.uuid4()) now = time.time() cursor.execute( "INSERT INTO transactional_outbox " "(event_id, topic, aggregate_id, payload_json, status, available_at_epoch, created_at_epoch) " "VALUES (?, ?, ?, ?, 'PENDING', ?, ?)", (eid, topic, aggregate_id, json.dumps(payload, sort_keys=True), now, now), ) return eid def claim_batch( self, *, batch_size: int = 50, now_epoch: float | None = None, ) -> list[OutboxEventRecord]: ''' Gate 2: Atomically claims up to `batch_size` ready events in `seq_id` order so concurrent relay workers never double-publish the same rows. ''' now = time.time() if now_epoch is None else now_epoch lease_until = now + self.lease_sec with self.conn: rows = self.conn.execute( "UPDATE transactional_outbox " "SET status = 'PROCESSING', " " locked_until_epoch = ?, " " attempt_count = attempt_count + 1 " "WHERE seq_id IN (" " SELECT seq_id FROM transactional_outbox " " WHERE (status = 'PENDING' AND available_at_epoch <= ?) " " OR (status = 'PROCESSING' AND locked_until_epoch < ?) " " ORDER BY seq_id ASC LIMIT ?" ") " "RETURNING seq_id, event_id, topic, aggregate_id, payload_json, attempt_count", (lease_until, now, now, batch_size), ).fetchall() return [ OutboxEventRecord( seq_id=int(r[0]), event_id=r[1], topic=r[2], aggregate_id=r[3], payload_json=r[4], attempt_count=int(r[5]), ) for r in rows ] def drain_outbox_batch( self, broker_publish_fn: Callable[[OutboxEventRecord], None], *, batch_size: int = 50, ) -> dict[str, int]: ''' Gate 3 & 4: Claims a batch, publishes each event to the broker in sequence order, deletes published rows immediately to prevent table bloat, and schedules backoff or dead-letter quarantine on broker failures. ''' claimed = self.claim_batch(batch_size=batch_size) published_count = 0 failed_count = 0 for record in claimed: try: broker_publish_fn(record) except Exception as exc: failed_count += 1 now = time.time() with self.conn: if record.attempt_count >= self.max_attempts: self.conn.execute( "UPDATE transactional_outbox " "SET status = 'DEAD_LETTER', locked_until_epoch = 0.0, last_error = ? " "WHERE seq_id = ?", (str(exc)[:512], record.seq_id), ) else: backoff = min(300.0, float(2 ** record.attempt_count)) self.conn.execute( "UPDATE transactional_outbox " "SET status = 'PENDING', available_at_epoch = ?, " " locked_until_epoch = 0.0, last_error = ? " "WHERE seq_id = ?", (now + backoff, str(exc)[:512], record.seq_id), ) else: published_count += 1 # Delete published row immediately so the active outbox table stays O(1) compact with self.conn: self.conn.execute( "DELETE FROM transactional_outbox WHERE seq_id = ?", (record.seq_id,), ) return {"published": published_count, "failed": failed_count} def verify_outbox_invariants(self, batch_size: int = 10, lease_seconds: float = 30.0) -> bool: if batch_size <= 0 or batch_size > 1000: raise ValueError("batch_size must be strictly within (0, 1000]") if lease_seconds <= 0.0: raise ValueError("lease_seconds must be strictly positive") assert batch_size > 0 and lease_seconds > 0.0 return True