Skip to content

Commit 2919a5d

Browse files
JuliaEdomclaude
andcommitted
feat(serving): replay outbox + dead-letter behind ControlPlaneStore (ADR 0010 slice 3)
Third rollout slice of ADR 0010: the replay outbox and dead-letter status transitions move behind the ControlPlaneStore port, preserving invariant 8 verbatim. Pure extraction — table DDL and SQL shapes are byte-compatible; no behavior change on the embedded (default) profile. - ControlPlaneStore gains claim_due_outbox_entries / get_pending_outbox_entry (reads), mark_outbox_sent / schedule_outbox_retry (invariant 8, one transaction each), enqueue_outbox_replay (the other invariant-8 half — EventReplayer.replay's dead-letter-update + outbox-insert), plus get_dead_letter_event_for_replay / dismiss_dead_letter_event / dead_letter_event_exists / get_dead_letter_event / list_dead_letter_events / get_dead_letter_stats for the router's read/write paths. - OutboxProcessor and EventReplayer gain an additive `store: ControlPlaneStore | None = None` kwarg; when omitted (every existing call site) they build a private embedded store from whichever conn/duckdb_path they already owned, preserving main.py's :memory:-vs-file dual-connection split verbatim. routers/deadletter.py resolves the app's shared store and passes it via store=, so it never reaches query_engine._conn directly. - ensure_outbox_table / ensure_dead_letter_table moved to control_plane/embedded.py. Unlike the webhook/alert log methods, the outbox/dead-letter write-transaction methods do NOT lazily ensure their tables — ensure_outbox_schema runs once at construction instead, matching pre-port timing exactly and keeping the fault-injection rollback tests (table-drop, connection-wrapper) meaningful. Tests: tests/unit/test_control_plane_store.py gains outbox/dead-letter coverage (claim ordering, mark-sent + rollback, retry/backoff/kafka-floor, replay enqueue + rollback, tenant-scoped reads, stats); structural ratchet extended to outbox.py/event_replayer.py/routers/deadletter.py; 6 test files' ensure_dead_letter_table imports rewired to the new module. Verified: full unit+property 1575 passed, mypy strict clean, ruff clean, targeted integration (test_outbox/test_deadletter/test_logical_correctness) 27 passed. Full integration suite re-run was interrupted by a tool error at session end — rerun tests/integration -m "not requires_docker and not live" before merging. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent e88d5b1 commit 2919a5d

14 files changed

Lines changed: 991 additions & 496 deletions

‎CHANGELOG.md‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,37 @@ All notable changes to AgentFlow are documented in this file.
44

55
## [Unreleased]
66

7+
### Added — Replay outbox + dead-letter behind the ControlPlaneStore port (ADR 0010 slice 3, 2026-07-02)
8+
9+
- **Invariant 8 preserved verbatim**: `mark_outbox_sent` flips an outbox row
10+
to `sent` and its dead-letter row to `replayed` in one transaction;
11+
`schedule_outbox_retry` bumps attempts/backoff or parks both rows `failed`
12+
once retries are exhausted, also in one transaction. Both moved into
13+
`EmbeddedControlPlaneStore` byte-for-byte (same SQL, same
14+
BEGIN/COMMIT/ROLLBACK shape) — confirmed by the existing rollback-simulation
15+
tests (table-drop and connection-wrapper fault injection) passing unchanged.
16+
- `OutboxProcessor` and `EventReplayer` gain an additive `store:
17+
ControlPlaneStore | None = None` constructor kwarg; when omitted (every
18+
existing call site) they build a private embedded store bound to whichever
19+
`conn`/`duckdb_path` they already owned — preserving `main.py`'s
20+
`:memory:`-vs-file dual-connection split verbatim. `routers/deadletter.py`
21+
now resolves the app's shared store and passes it via `store=`, so it never
22+
reaches `query_engine._conn` directly (stats/list/detail/replay/dismiss all
23+
routed through new store methods).
24+
- `ensure_outbox_table` / `ensure_dead_letter_table` moved to
25+
`control_plane/embedded.py` (same location as their webhook/alert
26+
siblings); `ensure_outbox_table` is deliberately NOT called lazily inside
27+
the write-transaction methods above (unlike the webhook/alert log methods)
28+
— it and `ensure_dead_letter_table` run once at `OutboxProcessor` /
29+
`EventReplayer` construction via a new `ensure_outbox_schema` port method,
30+
matching the pre-port eager-DDL timing exactly.
31+
- Structural ratchet test extended to `src/processing/outbox.py`,
32+
`src/processing/event_replayer.py` and `routers/deadletter.py`. New
33+
`tests/unit/test_control_plane_store.py` coverage for the outbox/
34+
dead-letter store methods (claim ordering, mark-sent + rollback,
35+
retry/backoff/kafka-floor, replay enqueue + rollback, tenant-scoped
36+
reads, stats).
37+
738
### Added — Alert history + alert-rule repository behind the ControlPlaneStore port (ADR 0010 slice 2, 2026-07-02)
839

940
- **Alert delivery history** (`alert_history`) moved behind `ControlPlaneStore`

‎src/processing/event_replayer.py‎

Lines changed: 30 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,11 @@
1212
import structlog
1313
from opentelemetry import trace
1414

15-
from src.db_concurrency import catalog_ddl_lock
16-
from src.processing.outbox import OutboxProcessor, ensure_outbox_table
15+
from src.processing.outbox import OutboxProcessor
1716
from src.processing.tracing import inject_trace_to_kafka_headers, telemetry_disabled
1817
from src.quality.validators.schema_validator import validate_event
1918
from src.quality.validators.semantic_validator import validate_semantics
19+
from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore
2020

2121
DEFAULT_KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
2222
DEFAULT_REPLAY_TOPIC = "events.raw"
@@ -40,45 +40,30 @@ class ReplayResult:
4040
payload: dict
4141

4242

43-
def ensure_dead_letter_table(conn: duckdb.DuckDBPyConnection) -> None:
44-
# Serialize the lazy DDL: the offloaded read handlers call this on worker
45-
# threads, and concurrent CREATE/ALTER on a cold DuckDB catalog conflicts.
46-
# (audit_30 A2 follow-up: #120 offload race)
47-
with catalog_ddl_lock:
48-
conn.execute(
49-
"""
50-
CREATE TABLE IF NOT EXISTS dead_letter_events (
51-
event_id TEXT PRIMARY KEY,
52-
tenant_id TEXT DEFAULT 'default',
53-
event_type TEXT,
54-
payload JSON,
55-
failure_reason TEXT,
56-
failure_detail TEXT,
57-
received_at TIMESTAMP,
58-
retry_count INTEGER DEFAULT 0,
59-
last_retried_at TIMESTAMP,
60-
status TEXT DEFAULT 'failed'
61-
)
62-
"""
63-
)
64-
conn.execute(
65-
"ALTER TABLE dead_letter_events "
66-
"ADD COLUMN IF NOT EXISTS tenant_id TEXT DEFAULT 'default'"
67-
)
68-
69-
7043
class EventReplayer:
7144
def __init__(
7245
self,
73-
conn: duckdb.DuckDBPyConnection,
46+
conn: duckdb.DuckDBPyConnection | None = None,
7447
producer: Callable[[str, dict], None] | None = None,
7548
bootstrap_servers: str | None = None,
49+
*,
50+
store: ControlPlaneStore | None = None,
7651
) -> None:
77-
self._conn = conn
52+
if conn is None and store is None:
53+
raise ValueError("conn or store is required")
7854
self._producer = producer or self._produce_to_kafka
7955
self._bootstrap_servers = bootstrap_servers or DEFAULT_KAFKA_BOOTSTRAP
80-
ensure_dead_letter_table(self._conn)
81-
ensure_outbox_table(self._conn)
56+
# ADR 0010 slice 3: table access goes through the ControlPlaneStore
57+
# port, same seam as OutboxProcessor. routers/deadletter.py resolves
58+
# the app's shared store and passes it via ``store=`` so it never
59+
# reaches the query engine's connection directly.
60+
if store is not None:
61+
self._store: ControlPlaneStore = store
62+
else:
63+
assert conn is not None # guaranteed by the check above
64+
resolved_conn = conn
65+
self._store = EmbeddedControlPlaneStore(conn_provider=lambda: resolved_conn)
66+
self._store.ensure_outbox_schema()
8267

8368
def replay(
8469
self,
@@ -95,55 +80,17 @@ def replay(
9580
replayed_at = datetime.now(UTC)
9681
retry_count = int(row["retry_count"] or 0) + 1
9782
outbox_id = str(uuid4())
98-
self._conn.execute("BEGIN TRANSACTION")
99-
try:
100-
self._conn.execute(
101-
"""
102-
UPDATE dead_letter_events
103-
SET payload = ?, status = 'replay_pending', retry_count = ?, last_retried_at = ?
104-
WHERE event_id = ?
105-
""",
106-
[
107-
json.dumps(candidate),
108-
retry_count,
109-
replayed_at,
110-
event_id,
111-
],
112-
)
113-
self._conn.execute(
114-
"""
115-
INSERT INTO outbox (
116-
id,
117-
event_id,
118-
payload,
119-
topic,
120-
created_at,
121-
sent_at,
122-
status,
123-
retry_count,
124-
next_attempt_at,
125-
last_error
126-
)
127-
VALUES (?, ?, ?, ?, ?, NULL, 'pending', 0, ?, NULL)
128-
""",
129-
[
130-
outbox_id,
131-
event_id,
132-
json.dumps(candidate),
133-
DEFAULT_REPLAY_TOPIC,
134-
replayed_at,
135-
replayed_at,
136-
],
137-
)
138-
self._conn.execute("COMMIT")
139-
# rollback must preserve the original replay failure
140-
except Exception: # nosec B110
141-
# Transaction rollback must happen before unexpected errors propagate.
142-
self._conn.execute("ROLLBACK")
143-
raise
83+
self._store.enqueue_outbox_replay(
84+
outbox_id=outbox_id,
85+
event_id=event_id,
86+
payload=candidate,
87+
topic=DEFAULT_REPLAY_TOPIC,
88+
retry_count=retry_count,
89+
replayed_at=replayed_at,
90+
)
14491
status = "replay_pending"
14592
processor = OutboxProcessor(
146-
conn=self._conn,
93+
store=self._store,
14794
producer=self._producer,
14895
bootstrap_servers=self._bootstrap_servers,
14996
)
@@ -159,30 +106,13 @@ def replay(
159106

160107
def dismiss(self, event_id: str) -> None:
161108
self._load_row(event_id)
162-
self._conn.execute(
163-
"UPDATE dead_letter_events SET status = 'dismissed' WHERE event_id = ?",
164-
[event_id],
165-
)
109+
self._store.dismiss_dead_letter_event(event_id)
166110

167111
def _load_row(self, event_id: str) -> dict:
168-
row = self._conn.execute(
169-
"""
170-
SELECT
171-
event_id,
172-
payload,
173-
retry_count
174-
FROM dead_letter_events
175-
WHERE event_id = ?
176-
""",
177-
[event_id],
178-
).fetchone()
112+
row = self._store.get_dead_letter_event_for_replay(event_id)
179113
if row is None:
180114
raise DeadLetterEventNotFoundError(event_id)
181-
return {
182-
"event_id": row[0],
183-
"payload": row[1],
184-
"retry_count": row[2],
185-
}
115+
return row
186116

187117
def _decoded_payload(self, payload: object) -> dict:
188118
if isinstance(payload, dict):

0 commit comments

Comments
 (0)