diff --git a/CHANGELOG.md b/CHANGELOG.md index 042aa586..37f2ee3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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` diff --git a/src/processing/event_replayer.py b/src/processing/event_replayer.py index b701374b..dca9232f 100644 --- a/src/processing/event_replayer.py +++ b/src/processing/event_replayer.py @@ -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" @@ -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, @@ -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, ) @@ -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): diff --git a/src/processing/outbox.py b/src/processing/outbox.py index 884a6cc5..c9735032 100644 --- a/src/processing/outbox.py +++ b/src/processing/outbox.py @@ -5,8 +5,6 @@ import os from collections.abc import Callable from contextlib import nullcontext -from datetime import UTC, datetime, timedelta -from typing import Any import duckdb import structlog @@ -14,6 +12,7 @@ from opentelemetry import trace from src.processing.tracing import inject_trace_to_kafka_headers, telemetry_disabled +from src.serving.control_plane import ControlPlaneStore, EmbeddedControlPlaneStore, OutboxEntry from src.serving.duckdb_connection import connect_duckdb logger = structlog.get_logger() @@ -22,25 +21,6 @@ DEFAULT_KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092") -def ensure_outbox_table(conn: duckdb.DuckDBPyConnection) -> None: - conn.execute( - """ - CREATE TABLE IF NOT EXISTS outbox ( - id TEXT PRIMARY KEY, - event_id TEXT NOT NULL, - payload JSON NOT NULL, - topic TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - sent_at TIMESTAMP, - status TEXT DEFAULT 'pending', - retry_count INTEGER DEFAULT 0, - next_attempt_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_error TEXT - ) - """ - ) - - class OutboxProcessor: def __init__( self, @@ -49,17 +29,30 @@ def __init__( producer: Callable[[str, dict], None] | None = None, bootstrap_servers: str | None = None, max_retries: int = 5, + *, + store: ControlPlaneStore | None = None, ) -> None: - if conn is None and duckdb_path is None: + if conn is None and duckdb_path is None and store is None: raise ValueError("duckdb_path or conn is required") - self._owns_conn = conn is None + self._owns_conn = conn is None and store is None self._conn: duckdb.DuckDBPyConnection | None = ( - conn if conn is not None else connect_duckdb(str(duckdb_path)) + None + if store is not None + else (conn if conn is not None else connect_duckdb(str(duckdb_path))) ) self._producer = producer or self._produce_to_kafka self._bootstrap_servers = bootstrap_servers or DEFAULT_KAFKA_BOOTSTRAP self._max_retries = max_retries - ensure_outbox_table(self._connection) + # ADR 0010 slice 3: table access goes through the ControlPlaneStore + # port. When no store is injected (the common case — main.py and every + # existing test construct via conn/duckdb_path), this builds a private + # embedded store bound to whichever connection this instance owns — + # not necessarily the app's shared query_engine connection (see + # main.py's file-vs-:memory: branch this preserves verbatim). + self._store: ControlPlaneStore = store or EmbeddedControlPlaneStore( + conn_provider=lambda: self._connection + ) + self._store.ensure_outbox_schema() @property def _connection(self) -> duckdb.DuckDBPyConnection: @@ -90,20 +83,10 @@ async def run_forever(self) -> None: self.close() def process_pending(self, limit: int = 100) -> int: - rows = self._connection.execute( - """ - SELECT id, event_id, payload, topic, retry_count - FROM outbox - WHERE status = 'pending' - AND (next_attempt_at IS NULL OR next_attempt_at <= ?) - ORDER BY created_at - LIMIT ? - """, - [datetime.now(UTC), limit], - ).fetchall() + entries = self._store.claim_due_outbox_entries(limit=limit) processed = 0 - for row in rows: - if self._process_row(row): + for entry in entries: + if self._process_entry(entry): processed += 1 return processed @@ -115,28 +98,17 @@ async def process_pending_async(self, limit: int = 100) -> int: but the blocking Kafka produce+flush(10) is offloaded so a slow or unreachable broker can't freeze the whole event loop. (audit_28_06_26.md #1) """ - rows = self._connection.execute( - """ - SELECT id, event_id, payload, topic, retry_count - FROM outbox - WHERE status = 'pending' - AND (next_attempt_at IS NULL OR next_attempt_at <= ?) - ORDER BY created_at - LIMIT ? - """, - [datetime.now(UTC), limit], - ).fetchall() + entries = self._store.claim_due_outbox_entries(limit=limit) processed = 0 - for row in rows: - if await self._process_row_async(row): + for entry in entries: + if await self._process_entry_async(entry): processed += 1 return processed - async def _process_row_async(self, row: tuple[Any, ...]) -> bool: - outbox_id, event_id, payload, topic, retry_count = row - decoded_payload = self._decode_payload(payload) + async def _process_entry_async(self, entry: OutboxEntry) -> bool: + decoded_payload = self._decode_payload(entry.payload) try: - await asyncio.to_thread(self._producer, topic, decoded_payload) + await asyncio.to_thread(self._producer, entry.topic, decoded_payload) except (BufferError, ConnectionError, TimeoutError, KafkaException, RuntimeError) as exc: error_message = str(exc) if isinstance(exc, RuntimeError) and not ( @@ -144,45 +116,37 @@ async def _process_row_async(self, row: tuple[Any, ...]) -> bool: or "Kafka message(s) were not delivered" in error_message ): raise - next_retry_count = int(retry_count or 0) + 1 + next_retry_count = int(entry.retry_count or 0) + 1 logger.warning( "outbox_delivery_retry_scheduled", - outbox_id=outbox_id, - event_id=event_id, - topic=topic, + outbox_id=entry.id, + event_id=entry.event_id, + topic=entry.topic, retry_count=next_retry_count, error=error_message, exc_info=True, ) - self._schedule_retry( - outbox_id=outbox_id, - event_id=event_id, + self._store.schedule_outbox_retry( + outbox_id=entry.id, + event_id=entry.event_id, retry_count=next_retry_count, error_message=error_message, + max_retries=self._max_retries, ) return False - self._mark_sent(outbox_id=outbox_id, event_id=event_id) + self._store.mark_outbox_sent(outbox_id=entry.id, event_id=entry.event_id) return True def process_entry(self, outbox_id: str) -> bool: - row = self._connection.execute( - """ - SELECT id, event_id, payload, topic, retry_count - FROM outbox - WHERE id = ? - AND status = 'pending' - """, - [outbox_id], - ).fetchone() - if row is None: + entry = self._store.get_pending_outbox_entry(outbox_id) + if entry is None: return False - return self._process_row(row) + return self._process_entry(entry) - def _process_row(self, row: tuple[Any, ...]) -> bool: - outbox_id, event_id, payload, topic, retry_count = row - decoded_payload = self._decode_payload(payload) + def _process_entry(self, entry: OutboxEntry) -> bool: + decoded_payload = self._decode_payload(entry.payload) try: - self._producer(topic, decoded_payload) + self._producer(entry.topic, decoded_payload) except (BufferError, ConnectionError, TimeoutError, KafkaException, RuntimeError) as exc: error_message = str(exc) if isinstance(exc, RuntimeError) and not ( @@ -190,97 +154,27 @@ def _process_row(self, row: tuple[Any, ...]) -> bool: or "Kafka message(s) were not delivered" in error_message ): raise - next_retry_count = int(retry_count or 0) + 1 + next_retry_count = int(entry.retry_count or 0) + 1 logger.warning( "outbox_delivery_retry_scheduled", - outbox_id=outbox_id, - event_id=event_id, - topic=topic, + outbox_id=entry.id, + event_id=entry.event_id, + topic=entry.topic, retry_count=next_retry_count, error=error_message, exc_info=True, ) - self._schedule_retry( - outbox_id=outbox_id, - event_id=event_id, + self._store.schedule_outbox_retry( + outbox_id=entry.id, + event_id=entry.event_id, retry_count=next_retry_count, error_message=error_message, + max_retries=self._max_retries, ) return False - self._mark_sent(outbox_id=outbox_id, event_id=event_id) + self._store.mark_outbox_sent(outbox_id=entry.id, event_id=entry.event_id) return True - def _mark_sent(self, outbox_id: str, event_id: str) -> None: - sent_at = datetime.now(UTC) - self._connection.execute("BEGIN TRANSACTION") - try: - self._connection.execute( - """ - UPDATE outbox - SET status = 'sent', - sent_at = ?, - last_error = NULL - WHERE id = ? - """, - [sent_at, outbox_id], - ) - self._connection.execute( - "UPDATE dead_letter_events SET status = 'replayed' WHERE event_id = ?", - [event_id], - ) - self._connection.execute("COMMIT") - # rollback must preserve the original replay failure - except Exception: # nosec B110 - # Transaction rollback must happen before unexpected errors propagate. - self._connection.execute("ROLLBACK") - raise - - def _schedule_retry( - self, - outbox_id: str, - event_id: str, - retry_count: int, - error_message: str, - ) -> None: - status = "pending" - retry_delay_seconds = 2**retry_count - is_kafka_error = ( - error_message.startswith("KafkaError{") - or "Kafka message(s) were not delivered" in error_message - ) - if is_kafka_error: - retry_delay_seconds = max(retry_delay_seconds, 30) - next_attempt_at: datetime | None = datetime.now(UTC) + timedelta( - seconds=retry_delay_seconds - ) - self._connection.execute("BEGIN TRANSACTION") - try: - if retry_count >= self._max_retries: - status = "failed" - next_attempt_at = None - self._connection.execute( - """ - UPDATE outbox - SET status = ?, - retry_count = ?, - next_attempt_at = ?, - last_error = ? - WHERE id = ? - """, - [status, retry_count, next_attempt_at, error_message, outbox_id], - ) - if status == "failed": - self._connection.execute( - "UPDATE dead_letter_events SET status = 'failed' WHERE event_id = ?", - [event_id], - ) - self._connection.execute("COMMIT") - # rollback must preserve the original retry scheduling failure - except Exception: # nosec B110 - # Transaction rollback must happen before unexpected errors propagate. - self._connection.execute("ROLLBACK") - raise - def _decode_payload(self, payload: object) -> dict: if isinstance(payload, dict): return payload diff --git a/src/serving/api/routers/deadletter.py b/src/serving/api/routers/deadletter.py index 9d742abc..39bf226f 100644 --- a/src/serving/api/routers/deadletter.py +++ b/src/serving/api/routers/deadletter.py @@ -3,9 +3,7 @@ import json import math from datetime import datetime -from typing import cast -import duckdb from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool @@ -14,8 +12,8 @@ DeadLetterEventNotFoundError, EventReplayer, ReplayValidationError, - ensure_dead_letter_table, ) +from src.serving.control_plane import get_control_plane_store router = APIRouter(prefix="/v1/deadletter", tags=["deadletter"]) @@ -62,31 +60,6 @@ class DismissResponse(BaseModel): status: str -def _conn(request: Request) -> duckdb.DuckDBPyConnection: - # app.state is dynamically typed; the query engine owns a real DuckDB handle. - conn = cast(duckdb.DuckDBPyConnection, request.app.state.query_engine._conn) - ensure_dead_letter_table(conn) - return conn - - -def _read_cursor(request: Request) -> duckdb.DuckDBPyConnection: - """A dedicated cursor for the GET handlers, used on a worker thread. - - The read handlers offload their scans with ``run_in_threadpool``; a separate - cursor — not the shared connection object — keeps concurrent reads on - different worker threads from colliding on the connection. (audit_30_06_26.md A2) - """ - cursor = cast(duckdb.DuckDBPyConnection, request.app.state.query_engine._conn).cursor() - try: - ensure_dead_letter_table(cursor) - except Exception: - # Don't leak the freshly-opened cursor if the lazy DDL fails before the - # handler's own try/finally takes over. - cursor.close() - raise - return cursor - - def _tenant_id(request: Request) -> str: tenant_key = getattr(request.state, "tenant_key", None) tenant_id = getattr(request.state, "tenant_id", None) @@ -105,7 +78,10 @@ def _decode_payload(payload: object) -> dict: def _replayer(request: Request) -> EventReplayer: producer = getattr(request.app.state, "deadletter_producer", None) - return EventReplayer(_conn(request), producer=producer if callable(producer) else None) + return EventReplayer( + store=get_control_plane_store(request.app), + producer=producer if callable(producer) else None, + ) def _require_deadletter_write_access(request: Request, event_id: str) -> None: @@ -117,19 +93,10 @@ def _require_deadletter_write_access(request: Request, event_id: str) -> None: status_code=403, detail="This API key has read-only access to dead-letter operations.", ) - row = ( - _conn(request) - .execute( - """ - SELECT event_id - FROM dead_letter_events - WHERE event_id = ? AND COALESCE(tenant_id, 'default') = ? - """, - [event_id, _tenant_id(request)], - ) - .fetchone() + exists = get_control_plane_store(request.app).dead_letter_event_exists( + event_id, _tenant_id(request) ) - if row is None: + if not exists: raise HTTPException(status_code=404, detail=f"Dead-letter event '{event_id}' not found.") @@ -139,55 +106,8 @@ async def deadletter_stats(request: Request) -> DeadLetterStatsResponse: def _deadletter_stats(request: Request) -> DeadLetterStatsResponse: - cursor = _read_cursor(request) - tenant_id = _tenant_id(request) - try: - rows = cursor.execute( - """ - SELECT failure_reason, COUNT(*) - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - GROUP BY failure_reason - ORDER BY failure_reason - """, - [tenant_id], - ).fetchall() - last_24h_row = cursor.execute( - """ - SELECT COUNT(*) - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - AND received_at >= NOW() - INTERVAL '24 hours' - """, - [tenant_id], - ).fetchone() - trend_rows = cursor.execute( - """ - SELECT DATE_TRUNC('hour', received_at) AS hour_bucket, COUNT(*) - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - AND received_at >= NOW() - INTERVAL '24 hours' - GROUP BY hour_bucket - ORDER BY hour_bucket - """, - [tenant_id], - ).fetchall() - finally: - cursor.close() - return DeadLetterStatsResponse( - counts={str(reason): int(count) for reason, count in rows if reason is not None}, - last_24h=int(last_24h_row[0]) if last_24h_row and last_24h_row[0] is not None else 0, - trend=[ - { - "hour": hour.isoformat() if hasattr(hour, "isoformat") else str(hour), - "count": int(count), - } - for hour, count in trend_rows - ], - ) + stats = get_control_plane_store(request.app).get_dead_letter_stats(_tenant_id(request)) + return DeadLetterStatsResponse(**stats) @router.get("", response_model=DeadLetterListResponse) @@ -203,93 +123,11 @@ async def list_deadletter_events( def _list_deadletter_events( request: Request, page: int, page_size: int, reason: str | None ) -> DeadLetterListResponse: - cursor = _read_cursor(request) - tenant_id = _tenant_id(request) - try: - params: list[object] - if reason is not None: - params = [tenant_id, reason] - total_row = cursor.execute( - """ - SELECT COUNT(*) - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - AND failure_reason = ? - """, - params, - ).fetchone() - else: - params = [tenant_id] - total_row = cursor.execute( - """ - SELECT COUNT(*) - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - """, - params, - ).fetchone() - total = int(total_row[0]) if total_row and total_row[0] is not None else 0 - offset = (page - 1) * page_size - if reason is not None: - rows = cursor.execute( - """ - SELECT - event_id, - event_type, - failure_reason, - failure_detail, - received_at, - retry_count, - last_retried_at, - status - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - AND failure_reason = ? - ORDER BY received_at DESC, event_id ASC - LIMIT ? OFFSET ? - """, - [tenant_id, reason, page_size, offset], - ).fetchall() - else: - rows = cursor.execute( - """ - SELECT - event_id, - event_type, - failure_reason, - failure_detail, - received_at, - retry_count, - last_retried_at, - status - FROM dead_letter_events - WHERE status = 'failed' - AND COALESCE(tenant_id, 'default') = ? - ORDER BY received_at DESC, event_id ASC - LIMIT ? OFFSET ? - """, - [tenant_id, page_size, offset], - ).fetchall() - finally: - cursor.close() - + items, total = get_control_plane_store(request.app).list_dead_letter_events( + tenant_id=_tenant_id(request), reason=reason, page=page, page_size=page_size + ) return DeadLetterListResponse( - items=[ - DeadLetterSummary( - event_id=row[0], - event_type=row[1], - failure_reason=row[2], - failure_detail=row[3], - received_at=row[4], - retry_count=int(row[5] or 0), - last_retried_at=row[6], - status=row[7], - ) - for row in rows - ], + items=[DeadLetterSummary(**item) for item in items], pagination={ "page": page, "page_size": page_size, @@ -305,41 +143,10 @@ async def get_deadletter_event(event_id: str, request: Request) -> DeadLetterDet def _get_deadletter_event(request: Request, event_id: str) -> DeadLetterDetail: - cursor = _read_cursor(request) - try: - row = cursor.execute( - """ - SELECT - event_id, - event_type, - payload, - failure_reason, - failure_detail, - received_at, - retry_count, - last_retried_at, - status - FROM dead_letter_events - WHERE event_id = ? - AND COALESCE(tenant_id, 'default') = ? - """, - [event_id, _tenant_id(request)], - ).fetchone() - finally: - cursor.close() + row = get_control_plane_store(request.app).get_dead_letter_event(event_id, _tenant_id(request)) if row is None: raise HTTPException(status_code=404, detail=f"Dead-letter event '{event_id}' not found.") - return DeadLetterDetail( - event_id=row[0], - event_type=row[1], - payload=_decode_payload(row[2]), - failure_reason=row[3], - failure_detail=row[4], - received_at=row[5], - retry_count=int(row[6] or 0), - last_retried_at=row[7], - status=row[8], - ) + return DeadLetterDetail(**{**row, "payload": _decode_payload(row["payload"])}) @router.post("/{event_id}/replay", response_model=ReplayResponse) diff --git a/src/serving/control_plane/__init__.py b/src/serving/control_plane/__init__.py index faacd33b..9db0f985 100644 --- a/src/serving/control_plane/__init__.py +++ b/src/serving/control_plane/__init__.py @@ -9,12 +9,15 @@ from .embedded import ( EmbeddedControlPlaneStore, ensure_alert_history_table, + ensure_dead_letter_table, + ensure_outbox_table, ensure_webhook_deliveries_table, ensure_webhook_delivery_queue_table, ) from .store import ( CONTROL_PLANE_STORE_ENV, ControlPlaneStore, + OutboxEntry, WebhookQueueRow, get_control_plane_store, ) @@ -23,8 +26,11 @@ "CONTROL_PLANE_STORE_ENV", "ControlPlaneStore", "EmbeddedControlPlaneStore", + "OutboxEntry", "WebhookQueueRow", "ensure_alert_history_table", + "ensure_dead_letter_table", + "ensure_outbox_table", "ensure_webhook_deliveries_table", "ensure_webhook_delivery_queue_table", "get_control_plane_store", diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index 907db0d5..709144f8 100644 --- a/src/serving/control_plane/embedded.py +++ b/src/serving/control_plane/embedded.py @@ -23,7 +23,7 @@ from src.db_concurrency import catalog_ddl_lock -from .store import ControlPlaneStore, WebhookQueueRow +from .store import ControlPlaneStore, OutboxEntry, WebhookQueueRow try: import yaml @@ -122,6 +122,57 @@ def ensure_alert_history_table(conn: duckdb.DuckDBPyConnection) -> None: ) +def ensure_outbox_table(conn: duckdb.DuckDBPyConnection) -> None: + # Moved verbatim from processing/outbox.py in ADR 0010 slice 3. Unlike its + # ensure_* siblings above, this one is NOT called lazily by the methods + # below — only once, from ensure_outbox_schema at OutboxProcessor / + # EventReplayer construction (see the store module docstring for why). + conn.execute( + """ + CREATE TABLE IF NOT EXISTS outbox ( + id TEXT PRIMARY KEY, + event_id TEXT NOT NULL, + payload JSON NOT NULL, + topic TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + sent_at TIMESTAMP, + status TEXT DEFAULT 'pending', + retry_count INTEGER DEFAULT 0, + next_attempt_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_error TEXT + ) + """ + ) + + +def ensure_dead_letter_table(conn: duckdb.DuckDBPyConnection) -> None: + # Moved verbatim from processing/event_replayer.py in ADR 0010 slice 3; + # same catalog-DDL-lock discipline as the ensure_* siblings above (the + # deadletter router's read handlers call this lazily per offloaded scan — + # 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 EmbeddedControlPlaneStore(ControlPlaneStore): """Control-plane state on the embedded serving DuckDB connection (queue, log and history tables) plus the YAML-backed alert-rule repository. @@ -437,3 +488,389 @@ def save_alert_rules(self, rules: list[dict]) -> None: else json.dumps(payload, indent=2) ) path.write_text(content, encoding="utf-8", newline="\n") + + # --- replay outbox + dead-letter (invariant 8: one transaction) ----------- + + def ensure_outbox_schema(self) -> None: + ensure_outbox_table(self._conn) + ensure_dead_letter_table(self._conn) + + def claim_due_outbox_entries(self, *, limit: int = 100) -> list[OutboxEntry]: + rows = self._conn.execute( + """ + SELECT id, event_id, payload, topic, retry_count + FROM outbox + WHERE status = 'pending' + AND (next_attempt_at IS NULL OR next_attempt_at <= ?) + ORDER BY created_at + LIMIT ? + """, + [datetime.now(UTC), limit], + ).fetchall() + return [ + OutboxEntry( + id=row_id, event_id=event_id, payload=payload, topic=topic, retry_count=retry_count + ) + for row_id, event_id, payload, topic, retry_count in rows + ] + + def get_pending_outbox_entry(self, outbox_id: str) -> OutboxEntry | None: + row = self._conn.execute( + """ + SELECT id, event_id, payload, topic, retry_count + FROM outbox + WHERE id = ? + AND status = 'pending' + """, + [outbox_id], + ).fetchone() + if row is None: + return None + row_id, event_id, payload, topic, retry_count = row + return OutboxEntry( + id=row_id, event_id=event_id, payload=payload, topic=topic, retry_count=retry_count + ) + + def mark_outbox_sent(self, *, outbox_id: str, event_id: str) -> None: + conn = self._conn + sent_at = datetime.now(UTC) + conn.execute("BEGIN TRANSACTION") + try: + conn.execute( + """ + UPDATE outbox + SET status = 'sent', + sent_at = ?, + last_error = NULL + WHERE id = ? + """, + [sent_at, outbox_id], + ) + conn.execute( + "UPDATE dead_letter_events SET status = 'replayed' WHERE event_id = ?", + [event_id], + ) + conn.execute("COMMIT") + # rollback must preserve the original replay failure + except Exception: # nosec B110 + # Transaction rollback must happen before unexpected errors propagate. + conn.execute("ROLLBACK") + raise + + def schedule_outbox_retry( + self, + *, + outbox_id: str, + event_id: str, + retry_count: int, + error_message: str, + max_retries: int, + ) -> None: + conn = self._conn + status = "pending" + retry_delay_seconds = 2**retry_count + is_kafka_error = ( + error_message.startswith("KafkaError{") + or "Kafka message(s) were not delivered" in error_message + ) + if is_kafka_error: + retry_delay_seconds = max(retry_delay_seconds, 30) + next_attempt_at: datetime | None = datetime.now(UTC) + timedelta( + seconds=retry_delay_seconds + ) + conn.execute("BEGIN TRANSACTION") + try: + if retry_count >= max_retries: + status = "failed" + next_attempt_at = None + conn.execute( + """ + UPDATE outbox + SET status = ?, + retry_count = ?, + next_attempt_at = ?, + last_error = ? + WHERE id = ? + """, + [status, retry_count, next_attempt_at, error_message, outbox_id], + ) + if status == "failed": + conn.execute( + "UPDATE dead_letter_events SET status = 'failed' WHERE event_id = ?", + [event_id], + ) + conn.execute("COMMIT") + # rollback must preserve the original retry scheduling failure + except Exception: # nosec B110 + # Transaction rollback must happen before unexpected errors propagate. + conn.execute("ROLLBACK") + raise + + def enqueue_outbox_replay( + self, + *, + outbox_id: str, + event_id: str, + payload: dict, + topic: str, + retry_count: int, + replayed_at: datetime, + ) -> None: + conn = self._conn + encoded_payload = json.dumps(payload) + conn.execute("BEGIN TRANSACTION") + try: + conn.execute( + """ + UPDATE dead_letter_events + SET payload = ?, status = 'replay_pending', retry_count = ?, last_retried_at = ? + WHERE event_id = ? + """, + [encoded_payload, retry_count, replayed_at, event_id], + ) + 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, encoded_payload, topic, replayed_at, replayed_at], + ) + conn.execute("COMMIT") + # rollback must preserve the original replay failure + except Exception: # nosec B110 + # Transaction rollback must happen before unexpected errors propagate. + conn.execute("ROLLBACK") + raise + + def get_dead_letter_event_for_replay(self, event_id: str) -> dict | None: + row = self._conn.execute( + """ + SELECT + event_id, + payload, + retry_count + FROM dead_letter_events + WHERE event_id = ? + """, + [event_id], + ).fetchone() + if row is None: + return None + return {"event_id": row[0], "payload": row[1], "retry_count": row[2]} + + def dismiss_dead_letter_event(self, event_id: str) -> None: + self._conn.execute( + "UPDATE dead_letter_events SET status = 'dismissed' WHERE event_id = ?", + [event_id], + ) + + def dead_letter_event_exists(self, event_id: str, tenant_id: str) -> bool: + cursor = self._conn.cursor() + try: + ensure_dead_letter_table(cursor) + row = cursor.execute( + """ + SELECT event_id + FROM dead_letter_events + WHERE event_id = ? AND COALESCE(tenant_id, 'default') = ? + """, + [event_id, tenant_id], + ).fetchone() + finally: + cursor.close() + return row is not None + + def get_dead_letter_event(self, event_id: str, tenant_id: str) -> dict | None: + cursor = self._conn.cursor() + try: + ensure_dead_letter_table(cursor) + row = cursor.execute( + """ + SELECT + event_id, + event_type, + payload, + failure_reason, + failure_detail, + received_at, + retry_count, + last_retried_at, + status + FROM dead_letter_events + WHERE event_id = ? + AND COALESCE(tenant_id, 'default') = ? + """, + [event_id, tenant_id], + ).fetchone() + finally: + cursor.close() + if row is None: + return None + return { + "event_id": row[0], + "event_type": row[1], + "payload": row[2], + "failure_reason": row[3], + "failure_detail": row[4], + "received_at": row[5], + "retry_count": int(row[6] or 0), + "last_retried_at": row[7], + "status": row[8], + } + + def list_dead_letter_events( + self, + *, + tenant_id: str, + reason: str | None, + page: int, + page_size: int, + ) -> tuple[list[dict], int]: + cursor = self._conn.cursor() + try: + ensure_dead_letter_table(cursor) + params: list[object] + if reason is not None: + params = [tenant_id, reason] + total_row = cursor.execute( + """ + SELECT COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + AND failure_reason = ? + """, + params, + ).fetchone() + else: + params = [tenant_id] + total_row = cursor.execute( + """ + SELECT COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + """, + params, + ).fetchone() + total = int(total_row[0]) if total_row and total_row[0] is not None else 0 + offset = (page - 1) * page_size + if reason is not None: + rows = cursor.execute( + """ + SELECT + event_id, + event_type, + failure_reason, + failure_detail, + received_at, + retry_count, + last_retried_at, + status + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + AND failure_reason = ? + ORDER BY received_at DESC, event_id ASC + LIMIT ? OFFSET ? + """, + [tenant_id, reason, page_size, offset], + ).fetchall() + else: + rows = cursor.execute( + """ + SELECT + event_id, + event_type, + failure_reason, + failure_detail, + received_at, + retry_count, + last_retried_at, + status + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + ORDER BY received_at DESC, event_id ASC + LIMIT ? OFFSET ? + """, + [tenant_id, page_size, offset], + ).fetchall() + finally: + cursor.close() + items = [ + { + "event_id": row[0], + "event_type": row[1], + "failure_reason": row[2], + "failure_detail": row[3], + "received_at": row[4], + "retry_count": int(row[5] or 0), + "last_retried_at": row[6], + "status": row[7], + } + for row in rows + ] + return items, total + + def get_dead_letter_stats(self, tenant_id: str) -> dict: + cursor = self._conn.cursor() + try: + ensure_dead_letter_table(cursor) + rows = cursor.execute( + """ + SELECT failure_reason, COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + GROUP BY failure_reason + ORDER BY failure_reason + """, + [tenant_id], + ).fetchall() + last_24h_row = cursor.execute( + """ + SELECT COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + AND received_at >= NOW() - INTERVAL '24 hours' + """, + [tenant_id], + ).fetchone() + trend_rows = cursor.execute( + """ + SELECT DATE_TRUNC('hour', received_at) AS hour_bucket, COUNT(*) + FROM dead_letter_events + WHERE status = 'failed' + AND COALESCE(tenant_id, 'default') = ? + AND received_at >= NOW() - INTERVAL '24 hours' + GROUP BY hour_bucket + ORDER BY hour_bucket + """, + [tenant_id], + ).fetchall() + finally: + cursor.close() + return { + "counts": {str(reason): int(count) for reason, count in rows if reason is not None}, + "last_24h": int(last_24h_row[0]) if last_24h_row and last_24h_row[0] is not None else 0, + "trend": [ + { + "hour": hour.isoformat() if hasattr(hour, "isoformat") else str(hour), + "count": int(count), + } + for hour, count in trend_rows + ], + } diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index 2786f63e..ad46cc86 100644 --- a/src/serving/control_plane/store.py +++ b/src/serving/control_plane/store.py @@ -20,18 +20,34 @@ N replicas work-steal without leader election. Rollout slice 1 covers the webhook delivery queue + attempt log. Slice 2 -(this module's alert methods) covers the alert delivery/history log and the -alert-rule repository — including the rules' mutable runtime state (``state``, -``fired_at``, ``last_escalation_level``, flap window, cooldown), the sharpest -per-pod split-brain in the ADR's inventory. Outbox/dead-letter and usage -migrate behind this port in follow-up slices (see the ADR's rollout section). +covers the alert delivery/history log and the alert-rule repository — +including the rules' mutable runtime state (``state``, ``fired_at``, +``last_escalation_level``, flap window, cooldown), the sharpest per-pod +split-brain in the ADR's inventory. Slice 3 (this module's outbox/dead-letter +methods) covers the replay outbox and dead-letter status transitions, +preserving invariant 8 verbatim: ``mark_outbox_sent`` flips a delivered +outbox row *and* its dead-letter row to ``replayed`` in one transaction; +``schedule_outbox_retry`` flips both to ``failed`` once retries are exhausted, +in one transaction. Usage/sessions migrate behind this port in the next +slice (see the ADR's rollout section). The alert-rule repository methods (``load_alert_rules`` / ``save_alert_rules``) operate on plain JSON-shaped ``dict`` records rather than the ``AlertRule`` pydantic model: the port must not import ``src.serving.api.alerts`` (that module imports this one to resolve the store — see ``get_control_plane_store`` below), so callers validate/serialize at the boundary, exactly like the -embedded adapter's YAML round-trip already did before the port existed. +embedded adapter's YAML round-trip already did before the port existed. The +outbox/dead-letter methods follow the same rule for the same reason. + +The outbox/dead-letter write methods (``mark_outbox_sent``, +``schedule_outbox_retry``, ``enqueue_outbox_replay``, +``dismiss_dead_letter_event``) do NOT lazily create their tables on every +call, unlike the webhook/alert log methods above: ``OutboxProcessor`` and +``EventReplayer`` call ``ensure_outbox_schema`` once at construction (mirrors +the pre-port ``OutboxProcessor.__init__`` behavior verbatim). A lazy +``CREATE TABLE IF NOT EXISTS`` inside these methods would silently recreate a +table a test dropped mid-scenario to simulate a transaction failure, +defeating that fault injection. """ from __future__ import annotations @@ -43,6 +59,8 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: + from datetime import datetime + from fastapi import FastAPI CONTROL_PLANE_STORE_ENV = "AGENTFLOW_CONTROLPLANE_STORE" @@ -59,6 +77,17 @@ class WebhookQueueRow: body: str | None +@dataclass(frozen=True) +class OutboxEntry: + """One pending replay-outbox row (a Kafka message owed to a topic).""" + + id: str + event_id: str + payload: object + topic: str + retry_count: int + + class ControlPlaneStore(ABC): """Port for control-plane state (ADR 0010). See the module docstring for the claim-semantics contract adapters must uphold.""" @@ -176,6 +205,100 @@ def save_alert_rules(self, rules: list[dict]) -> None: """Persist the full alert-rule set verbatim (JSON-shaped records, the caller's serialized ``AlertRule.model_dump(mode="json")`` output).""" + # --- replay outbox + dead-letter (invariant 8: one transaction) ----------- + + @abstractmethod + def ensure_outbox_schema(self) -> None: + """Idempotently create the ``outbox`` and ``dead_letter_events`` + tables. Called once at ``OutboxProcessor`` / ``EventReplayer`` + construction — not lazily inside the methods below (see the module + docstring).""" + + @abstractmethod + def claim_due_outbox_entries(self, *, limit: int = 100) -> list[OutboxEntry]: + """Return up to ``limit`` due ``pending`` outbox rows, oldest first.""" + + @abstractmethod + def get_pending_outbox_entry(self, outbox_id: str) -> OutboxEntry | None: + """Fetch one ``pending`` outbox row by id, or ``None`` if it does not + exist or has already left the ``pending`` state.""" + + @abstractmethod + def mark_outbox_sent(self, *, outbox_id: str, event_id: str) -> None: + """Flip an outbox row to ``sent`` and its dead-letter row (if any) to + ``replayed``, in one transaction (invariant 8). Rolls back and + re-raises if either update fails.""" + + @abstractmethod + def schedule_outbox_retry( + self, + *, + outbox_id: str, + event_id: str, + retry_count: int, + error_message: str, + max_retries: int, + ) -> None: + """Bump an outbox row's attempts and back off (exponential, floored at + 30s for Kafka-shaped errors), or park it ``failed`` and flip its + dead-letter row to ``failed`` too once ``max_retries`` is reached — + both updates in one transaction (invariant 8).""" + + @abstractmethod + def enqueue_outbox_replay( + self, + *, + outbox_id: str, + event_id: str, + payload: dict, + topic: str, + retry_count: int, + replayed_at: datetime, + ) -> None: + """Mark a dead-letter row ``replay_pending`` (with the corrected + payload and bumped retry count) and insert its outbox replay row, in + one transaction (invariant 8, the other half of ``mark_outbox_sent``: + this is the write ``EventReplayer.replay`` performs before inline + delivery).""" + + @abstractmethod + def get_dead_letter_event_for_replay(self, event_id: str) -> dict | None: + """Fetch ``{event_id, payload, retry_count}`` for a replay/dismiss + candidate, or ``None`` if the event does not exist.""" + + @abstractmethod + def dismiss_dead_letter_event(self, event_id: str) -> None: + """Mark a dead-letter row ``dismissed``.""" + + @abstractmethod + def dead_letter_event_exists(self, event_id: str, tenant_id: str) -> bool: + """Whether a dead-letter row exists for this event, scoped to + ``tenant_id`` (write-access and tenant-isolation gate).""" + + @abstractmethod + def get_dead_letter_event(self, event_id: str, tenant_id: str) -> dict | None: + """Full detail record for one dead-letter event, tenant-scoped, or + ``None`` if not found. ``payload`` is returned as stored (string or + dict) — the caller decodes it.""" + + @abstractmethod + def list_dead_letter_events( + self, + *, + tenant_id: str, + reason: str | None, + page: int, + page_size: int, + ) -> tuple[list[dict], int]: + """Paginated ``failed`` dead-letter events for one tenant (optionally + filtered by ``failure_reason``), newest first, plus the total count + matching the filter.""" + + @abstractmethod + def get_dead_letter_stats(self, tenant_id: str) -> dict: + """``{"counts": {reason: count}, "last_24h": int, "trend": [...]}`` + for one tenant's active (``failed``) dead-letter events.""" + def get_control_plane_store(app: FastAPI) -> ControlPlaneStore: """Resolve the app's control-plane store, creating the configured one on diff --git a/tests/chaos/conftest.py b/tests/chaos/conftest.py index 48505655..ee66c7fe 100644 --- a/tests/chaos/conftest.py +++ b/tests/chaos/conftest.py @@ -16,8 +16,8 @@ from fastapi.testclient import TestClient import src.serving.api.main as main_module -from src.processing.event_replayer import ensure_dead_letter_table from src.serving.cache import QueryCache +from src.serving.control_plane import ensure_dead_letter_table PROJECT_ROOT = Path(__file__).resolve().parents[2] CHAOS_COMPOSE_FILE = PROJECT_ROOT / "docker-compose.chaos.yml" diff --git a/tests/integration/test_logical_correctness.py b/tests/integration/test_logical_correctness.py index 84c7a4f9..bf7afb17 100644 --- a/tests/integration/test_logical_correctness.py +++ b/tests/integration/test_logical_correctness.py @@ -12,8 +12,9 @@ from pyiceberg.exceptions import ServiceUnavailableError from src.processing import local_pipeline as local_pipeline_module -from src.processing.event_replayer import EventReplayer, ensure_dead_letter_table +from src.processing.event_replayer import EventReplayer from src.serving.api.main import app +from src.serving.control_plane import ensure_dead_letter_table pytestmark = pytest.mark.integration diff --git a/tests/integration/test_outbox.py b/tests/integration/test_outbox.py index e4ac0118..325d4093 100644 --- a/tests/integration/test_outbox.py +++ b/tests/integration/test_outbox.py @@ -15,7 +15,8 @@ from opentelemetry.sdk.trace import TracerProvider import src.serving.api.main as main_module -from src.processing.event_replayer import EventReplayer, ensure_dead_letter_table +from src.processing.event_replayer import EventReplayer +from src.serving.control_plane import ensure_dead_letter_table pytestmark = pytest.mark.integration diff --git a/tests/unit/test_catalog_ddl_concurrency.py b/tests/unit/test_catalog_ddl_concurrency.py index 08e35df2..606836ef 100644 --- a/tests/unit/test_catalog_ddl_concurrency.py +++ b/tests/unit/test_catalog_ddl_concurrency.py @@ -6,12 +6,11 @@ import duckdb import pytest -import src.processing.event_replayer as event_replayer_module import src.serving.control_plane.embedded as control_plane_embedded_module from src.db_concurrency import catalog_ddl_lock -from src.processing.event_replayer import ensure_dead_letter_table from src.serving.control_plane import ( ensure_alert_history_table, + ensure_dead_letter_table, ensure_webhook_deliveries_table, ensure_webhook_delivery_queue_table, ) @@ -82,11 +81,8 @@ def test_ensure_tables_concurrency_safe_across_tables_cold_db() -> None: def test_catalog_ddl_lock_is_a_single_shared_instance() -> None: - # Every ensure_* helper (across both modules) must guard DDL with the - # *same* lock instance, or the cross-table conflict resurfaces. Pins against a - # future per-module lock. - assert ( - event_replayer_module.catalog_ddl_lock - is control_plane_embedded_module.catalog_ddl_lock - is catalog_ddl_lock - ) + # Every lazily-called ensure_* helper (webhook/alert/dead-letter — all now + # live in control_plane.embedded since ADR 0010 slice 3) must guard DDL + # with the *same* lock instance, or the cross-table conflict resurfaces. + # Pins against a future per-module lock. + assert control_plane_embedded_module.catalog_ddl_lock is catalog_ddl_lock diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index e762ca38..3eb8733e 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -1,11 +1,13 @@ -"""ADR 0010 slices 1-2: the ControlPlaneStore port and its embedded adapter. +"""ADR 0010 slices 1-3: the ControlPlaneStore port and its embedded adapter. Covers the store-level contract the dispatcher relies on (enqueue-win semantics, due-claim ordering and bounds, the outcome state machine, parking, attempt-log roundtrip), the alert-delivery history log and the YAML-backed -alert-rule repository (slice 2), the ``get_control_plane_store`` -resolution/ratchet, and the structural pin that keeps the webhook and alert -paths from re-growing direct ``query_engine._conn`` reaches. +alert-rule repository (slice 2), the replay outbox + dead-letter transitions +including invariant 8's transactional atomicity (slice 3), the +``get_control_plane_store`` resolution/ratchet, and the structural pin that +keeps the webhook/alert/outbox/dead-letter paths from re-growing direct +``query_engine._conn`` reaches. """ from __future__ import annotations @@ -49,6 +51,13 @@ def alert_store(conn: duckdb.DuckDBPyConnection, tmp_path: Path) -> EmbeddedCont ) +@pytest.fixture +def outbox_store(conn: duckdb.DuckDBPyConnection) -> EmbeddedControlPlaneStore: + store = EmbeddedControlPlaneStore(conn_provider=lambda: conn) + store.ensure_outbox_schema() + return store + + def _enqueue(store: EmbeddedControlPlaneStore, webhook_id: str, event_id: str) -> bool: return store.enqueue_webhook_delivery( webhook_id=webhook_id, @@ -308,6 +317,263 @@ def test_alert_rules_methods_require_a_path_provider(store: EmbeddedControlPlane store.load_alert_rules() +# --- replay outbox + dead-letter (invariant 8) ------------------------------------ + + +def _seed_dead_letter( + conn: duckdb.DuckDBPyConnection, + *, + event_id: str, + tenant_id: str = "acme", + status: str = "failed", +) -> None: + conn.execute( + """ + INSERT INTO dead_letter_events ( + event_id, tenant_id, event_type, payload, failure_reason, + failure_detail, received_at, retry_count, last_retried_at, status + ) VALUES (?, ?, 'order.created', '{"a": 1}', 'semantic', 'x', ?, 0, NULL, ?) + """, + [event_id, tenant_id, datetime.now(UTC), status], + ) + + +def _seed_outbox( + conn: duckdb.DuckDBPyConnection, + *, + outbox_id: str, + event_id: str = "evt-1", + status: str = "pending", +) -> None: + conn.execute( + """ + INSERT INTO outbox (id, event_id, payload, topic, status, retry_count, next_attempt_at) + VALUES (?, ?, '{"a": 1}', 'agentflow.orders', ?, 0, ?) + """, + [outbox_id, event_id, status, datetime.now(UTC)], + ) + + +def test_claim_due_outbox_entries_returns_oldest_first( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + for index in range(3): + _seed_outbox(conn, outbox_id=f"o{index}", event_id=f"e{index}") + conn.execute( + "UPDATE outbox SET created_at = ? WHERE id = ?", + [datetime.now(UTC) + timedelta(seconds=index), f"o{index}"], + ) + + claimed = outbox_store.claim_due_outbox_entries(limit=2) + + assert [entry.id for entry in claimed] == ["o0", "o1"] + assert claimed[0].topic == "agentflow.orders" + + +def test_get_pending_outbox_entry_returns_none_when_not_pending_or_missing( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_outbox(conn, outbox_id="o1", status="sent") + assert outbox_store.get_pending_outbox_entry("o1") is None + assert outbox_store.get_pending_outbox_entry("does-not-exist") is None + + +def test_mark_outbox_sent_flips_both_rows_in_one_transaction( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_outbox(conn, outbox_id="o1", event_id="e1") + _seed_dead_letter(conn, event_id="e1") + + outbox_store.mark_outbox_sent(outbox_id="o1", event_id="e1") + + assert conn.execute("SELECT status FROM outbox WHERE id = 'o1'").fetchone()[0] == "sent" + assert ( + conn.execute("SELECT status FROM dead_letter_events WHERE event_id = 'e1'").fetchone()[0] + == "replayed" + ) + + +def test_mark_outbox_sent_rolls_back_when_dead_letter_update_fails( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_outbox(conn, outbox_id="o1", event_id="e1") + conn.execute("DROP TABLE dead_letter_events") + + with pytest.raises(duckdb.Error): + outbox_store.mark_outbox_sent(outbox_id="o1", event_id="e1") + + assert conn.execute("SELECT status FROM outbox WHERE id = 'o1'").fetchone()[0] == "pending" + + +def test_schedule_outbox_retry_backs_off_then_fails_and_dead_letters( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_outbox(conn, outbox_id="o1", event_id="e1") + _seed_dead_letter(conn, event_id="e1") + + outbox_store.schedule_outbox_retry( + outbox_id="o1", event_id="e1", retry_count=1, error_message="boom", max_retries=2 + ) + status, next_at = conn.execute( + "SELECT status, next_attempt_at FROM outbox WHERE id = 'o1'" + ).fetchone() + assert status == "pending" + assert next_at is not None + + outbox_store.schedule_outbox_retry( + outbox_id="o1", event_id="e1", retry_count=2, error_message="boom", max_retries=2 + ) + status, next_at = conn.execute( + "SELECT status, next_attempt_at FROM outbox WHERE id = 'o1'" + ).fetchone() + assert status == "failed" + assert next_at is None + assert ( + conn.execute("SELECT status FROM dead_letter_events WHERE event_id = 'e1'").fetchone()[0] + == "failed" + ) + + +def test_schedule_outbox_retry_floors_kafka_shaped_errors_at_30s( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_outbox(conn, outbox_id="o1", event_id="e1") + _seed_dead_letter(conn, event_id="e1") + + outbox_store.schedule_outbox_retry( + outbox_id="o1", + event_id="e1", + retry_count=1, + error_message="KafkaError{code=_MSG_TIMED_OUT}", + max_retries=5, + ) + + next_at = conn.execute("SELECT next_attempt_at FROM outbox WHERE id = 'o1'").fetchone()[0] + assert next_at >= datetime.now(UTC).replace(tzinfo=next_at.tzinfo) + timedelta(seconds=20) + + +def test_enqueue_outbox_replay_marks_pending_and_inserts_row_in_one_transaction( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", status="failed") + replayed_at = datetime.now(UTC) + + outbox_store.enqueue_outbox_replay( + outbox_id="o1", + event_id="e1", + payload={"event_id": "e1", "total_amount": "9.99"}, + topic="events.raw", + retry_count=1, + replayed_at=replayed_at, + ) + + dl_status, dl_retry = conn.execute( + "SELECT status, retry_count FROM dead_letter_events WHERE event_id = 'e1'" + ).fetchone() + assert dl_status == "replay_pending" + assert dl_retry == 1 + outbox_row = conn.execute( + "SELECT event_id, topic, status FROM outbox WHERE id = 'o1'" + ).fetchone() + assert outbox_row == ("e1", "events.raw", "pending") + + +def test_enqueue_outbox_replay_rolls_back_when_outbox_insert_fails( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", status="failed") + conn.execute("DROP TABLE outbox") + + with pytest.raises(duckdb.Error): + outbox_store.enqueue_outbox_replay( + outbox_id="o1", + event_id="e1", + payload={"event_id": "e1"}, + topic="events.raw", + retry_count=1, + replayed_at=datetime.now(UTC), + ) + + assert ( + conn.execute("SELECT status FROM dead_letter_events WHERE event_id = 'e1'").fetchone()[0] + == "failed" + ) + + +def test_get_dead_letter_event_for_replay_returns_none_when_missing( + outbox_store: EmbeddedControlPlaneStore, +) -> None: + assert outbox_store.get_dead_letter_event_for_replay("ghost") is None + + +def test_dismiss_dead_letter_event_marks_dismissed( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1") + outbox_store.dismiss_dead_letter_event("e1") + assert ( + conn.execute("SELECT status FROM dead_letter_events WHERE event_id = 'e1'").fetchone()[0] + == "dismissed" + ) + + +def test_dead_letter_event_exists_is_tenant_scoped( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", tenant_id="acme") + assert outbox_store.dead_letter_event_exists("e1", "acme") is True + assert outbox_store.dead_letter_event_exists("e1", "beta") is False + assert outbox_store.dead_letter_event_exists("ghost", "acme") is False + + +def test_get_dead_letter_event_is_tenant_scoped_and_shapes_detail( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", tenant_id="acme") + + record = outbox_store.get_dead_letter_event("e1", "acme") + + assert record is not None + assert record["event_id"] == "e1" + assert record["failure_reason"] == "semantic" + assert outbox_store.get_dead_letter_event("e1", "beta") is None + + +def test_list_dead_letter_events_paginates_and_filters_by_reason( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", tenant_id="acme") + _seed_dead_letter(conn, event_id="e2", tenant_id="acme") + _seed_dead_letter(conn, event_id="e-beta", tenant_id="beta") + conn.execute("UPDATE dead_letter_events SET failure_reason = 'schema' WHERE event_id = 'e2'") + + items, total = outbox_store.list_dead_letter_events( + tenant_id="acme", reason=None, page=1, page_size=1 + ) + assert total == 2 + assert len(items) == 1 + + items, total = outbox_store.list_dead_letter_events( + tenant_id="acme", reason="schema", page=1, page_size=10 + ) + assert total == 1 + assert items[0]["event_id"] == "e2" + + +def test_get_dead_letter_stats_counts_by_reason_and_last_24h( + outbox_store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _seed_dead_letter(conn, event_id="e1", tenant_id="acme") + _seed_dead_letter(conn, event_id="e2", tenant_id="acme") + _seed_dead_letter(conn, event_id="e-beta", tenant_id="beta") + conn.execute("UPDATE dead_letter_events SET failure_reason = 'schema' WHERE event_id = 'e2'") + + stats = outbox_store.get_dead_letter_stats("acme") + + assert stats["counts"] == {"semantic": 1, "schema": 1} + assert stats["last_24h"] == 2 + + # --- get_control_plane_store resolution + ratchet ------------------------------- @@ -372,16 +638,19 @@ def test_get_store_follows_a_swapped_query_engine( def test_webhook_path_does_not_reach_into_the_engine_connection() -> None: - """ADR 0010 slices 1-2 ratchet: the webhook/alert dispatchers and their - routers go through the ControlPlaneStore port; direct ``query_engine._conn`` - reaches must not re-grow there. The single sanctioned reach is the - composition seam in ``control_plane/store.py``.""" + """ADR 0010 slices 1-3 ratchet: the webhook/alert/outbox/dead-letter + dispatchers and their routers go through the ControlPlaneStore port; + direct ``query_engine._conn`` reaches must not re-grow there. The single + sanctioned reach is the composition seam in ``control_plane/store.py``.""" for relative in ( "src/serving/api/webhook_dispatcher.py", "src/serving/api/routers/webhooks.py", "src/serving/api/alerts/dispatcher.py", "src/serving/api/alerts/escalation.py", "src/serving/api/routers/alerts.py", + "src/processing/outbox.py", + "src/processing/event_replayer.py", + "src/serving/api/routers/deadletter.py", ): source = (PROJECT_ROOT / relative).read_text(encoding="utf-8") assert "query_engine._conn" not in source, relative diff --git a/tests/unit/test_event_replayer.py b/tests/unit/test_event_replayer.py index 295a9258..0953b480 100644 --- a/tests/unit/test_event_replayer.py +++ b/tests/unit/test_event_replayer.py @@ -21,8 +21,8 @@ DeadLetterEventNotFoundError, EventReplayer, ReplayValidationError, - ensure_dead_letter_table, ) +from src.serving.control_plane import ensure_dead_letter_table # Event schema pins event_id to a 36-char uuid pattern (^[a-f0-9\-]{36}$). EVENT_ID_1 = "11111111-1111-4111-8111-111111111111" diff --git a/tests/unit/test_outbox_processor.py b/tests/unit/test_outbox_processor.py index 59b031c6..dcfcdf10 100644 --- a/tests/unit/test_outbox_processor.py +++ b/tests/unit/test_outbox_processor.py @@ -18,8 +18,8 @@ import pytest import src.processing.outbox as outbox_module -from src.processing.event_replayer import ensure_dead_letter_table from src.processing.outbox import OutboxProcessor +from src.serving.control_plane import ensure_dead_letter_table TOPIC = "agentflow.orders"