From 4481436aa7513c7e58ab9a6e3a9cdc6ee803b04c Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Sat, 18 Jul 2026 03:42:49 +0300 Subject: [PATCH 1/2] =?UTF-8?q?fix(webhook):=20close=20two=20P3s=20?= =?UTF-8?q?=E2=80=94=20settle-invariant=20detector=20+=20idempotent=20outc?= =?UTF-8?q?ome?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P3 limitations left documented by the post-11.07 review, closed together (they share the webhook delivery path); the P3 DuckDB DST-fold delay is left documented as an accepted narrow limit. P3(a) — the settle watermark's operator invariant (AGENTFLOW_WEBHOOK_SETTLE_SECONDS > writer stamp-to-visibility lag + clock skew) was enforced only implicitly: a violation silently dropped the late-visible row behind the keyset frontier. Add a cheap sampled runtime detector that probes the bounded band immediately behind the frontier (a newest_first window under a new symmetric max_processed_at bound on fetch_pipeline_events, once per interval — no per-pass cost, no journal-wide scan) for rows the scan never marked seen, and raises the new agentflow_webhook_settle_violations_total counter with a webhook_settle_invariant_violation warning. Reads only fetch_pipeline_events and the in-memory seen-set, so it runs on both DuckDB and ClickHouse serving stores; silent under the settle=0 opt-out. P3(b) — record_webhook_delivery_outcome did a non-idempotent read-modify-write of attempts, so the PostgreSQL adapter's transient-error retry could count one failure twice (attempts+2 -> premature dead-letter) when an UPDATE committed but its commit-ack was lost. Thread the delivery round's delivery_id through and record it on the queue row (last_outcome_id); a repeat outcome write no-ops. Embedded and PostgreSQL adapters both stamp/guard on the id; PostgreSQL adds migration 2 for the column. Store-level regression test is red on the pre-fix path, green on the fix; detector tests cover fire / settle=0 silence / no false positive on already-delivered rows. Co-Authored-By: Claude Fable 5 --- src/serving/api/metrics.py | 15 ++ src/serving/api/webhook_dispatcher.py | 158 ++++++++++++++++++++- src/serving/control_plane/embedded.py | 38 +++-- src/serving/control_plane/postgres.py | 42 ++++-- src/serving/control_plane/store.py | 13 +- src/serving/semantic_layer/query/engine.py | 25 ++++ tests/unit/test_control_plane_store.py | 74 ++++++++++ tests/unit/test_webhook_dispatcher_unit.py | 96 +++++++++++++ 8 files changed, 437 insertions(+), 24 deletions(-) diff --git a/src/serving/api/metrics.py b/src/serving/api/metrics.py index ff8b61e..e9a82c6 100644 --- a/src/serving/api/metrics.py +++ b/src/serving/api/metrics.py @@ -41,3 +41,18 @@ "agentflow_usage_rows_dropped_total", "api_usage rows dropped because the off-path writer queue was full.", ) + +# The webhook dispatcher's settle watermark rests on an operator invariant: +# AGENTFLOW_WEBHOOK_SETTLE_SECONDS must exceed writer stamp-to-visibility lag + +# writer<->DB clock skew. A violation is otherwise SILENT — a row that becomes +# visible with a (processed_at, event_id) already behind the strict keyset +# frontier is excluded by every future forward scan and never delivered. This +# counter is incremented by the dispatcher's sampled behind-frontier probe for +# each such never-handed-out row it observes; sustained non-zero means settle is +# set below the writers' true visibility lag and webhook deliveries are being +# dropped (raise AGENTFLOW_WEBHOOK_SETTLE_SECONDS). Flat at 0 is healthy. +WEBHOOK_SETTLE_VIOLATIONS = Counter( + "agentflow_webhook_settle_violations_total", + "Journal rows observed strictly behind the webhook scan frontier yet never " + "handed out — a violated settle invariant (settle < write-visibility lag).", +) diff --git a/src/serving/api/webhook_dispatcher.py b/src/serving/api/webhook_dispatcher.py index 339614e..d4f349b 100644 --- a/src/serving/api/webhook_dispatcher.py +++ b/src/serving/api/webhook_dispatcher.py @@ -6,9 +6,10 @@ import json import os import secrets +import time import uuid from collections.abc import Awaitable, Callable -from datetime import UTC, date, datetime +from datetime import UTC, date, datetime, timedelta from decimal import Decimal from pathlib import Path @@ -19,6 +20,7 @@ from pydantic import BaseModel, Field from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url +from src.serving.api.metrics import WEBHOOK_SETTLE_VIOLATIONS from src.serving.backends import BackendExecutionError from src.serving.control_plane import get_control_plane_store from src.serving.seen_events import BoundedSeenSet @@ -178,6 +180,25 @@ def __init__( self.seen_event_ids: BoundedSeenSet = BoundedSeenSet(maxlen=seen_cache_size) self._scan_cursor: tuple[str, str] | None = None self._task: asyncio.Task | None = None + # Settle-invariant detector (P3 runtime check): the settle watermark + # only works while `settle_seconds` exceeds writer stamp-to-visibility + # lag + clock skew; a violation is otherwise silent (a late-visible row + # behind the frontier is never delivered). Once per interval the + # dispatcher probes a bounded band immediately BEHIND the keyset frontier + # for rows it never marked seen and warns + counts them — no per-pass + # cost, no journal-wide scan, and zero effect on delivery semantics. The + # probe stays out of open seconds by construction (its upper bound is the + # frontier, which is already settled), so it runs identically on DuckDB + # and ClickHouse. Silent when settle is opted out (`settle_seconds == 0`). + self._settle_check_interval_seconds = 30.0 + self._settle_probe_lookback_seconds = max(60.0, float(settle_seconds) * 20.0) + self._settle_probe_limit = 200 + # Seeded at construction (not None) so the probe first runs one interval + # into the dispatcher's life, never on the very first pass — startup + # warmup does not need a behind-frontier scan, and a single hand-driven + # dispatch in a test stays a single journal fetch unless it opts in by + # lowering the interval. + self._last_settle_check_at: float = time.monotonic() # S7: first-class subscribers notified when the journal scan marks new # events seen. Replaces the historical monkey-patch over # ``dispatch_new_events`` in main.py. Cache invalidation is owned by @@ -360,6 +381,13 @@ async def dispatch_new_events(self) -> None: if advance is not None: self._scan_cursor = advance + # Cheap, sampled runtime check that the settle invariant still holds. + # Isolated so a probe failure can never disturb delivery. + try: + self._check_settle_invariant() + except Exception as exc: # pragma: no cover - defensive; probe is read-only + logger.warning("webhook_settle_probe_error", error=str(exc)) + if newly_seen: for listener in self._on_new_events: try: @@ -367,6 +395,101 @@ async def dispatch_new_events(self) -> None: except Exception as exc: logger.warning("webhook_new_events_listener_failed", error=str(exc)) + def _check_settle_invariant(self) -> None: + """Detect a violated settle invariant cheaply, at runtime. + + The operator invariant behind the settle watermark is ``settle_seconds > + writer stamp-to-visibility lag + writer<->DB clock skew``. When it holds, + the strict keyset frontier only crosses seconds no writer will stamp + again. When it is violated, a row becomes visible with a + ``(processed_at, event_id)`` already behind the frontier; every future + forward scan excludes it and it is **never delivered** — a silent drop. + + The forward scan cannot see such a row (that is the whole problem), so + this probe looks the other way: a single bounded ``newest_first`` window + in the band immediately behind the current frontier + (``max_processed_at = frontier``), for rows the dispatcher never marked + seen. Each such row is a concrete never-handed-out delivery and bumps + ``agentflow_webhook_settle_violations_total`` with a warning. + + Cost control: it runs at most once per ``_settle_check_interval_seconds`` + (not every pass), fetches at most ``_settle_probe_limit`` rows within a + bounded lookback band, and never materializes the journal — so it adds no + per-pass cost and cannot regrow the RSS issue #183 fixed. It only reads + ``fetch_pipeline_events`` (the ClickHouse-safe chokepoint) and the + in-memory seen-set, so it works on both the DuckDB and ClickHouse serving + stores. Silent under the ``settle_seconds == 0`` opt-out. + + Residual limits (documented, not silent): membership is tested against + the bounded seen-set, so a genuine drop whose id was already evicted + (only under an extreme burst wider than the seen-set within the lookback + band) could be missed or, if delivered-then-evicted, over-counted; the + probe is read-only either way and never changes what is delivered. A + pathologically late arrival stamped older than the lookback band is also + outside the window — the band is sized for realistic lag near the settle + boundary. + """ + if self.settle_seconds <= 0: + return + cursor = self._scan_cursor + if cursor is None: + return + now = time.monotonic() + if (now - self._last_settle_check_at) < self._settle_check_interval_seconds: + return + self._last_settle_check_at = now + + frontier_ts, frontier_id = cursor + frontier_dt = _parse_cursor_timestamp(frontier_ts) + if frontier_dt is None: + return + band_lower = frontier_dt - timedelta(seconds=self._settle_probe_lookback_seconds) + try: + behind = self._fetch_pipeline_events( + limit=self._settle_probe_limit, + newest_first=True, + min_processed_at=band_lower, + max_processed_at=frontier_ts, + # The band's upper bound is the already-settled frontier, so the + # watermark is redundant here; disable it so the probe cannot be + # confused with the forward scan's settle bound. + settle_seconds=0, + ) + except Exception as exc: + logger.warning("webhook_settle_probe_failed", error=str(exc)) + return + + undelivered: list[dict] = [] + for event in behind: + key = _cursor_key(event) + if key is None: + continue + ev_ts, ev_id = key + ev_dt = _parse_cursor_timestamp(ev_ts) + if ev_dt is None: + continue + # Keep only rows STRICTLY behind the keyset frontier — the exact set + # the forward scan `(t > ts OR (t = ts AND id > id))` will never + # return. Rows at or ahead of the frontier are still deliverable. + if ev_dt > frontier_dt or (ev_dt == frontier_dt and ev_id >= frontier_id): + continue + if _seen_event_key(event) in self.seen_event_ids: + continue + undelivered.append(event) + + if undelivered: + WEBHOOK_SETTLE_VIOLATIONS.inc(len(undelivered)) + sample = undelivered[0] + logger.warning( + "webhook_settle_invariant_violation", + settle_seconds=self.settle_seconds, + frontier_processed_at=frontier_ts, + frontier_event_id=frontier_id, + undelivered_behind_frontier=len(undelivered), + sample_event_id=str(sample.get("event_id") or ""), + sample_processed_at=str(sample.get("processed_at") or ""), + ) + async def deliver(self, webhook: WebhookRegistration, event: dict) -> dict: """Deliver one event now (the ``/test`` endpoint and the inline dispatch path). Computes the canonical body from ``event`` and posts it. @@ -495,8 +618,10 @@ def _fetch_pipeline_events( *, limit: int | None = None, newest_first: bool = False, - min_processed_at: str | None = None, + min_processed_at: str | datetime | None = None, min_event_id: str | None = None, + max_processed_at: str | datetime | None = None, + settle_seconds: int | None = None, ) -> list[dict]: # Scan the journal through the serving backend, not the embedded DuckDB # connection: when serving is ClickHouse (ADR 0006) the events that @@ -511,7 +636,8 @@ def _fetch_pipeline_events( newest_first=newest_first, min_processed_at=min_processed_at, min_event_id=min_event_id, - settle_seconds=self.settle_seconds, + max_processed_at=max_processed_at, + settle_seconds=self.settle_seconds if settle_seconds is None else settle_seconds, ) return events @@ -538,7 +664,12 @@ def _record_delivery_outcome(self, webhook_id: str, event_id: str, result: dict) ``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. The transition itself lives in the - control-plane store; the retry policy stays dispatcher configuration.""" + control-plane store; the retry policy stays dispatcher configuration. + + ``delivery_id`` (the per-round id ``deliver``/``_deliver_body`` mints and + returns in ``result``) is the idempotency token the store uses so a + retry of this write after a lost commit-ack does not count the same + outcome twice — the attempts+2 → premature dead-letter bug (P3).""" if not event_id: return get_control_plane_store(self.app).record_webhook_delivery_outcome( @@ -549,6 +680,7 @@ def _record_delivery_outcome(self, webhook_id: str, event_id: str, result: dict) error=result.get("error"), max_attempts=self.max_delivery_attempts, backoff_seconds=self.backoff_seconds, + delivery_id=result.get("delivery_id"), ) async def process_delivery_queue(self) -> None: @@ -648,6 +780,24 @@ def _cursor_timestamp(event: dict) -> str | None: return None +def _parse_cursor_timestamp(text: str) -> datetime | None: + """Parse a normalized cursor string back into a naive datetime. + + Inverse of :func:`_cursor_timestamp` for the settle-invariant detector's + in-Python keyset comparison — comparing datetimes (not raw strings) avoids + any lexicographic edge between whole-second and sub-second stamps of the + same second. Returns ``None`` on anything that does not round-trip, so a + malformed row is skipped rather than crashing the probe. + """ + candidate = text.strip().replace("T", " ") + for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"): + try: + return datetime.strptime(candidate, fmt) + except ValueError: + continue + return None + + def _cursor_key(event: dict) -> tuple[str, str] | None: """Composite keyset cursor ``(processed_at, event_id)`` for a journal row. diff --git a/src/serving/control_plane/embedded.py b/src/serving/control_plane/embedded.py index d378c5b..86a6658 100644 --- a/src/serving/control_plane/embedded.py +++ b/src/serving/control_plane/embedded.py @@ -153,12 +153,21 @@ def ensure_webhook_delivery_queue_table(conn: duckdb.DuckDBPyConnection) -> None next_attempt_at TIMESTAMP, last_status_code INTEGER, last_error VARCHAR, + last_outcome_id VARCHAR, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (webhook_id, event_id) ) """ ) + # Idempotency token for the outcome write (P3): the delivery_id of the + # last outcome applied to this row, so a retry after a lost commit-ack + # is a no-op instead of a second attempts bump. ADD COLUMN IF NOT EXISTS + # upgrades a queue table created before this column existed (same + # pattern as ensure_dead_letter_table's tenant_id backfill). + conn.execute( + "ALTER TABLE webhook_delivery_queue ADD COLUMN IF NOT EXISTS last_outcome_id VARCHAR" + ) def ensure_alert_history_table(conn: duckdb.DuckDBPyConnection) -> None: @@ -471,39 +480,50 @@ def record_webhook_delivery_outcome( error: str | None, max_attempts: int, backoff_seconds: Sequence[float], + delivery_id: str | None = None, ) -> None: conn = self._conn now = datetime.now(UTC) + # Read the current attempts AND the last outcome id in one shot so the + # idempotency guard and the failure-branch increment share one snapshot. + row = conn.execute( + "SELECT attempts, last_outcome_id FROM webhook_delivery_queue " + "WHERE webhook_id = ? AND event_id = ?", + [webhook_id, event_id], + ).fetchone() + # Idempotency (P3): this exact delivery round's outcome already landed — + # a retry after a lost commit-ack. No-op so attempts is not bumped twice. + if delivery_id is not None and row is not None and row[1] == delivery_id: + return if success: conn.execute( "UPDATE webhook_delivery_queue SET status = 'delivered', " - "last_status_code = ?, last_error = NULL, updated_at = ? " + "last_status_code = ?, last_error = NULL, last_outcome_id = ?, updated_at = ? " "WHERE webhook_id = ? AND event_id = ?", - [status_code, now, webhook_id, event_id], + [status_code, delivery_id, 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 = ? " + "last_status_code = ?, last_error = ?, last_outcome_id = ?, " + "next_attempt_at = NULL, updated_at = ? " "WHERE webhook_id = ? AND event_id = ?", - [attempts, status_code, error, now, webhook_id, event_id], + [attempts, status_code, error, delivery_id, 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 = ? " + "last_status_code = ?, last_error = ?, last_outcome_id = ?, " + "next_attempt_at = ?, updated_at = ? " "WHERE webhook_id = ? AND event_id = ?", [ attempts, status_code, error, + delivery_id, now + timedelta(seconds=delay), now, webhook_id, diff --git a/src/serving/control_plane/postgres.py b/src/serving/control_plane/postgres.py index 861a8d4..c6b17eb 100644 --- a/src/serving/control_plane/postgres.py +++ b/src/serving/control_plane/postgres.py @@ -298,6 +298,11 @@ _MIGRATIONS: tuple[tuple[int, str, tuple[str, ...]], ...] = ( (1, "baseline: six control-plane state classes (ADR 0010 slice 5)", _SCHEMA_STATEMENTS), + ( + 2, + "webhook_delivery_queue.last_outcome_id — idempotent outcome write (P3)", + ("ALTER TABLE webhook_delivery_queue ADD COLUMN IF NOT EXISTS last_outcome_id TEXT",), + ), ) if tuple(version for version, _, _ in _MIGRATIONS) != tuple(range(1, len(_MIGRATIONS) + 1)): @@ -593,43 +598,59 @@ def record_webhook_delivery_outcome( error: str | None, max_attempts: int, backoff_seconds: Sequence[float], + delivery_id: str | None = None, ) -> None: # Bounded retry on transient connection errors, same shape as # record_api_usage: without it, a POST that succeeded but then hit a # momentary DB blip on this outcome write never clears the enqueue # lease, stranding the row pending+leased for the full claim lease # window instead of a fast redrive (audit finding #4). + # + # That retry is exactly what could count one failure twice (P3): attempt + # 0's UPDATE commits on the server but the commit-ack is lost, so the + # except-branch retries and attempt 1 re-reads the already-bumped + # attempts and bumps it again — attempts+2, a premature dead-letter. + # delivery_id makes the round idempotent: the row records the last + # applied outcome id under FOR UPDATE, and a repeat is a no-op. Because + # the guard and the increment read the same locked row inside one + # transaction, the retry sees attempt 0's committed stamp (skip) or its + # rollback (apply once) — never a double bump. last_error: Exception | None = None for attempt in range(3): try: with self._connect() as conn: + row = conn.execute( + "SELECT attempts, last_outcome_id FROM webhook_delivery_queue " + "WHERE webhook_id = %s AND event_id = %s FOR UPDATE", + (webhook_id, event_id), + ).fetchone() + if delivery_id is not None and row is not None and row[1] == delivery_id: + # This delivery round's outcome already landed (a retry + # after a lost commit-ack). No-op. + return if success: conn.execute( """ UPDATE webhook_delivery_queue SET status = 'delivered', last_status_code = %s, - last_error = NULL, lease_expires_at = NULL, updated_at = now() + last_error = NULL, last_outcome_id = %s, + lease_expires_at = NULL, updated_at = now() WHERE webhook_id = %s AND event_id = %s """, - (status_code, webhook_id, event_id), + (status_code, delivery_id, webhook_id, event_id), ) return - row = conn.execute( - "SELECT attempts FROM webhook_delivery_queue " - "WHERE webhook_id = %s AND event_id = %s FOR UPDATE", - (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 = %s, last_status_code = %s, - last_error = %s, next_attempt_at = NULL, + last_error = %s, last_outcome_id = %s, next_attempt_at = NULL, lease_expires_at = NULL, updated_at = now() WHERE webhook_id = %s AND event_id = %s """, - (attempts, status_code, error, webhook_id, event_id), + (attempts, status_code, error, delivery_id, webhook_id, event_id), ) return delay = backoff_seconds[min(attempts - 1, len(backoff_seconds) - 1)] @@ -637,7 +658,7 @@ def record_webhook_delivery_outcome( """ UPDATE webhook_delivery_queue SET status = 'pending', attempts = %s, last_status_code = %s, - last_error = %s, + last_error = %s, last_outcome_id = %s, next_attempt_at = now() + make_interval(secs => %s), lease_expires_at = NULL, updated_at = now() WHERE webhook_id = %s AND event_id = %s @@ -646,6 +667,7 @@ def record_webhook_delivery_outcome( attempts, status_code, error, + delivery_id, delay, webhook_id, event_id, diff --git a/src/serving/control_plane/store.py b/src/serving/control_plane/store.py index c205857..57bc4f0 100644 --- a/src/serving/control_plane/store.py +++ b/src/serving/control_plane/store.py @@ -235,10 +235,21 @@ def record_webhook_delivery_outcome( error: str | None, max_attempts: int, backoff_seconds: Sequence[float], + delivery_id: str | None = None, ) -> 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.""" + park as ``dead`` once ``max_attempts`` is reached. + + ``delivery_id`` makes the write **idempotent per delivery round**. The + failure branch is a read-modify-write of ``attempts``; a store that + retries this call after the DB committed but the commit-ack was lost + (the PostgreSQL adapter's transient-error retry) would otherwise re-read + the already-bumped ``attempts`` and bump it again — attempts+2, a + premature dead-letter (P3). Adapters persist the last applied + ``delivery_id`` on the queue row and no-op when it repeats, so the same + round's outcome lands exactly once. ``None`` (a caller with no round id) + keeps the pre-idempotency behaviour — every call applies.""" @abstractmethod def park_webhook_delivery(self, *, webhook_id: str, event_id: str, error: str) -> None: diff --git a/src/serving/semantic_layer/query/engine.py b/src/serving/semantic_layer/query/engine.py index 155c282..fccce13 100644 --- a/src/serving/semantic_layer/query/engine.py +++ b/src/serving/semantic_layer/query/engine.py @@ -180,6 +180,7 @@ def fetch_pipeline_events( newest_first: bool = False, min_processed_at: datetime | str | None = None, min_event_id: str | None = None, + max_processed_at: datetime | str | None = None, settle_seconds: int | None = None, ) -> list[dict]: """Read the ``pipeline_events`` journal through the serving backend. @@ -232,6 +233,15 @@ def fetch_pipeline_events( both backends. ``min_event_id`` alone (no ``min_processed_at``) is ignored — a keyset needs both halves. + ``max_processed_at`` is the symmetric **inclusive upper bound** + (``processed_at <= cursor``, sub-second precision preserved). It is the + seam the webhook dispatcher's settle-invariant detector reads through: a + bounded ``newest_first`` window under this bound scans only the band + immediately *behind* the keyset frontier, never the whole journal, so + the detector's probe stays O(window). Orthogonal to ``min_processed_at`` + (pair them for a bounded band); ignored when the journal has no time + column. Same strict-parse contract as the other cursors. + ``settle_seconds`` is the scan's settle watermark (audit 2026-07-17 #1 follow-up): when set, only rows with ``processed_at`` at least that many seconds in the past are visible — ``processed_at <= @@ -385,6 +395,21 @@ def render(value: str) -> str: cursor = _coerce_journal_timestamp(min_processed_at) rendered = render(cursor.strftime("%Y-%m-%d %H:%M:%S")) where_clauses.append(f"{time_column} >= CAST({rendered} AS TIMESTAMP)") + if max_processed_at is not None and time_column is not None: + # Inclusive upper bound, mirror of the min_processed_at cursor. + # Sub-second precision is preserved (floor_seconds=False) so the + # detector's band ends exactly at the keyset frontier rather than + # one second past it; on ClickHouse processed_at is second-granular + # so the literal is a whole second either way. Same CAST/render + # discipline as every other bound (allowlisted, transpiles to both + # dialects). + upper = _coerce_journal_timestamp(max_processed_at, floor_seconds=False) + if upper.microsecond: + upper_text = upper.strftime("%Y-%m-%d %H:%M:%S.%f") + else: + upper_text = upper.strftime("%Y-%m-%d %H:%M:%S") + rendered_max = render(upper_text) + where_clauses.append(f"{time_column} <= CAST({rendered_max} AS TIMESTAMP)") if settle_seconds is not None and int(settle_seconds) > 0 and time_column is not None: # DB-clock watermark (see docstring): the keyset frontier must not # enter a second writers can still stamp. 0 is a true opt-out (no diff --git a/tests/unit/test_control_plane_store.py b/tests/unit/test_control_plane_store.py index 87f45c6..5ce2a24 100644 --- a/tests/unit/test_control_plane_store.py +++ b/tests/unit/test_control_plane_store.py @@ -184,6 +184,80 @@ def test_outcome_failure_backs_off_then_parks_dead_at_max( assert next_at is None # parked for good +# --- P3(b): idempotent outcome recording (no attempts+2 on a lost-ack retry) --- + + +def test_outcome_is_idempotent_per_delivery_id_no_double_count( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + """A retry of the SAME delivery round's outcome must not count twice. + + Models the P3 failure: the outcome write commits on the DB but the + commit-ack is lost, so the caller (the postgres adapter's transient-error + retry) re-applies the identical outcome — same ``delivery_id``. Without the + idempotency guard the second application re-reads the already-bumped + ``attempts`` and bumps it again (attempts+2), prematurely parking the row + ``dead`` one real failure early. With the guard the repeat is a no-op. + + Proving property: on the pre-fix store (guard removed) the second call bumps + ``attempts`` to 2 and, at ``max_attempts=2``, flips ``status`` to ``dead`` — + this test then fails on both asserts. + """ + _enqueue(store, "wh-1", "e1") + + outcome = { + "webhook_id": "wh-1", + "event_id": "e1", + "success": False, + "status_code": 500, + "error": "boom", + "max_attempts": 2, # a double count here would immediately dead-letter + "backoff_seconds": [10.0], + "delivery_id": "round-1", + } + store.record_webhook_delivery_outcome(**outcome) + status, attempts, next_at, _ = _row(conn, "wh-1", "e1") + assert (status, attempts) == ("pending", 1) # one real failure counted once + assert next_at is not None # still scheduled for redrive, NOT dead + + # The lost-ack retry: identical outcome, identical delivery_id. + store.record_webhook_delivery_outcome(**outcome) + status, attempts, next_at, _ = _row(conn, "wh-1", "e1") + assert (status, attempts) == ("pending", 1) # idempotent — still 1, not 2 + assert next_at is not None # not prematurely dead-lettered + + +def test_outcome_distinct_delivery_ids_still_count_each_round( + store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection +) -> None: + """Two genuinely different delivery rounds (distinct ids) still each count — + the guard suppresses only a repeat of the same round, never a real redrive.""" + _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=5, + backoff_seconds=[10.0], + delivery_id="round-1", + ) + store.record_webhook_delivery_outcome( + webhook_id="wh-1", + event_id="e1", + success=False, + status_code=500, + error="boom", + max_attempts=5, + backoff_seconds=[10.0], + delivery_id="round-2", + ) + status, attempts, _next_at, _ = _row(conn, "wh-1", "e1") + assert (status, attempts) == ("pending", 2) # two distinct rounds, two counts + + def test_park_marks_dead_with_reason( store: EmbeddedControlPlaneStore, conn: duckdb.DuckDBPyConnection ) -> None: diff --git a/tests/unit/test_webhook_dispatcher_unit.py b/tests/unit/test_webhook_dispatcher_unit.py index 4508151..39ab3b9 100644 --- a/tests/unit/test_webhook_dispatcher_unit.py +++ b/tests/unit/test_webhook_dispatcher_unit.py @@ -22,6 +22,7 @@ import duckdb import httpx import pytest +from prometheus_client import REGISTRY from src.serving.api.webhook_dispatcher import ( WebhookDispatcher, @@ -532,6 +533,101 @@ def test_mark_existing_does_not_seed_unsettled_rows(config_path: Path) -> None: conn.close() +# --- P3(a): runtime settle-invariant detector -------------------------------- + +_SETTLE_METRIC = "agentflow_webhook_settle_violations_total" + + +def _settle_violations() -> float: + return REGISTRY.get_sample_value(_SETTLE_METRIC) or 0.0 + + +@pytest.mark.asyncio +async def test_settle_detector_flags_undelivered_row_behind_frontier() -> None: + # The silent failure the detector exists to surface: the settle invariant is + # violated, so a row becomes visible with a processed_at already behind the + # settled keyset frontier. The forward scan's strict keyset excludes it + # forever — it is never delivered. The behind-frontier probe catches it and + # bumps agentflow_webhook_settle_violations_total. + conn = _journal_conn([("e-front", "acme", "order.created", "2026-07-10 10:00:10")]) + try: + dispatcher = WebhookDispatcher(_stub_app(conn), settle_seconds=3) + dispatcher._settle_check_interval_seconds = 0.0 # probe every pass in the test + + # Pass 1: e-front is long settled, so the frontier advances onto it and + # it is marked seen. Nothing is behind the frontier yet. + await dispatcher.dispatch_new_events() + assert dispatcher._scan_cursor == ("2026-07-10 10:00:10", "e-front") + base = _settle_violations() + + # A late-visible row lands BEHIND the settled frontier (lower second). + conn.execute( + "INSERT INTO pipeline_events VALUES " + "('e-late', 'orders.raw', 'acme', 'order.created', '2026-07-10 10:00:05')" + ) + await dispatcher.dispatch_new_events() + + # The forward scan never handed it out, and the probe flagged exactly it. + assert "acme:e-late" not in dispatcher.seen_event_ids + assert _settle_violations() - base == 1.0 + finally: + conn.close() + + +@pytest.mark.asyncio +async def test_settle_detector_silent_when_settle_opted_out() -> None: + # settle_seconds == 0 is a deliberate opt-out (tests, or an operator who + # accepts the drop risk); the detector must stay completely silent even when + # a row is demonstrably behind the frontier. + conn = _journal_conn([("e-front", "acme", "order.created", "2026-07-10 10:00:10")]) + try: + dispatcher = WebhookDispatcher(_stub_app(conn), settle_seconds=0) + dispatcher._settle_check_interval_seconds = 0.0 + + await dispatcher.dispatch_new_events() + assert dispatcher._scan_cursor == ("2026-07-10 10:00:10", "e-front") + base = _settle_violations() + + conn.execute( + "INSERT INTO pipeline_events VALUES " + "('e-late', 'orders.raw', 'acme', 'order.created', '2026-07-10 10:00:05')" + ) + await dispatcher.dispatch_new_events() + + assert _settle_violations() == base # no warning, no counter movement + finally: + conn.close() + + +@pytest.mark.asyncio +async def test_settle_detector_does_not_flag_already_delivered_rows() -> None: + # No false positives: a row behind the frontier that the scan DID hand out + # (it is in the seen-set) must not be counted. This is the common case — + # every settled row sits behind the frontier once the cursor moves past it. + conn = _journal_conn( + [ + ("e-early", "acme", "order.created", "2026-07-10 10:00:05"), + ("e-front", "acme", "order.created", "2026-07-10 10:00:10"), + ] + ) + try: + dispatcher = WebhookDispatcher(_stub_app(conn), settle_seconds=3) + dispatcher._settle_check_interval_seconds = 0.0 + base = _settle_violations() + + # Both settle and are handed out; the frontier ends at e-front with + # e-early sitting (delivered) behind it. + await dispatcher.dispatch_new_events() + assert dispatcher._scan_cursor == ("2026-07-10 10:00:10", "e-front") + assert "acme:e-early" in dispatcher.seen_event_ids + + # A second pass re-runs the probe over the same behind-frontier band. + await dispatcher.dispatch_new_events() + assert _settle_violations() == base # delivered-behind-frontier is not a violation + finally: + conn.close() + + @pytest.mark.asyncio async def test_dispatch_cursor_freezes_on_enqueue_failure_then_retries( config_path: Path, monkeypatch: pytest.MonkeyPatch From e1857e0ea9edae91ecf9e97a167d8e8b0b9d8aea Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Sat, 18 Jul 2026 03:43:02 +0300 Subject: [PATCH 2/2] docs(webhook): record the P3 settle detector and idempotent outcome fixes serving-bridge.md: the settle-invariant violation is no longer silent (runtime detector + agentflow_webhook_settle_violations_total), and delivery outcomes are idempotent per round (no attempts+2). The DuckDB autumn DST-fold delivery-delay paragraph is left unchanged (still an accepted limit). CHANGELOG: unreleased P3 follow-ups entry. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 32 ++++++++++++++++++++++++++++++++ docs/serving-bridge.md | 25 +++++++++++++++++++++---- 2 files changed, 53 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 643fffb..37fc147 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,38 @@ All notable changes to AgentFlow are documented in this file. ## [Unreleased] +### Fixes — P3 follow-ups from the post-11.07 review (webhook) + +Two of the three P3 limitations the post-11.07 review left documented are now +closed; the third (DuckDB autumn DST-fold delivery *delay* — never a drop) stays +documented as an accepted narrow limit. Locally verified (ruff check + ruff +format + targeted webhook/control-plane unit suites); the PostgreSQL adapter's +symmetric change is verified by inspection against the shared contract (no +live-cluster run on this host). + +- **Settle-invariant violation is no longer silent (P3a).** The webhook scan's + settle watermark rests on an operator invariant + (`AGENTFLOW_WEBHOOK_SETTLE_SECONDS` must exceed writer stamp-to-visibility lag + + writer↔DB clock skew); a violation used to drop the late-visible row behind the + keyset frontier with no signal at all. A cheap **sampled runtime detector** now + probes the bounded band immediately behind the frontier (a `newest_first` + window under a new `max_processed_at` bound, once per interval — no per-pass + cost, no journal-wide scan) for rows the scan never marked seen, and raises + `agentflow_webhook_settle_violations_total` with a + `webhook_settle_invariant_violation` warning. It reads only + `fetch_pipeline_events` + the in-memory seen-set, so it runs on both the DuckDB + and ClickHouse serving stores; silent under the `0` opt-out. +- **Idempotent webhook outcome recording (P3b).** + `record_webhook_delivery_outcome` now carries the delivery round's + `delivery_id` and records it on the queue row (`last_outcome_id`), so a repeat + of the outcome write no-ops instead of bumping `attempts` a second time. This + closes attempts+2 → premature dead-letter, where the PostgreSQL adapter's + transient-error retry re-applied a failure whose UPDATE had committed but whose + commit-ack was lost. Embedded (DuckDB) and PostgreSQL adapters both stamp and + guard on the id; PostgreSQL migration 2 adds the column. A store-level + regression test is red on the pre-fix (non-idempotent) path and green on the + fix. + ### Fixes — post-11.07 adversarial audit (2 latent P1, 4 P2, 1 low) Four-surface adversarial review of everything merged since the 11.07 audit diff --git a/docs/serving-bridge.md b/docs/serving-bridge.md index 99f605a..1244b46 100644 --- a/docs/serving-bridge.md +++ b/docs/serving-bridge.md @@ -196,16 +196,33 @@ now bounded: (`processed_at <= now() - INTERVAL N SECOND`, `AGENTFLOW_WEBHOOK_SETTLE_SECONDS`, default 3 s; `0` disables the bound entirely — tests only; must exceed writer stamp-to-visibility lag + - writer↔DB clock skew, and a violation is silent (a late-visible row behind - the frontier is simply never delivered), so treat the invariant as an - operating requirement, not a tunable. On a non-UTC DuckDB host the + writer↔DB clock skew — treat the invariant as an operating requirement, not + a tunable, because a violation still drops the late-visible row behind the + frontier (it is simply never delivered). That drop is no longer *silent*: a + cheap sampled runtime detector probes the band immediately behind the frontier + (`max_processed_at = frontier`, bounded `newest_first` window, once per + interval — no per-pass cost, no journal-wide scan) for rows the scan never + marked seen, and raises `agentflow_webhook_settle_violations_total` with a + `webhook_settle_invariant_violation` warning naming the frontier and a sample + row. It reads only `fetch_pipeline_events` and the in-memory seen-set, so it + runs on both the DuckDB and ClickHouse serving stores; it is silent under the + `0` opt-out. (Residual, documented: membership is tested against the bounded + seen-set, so an id evicted under an extreme burst within the lookback band, or + an arrival stamped older than that band, can be missed — the probe is + read-only and never changes what is delivered.) On a non-UTC DuckDB host the session-local frame is non-monotonic across the autumn DST fold: delivery of rows stamped inside the fold window is delayed (never dropped) by up to the fold width. Worst-case added delivery latency = settle. Startup (`mark_existing_events_seen`) seeds the cursor from the newest **settled** batch instead of enumerating the journal — unsettled rows deliver once settled (a restart race is not lost; the durable enqueue's idempotent key suppresses re-POSTs); the seen-set is now a - secondary safety net, with the keyset as the primary dedup. + secondary safety net, with the keyset as the primary dedup. Delivery + *outcomes* are idempotent per round too: `record_webhook_delivery_outcome` + carries that round's `delivery_id` and records it on the queue row + (`last_outcome_id`), so a repeat of the outcome write no-ops instead of + bumping `attempts` a second time. This closes attempts+2 → premature + dead-letter, where the PostgreSQL adapter's transient-error retry re-applied a + failure whose UPDATE had committed but whose commit-ack was lost. - **Seen-sets** — `BoundedSeenSet` (`src/serving/seen_events.py`): capped, FIFO-with-refresh eviction. Eviction is safe because webhook enqueue is idempotent on its primary key (inline delivery fires only for freshly