Skip to content

feat(serving): replay outbox + dead-letter behind ControlPlaneStore (ADR 0010 slice 3) #133

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brownjuly2003-code merged 1 commit into from
Jul 2, 2026
Merged

feat(serving): replay outbox + dead-letter behind ControlPlaneStore (ADR 0010 slice 3) #133

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,37 @@ All notable changes to AgentFlow are documented in this file.

## [Unreleased]

### Added — Replay outbox + dead-letter behind the ControlPlaneStore port (ADR 0010 slice 3, 2026-07-02)

- **Invariant 8 preserved verbatim**: `mark_outbox_sent` flips an outbox row
to `sent` and its dead-letter row to `replayed` in one transaction;
`schedule_outbox_retry` bumps attempts/backoff or parks both rows `failed`
once retries are exhausted, also in one transaction. Both moved into
`EmbeddedControlPlaneStore` byte-for-byte (same SQL, same
BEGIN/COMMIT/ROLLBACK shape) — confirmed by the existing rollback-simulation
tests (table-drop and connection-wrapper fault injection) passing unchanged.
- `OutboxProcessor` and `EventReplayer` gain an additive `store:
ControlPlaneStore | None = None` constructor kwarg; when omitted (every
existing call site) they build a private embedded store bound to whichever
`conn`/`duckdb_path` they already owned — preserving `main.py`'s
`:memory:`-vs-file dual-connection split verbatim. `routers/deadletter.py`
now resolves the app's shared store and passes it via `store=`, so it never
reaches `query_engine._conn` directly (stats/list/detail/replay/dismiss all
routed through new store methods).
- `ensure_outbox_table` / `ensure_dead_letter_table` moved to
`control_plane/embedded.py` (same location as their webhook/alert
siblings); `ensure_outbox_table` is deliberately NOT called lazily inside
the write-transaction methods above (unlike the webhook/alert log methods)
— it and `ensure_dead_letter_table` run once at `OutboxProcessor` /
`EventReplayer` construction via a new `ensure_outbox_schema` port method,
matching the pre-port eager-DDL timing exactly.
- Structural ratchet test extended to `src/processing/outbox.py`,
`src/processing/event_replayer.py` and `routers/deadletter.py`. New
`tests/unit/test_control_plane_store.py` coverage for the outbox/
dead-letter store methods (claim ordering, mark-sent + rollback,
retry/backoff/kafka-floor, replay enqueue + rollback, tenant-scoped
reads, stats).

### Added — Alert history + alert-rule repository behind the ControlPlaneStore port (ADR 0010 slice 2, 2026-07-02)

- **Alert delivery history** (`alert_history`) moved behind `ControlPlaneStore`
Expand Down
130 changes: 30 additions & 100 deletions src/processing/event_replayer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,11 @@
import structlog
from opentelemetry import trace

from src.db_concurrency import catalog_ddl_lock
from src.processing.outbox import OutboxProcessor, ensure_outbox_table
from src.processing.outbox import OutboxProcessor
from src.processing.tracing import inject_trace_to_kafka_headers, telemetry_disabled
from src.quality.validators.schema_validator import validate_event
from src.quality.validators.semantic_validator import validate_semantics
from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore

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


