diff --git a/CHANGELOG.md b/CHANGELOG.md index 634358a1..a344d8b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,25 @@ All notable changes to AgentFlow are documented in this file. ## [Unreleased] +### Added — ControlPlaneStore port + embedded adapter: webhook queue/log behind it (ADR 0010 slice 1, 2026-07-02) + +- **New `src/serving/control_plane/`** — the `ControlPlaneStore` port and its + `EmbeddedControlPlaneStore` (DuckDB) adapter; the webhook durable delivery + queue and the delivery attempt log are the first subsystem behind it. Pure + extraction: DDL, SQL shapes, the catalog-DDL-lock discipline and the + dispatcher's pinned method signatures are byte-compatible — no behavior + change on the embedded (default) profile. +- Claim semantics are part of the port contract (enqueue-winner-only inline + delivery; `claim_due` ownership — degenerate in one process, `FOR UPDATE + SKIP LOCKED` + lease in the slice-5 PostgreSQL adapter); the outcome state + machine moved into the store so it can be one transaction on PostgreSQL, + while retry policy stays dispatcher configuration. +- `webhook_dispatcher` and `routers/webhooks.py` no longer reach into + `query_engine._conn` — pinned by a structural ratchet test; the + `AGENTFLOW_CONTROLPLANE_STORE` knob resolves the adapter (`postgres` raises + until slice 5 ships — fail-closed, pinned by test; unknown values fail the + boot). New `tests/unit/test_control_plane_store.py` (12 tests). + ### Added — Control-plane externalization decided; scaling gate enforced at render time (ADR 0010, 2026-07-02) - **New [ADR 0010](docs/decisions/0010-control-plane-externalization-postgres.md)** diff --git a/src/serving/api/main.py b/src/serving/api/main.py index e73f81a9..18480b53 100644 --- a/src/serving/api/main.py +++ b/src/serving/api/main.py @@ -53,6 +53,7 @@ ) from src.serving.api.webhook_dispatcher import WebhookDispatcher from src.serving.cache import QueryCache +from src.serving.control_plane import get_control_plane_store from src.serving.db_pool import DuckDBPool from src.serving.semantic_layer.catalog import DataCatalog from src.serving.semantic_layer.query_engine import QueryEngine @@ -195,6 +196,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app.state.auth_manager.register_signal_handlers() app.state.auth_manager.ensure_usage_table() ensure_analytics_table(app.state.auth_manager.db_path) + # Control-plane store (ADR 0010): resolve eagerly so a misconfigured + # AGENTFLOW_CONTROLPLANE_STORE fails the boot, not the first delivery. + # Reset any instance cached by a previous lifespan of this process-wide + # app — the query engine above is fresh, the store must bind to it. + app.state.control_plane_store = None + get_control_plane_store(app) app.state.webhook_dispatcher = WebhookDispatcher(app) original_dispatch_new_events = app.state.webhook_dispatcher.dispatch_new_events diff --git a/src/serving/api/routers/webhooks.py b/src/serving/api/routers/webhooks.py index c1b87da2..f066434a 100644 --- a/src/serving/api/routers/webhooks.py +++ b/src/serving/api/routers/webhooks.py @@ -12,11 +12,11 @@ WebhookFilters, create_webhook, deactivate_webhook, - get_delivery_logs, get_webhook, get_webhook_config_path, list_webhooks, ) +from src.serving.control_plane import get_control_plane_store router = APIRouter(prefix="/v1/webhooks", tags=["webhooks"]) @@ -106,11 +106,8 @@ async def webhook_logs(webhook_id: str, request: Request) -> dict[str, object]: def _read_delivery_logs(request: Request, webhook_id: str) -> list[dict]: - # The log scan runs on a worker thread (run_in_threadpool); a dedicated - # cursor — not the shared connection — keeps concurrent reads on different - # threads from colliding on the connection. (audit_30_06_26.md A2) - cursor = request.app.state.query_engine._conn.cursor() - try: - return get_delivery_logs(cursor, webhook_id) - finally: - cursor.close() + # Runs on a worker thread (run_in_threadpool); the control-plane store + # isolates the read (a dedicated cursor per call in the embedded adapter) + # so concurrent reads on different threads don't collide on the shared + # connection. (audit_30_06_26.md A2) + return get_control_plane_store(request.app).get_webhook_delivery_logs(webhook_id) diff --git a/src/serving/api/webhook_dispatcher.py b/src/serving/api/webhook_dispatcher.py index 6616f486..37327e35 100644 --- a/src/serving/api/webhook_dispatcher.py +++ b/src/serving/api/webhook_dispatcher.py @@ -7,7 +7,7 @@ import os import secrets import uuid -from datetime import UTC, date, datetime, timedelta +from datetime import UTC, date, datetime from decimal import Decimal from pathlib import Path @@ -17,9 +17,9 @@ from fastapi import FastAPI from pydantic import BaseModel, Field -from src.db_concurrency import catalog_ddl_lock from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url from src.serving.backends import BackendExecutionError +from src.serving.control_plane import get_control_plane_store try: import yaml @@ -125,84 +125,6 @@ def deactivate_webhook(path: Path, webhook_id: str, tenant: str) -> bool: return changed -def ensure_webhook_deliveries_table(conn: duckdb.DuckDBPyConnection) -> None: - # Serialize the lazy DDL: the offloaded read handler calls this on a worker - # thread, and concurrent CREATE on a cold DuckDB catalog conflicts (across - # tables too). (audit_30 A2 follow-up: #120 offload race) - with catalog_ddl_lock: - conn.execute( - """ - CREATE TABLE IF NOT EXISTS webhook_deliveries ( - delivery_id VARCHAR, - webhook_id VARCHAR, - event_id VARCHAR, - event_type VARCHAR, - attempt INTEGER, - status_code INTEGER, - success BOOLEAN, - error TEXT, - delivered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - -def get_delivery_logs(conn: duckdb.DuckDBPyConnection, webhook_id: str) -> list[dict]: - ensure_webhook_deliveries_table(conn) - cursor = conn.execute( - """ - SELECT delivery_id, webhook_id, event_id, event_type, attempt, - status_code, success, error, delivered_at - FROM webhook_deliveries - WHERE webhook_id = ? - ORDER BY delivered_at DESC - LIMIT 20 - """, - [webhook_id], - ) - columns = [description[0] for description in cursor.description] - return [dict(zip(columns, row, strict=False)) for row in cursor.fetchall()] - - -def ensure_webhook_delivery_queue_table(conn: duckdb.DuckDBPyConnection) -> None: - """Durable per-(webhook, event) delivery state for re-drive. - - Distinct from ``webhook_deliveries`` (an append-only attempt *log*): this is - the *state* table whose ``(webhook_id, event_id)`` primary key dedupes - enqueues and whose ``status`` / ``next_attempt_at`` drive retries that - survive a process restart. ``body`` stores the canonical payload so a - delivery can be replayed without re-reading ``pipeline_events``. - """ - # Serialize the lazy DDL behind the shared catalog lock, exactly like the - # three #123-locked ``ensure_*`` siblings: the dispatcher creates this table - # on the shared serving connection from the event loop while an offloaded - # read handler runs its own ``ensure_*`` on a worker thread, and concurrent - # CREATE on a cold DuckDB catalog conflicts across *different* tables too. - # Omitting the lock here left the cross-table "Catalog write-write conflict" - # the #123 fix set out to remove still reachable on a cold restart. - # (audit_30 D2/A2 follow-up residual) - with catalog_ddl_lock: - conn.execute( - """ - CREATE TABLE IF NOT EXISTS webhook_delivery_queue ( - webhook_id VARCHAR NOT NULL, - event_id VARCHAR NOT NULL, - tenant VARCHAR, - event_type VARCHAR, - body VARCHAR, - status VARCHAR NOT NULL DEFAULT 'pending', - attempts INTEGER NOT NULL DEFAULT 0, - next_attempt_at TIMESTAMP, - last_status_code INTEGER, - last_error VARCHAR, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (webhook_id, event_id) - ) - """ - ) - - class WebhookDispatcher: def __init__(self, app: FastAPI, poll_interval_seconds: float = 2.0) -> None: self.app = app @@ -345,8 +267,7 @@ async def _deliver_body( per-attempt logging. Shared by :meth:`deliver` and the durable re-drive (:meth:`process_delivery_queue`), which replays the stored body verbatim. """ - conn = self.app.state.query_engine._conn - ensure_webhook_deliveries_table(conn) + store = get_control_plane_store(self.app) delivery_id = str(uuid.uuid4()) if not event_id: @@ -371,8 +292,7 @@ async def _deliver_body( await asyncio.to_thread(validate_public_url, webhook.url) except UnsafeEgressURLError as exc: error = f"unsafe egress URL: {exc}" - _log_delivery( - conn, + store.log_webhook_delivery( delivery_id=delivery_id, webhook_id=webhook.id, event_id=event_id, @@ -405,8 +325,7 @@ async def _deliver_body( ) status_code = response.status_code success = 200 <= response.status_code < 300 - _log_delivery( - conn, + store.log_webhook_delivery( delivery_id=delivery_id, webhook_id=webhook.id, event_id=event_id, @@ -422,8 +341,7 @@ async def _deliver_body( status_code = None success = False error = str(exc) - _log_delivery( - conn, + store.log_webhook_delivery( delivery_id=delivery_id, webhook_id=webhook.id, event_id=event_id, @@ -467,82 +385,30 @@ def _enqueue_delivery(self, webhook: WebhookRegistration, event: dict) -> bool: event_id = str(event.get("event_id") or "") if not event_id: return False - conn = self.app.state.query_engine._conn - ensure_webhook_delivery_queue_table(conn) - existing = conn.execute( - "SELECT 1 FROM webhook_delivery_queue WHERE webhook_id = ? AND event_id = ?", - [webhook.id, event_id], - ).fetchone() - if existing is not None: - return False - now = datetime.now(UTC) - conn.execute( - """ - INSERT INTO webhook_delivery_queue - (webhook_id, event_id, tenant, event_type, body, status, attempts, - next_attempt_at, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?) - ON CONFLICT DO NOTHING - """, - [ - webhook.id, - event_id, - str(event.get("tenant_id") or "default"), - str(event.get("event_type") or event.get("topic") or "unknown"), - _event_body(event).decode("utf-8"), - now, - now, - now, - ], + return get_control_plane_store(self.app).enqueue_webhook_delivery( + webhook_id=webhook.id, + event_id=event_id, + tenant=str(event.get("tenant_id") or "default"), + event_type=str(event.get("event_type") or event.get("topic") or "unknown"), + body=_event_body(event).decode("utf-8"), ) - return True def _record_delivery_outcome(self, webhook_id: str, event_id: str, result: dict) -> None: """Advance a queue row from the outcome of one delivery round: success → ``delivered``; failure → bump attempts and re-schedule (back to ``pending`` with a backoff ``next_attempt_at``), or park as ``dead`` once - ``max_delivery_attempts`` is reached.""" + ``max_delivery_attempts`` is reached. The transition itself lives in the + control-plane store; the retry policy stays dispatcher configuration.""" if not event_id: return - conn = self.app.state.query_engine._conn - now = datetime.now(UTC) - status_code = result.get("status_code") - if result.get("success"): - conn.execute( - "UPDATE webhook_delivery_queue SET status = 'delivered', " - "last_status_code = ?, last_error = NULL, updated_at = ? " - "WHERE webhook_id = ? AND event_id = ?", - [status_code, now, webhook_id, event_id], - ) - return - row = conn.execute( - "SELECT attempts FROM webhook_delivery_queue WHERE webhook_id = ? AND event_id = ?", - [webhook_id, event_id], - ).fetchone() - attempts = (row[0] if row else 0) + 1 - error = result.get("error") - if attempts >= self.max_delivery_attempts: - conn.execute( - "UPDATE webhook_delivery_queue SET status = 'dead', attempts = ?, " - "last_status_code = ?, last_error = ?, next_attempt_at = NULL, updated_at = ? " - "WHERE webhook_id = ? AND event_id = ?", - [attempts, status_code, error, now, webhook_id, event_id], - ) - return - delay = self.backoff_seconds[min(attempts - 1, len(self.backoff_seconds) - 1)] - conn.execute( - "UPDATE webhook_delivery_queue SET status = 'pending', attempts = ?, " - "last_status_code = ?, last_error = ?, next_attempt_at = ?, updated_at = ? " - "WHERE webhook_id = ? AND event_id = ?", - [ - attempts, - status_code, - error, - now + timedelta(seconds=delay), - now, - webhook_id, - event_id, - ], + get_control_plane_store(self.app).record_webhook_delivery_outcome( + webhook_id=webhook_id, + event_id=event_id, + success=bool(result.get("success")), + status_code=result.get("status_code"), + error=result.get("error"), + max_attempts=self.max_delivery_attempts, + backoff_seconds=self.backoff_seconds, ) async def process_delivery_queue(self) -> None: @@ -550,68 +416,24 @@ async def process_delivery_queue(self) -> None: removed/deactivated parks its row as ``dead`` rather than retrying forever. Bounded by ``redrive_batch_size`` so one pass can't stall the loop on a large backlog.""" - conn = self.app.state.query_engine._conn - ensure_webhook_delivery_queue_table(conn) + store = get_control_plane_store(self.app) path = get_webhook_config_path(self.app) - now = datetime.now(UTC) - due = conn.execute( - "SELECT webhook_id, event_id, tenant, event_type, body " - "FROM webhook_delivery_queue " - "WHERE status = 'pending' AND (next_attempt_at IS NULL OR next_attempt_at <= ?) " - "ORDER BY created_at ASC LIMIT ?", - [now, self.redrive_batch_size], - ).fetchall() - for webhook_id, event_id, tenant, event_type, body in due: - webhook = get_webhook(path, webhook_id, str(tenant or "default")) + for row in store.claim_due_webhook_deliveries(limit=self.redrive_batch_size): + webhook = get_webhook(path, row.webhook_id, str(row.tenant or "default")) if webhook is None: - conn.execute( - "UPDATE webhook_delivery_queue SET status = 'dead', " - "last_error = 'webhook inactive or removed', next_attempt_at = NULL, " - "updated_at = ? WHERE webhook_id = ? AND event_id = ?", - [datetime.now(UTC), webhook_id, event_id], + store.park_webhook_delivery( + webhook_id=row.webhook_id, + event_id=row.event_id, + error="webhook inactive or removed", ) continue result = await self._deliver_body( webhook, - body=(body or "").encode("utf-8"), - event_id=event_id, - event_type=event_type or "unknown", + body=(row.body or "").encode("utf-8"), + event_id=row.event_id, + event_type=row.event_type or "unknown", ) - self._record_delivery_outcome(webhook_id, event_id, result) - - -def _log_delivery( - conn: duckdb.DuckDBPyConnection, - *, - delivery_id: str, - webhook_id: str, - event_id: str, - event_type: str, - attempt: int, - status_code: int | None, - success: bool, - error: str | None, -) -> None: - conn.execute( - """ - INSERT INTO webhook_deliveries ( - delivery_id, webhook_id, event_id, event_type, attempt, - status_code, success, error, delivered_at - ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - [ - delivery_id, - webhook_id, - event_id, - event_type, - attempt, - status_code, - success, - error, - datetime.now(UTC), - ], - ) + self._record_delivery_outcome(row.webhook_id, row.event_id, result) def _matches_filters(event: dict, filters: WebhookFilters) -> bool: diff --git a/src/serving/control_plane/__init__.py b/src/serving/control_plane/__init__.py new file mode 100644 index 00000000..9f1b1691 --- /dev/null +++ b/src/serving/control_plane/__init__.py @@ -0,0 +1,29 @@ +"""Control-plane state store (ADR 0009 / ADR 0010). + +``ControlPlaneStore`` is the port; ``EmbeddedControlPlaneStore`` (DuckDB, +single-replica default) is the shipped adapter; ``PostgresControlPlaneStore`` +(scale profile) arrives with ADR 0010 rollout slice 5. Resolve the app's +store via ``get_control_plane_store`` — never through ``query_engine._conn``. +""" + +from .embedded import ( + EmbeddedControlPlaneStore, + ensure_webhook_deliveries_table, + ensure_webhook_delivery_queue_table, +) +from .store import ( + CONTROL_PLANE_STORE_ENV, + ControlPlaneStore, + WebhookQueueRow, + get_control_plane_store, +) + +__all__ = [ + "CONTROL_PLANE_STORE_ENV", + "ControlPlaneStore", + "EmbeddedControlPlaneStore", + "WebhookQueueRow", + "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 new file mode 100644 index 00000000..2bdc29be --- /dev/null +++ b/src/serving/control_plane/embedded.py @@ -0,0 +1,272 @@ +"""Embedded (DuckDB) control-plane store — the single-replica default profile. + +Extracted from ``webhook_dispatcher`` in ADR 0010 rollout slice 1: the table +DDL, SQL shapes and catalog-DDL-lock behavior are byte-compatible with the +pre-port code, so existing deployments and the pinned regression suites see +no behavior change. + +"Claims" here are trivially exclusive: this store is only ever used by one +process (the Helm chart refuses multi-replica renders on the embedded +profile), so ``claim_due_webhook_deliveries`` is a plain due-scan without +marking rows in-flight. The PostgreSQL adapter (slice 5) implements the same +contract with ``FOR UPDATE SKIP LOCKED`` plus a lease column. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from datetime import UTC, datetime, timedelta + +import duckdb + +from src.db_concurrency import catalog_ddl_lock + +from .store import ControlPlaneStore, WebhookQueueRow + + +def ensure_webhook_deliveries_table(conn: duckdb.DuckDBPyConnection) -> None: + # Serialize the lazy DDL: the offloaded read handler calls this on a worker + # thread, and concurrent CREATE on a cold DuckDB catalog conflicts (across + # tables too). (audit_30 A2 follow-up: #120 offload race) + with catalog_ddl_lock: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS webhook_deliveries ( + delivery_id VARCHAR, + webhook_id VARCHAR, + event_id VARCHAR, + event_type VARCHAR, + attempt INTEGER, + status_code INTEGER, + success BOOLEAN, + error TEXT, + delivered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + + +def ensure_webhook_delivery_queue_table(conn: duckdb.DuckDBPyConnection) -> None: + """Durable per-(webhook, event) delivery state for re-drive. + + Distinct from ``webhook_deliveries`` (an append-only attempt *log*): this is + the *state* table whose ``(webhook_id, event_id)`` primary key dedupes + enqueues and whose ``status`` / ``next_attempt_at`` drive retries that + survive a process restart. ``body`` stores the canonical payload so a + delivery can be replayed without re-reading ``pipeline_events``. + """ + # Serialize the lazy DDL behind the shared catalog lock, exactly like the + # three #123-locked ``ensure_*`` siblings: the dispatcher creates this table + # on the shared serving connection from the event loop while an offloaded + # read handler runs its own ``ensure_*`` on a worker thread, and concurrent + # CREATE on a cold DuckDB catalog conflicts across *different* tables too. + # Omitting the lock here left the cross-table "Catalog write-write conflict" + # the #123 fix set out to remove still reachable on a cold restart. + # (audit_30 D2/A2 follow-up residual) + with catalog_ddl_lock: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS webhook_delivery_queue ( + webhook_id VARCHAR NOT NULL, + event_id VARCHAR NOT NULL, + tenant VARCHAR, + event_type VARCHAR, + body VARCHAR, + status VARCHAR NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMP, + last_status_code INTEGER, + last_error VARCHAR, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (webhook_id, event_id) + ) + """ + ) + + +class EmbeddedControlPlaneStore(ControlPlaneStore): + """Control-plane state on the embedded serving DuckDB connection. + + ``conn_provider`` is resolved per call (not captured once): tests and the + lifespan may swap ``app.state.query_engine``, and the store must follow + the live connection exactly like the pre-port ``_conn`` lookups did. + """ + + def __init__(self, conn_provider: Callable[[], duckdb.DuckDBPyConnection]) -> None: + self._conn_provider = conn_provider + + @property + def _conn(self) -> duckdb.DuckDBPyConnection: + return self._conn_provider() + + # --- webhook durable delivery queue -------------------------------------- + + def enqueue_webhook_delivery( + self, + *, + webhook_id: str, + event_id: str, + tenant: str, + event_type: str, + body: str, + ) -> bool: + conn = self._conn + ensure_webhook_delivery_queue_table(conn) + existing = conn.execute( + "SELECT 1 FROM webhook_delivery_queue WHERE webhook_id = ? AND event_id = ?", + [webhook_id, event_id], + ).fetchone() + if existing is not None: + return False + now = datetime.now(UTC) + conn.execute( + """ + INSERT INTO webhook_delivery_queue + (webhook_id, event_id, tenant, event_type, body, status, attempts, + next_attempt_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?) + ON CONFLICT DO NOTHING + """, + [webhook_id, event_id, tenant, event_type, body, now, now, now], + ) + return True + + def claim_due_webhook_deliveries(self, *, limit: int) -> list[WebhookQueueRow]: + conn = self._conn + ensure_webhook_delivery_queue_table(conn) + rows = conn.execute( + "SELECT webhook_id, event_id, tenant, event_type, body " + "FROM webhook_delivery_queue " + "WHERE status = 'pending' AND (next_attempt_at IS NULL OR next_attempt_at <= ?) " + "ORDER BY created_at ASC LIMIT ?", + [datetime.now(UTC), limit], + ).fetchall() + return [ + WebhookQueueRow( + webhook_id=webhook_id, + event_id=event_id, + tenant=tenant, + event_type=event_type, + body=body, + ) + for webhook_id, event_id, tenant, event_type, body in rows + ] + + def record_webhook_delivery_outcome( + self, + *, + webhook_id: str, + event_id: str, + success: bool, + status_code: int | None, + error: str | None, + max_attempts: int, + backoff_seconds: Sequence[float], + ) -> None: + conn = self._conn + now = datetime.now(UTC) + if success: + conn.execute( + "UPDATE webhook_delivery_queue SET status = 'delivered', " + "last_status_code = ?, last_error = NULL, updated_at = ? " + "WHERE webhook_id = ? AND event_id = ?", + [status_code, now, webhook_id, event_id], + ) + return + row = conn.execute( + "SELECT attempts FROM webhook_delivery_queue WHERE webhook_id = ? AND event_id = ?", + [webhook_id, event_id], + ).fetchone() + attempts = (row[0] if row else 0) + 1 + if attempts >= max_attempts: + conn.execute( + "UPDATE webhook_delivery_queue SET status = 'dead', attempts = ?, " + "last_status_code = ?, last_error = ?, next_attempt_at = NULL, updated_at = ? " + "WHERE webhook_id = ? AND event_id = ?", + [attempts, status_code, error, now, webhook_id, event_id], + ) + return + delay = backoff_seconds[min(attempts - 1, len(backoff_seconds) - 1)] + conn.execute( + "UPDATE webhook_delivery_queue SET status = 'pending', attempts = ?, " + "last_status_code = ?, last_error = ?, next_attempt_at = ?, updated_at = ? " + "WHERE webhook_id = ? AND event_id = ?", + [ + attempts, + status_code, + error, + now + timedelta(seconds=delay), + now, + webhook_id, + event_id, + ], + ) + + def park_webhook_delivery(self, *, webhook_id: str, event_id: str, error: str) -> None: + self._conn.execute( + "UPDATE webhook_delivery_queue SET status = 'dead', " + "last_error = ?, next_attempt_at = NULL, updated_at = ? " + "WHERE webhook_id = ? AND event_id = ?", + [error, datetime.now(UTC), webhook_id, event_id], + ) + + # --- webhook delivery attempt log ---------------------------------------- + + def log_webhook_delivery( + self, + *, + delivery_id: str, + webhook_id: str, + event_id: str, + event_type: str, + attempt: int, + status_code: int | None, + success: bool, + error: str | None, + ) -> None: + conn = self._conn + ensure_webhook_deliveries_table(conn) + conn.execute( + """ + INSERT INTO webhook_deliveries ( + delivery_id, webhook_id, event_id, event_type, attempt, + status_code, success, error, delivered_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + delivery_id, + webhook_id, + event_id, + event_type, + attempt, + status_code, + success, + error, + datetime.now(UTC), + ], + ) + + def get_webhook_delivery_logs(self, webhook_id: str, *, limit: int = 20) -> list[dict]: + # A dedicated cursor per read — not the shared connection — keeps + # concurrent reads on worker threads (run_in_threadpool) from colliding + # on the connection. (audit_30_06_26.md A2) + cursor = self._conn.cursor() + try: + ensure_webhook_deliveries_table(cursor) + result = cursor.execute( + """ + SELECT delivery_id, webhook_id, event_id, event_type, attempt, + status_code, success, error, delivered_at + FROM webhook_deliveries + WHERE webhook_id = ? + ORDER BY delivered_at DESC + LIMIT ? + """, + [webhook_id, limit], + ) + columns = [description[0] for description in result.description] + return [dict(zip(columns, row, strict=False)) for row in result.fetchall()] + finally: + cursor.close() diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py new file mode 100644 index 00000000..7b5f0953 --- /dev/null +++ b/src/serving/control_plane/store.py @@ -0,0 +1,158 @@ +"""Control-plane store port (ADR 0010). + +The API process hosts a control plane — webhook delivery queue and attempt +log, alert rules/history, replay outbox, dead-letter transitions, usage +accounting — whose state must not live embedded per-pod once +``replicaCount > 1`` (ADR 0009). This port is the seam that makes the state +location swappable: an embedded DuckDB adapter is the single-replica default +profile; a PostgreSQL adapter is the scale profile (ADR 0010 rollout +slice 5). + +Claim semantics are part of this contract, not adapter detail: + +- ``enqueue_webhook_delivery`` returns ``True`` only when *this caller* + inserted the row — inline delivery belongs to the enqueue winner alone, so + a re-scan (or a second replica scanning the same journal) never re-POSTs a + (webhook, event) that is already queued. +- ``claim_due_webhook_deliveries`` returns rows this worker now owns. The + embedded adapter satisfies exclusivity degenerately (one process); the + PostgreSQL adapter uses ``FOR UPDATE SKIP LOCKED`` plus a lease column so + N replicas work-steal without leader election. + +Rollout slice 1 covers the webhook delivery queue + attempt log; alert +rules/history, outbox/dead-letter and usage migrate behind this port in +follow-up slices (see the ADR's rollout section). +""" + +from __future__ import annotations + +import os +from abc import ABC, abstractmethod +from collections.abc import Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from fastapi import FastAPI + +CONTROL_PLANE_STORE_ENV = "AGENTFLOW_CONTROLPLANE_STORE" + + +@dataclass(frozen=True) +class WebhookQueueRow: + """One durable (webhook, event) delivery owed to a receiver.""" + + webhook_id: str + event_id: str + tenant: str | None + event_type: str | None + body: str | None + + +class ControlPlaneStore(ABC): + """Port for control-plane state (ADR 0010). See the module docstring for + the claim-semantics contract adapters must uphold.""" + + # --- webhook durable delivery queue (re-drive state) --------------------- + + @abstractmethod + def enqueue_webhook_delivery( + self, + *, + webhook_id: str, + event_id: str, + tenant: str, + event_type: str, + body: str, + ) -> bool: + """Durably record a (webhook, event) delivery as ``pending``. + + Idempotent on ``(webhook_id, event_id)``; returns ``True`` only when a + new row was inserted by this call (the caller may then inline-deliver). + """ + + @abstractmethod + def claim_due_webhook_deliveries(self, *, limit: int) -> list[WebhookQueueRow]: + """Return up to ``limit`` due ``pending`` deliveries this worker now + owns, oldest first.""" + + @abstractmethod + def record_webhook_delivery_outcome( + self, + *, + webhook_id: str, + event_id: str, + success: bool, + status_code: int | None, + error: str | None, + max_attempts: int, + backoff_seconds: Sequence[float], + ) -> None: + """Advance a queue row from one delivery round's outcome: success → + ``delivered``; failure → bump attempts and re-schedule with backoff, or + park as ``dead`` once ``max_attempts`` is reached.""" + + @abstractmethod + def park_webhook_delivery(self, *, webhook_id: str, event_id: str, error: str) -> None: + """Park a queue row as ``dead`` (e.g. its webhook was removed or + deactivated) so it is never re-driven again.""" + + # --- webhook delivery attempt log (append-only) -------------------------- + + @abstractmethod + def log_webhook_delivery( + self, + *, + delivery_id: str, + webhook_id: str, + event_id: str, + event_type: str, + attempt: int, + status_code: int | None, + success: bool, + error: str | None, + ) -> None: + """Append one delivery attempt to the ``webhook_deliveries`` log.""" + + @abstractmethod + def get_webhook_delivery_logs(self, webhook_id: str, *, limit: int = 20) -> list[dict]: + """Most recent attempt-log entries for one webhook, newest first. + + Safe to call from a worker thread (adapters isolate the read — the + embedded store opens a dedicated cursor per call, audit_30 A2).""" + + +def get_control_plane_store(app: FastAPI) -> ControlPlaneStore: + """Resolve the app's control-plane store, creating the configured one on + first use and caching it on ``app.state`` (the lazy pattern mirrors + ``ensure_alert_dispatcher`` so lightweight test stubs keep working). + + ``AGENTFLOW_CONTROLPLANE_STORE`` selects the adapter. Fail-closed ratchet: + ``postgres`` is the ADR 0010 target profile and raises until the + ``PostgresControlPlaneStore`` adapter ships (rollout slice 5); anything + else but ``embedded`` is a configuration error. + """ + store: ControlPlaneStore | None = getattr(app.state, "control_plane_store", None) + if store is not None: + return store + kind = (os.getenv(CONTROL_PLANE_STORE_ENV) or "embedded").strip().lower() + if kind == "embedded": + from .embedded import EmbeddedControlPlaneStore + + # The one sanctioned reach into the engine's embedded connection: the + # composition seam that binds the default profile's store to the + # serving DuckDB. Everything above the port goes through the store. + store = EmbeddedControlPlaneStore(conn_provider=lambda: app.state.query_engine._conn) + elif kind == "postgres": + raise NotImplementedError( + "AGENTFLOW_CONTROLPLANE_STORE=postgres is the ADR 0010 scale profile; " + "the PostgresControlPlaneStore adapter ships in rollout slice 5 — " + "until then only 'embedded' runs." + ) + else: + raise ValueError( + f"Unknown control-plane store {kind!r} " + f"(set {CONTROL_PLANE_STORE_ENV} to 'embedded', or leave it unset)." + ) + app.state.control_plane_store = store + return store diff --git a/tests/integration/test_webhooks.py b/tests/integration/test_webhooks.py index 2db5ca17..76362ee6 100644 --- a/tests/integration/test_webhooks.py +++ b/tests/integration/test_webhooks.py @@ -13,6 +13,7 @@ from src.serving.api import webhook_dispatcher from src.serving.api.auth import TenantKey from src.serving.api.main import app +from src.serving.control_plane import ensure_webhook_delivery_queue_table class _HTTPXMock: @@ -434,7 +435,7 @@ async def test_failed_delivery_is_requeued_and_redriven(self, client, httpx_mock dispatcher = client.app.state.webhook_dispatcher dispatcher.seen_event_ids.clear() conn = client.app.state.query_engine._conn - webhook_dispatcher.ensure_webhook_delivery_queue_table(conn) + ensure_webhook_delivery_queue_table(conn) conn.execute("DELETE FROM webhook_delivery_queue") # endpoint down: every attempt of the inline burst returns 5xx diff --git a/tests/unit/test_catalog_ddl_concurrency.py b/tests/unit/test_catalog_ddl_concurrency.py index 856bdc86..02d201e8 100644 --- a/tests/unit/test_catalog_ddl_concurrency.py +++ b/tests/unit/test_catalog_ddl_concurrency.py @@ -8,20 +8,21 @@ import src.processing.event_replayer as event_replayer_module import src.serving.api.alerts.history as history_module -import src.serving.api.webhook_dispatcher as webhook_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.api.alerts.history import ensure_alert_history_table -from src.serving.api.webhook_dispatcher import ( +from src.serving.control_plane import ( ensure_webhook_deliveries_table, ensure_webhook_delivery_queue_table, ) Ensurer = Callable[[duckdb.DuckDBPyConnection], None] -# ``ensure_webhook_delivery_queue_table`` is the dispatcher's own lazy CREATE -# (run on the shared serving connection from the event loop). It was left out of -# the #123 lock and so still raced the offloaded read-handler DDL across tables; +# ``ensure_webhook_delivery_queue_table`` is the control-plane store's lazy +# CREATE (run on the shared serving connection from the event loop; lives in +# ``control_plane.embedded`` since ADR 0010 slice 1). It was left out of the +# #123 lock and so still raced the offloaded read-handler DDL across tables; # include it here so both the same-table and cross-table hammers cover it. # (audit_30 D2/A2 follow-up residual) _ENSURERS: list[Ensurer] = [ @@ -88,6 +89,6 @@ def test_catalog_ddl_lock_is_a_single_shared_instance() -> None: assert ( event_replayer_module.catalog_ddl_lock is history_module.catalog_ddl_lock - is webhook_module.catalog_ddl_lock + is 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 new file mode 100644 index 00000000..98a87f66 --- /dev/null +++ b/tests/unit/test_control_plane_store.py @@ -0,0 +1,281 @@ +"""ADR 0010 slice 1: 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 ``get_control_plane_store`` resolution/ratchet, +and the structural pin that keeps the webhook path from re-growing direct +``query_engine._conn`` reaches. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import duckdb +import pytest + +from src.serving.control_plane import ( + CONTROL_PLANE_STORE_ENV, + EmbeddedControlPlaneStore, + get_control_plane_store, +) + +PROJECT_ROOT = Path(__file__).resolve().parents[2] + + +@pytest.fixture +def conn() -> Iterator[duckdb.DuckDBPyConnection]: + connection = duckdb.connect(":memory:") + try: + yield connection + finally: + connection.close() + + +@pytest.fixture +def store(conn: duckdb.DuckDBPyConnection) -> EmbeddedControlPlaneStore: + return EmbeddedControlPlaneStore(conn_provider=lambda: conn) + + +def _enqueue(store: EmbeddedControlPlaneStore, webhook_id: str, event_id: str) -> bool: + return store.enqueue_webhook_delivery( + webhook_id=webhook_id, + event_id=event_id, + tenant="acme", + event_type="order.created", + body='{"event_id":"' + event_id + '"}', + ) + + +def _row(conn: duckdb.DuckDBPyConnection, webhook_id: str, event_id: str): + return conn.execute( + "SELECT status, attempts, next_attempt_at, last_error FROM webhook_delivery_queue " + "WHERE webhook_id = ? AND event_id = ?", + [webhook_id, event_id], + ).fetchone() + + +# --- enqueue: the winner-only contract --------------------------------------- + + +def test_enqueue_returns_true_only_for_the_inserting_call( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + assert _enqueue(store, "wh-1", "e1") is True + # Idempotent on the (webhook, event) key: a re-scan is not a win, so the + # caller never inline-re-POSTs an already-queued delivery. + assert _enqueue(store, "wh-1", "e1") is False + assert conn.execute("SELECT count(*) FROM webhook_delivery_queue").fetchone()[0] == 1 + + +# --- claim_due: ordering, bounds, due-ness ------------------------------------ + + +def test_claim_due_returns_oldest_first_and_respects_limit( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + for index in range(3): + _enqueue(store, "wh-1", f"e{index}") + # created_at drives the order; stagger it explicitly (CURRENT_TIMESTAMP + # has coarse resolution within one transaction). + conn.execute( + "UPDATE webhook_delivery_queue SET created_at = ?, next_attempt_at = NULL " + "WHERE event_id = ?", + [datetime.now(UTC) + timedelta(seconds=index), f"e{index}"], + ) + + claimed = store.claim_due_webhook_deliveries(limit=2) + + assert [row.event_id for row in claimed] == ["e0", "e1"] + assert claimed[0].webhook_id == "wh-1" + assert claimed[0].tenant == "acme" + assert claimed[0].body is not None + + +def test_claim_due_skips_future_and_non_pending_rows( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _enqueue(store, "wh-1", "future") + _enqueue(store, "wh-1", "done") + conn.execute( + "UPDATE webhook_delivery_queue SET next_attempt_at = ? WHERE event_id = 'future'", + [datetime.now(UTC) + timedelta(hours=1)], + ) + conn.execute( + "UPDATE webhook_delivery_queue SET status = 'delivered', next_attempt_at = NULL " + "WHERE event_id = 'done'" + ) + + assert store.claim_due_webhook_deliveries(limit=10) == [] + + +# --- outcome state machine ----------------------------------------------------- + + +def test_outcome_success_marks_delivered_and_clears_error( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _enqueue(store, "wh-1", "e1") + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=True, + status_code=200, + error=None, + max_attempts=5, + backoff_seconds=[1.0], + ) + status, _attempts, _next_at, last_error = _row(conn, "wh-1", "e1") + assert status == "delivered" + assert last_error is None + + +def test_outcome_failure_backs_off_then_parks_dead_at_max( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _enqueue(store, "wh-1", "e1") + + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=2, + backoff_seconds=[10.0], + ) + status, attempts, next_at, _ = _row(conn, "wh-1", "e1") + assert (status, attempts) == ("pending", 1) + assert next_at is not None # re-drive scheduled with backoff + + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=2, + backoff_seconds=[10.0], + ) + status, attempts, next_at, _ = _row(conn, "wh-1", "e1") + assert (status, attempts) == ("dead", 2) + assert next_at is None # parked for good + + +def test_park_marks_dead_with_reason( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + _enqueue(store, "ghost", "e1") + store.park_webhook_delivery( + webhook_id="ghost", event_id="e1", error="webhook inactive or removed" + ) + status, _attempts, next_at, last_error = _row(conn, "ghost", "e1") + assert status == "dead" + assert next_at is None + assert last_error == "webhook inactive or removed" + + +# --- attempt log --------------------------------------------------------------- + + +def test_delivery_log_roundtrip_is_isolated_per_webhook_and_newest_first( + store: EmbeddedControlPlaneStore, +) -> None: + for attempt in (1, 2): + store.log_webhook_delivery( + delivery_id=f"d{attempt}", + webhook_id="wh-1", + event_id="e1", + event_type="order.created", + attempt=attempt, + status_code=500 if attempt == 1 else 200, + success=attempt == 2, + error="boom" if attempt == 1 else None, + ) + + logs = store.get_webhook_delivery_logs("wh-1") + + assert [entry["attempt"] for entry in logs] == [2, 1] or [ + entry["delivery_id"] for entry in logs + ] == ["d2", "d1"] + assert store.get_webhook_delivery_logs("wh-other") == [] + assert store.get_webhook_delivery_logs("wh-1", limit=1)[0]["delivery_id"] in {"d1", "d2"} + + +# --- get_control_plane_store resolution + ratchet ------------------------------- + + +def _stub_app(conn: duckdb.DuckDBPyConnection) -> SimpleNamespace: + engine = SimpleNamespace(_conn=conn) + return SimpleNamespace(state=SimpleNamespace(query_engine=engine)) + + +def test_get_store_defaults_to_embedded_and_caches_on_app_state( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv(CONTROL_PLANE_STORE_ENV, raising=False) + app = _stub_app(conn) + + store = get_control_plane_store(app) # type: ignore[arg-type] + + assert isinstance(store, EmbeddedControlPlaneStore) + assert app.state.control_plane_store is store + assert get_control_plane_store(app) is store # type: ignore[arg-type] + + +def test_get_store_postgres_is_a_fail_closed_ratchet_until_slice_5( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: + # ADR 0010: the scale profile must not silently fall back to the embedded + # (split-brain at replicaCount>1) store — it fails the boot instead, and + # this test is deleted only when PostgresControlPlaneStore ships. + monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "postgres") + with pytest.raises(NotImplementedError, match="slice 5"): + get_control_plane_store(_stub_app(conn)) # type: ignore[arg-type] + + +def test_get_store_rejects_unknown_kind( + conn: duckdb.DuckDBPyConnection, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv(CONTROL_PLANE_STORE_ENV, "sqlite") + with pytest.raises(ValueError, match="Unknown control-plane store"): + get_control_plane_store(_stub_app(conn)) # type: ignore[arg-type] + + +def test_get_store_follows_a_swapped_query_engine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The embedded adapter must bind to the *live* engine connection (tests and + # lifespans swap app.state.query_engine), exactly like the pre-port direct + # ``_conn`` lookups did. + monkeypatch.delenv(CONTROL_PLANE_STORE_ENV, raising=False) + first = duckdb.connect(":memory:") + second = duckdb.connect(":memory:") + try: + app = _stub_app(first) + store = get_control_plane_store(app) # type: ignore[arg-type] + assert store._conn is first + app.state.query_engine = SimpleNamespace(_conn=second) + assert store._conn is second + finally: + first.close() + second.close() + + +# --- structural ratchet ---------------------------------------------------------- + + +def test_webhook_path_does_not_reach_into_the_engine_connection() -> None: + """ADR 0010 slice 1 ratchet: the dispatcher and the webhooks router 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", + ): + source = (PROJECT_ROOT / relative).read_text(encoding="utf-8") + assert "query_engine._conn" not in source, relative diff --git a/tests/unit/test_webhook_dispatcher_unit.py b/tests/unit/test_webhook_dispatcher_unit.py index 66e3575e..f884622e 100644 --- a/tests/unit/test_webhook_dispatcher_unit.py +++ b/tests/unit/test_webhook_dispatcher_unit.py @@ -29,19 +29,17 @@ _event_body, _event_type_matches, _json_default, - _log_delivery, _matches_filters, _seen_event_key, _signature, create_webhook, deactivate_webhook, - ensure_webhook_deliveries_table, - get_delivery_logs, get_webhook, list_webhooks, load_webhooks, ) from src.serving.backends.duckdb_backend import DuckDBBackend +from src.serving.control_plane import EmbeddedControlPlaneStore from src.serving.semantic_layer.query import QueryEngine @@ -237,9 +235,8 @@ def test_mark_existing_events_seen_populates_seen_ids( def test_delivery_logs_roundtrip() -> None: conn = duckdb.connect(":memory:") try: - ensure_webhook_deliveries_table(conn) - _log_delivery( - conn, + store = EmbeddedControlPlaneStore(conn_provider=lambda: conn) + store.log_webhook_delivery( delivery_id="d1", webhook_id="wh-1", event_id="e1", @@ -249,12 +246,12 @@ def test_delivery_logs_roundtrip() -> None: success=True, error=None, ) - logs = get_delivery_logs(conn, "wh-1") + logs = store.get_webhook_delivery_logs("wh-1") assert len(logs) == 1 assert logs[0]["webhook_id"] == "wh-1" assert logs[0]["success"] is True # unrelated webhook id sees nothing - assert get_delivery_logs(conn, "wh-other") == [] + assert store.get_webhook_delivery_logs("wh-other") == [] finally: conn.close()