← back to Unclaimed Property Platform
Cycle 9: outbox worker — completes the transactional-outbox pattern
e3a39c32c95eba763030ddc244ead4d91af5bb97 · 2026-08-01 21:01:39 -0700 · Steve Abrams
- services/claims/outbox_worker.py: drains pending outbox_event -> StateAdapter with
AT-LEAST-ONCE semantics + bounded retry. Transient fail -> stays pending (attempts++);
exhausted -> 'failed' (claim NOT lost, stays SUBMITTING for ops); already-SUBMITTED ->
treated as idempotent success (no retry loop).
- schema: outbox_event gains attempts + last_error; SqliteClaimRepository fetch_pending_outbox
/ mark_outbox / outbox_row.
- tests/test_cycle10_outbox_worker.py: transient-then-success, no-op re-drain, permanent-fail
exhaustion. Now events are not just WRITTEN (Cycle 8) but reliably DELIVERED.
Tests: 10/10 suites green. All local/synthetic/$0.
TK-10097
Files touched
M db/schema.sqlA services/claims/outbox_worker.pyM services/claims/sqlite_claim_repo.pyA tests/test_cycle10_outbox_worker.py
Diff
commit e3a39c32c95eba763030ddc244ead4d91af5bb97
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Aug 1 21:01:39 2026 -0700
Cycle 9: outbox worker — completes the transactional-outbox pattern
- services/claims/outbox_worker.py: drains pending outbox_event -> StateAdapter with
AT-LEAST-ONCE semantics + bounded retry. Transient fail -> stays pending (attempts++);
exhausted -> 'failed' (claim NOT lost, stays SUBMITTING for ops); already-SUBMITTED ->
treated as idempotent success (no retry loop).
- schema: outbox_event gains attempts + last_error; SqliteClaimRepository fetch_pending_outbox
/ mark_outbox / outbox_row.
- tests/test_cycle10_outbox_worker.py: transient-then-success, no-op re-drain, permanent-fail
exhaustion. Now events are not just WRITTEN (Cycle 8) but reliably DELIVERED.
Tests: 10/10 suites green. All local/synthetic/$0.
TK-10097
---
db/schema.sql | 4 +-
services/claims/outbox_worker.py | 62 +++++++++++++++++++++
services/claims/sqlite_claim_repo.py | 25 +++++++++
tests/test_cycle10_outbox_worker.py | 103 +++++++++++++++++++++++++++++++++++
4 files changed, 193 insertions(+), 1 deletion(-)
diff --git a/db/schema.sql b/db/schema.sql
index 71f2365..b5efb41 100644
--- a/db/schema.sql
+++ b/db/schema.sql
@@ -145,7 +145,9 @@ CREATE TABLE outbox_event ( -- transactional outbox → reliable stat
aggregate_id TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
- delivery_state TEXT NOT NULL DEFAULT 'pending' -- pending|sent|failed
+ delivery_state TEXT NOT NULL DEFAULT 'pending', -- pending|sent|failed
+ attempts INTEGER NOT NULL DEFAULT 0,
+ last_error TEXT
);
-- ---------------------------------------------------------------------------
diff --git a/services/claims/outbox_worker.py b/services/claims/outbox_worker.py
new file mode 100644
index 0000000..77c63be
--- /dev/null
+++ b/services/claims/outbox_worker.py
@@ -0,0 +1,62 @@
+"""Outbox worker — drains pending outbox_event rows to the state, completing the pattern.
+
+queue_state_submission writes a 'state_claim_submission_requested' outbox row in the same
+transaction as the SUBMITTING transition. This worker delivers it with AT-LEAST-ONCE
+semantics + bounded retry:
+
+ - success -> mark 'sent'
+ - transient failure -> leave 'pending', bump attempts (retried next drain)
+ - attempts exhausted -> mark 'failed' (claim NOT lost — stays SUBMITTING for ops review)
+
+At-least-once means a row may be delivered more than once (e.g. adapter succeeded but the
+'sent' write didn't land). So "claim already SUBMITTED_TO_STATE" is treated as SUCCESS, not
+an error — otherwise a retry after a partial failure would loop forever.
+"""
+from __future__ import annotations
+
+import json
+from uuid import UUID
+
+from services.claims.claim_workflow import ClaimStatus, complete_state_submission
+
+
+class OutboxWorker:
+ def __init__(self, repo, adapter, max_attempts: int = 3) -> None:
+ self.repo = repo
+ self.adapter = adapter
+ self.max_attempts = max_attempts
+
+ def drain_once(self) -> dict:
+ sent = failed = retried = 0
+ for row in self.repo.fetch_pending_outbox():
+ attempts = row["attempts"] + 1
+ try:
+ self._dispatch(row["event_type"], json.loads(row["payload"]))
+ self.repo.mark_outbox(row["outbox_event_id"], "sent", attempts)
+ sent += 1
+ except Exception as exc: # noqa: BLE001 - worker isolates each row's failure
+ if attempts >= self.max_attempts:
+ self.repo.mark_outbox(row["outbox_event_id"], "failed", attempts, str(exc))
+ failed += 1
+ else:
+ self.repo.mark_outbox(row["outbox_event_id"], "pending", attempts, str(exc))
+ retried += 1
+ return {"sent": sent, "failed": failed, "retried": retried}
+
+ def _dispatch(self, event_type: str, payload: dict) -> None:
+ if event_type == "state_claim_submission_requested":
+ self._deliver_submission(payload)
+ else:
+ raise NotImplementedError(f"no outbox handler for {event_type!r}")
+
+ def _deliver_submission(self, payload: dict) -> None:
+ claim_id = UUID(payload["claim_id"])
+ key = payload["idempotency_key"]
+ try:
+ complete_state_submission(self.repo, self.adapter, claim_id, key)
+ except ValueError:
+ # complete raises if the claim isn't awaiting submission. If it already reached
+ # SUBMITTED_TO_STATE, a prior attempt delivered it — idempotent success.
+ if self.repo.get_for_update(claim_id).status == ClaimStatus.SUBMITTED_TO_STATE:
+ return
+ raise
diff --git a/services/claims/sqlite_claim_repo.py b/services/claims/sqlite_claim_repo.py
index 768dd61..43386de 100644
--- a/services/claims/sqlite_claim_repo.py
+++ b/services/claims/sqlite_claim_repo.py
@@ -106,3 +106,28 @@ class SqliteClaimRepository:
(str(aggregate_id),),
).fetchall()
return [dict(r) for r in rows]
+
+ # --- outbox worker surface --------------------------------------------
+ def fetch_pending_outbox(self, limit: int = 100) -> list[dict]:
+ rows = self.conn.execute(
+ """SELECT outbox_event_id, aggregate_id, event_type, payload, attempts
+ FROM outbox_event WHERE delivery_state='pending'
+ ORDER BY rowid LIMIT ?""",
+ (limit,),
+ ).fetchall()
+ return [dict(r) for r in rows]
+
+ def mark_outbox(self, outbox_event_id: str, delivery_state: str,
+ attempts: int, last_error: str | None = None) -> None:
+ self.conn.execute(
+ "UPDATE outbox_event SET delivery_state=?, attempts=?, last_error=? WHERE outbox_event_id=?",
+ (delivery_state, attempts, last_error, outbox_event_id),
+ )
+ self.conn.commit()
+
+ def outbox_row(self, outbox_event_id: str) -> dict | None:
+ r = self.conn.execute(
+ "SELECT delivery_state, attempts, last_error FROM outbox_event WHERE outbox_event_id=?",
+ (outbox_event_id,),
+ ).fetchone()
+ return dict(r) if r else None
diff --git a/tests/test_cycle10_outbox_worker.py b/tests/test_cycle10_outbox_worker.py
new file mode 100644
index 0000000..68ce6c3
--- /dev/null
+++ b/tests/test_cycle10_outbox_worker.py
@@ -0,0 +1,103 @@
+"""Cycle 9 test — outbox worker: at-least-once delivery + bounded retry.
+
+Run: python -m tests.test_cycle10_outbox_worker
+
+Proves:
+ 1. a transient state-API failure leaves the row pending (claim not advanced), and a later
+ drain succeeds -> claim SUBMITTED_TO_STATE, outbox 'sent';
+ 2. re-draining after success is a no-op (idempotent);
+ 3. permanent failure exhausts retries -> outbox 'failed', claim NOT lost (stays SUBMITTING).
+"""
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+from uuid import uuid4
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.claims.claim_workflow import (
+ ClaimStatus, queue_state_submission, transition_claim,
+)
+from services.claims.outbox_worker import OutboxWorker
+from services.claims.sqlite_claim_repo import SqliteClaimRepository
+from services.common.sqlite_repo import SqliteRepository
+
+
+class FlakyAdapter:
+ def __init__(self, fail_times: int) -> None:
+ self.calls = 0
+ self.fail_times = fail_times
+
+ def submit_claim(self, claim, idempotency_key) -> str:
+ self.calls += 1
+ if self.calls <= self.fail_times:
+ raise RuntimeError("state API timeout")
+ return "STATE-CASE-9"
+
+
+class AlwaysFail:
+ def submit_claim(self, claim, idempotency_key) -> str:
+ raise RuntimeError("state API permanently down")
+
+
+def ok(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def _ready_claim(repo: SqliteClaimRepository):
+ c = repo.create_claim("SAMPLE", public_property_reference="ref", claimant_id=uuid4())
+ for target, key in [(ClaimStatus.IDENTITY_PENDING, "k1"),
+ (ClaimStatus.EVIDENCE_PENDING, "k2"),
+ (ClaimStatus.READY_FOR_SUBMISSION, "k3")]:
+ transition_claim(repo, c.claim_id, target, actor_id="worker", idempotency_key=key)
+ queue_state_submission(repo, c.claim_id, actor_id="worker", idempotency_key="sub")
+ return c
+
+
+def main() -> int:
+ tmp = Path(tempfile.mkdtemp(prefix="upp-cycle10-"))
+ base = SqliteRepository(str(tmp / "p.db"))
+ repo = SqliteClaimRepository(base.conn)
+
+ print("1) transient failure -> retry -> success")
+ c1 = _ready_claim(repo)
+ assert repo.get_for_update(c1.claim_id).status == ClaimStatus.SUBMITTING
+ worker = OutboxWorker(repo, FlakyAdapter(fail_times=1), max_attempts=3)
+ r1 = worker.drain_once()
+ assert r1["retried"] == 1 and r1["sent"] == 0, r1
+ assert repo.get_for_update(c1.claim_id).status == ClaimStatus.SUBMITTING, "must not advance on failure"
+ ok(f"drain#1 adapter failed -> retried=1, claim still SUBMITTING")
+ r2 = worker.drain_once()
+ assert r2["sent"] == 1, r2
+ fin = repo.get_for_update(c1.claim_id)
+ assert fin.status == ClaimStatus.SUBMITTED_TO_STATE and fin.state_case_id == "STATE-CASE-9", fin
+ ok("drain#2 adapter recovered -> SUBMITTED_TO_STATE, outbox sent")
+
+ print("2) re-drain after success is a no-op")
+ r3 = worker.drain_once()
+ assert r3 == {"sent": 0, "failed": 0, "retried": 0}, r3
+ ok("no pending rows -> nothing re-delivered")
+
+ print("3) permanent failure exhausts retries -> failed, claim not lost")
+ c2 = _ready_claim(repo)
+ w2 = OutboxWorker(repo, AlwaysFail(), max_attempts=2)
+ a = w2.drain_once() # attempt 1 -> pending
+ assert a["retried"] == 1, a
+ b = w2.drain_once() # attempt 2 -> failed
+ assert b["failed"] == 1, b
+ # find c2's outbox row state
+ oid = [row for row in base.conn.execute(
+ "SELECT outbox_event_id FROM outbox_event WHERE aggregate_id=?",
+ (str(c2.claim_id),))][0]["outbox_event_id"]
+ assert repo.outbox_row(oid)["delivery_state"] == "failed"
+ assert repo.get_for_update(c2.claim_id).status == ClaimStatus.SUBMITTING, "claim must not be lost"
+ ok("2 attempts exhausted -> outbox 'failed', claim still SUBMITTING (recoverable)")
+
+ print("\nALL CYCLE-9 OUTBOX-WORKER ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
← e7a163f docs: ledger Cycle-8 record (claims persistence); TK-10097
·
back to Unclaimed Property Platform
·
docs: ledger Cycle-9 record (outbox worker); TK-10097 2a43af4 →