def ensure_dead_letter_table(conn: duckdb.DuckDBPyConnection) -> None:
# Serialize the lazy DDL: the offloaded read handlers call this on worker
# threads, and concurrent CREATE/ALTER on a cold DuckDB catalog conflicts.
# (audit_30 A2 follow-up: #120 offload race)
with catalog_ddl_lock:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS dead_letter_events (
event_id TEXT PRIMARY KEY,
tenant_id TEXT DEFAULT 'default',
event_type TEXT,
payload JSON,
failure_reason TEXT,
failure_detail TEXT,
received_at TIMESTAMP,
retry_count INTEGER DEFAULT 0,
last_retried_at TIMESTAMP,
status TEXT DEFAULT 'failed'
)
"""
)
conn.execute(
"ALTER TABLE dead_letter_events "
"ADD COLUMN IF NOT EXISTS tenant_id TEXT DEFAULT 'default'"
)


class EventReplayer:
def __init__(
self,
conn: duckdb.DuckDBPyConnection,
conn: duckdb.DuckDBPyConnection | None = None,
producer: Callable[[str, dict], None] | None = None,
bootstrap_servers: str | None = None,
*,
store: ControlPlaneStore | None = None,
) -> None:
self._conn = conn
if conn is None and store is None:
raise ValueError("conn or store is required")
self._producer = producer or self._produce_to_kafka
self._bootstrap_servers = bootstrap_servers or DEFAULT_KAFKA_BOOTSTRAP
ensure_dead_letter_table(self._conn)
ensure_outbox_table(self._conn)
# ADR 0010 slice 3: table access goes through the ControlPlaneStore
# port, same seam as OutboxProcessor. routers/deadletter.py resolves
# the app's shared store and passes it via ``store=`` so it never
# reaches the query engine's connection directly.
if store is not None:
self._store: ControlPlaneStore = store
else:
assert conn is not None # guaranteed by the check above
resolved_conn = conn
self._store = EmbeddedControlPlaneStore(conn_provider=lambda: resolved_conn)
self._store.ensure_outbox_schema()

def replay(
self,
Expand All @@ -95,55 +80,17 @@ def replay(
replayed_at = datetime.now(UTC)
retry_count = int(row["retry_count"] or 0) + 1
outbox_id = str(uuid4())
self._conn.execute("BEGIN TRANSACTION")
try:
self._conn.execute(
"""
UPDATE dead_letter_events
SET payload = ?, status = 'replay_pending', retry_count = ?, last_retried_at = ?
WHERE event_id = ?
""",
[
json.dumps(candidate),
retry_count,
replayed_at,
event_id,
],
)
self._conn.execute(
"""
INSERT INTO outbox (
id,
event_id,
payload,
topic,
created_at,
sent_at,
status,
retry_count,
next_attempt_at,
last_error
)
VALUES (?, ?, ?, ?, ?, NULL, 'pending', 0, ?, NULL)
""",
[
outbox_id,
event_id,
json.dumps(candidate),
DEFAULT_REPLAY_TOPIC,
replayed_at,
replayed_at,
],
)
self._conn.execute("COMMIT")
# rollback must preserve the original replay failure
except Exception: # nosec B110
# Transaction rollback must happen before unexpected errors propagate.
self._conn.execute("ROLLBACK")
raise
self._store.enqueue_outbox_replay(
outbox_id=outbox_id,
event_id=event_id,
payload=candidate,
topic=DEFAULT_REPLAY_TOPIC,
retry_count=retry_count,
replayed_at=replayed_at,
)
status = "replay_pending"
processor = OutboxProcessor(
conn=self._conn,
store=self._store,
producer=self._producer,
bootstrap_servers=self._bootstrap_servers,
)
Expand All @@ -159,30 +106,13 @@ def replay(

def dismiss(self, event_id: str) -> None:
self._load_row(event_id)
self._conn.execute(
"UPDATE dead_letter_events SET status = 'dismissed' WHERE event_id = ?",
[event_id],
)
self._store.dismiss_dead_letter_event(event_id)

def _load_row(self, event_id: str) -> dict:
row = self._conn.execute(
"""
SELECT
event_id,
payload,
retry_count
FROM dead_letter_events
WHERE event_id = ?
""",
[event_id],
).fetchone()
row = self._store.get_dead_letter_event_for_replay(event_id)
if row is None:
raise DeadLetterEventNotFoundError(event_id)
return {
"event_id": row[0],
"payload": row[1],
"retry_count": row[2],
}
return row

def _decoded_payload(self, payload: object) -> dict:
if isinstance(payload, dict):
Expand Down
Loading