diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 7d05a50e..7bf83c4b 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -46,6 +46,16 @@ updates: # have explicit floors in pyproject.toml driven by CVE fixes # (see commit e8b1237 and the A06 dependency-profiles block). # Allow minor + patch within the >= floor, never major. + # + # NOTE (audit 2026-07-17 #8): a Dependabot `ignore` applies to BOTH + # version updates AND security updates — there is no per-ignore + # `applies-to` (that key exists only on `groups`). So a CVE fixed ONLY in + # a major bump of these deps gets no automatic security-update PR. This is + # an ACCEPTED tradeoff: the `safety` + `pip-audit` jobs in security.yml + # scan these deps and FAIL on such a CVE, surfacing it for a deliberate + # manual major bump (the whole reason their majors are gated). Do NOT + # "fix" by dropping the ignore — that re-introduces the unwanted + # major-version PR churn these floors exist to prevent. - dependency-name: "langchain-core" update-types: ["version-update:semver-major"] - dependency-name: "langchain-text-splitters" diff --git a/CHANGELOG.md b/CHANGELOG.md index 06c6545f..643fffb8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,54 @@ All notable changes to AgentFlow are documented in this file. ## [Unreleased] +### 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 +(control-plane, serving/webhook, helm/k8s, CI). All findings fixed and locally +verified (ruff/lint/targeted unit tests + helm render); no live-cluster run. +Full report: `audit_2026-07-17_post11.md`. + +- **Webhook journal scan — composite keyset cursor (P1, was a silent wedge).** + The scan cursor is now `(processed_at, event_id)` and fetches strictly after + it, so a single second holding ≥ `scan_batch_size` rows can no longer re-pin a + second-granular cursor and silently drop every later webhook. Portable + OR-decomposition predicate (row-value tuples don't transpile to ClickHouse); a + regression test seeds a larger-than-one-batch second and is red on the old + code, green on the fix. Follow-up (adversarial review): a strict keyset alone + would silently drop same-second rows that become visible after the frontier + passed their second (UUID event ids are not monotonic; ClickHouse + `processed_at` is second-granular), so every dispatcher fetch is bounded by a + DB-clock **settle watermark** — `processed_at <= now() - INTERVAL + AGENTFLOW_WEBHOOK_SETTLE_SECONDS` (default 3 s) — and the frontier crosses + only closed seconds. The settle value must exceed writer stamp-to-visibility + lag plus writer↔DB clock skew. +- **NetworkPolicy — PostgreSQL egress (P1, was a hardened-prod outage).** A + `:5432` egress rule now renders when `controlPlane.store=postgres`; without it, + `networkPolicy.enabled=true` on the postgres profile denied every pod's PG + connect and the control-plane pool never opened. +- **Control-plane retry backoff — DB clock (P2).** `next_attempt_at` is computed + in SQL (`now() + make_interval`) instead of the app pod's clock, mirroring the + lease columns, so NTP skew no longer collapses or over-delays the + webhook/outbox backoff. +- **Replica-correctness verify — settle window (P2).** E4 Checks 3/4 dwell a + settle window and assert the count never exceeds one, instead of asserting on + the first sighting — which passed a delayed split delivery/page undetected. +- **PodDisruptionBudget — api-scoped (P2).** The PDB selects `component: api`, so + a drain cannot evict both API pods while a lone worker satisfies the budget. +- **API Deployment selector — stable component label (P2).** + `app.kubernetes.io/component: api` is now unconditional, so the immutable + `spec.selector` no longer changes when `worker.enabled` is toggled. One-time + recreate for pre-split installs is documented in + `docs/migration/helm-component-selector.md` (no pre-upgrade hook — it would + fire on every upgrade). +- **Webhook outcome write — transient retry (P2).** + `record_webhook_delivery_outcome` retries transient DB errors like + `record_api_usage`, so a blip after a successful POST no longer strands the + row+lease for the full claim window. +- **Dependabot `ignore` — documented security-update scope (low).** Commented + that an `ignore` also suppresses security-update PRs (there is no per-ignore + `applies-to`); an accepted tradeoff, still caught by the safety/pip-audit jobs. + ### Perf — paced 1 h @ 100 eps produce (apply 99.5) - Mac compose (1 Flink TM), fresh volumes: 360 000 events at `--pace-eps 100` diff --git a/docs/migration/helm-component-selector.md b/docs/migration/helm-component-selector.md new file mode 100644 index 00000000..83f6467a --- /dev/null +++ b/docs/migration/helm-component-selector.md @@ -0,0 +1,86 @@ +# Helm migration: API Deployment selector now carries `component: api` unconditionally + +## Applies to + +Operators upgrading an **existing** AgentFlow Helm release that was installed +with a chart **before** this one, where `worker.enabled=false` (the default). +Fresh installs and releases already running `worker.enabled=true` are **not** +affected. + +## What changed and why + +The API Deployment (`templates/deployment.yaml`) now adds +`app.kubernetes.io/component: api` to its pod template labels **and** its +`spec.selector.matchLabels` **unconditionally**. Previously that label was added +only when `worker.enabled=true`. + +This fixes two audit findings: + +- **#7 — immutable selector toggle-break.** `spec.selector` is immutable in + Kubernetes. When the label was gated on `worker.enabled`, flipping that value + on a live release changed the selector, so `helm upgrade` aborted with + `spec.selector is immutable`. Making the label unconditional keeps the selector + **constant** across the `worker.enabled` toggle, so the split can be enabled or + disabled in place. +- **#6 — component-scoped PodDisruptionBudget.** The PDB now selects + `component: api`, guaranteeing at least one *API* pod survives a voluntary + disruption. That guarantee only holds if API pods **always** carry the label, + which is why the pod-template label had to become unconditional too. + +It also keeps API and worker pods in disjoint selector sets, so the two +Deployments can never adopt each other's pods. + +## The one-time breaking upgrade + +Because `spec.selector` is immutable, the **first** upgrade of a pre-existing +`worker.enabled=false` release onto this chart changes the selector from +`{name, instance}` to `{name, instance, component=api}`. The API server rejects +the in-place change: + +``` +Deployment.apps "-agentflow" is invalid: spec.selector: Invalid value: +v1.LabelSelector{...}: field is immutable +``` + +`helm upgrade` fails and rolls back; nothing is left half-applied. Recreate the +API Deployment **once** to move past it. This is a one-time step — subsequent +upgrades are normal. + +### Option A — zero-downtime (recommended for multi-replica / production) + +Orphan the existing pods so they keep serving, let the recreated Deployment +stand up new pods, then reap the orphans: + +```bash +kubectl delete deployment -agentflow -n --cascade=orphan +helm upgrade [same flags as before] +# wait for the new (component=api) pods to become Ready, then remove the +# orphaned old pods — they lack component=api so the new Deployment won't adopt +# them and they will not be cleaned up automatically: +kubectl get pods -n -l app.kubernetes.io/name=agentflow \ + -L app.kubernetes.io/component +kubectl delete pod -n +``` + +### Option B — simplest (brief serving gap) + +For a single-replica default install where a short gap is acceptable: + +```bash +kubectl delete deployment -agentflow -n +helm upgrade [same flags as before] +``` + +## Notes + +- The chart does **not** ship a `pre-upgrade` hook to automate the delete. A hook + that deletes the primary API Deployment would fire on **every** upgrade (helm + hooks are not conditional on "did the selector change"), turning a one-time + migration into a recurring outage. The manual step above runs exactly once, by + operator decision. +- The Service selector is unaffected (Service selectors are mutable): it already + narrows to `component: api` when `worker.enabled=true` and otherwise selects on + `{name, instance}`, which still matches the API pods. +- Related NetworkPolicy change in the same release: a PostgreSQL `:5432` egress + rule is now rendered when `controlPlane.store=postgres`. No migration action — + it only adds an allowed egress. diff --git a/docs/serving-bridge.md b/docs/serving-bridge.md index 01a2773c..99f605ae 100644 --- a/docs/serving-bridge.md +++ b/docs/serving-bridge.md @@ -177,16 +177,35 @@ scans and pushes kept one entry per event forever. All journal consumers are now bounded: - **Webhook dispatcher** — incremental cursor scan: each pass fetches at most - `scan_batch_size` (1000) rows at/after the `processed_at` high-water mark - (`min_processed_at`, inclusive — the journal is second-granular on - ClickHouse, and the seen-set dedups the re-fetched cursor second). The - cursor advances over the contiguous prefix of rows that end the pass seen - and **freezes at the first row whose durable enqueue failed**, so that row - is re-fetched and retried next pass — the retry-forever semantics the full - scan provided. An all-seen full batch still advances the cursor, so a seen - frontier wider than one batch cannot pin the window. Startup - (`mark_existing_events_seen`) seeds the cursor from the newest batch instead - of enumerating the journal. + `scan_batch_size` (1000) rows **strictly after** a composite + `(processed_at, event_id)` keyset cursor, in `processed_at, event_id ASC` + order. The cursor advances over the contiguous prefix of rows that end the + pass seen and **freezes at the first row whose durable enqueue failed**, so + that row is re-fetched and retried next pass — the retry-forever semantics the + full scan provided. The keyset (not a bare second-granular high-water mark) is + what lets the frontier advance **within** a single second: an inclusive + `processed_at >=` cursor floored to the second was re-pinned forever by any + second holding ≥ `scan_batch_size` rows, silently dropping every later webhook + for every tenant (audit 2026-07-17 #1). The predicate is the portable + OR-decomposition `t > ts OR (t = ts AND event_id > id)` — a row-value tuple + does not transpile to ClickHouse. A strict keyset is only lossless over + seconds no writer will stamp again — event ids are UUIDs (not monotonic), so + a frontier inside the still-open second would permanently exclude any + same-second row that becomes visible later with a lower id. Every dispatcher + fetch is therefore bounded by a DB-clock **settle watermark** + (`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 + 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. - **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 diff --git a/helm/agentflow/templates/NOTES.txt b/helm/agentflow/templates/NOTES.txt new file mode 100644 index 00000000..ceb197a6 --- /dev/null +++ b/helm/agentflow/templates/NOTES.txt @@ -0,0 +1,45 @@ +AgentFlow {{ .Chart.AppVersion }} has been deployed as release "{{ .Release.Name }}". + +Service (in-cluster): + {{ include "agentflow.fullname" . }}.{{ .Release.Namespace }}.svc:{{ .Values.service.port }} + +Control-plane store: {{ .Values.controlPlane.store }} +{{- if .Values.worker.enabled }} +Process-role split: ENABLED (API Deployment role=api, worker Deployment role=worker). +{{- end }} +{{- if .Values.networkPolicy.enabled }} +NetworkPolicy: ENABLED (default-deny + explicit egress{{ if eq .Values.controlPlane.store "postgres" }}, including PostgreSQL:{{ .Values.networkPolicy.egressPorts.postgres }}{{ end }}). +{{- end }} + +{{- if .Release.IsUpgrade }} + +>>> UPGRADE NOTE — one-time API Deployment recreate may be required <<< + +This release makes the API Deployment's spec.selector carry +`app.kubernetes.io/component: api` UNCONDITIONALLY. spec.selector is IMMUTABLE. +If the release you just upgraded FROM was installed with an earlier chart where +`worker.enabled=false` (selector was name+instance only), the API server will +reject this upgrade with: + + Deployment.apps "{{ include "agentflow.fullname" . }}" is invalid: + spec.selector: Invalid value: ... field is immutable + +If that happened, the upgrade did NOT apply. Recreate the API Deployment once — +pick ONE: + + # A) zero-downtime: orphan the old pods, let the new Deployment take over, + # then reap the orphans after the new pods are Ready. + kubectl delete deployment {{ include "agentflow.fullname" . }} \ + -n {{ .Release.Namespace }} --cascade=orphan + helm upgrade {{ .Release.Name }} [same flags] + # ...verify new pods Ready, then delete leftover old pods by hand. + + # B) simplest (brief serving gap on a single-replica install): + kubectl delete deployment {{ include "agentflow.fullname" . }} \ + -n {{ .Release.Namespace }} + helm upgrade {{ .Release.Name }} [same flags] + +Releases already running the split (worker.enabled=true) are unaffected — their +selector already carried component=api. Fresh installs are unaffected. +See docs/migration/helm-component-selector.md for detail. +{{- end }} diff --git a/helm/agentflow/templates/deployment.yaml b/helm/agentflow/templates/deployment.yaml index 247de310..5589c5bd 100644 --- a/helm/agentflow/templates/deployment.yaml +++ b/helm/agentflow/templates/deployment.yaml @@ -15,9 +15,12 @@ metadata: name: {{ include "agentflow.fullname" . }} labels: {{- include "agentflow.labels" . | nindent 4 }} - {{- if .Values.worker.enabled }} + # component=api is UNCONDITIONAL (not gated on worker.enabled): it keeps this + # Deployment's immutable spec.selector stable across a worker.enabled toggle + # (helm upgrade would otherwise fail "spec.selector is immutable"), keeps API + # vs worker pods disjoint so the two Deployments never adopt each other's + # pods, and lets the api-scoped PodDisruptionBudget always match these pods. app.kubernetes.io/component: api - {{- end }} spec: replicas: {{ .Values.replicaCount }} strategy: @@ -28,16 +31,12 @@ spec: selector: matchLabels: {{- include "agentflow.selectorLabels" . | nindent 6 }} - {{- if .Values.worker.enabled }} app.kubernetes.io/component: api - {{- end }} template: metadata: labels: {{- include "agentflow.selectorLabels" . | nindent 8 }} - {{- if .Values.worker.enabled }} app.kubernetes.io/component: api - {{- end }} {{- with .Values.podLabels }} {{- toYaml . | nindent 8 }} {{- end }} diff --git a/helm/agentflow/templates/networkpolicy.yaml b/helm/agentflow/templates/networkpolicy.yaml index 249ccbef..43cf64a2 100644 --- a/helm/agentflow/templates/networkpolicy.yaml +++ b/helm/agentflow/templates/networkpolicy.yaml @@ -50,4 +50,12 @@ spec: - ports: - protocol: TCP port: {{ .Values.networkPolicy.egressPorts.otlp }} + {{- if eq .Values.controlPlane.store "postgres" }} + # PostgreSQL control-plane store (when controlPlane.store=postgres): the + # shared webhook/alert/outbox/usage pool. Without this rule every API and + # worker pod's connect to PG:5432 is denied and the pool never opens. + - ports: + - protocol: TCP + port: {{ .Values.networkPolicy.egressPorts.postgres }} + {{- end }} {{- end }} diff --git a/helm/agentflow/templates/poddisruptionbudget.yaml b/helm/agentflow/templates/poddisruptionbudget.yaml index 0d63da8a..db40526f 100644 --- a/helm/agentflow/templates/poddisruptionbudget.yaml +++ b/helm/agentflow/templates/poddisruptionbudget.yaml @@ -10,4 +10,13 @@ spec: selector: matchLabels: {{- include "agentflow.selectorLabels" . | nindent 6 }} + # Scope to the API role only. selectorLabels alone (name+instance) also + # match the worker pods, so with worker.enabled a single minAvailable:1 + # budget spanning {api ∪ worker} lets a drain evict BOTH API pods while the + # lone worker satisfies it → serving surface hits zero. component=api keeps + # this budget a guarantee about request-serving pods. The worker is a + # background singleton (delivery loops claim rows in PG and resume after a + # reschedule); a minAvailable:1 PDB on a single-replica worker would instead + # make its node undrainable, so workers deliberately get no PDB. + app.kubernetes.io/component: api {{- end }} diff --git a/helm/agentflow/values.yaml b/helm/agentflow/values.yaml index 30d0a371..d2ce5e45 100644 --- a/helm/agentflow/values.yaml +++ b/helm/agentflow/values.yaml @@ -101,8 +101,10 @@ podDisruptionBudget: # NetworkPolicy enforces a default-deny baseline plus explicit ingress on the # HTTP port and egress to dependencies (DNS, Kubernetes API, Redis, Kafka, -# ClickHouse, OTLP collector). Disabled by default so existing clusters -# without a NetworkPolicy controller are not broken; turn on per environment. +# ClickHouse, OTLP collector, and — on the postgres control-plane profile — +# PostgreSQL). Disabled by default so existing clusters without a NetworkPolicy +# controller are not broken; turn on per environment. The postgres:5432 egress +# rule is rendered only when controlPlane.store=postgres (harmless otherwise). networkPolicy: enabled: false ingressFromNamespaces: @@ -112,6 +114,7 @@ networkPolicy: kafka: 9092 clickhouse: 8123 otlp: 4317 + postgres: 5432 # Serving engine (ADR 0006 / 0007 / 0009). The shipped demo runs ClickHouse; # the chart's default stays the safe single-node DuckDB profile because the diff --git a/scripts/k8s_replica_correctness_verify.sh b/scripts/k8s_replica_correctness_verify.sh index 6b7a1fd5..91c0e725 100644 --- a/scripts/k8s_replica_correctness_verify.sh +++ b/scripts/k8s_replica_correctness_verify.sh @@ -63,6 +63,9 @@ CLICKHOUSE_PASSWORD_KEY="${CLICKHOUSE_PASSWORD_KEY:-clickhouse-password}" CLICKHOUSE_POD="${CLICKHOUSE_POD:-}" DELIVERY_WAIT_SECONDS="${DELIVERY_WAIT_SECONDS:-45}" DELIVERY_POLL_SECONDS="${DELIVERY_POLL_SECONDS:-2}" +# Once the first delivery is seen, keep polling this long and assert no second +# delivery_id appears — a split delivery can land a poll cycle or two later. +DELIVERY_SETTLE_SECONDS="${DELIVERY_SETTLE_SECONDS:-12}" # Tenant of the inserted event — must match the API key's tenant (staging: default). EVENT_TENANT="${EVENT_TENANT:-default}" @@ -70,6 +73,9 @@ EVENT_TENANT="${EVENT_TENANT:-default}" # pods have had a chance to race claim_alert_tick for the same rule. ALERT_WAIT_SECONDS="${ALERT_WAIT_SECONDS:-150}" ALERT_POLL_SECONDS="${ALERT_POLL_SECONDS:-5}" +# After the first page, dwell > one full alert tick (60s) and assert no second +# page appears — the other pod's dispatcher could page up to one poll later. +ALERT_SETTLE_SECONDS="${ALERT_SETTLE_SECONDS:-75}" # error_rate over 1h is 0 when the only journal rows are non-deadletter # (Check 3 insert is enough). below 1.0 therefore always fires on a healthy stand. ALERT_METRIC="${ALERT_METRIC:-error_rate}" @@ -184,10 +190,20 @@ echo " inserted pipeline_events row" # Both pods poll ~2s; enqueue is insert-win on (webhook_id, event_id). Only the # winner calls deliver() and writes webhook_deliveries rows for this event. -deadline=$(( $(date +%s) + DELIVERY_WAIT_SECONDS )) +end=$(( $(date +%s) + DELIVERY_WAIT_SECONDS )) +settle_end=0 unique_ids=0 log_count=0 -while (( $(date +%s) < deadline )); do +# Wait for the first delivery, then DWELL a settle window and assert the count +# never rises to 2. Asserting on the first sighting cannot tell "exactly one" +# from "one so far" (audit 2026-07-17 #3, vacuous break-then-assert). +while :; do + now=$(date +%s) + if (( settle_end == 0 )); then + (( now < end )) || break + else + (( now < settle_end )) || break + fi logs_json=$(curl -fsS -H "X-API-Key: $API_KEY" \ "$BASE_URL/v1/webhooks/$webhook_id/logs") # Count distinct delivery_id for this event_id. Multiple attempt rows may share @@ -205,15 +221,17 @@ print(len(ids), len(matched)) set -- $counts unique_ids=${1:-0} log_count=${2:-0} - if (( unique_ids >= 1 )); then - break + # Fail the moment a second delivery_id appears — that IS the split we guard against. + (( unique_ids <= 1 )) || fail "expected exactly 1 delivery_id for event_id=$event_id, got ${unique_ids} (split delivery across pods)" + if (( unique_ids == 1 && settle_end == 0 )); then + settle_end=$(( now + DELIVERY_SETTLE_SECONDS )) fi sleep "$DELIVERY_POLL_SECONDS" done (( unique_ids >= 1 )) || fail "no delivery log for event_id=$event_id within ${DELIVERY_WAIT_SECONDS}s (scanners idle or CH not shared)" (( unique_ids == 1 )) || fail "expected exactly 1 delivery_id for event_id=$event_id, got ${unique_ids} (split delivery across pods)" -pass "exactly one delivery_id for event_id=$event_id (${log_count} log row(s))" +pass "exactly one delivery_id for event_id=$event_id, stable over ${DELIVERY_SETTLE_SECONDS}s (${log_count} log row(s))" # --- Check 4: one alert page per incident (claim_alert_tick single-flight) ---- echo "==> [Check 4] exactly one alert.triggered page for one firing rule across pods" @@ -244,10 +262,21 @@ alert_id=$(printf '%s' "$alert_reg" | python3 -c 'import json,sys; print(json.lo [[ -n "$alert_id" ]] || fail "alert create returned no id: $alert_reg" echo " registered alert_id=$alert_id metric=$ALERT_METRIC $ALERT_CONDITION $ALERT_THRESHOLD window=$ALERT_WINDOW" -deadline=$(( $(date +%s) + ALERT_WAIT_SECONDS )) +end=$(( $(date +%s) + ALERT_WAIT_SECONDS )) +settle_end=0 triggered_ok=0 history_count=0 -while (( $(date +%s) < deadline )); do +# Wait for the first page, then DWELL > one full alert tick (60s) and assert the +# count never rises to 2 — the other pod's dispatcher could page up to one poll +# later. Asserting on the first sighting cannot tell "one" from "one so far" +# (audit 2026-07-17 #3, vacuous break-then-assert). +while :; do + now=$(date +%s) + if (( settle_end == 0 )); then + (( now < end )) || break + else + (( now < settle_end )) || break + fi hist_json=$(curl -fsS -H "X-API-Key: $API_KEY" \ "$BASE_URL/v1/alerts/$alert_id/history") counts=$(printf '%s' "$hist_json" | python3 -c ' @@ -265,14 +294,16 @@ print(len(ok), len(hist)) set -- $counts triggered_ok=${1:-0} history_count=${2:-0} - if (( triggered_ok >= 1 )); then - break + # Fail the moment a second page appears — that IS the split we guard against. + (( triggered_ok <= 1 )) || fail "expected exactly 1 successful alert.triggered page, got ${triggered_ok} (split page across pods)" + if (( triggered_ok == 1 && settle_end == 0 )); then + settle_end=$(( now + ALERT_SETTLE_SECONDS )) fi sleep "$ALERT_POLL_SECONDS" done (( triggered_ok >= 1 )) || fail "no successful alert.triggered within ${ALERT_WAIT_SECONDS}s (dispatcher idle, metric not firing, or delivery failing)" (( triggered_ok == 1 )) || fail "expected exactly 1 successful alert.triggered page, got ${triggered_ok} (split page across pods?)" -pass "exactly one alert.triggered page for alert_id=$alert_id (${history_count} history row(s))" +pass "exactly one alert.triggered page for alert_id=$alert_id, stable over ${ALERT_SETTLE_SECONDS}s (${history_count} history row(s))" echo "==> replica-correctness verify OK (Checks 1-4)" diff --git a/src/serving/api/main.py b/src/serving/api/main.py index 6fa52cf8..e58597db 100644 --- a/src/serving/api/main.py +++ b/src/serving/api/main.py @@ -320,7 +320,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # postgres adapter creates its whole schema (sessions included) once # per process, so a stray local DuckDB file would be dead weight. ensure_analytics_table(app.state.auth_manager.db_path) - app.state.webhook_dispatcher = WebhookDispatcher(app) + # Settle watermark: the journal scan only crosses seconds settled on the + # DB clock (see WebhookDispatcher). Must exceed writer stamp-to-visibility + # lag + clock skew; 0 opts out (tests only). + app.state.webhook_dispatcher = WebhookDispatcher( + app, settle_seconds=int(os.getenv("AGENTFLOW_WEBHOOK_SETTLE_SECONDS", "3")) + ) # S7: metric-cache invalidation is a first-class controller (push from the # bridge + independent journal scan). It always starts — even when the # webhook dispatcher is held back for tests — so event-driven freshness is diff --git a/src/serving/api/webhook_dispatcher.py b/src/serving/api/webhook_dispatcher.py index a5a234d3..339614e3 100644 --- a/src/serving/api/webhook_dispatcher.py +++ b/src/serving/api/webhook_dispatcher.py @@ -131,6 +131,7 @@ def __init__( poll_interval_seconds: float = 2.0, scan_batch_size: int = 1000, seen_cache_size: int = 50_000, + settle_seconds: int = 3, ) -> None: self.app = app self.poll_interval_seconds = poll_interval_seconds @@ -141,19 +142,41 @@ def __init__( self.max_delivery_attempts = 5 self.redrive_batch_size = 100 # Journal scan is incremental and bounded (issue #183): each pass reads - # at most `scan_batch_size` rows at/after `_scan_cursor` instead of the - # whole journal — the unbounded scan is what grew the API process to - # 1.67 GB over the 4 h S11 soak. `scan_batch_size` must exceed the - # largest same-second event cohort the pipeline produces (journal - # timestamps are second-granular on ClickHouse; at the measured - # 87 eps a cohort is ~100 rows, so 1000 leaves ×10 headroom) — an - # all-seen full batch advances the cursor, so a cohort smaller than - # one batch can never pin the window. The seen-set is capped too: - # eviction is safe because enqueue is idempotent on its primary key - # and inline delivery fires only for freshly inserted rows. + # at most `scan_batch_size` rows strictly after `_scan_cursor` instead of + # the whole journal — the unbounded scan is what grew the API process to + # 1.67 GB over the 4 h S11 soak. + # + # `_scan_cursor` is a COMPOSITE keyset — the (processed_at, event_id) of + # the last contiguously-handled row — not a bare timestamp (audit + # 2026-07-17 #1). A bare second-granular cursor advanced only to the + # whole second and re-fetched `WHERE processed_at >= cursor`; a single + # second holding >= `scan_batch_size` rows then filled every batch with + # that second's lowest-event_id rows, re-pinned the cursor to the SAME + # second, and silently dropped every webhook for every event at/after it + # — permanently, with a healthy-looking cursor. The keyset advances + # WITHIN a saturated second (`(processed_at, event_id) > cursor`), so the + # batch size no longer has to exceed the largest same-second cohort for + # the scan to make progress. The seen-set is capped and is now a + # secondary safety net (the keyset is the primary dedup): eviction is + # safe because enqueue is idempotent on its primary key and inline + # delivery fires only for freshly inserted rows. + # + # `settle_seconds` guards the keyset's blind spot: a STRICT cursor is + # only lossless over seconds no writer will stamp again. ClickHouse + # `processed_at` is second-granular and event ids are UUIDs (not + # monotonic), so a cursor advanced into the still-open wall-clock + # second would permanently exclude any same-second row that becomes + # visible later with a lower event_id — a silent drop at ORDINARY + # load, worse than the wedge the keyset fixed. Every journal fetch is + # therefore bounded to rows settled at least `settle_seconds` on the + # DB clock; the frontier crosses only closed seconds. The value must + # exceed writer stamp-to-visibility lag + writer↔DB clock skew + # (AGENTFLOW_WEBHOOK_SETTLE_SECONDS; 0 opts out, accepting the drop + # risk — tests only). Worst-case added delivery latency = settle. + self.settle_seconds = settle_seconds self.scan_batch_size = scan_batch_size self.seen_event_ids: BoundedSeenSet = BoundedSeenSet(maxlen=seen_cache_size) - self._scan_cursor: str | None = None + self._scan_cursor: tuple[str, str] | None = None self._task: asyncio.Task | None = None # S7: first-class subscribers notified when the journal scan marks new # events seen. Replaces the historical monkey-patch over @@ -201,11 +224,15 @@ async def run(self) -> None: def mark_existing_events_seen(self) -> None: """Initialize the scan frontier at the journal tail. - Startup semantics are unchanged — events already in the journal are - not delivered — but the initialization is O(scan_batch_size), not - O(journal): the newest batch seeds the seen-set and the cursor, and - everything older is excluded by the cursor's ``>=`` bound instead of - by enumerating it (issue #183). + Startup semantics: settled events already in the journal are not + delivered, and the initialization is O(scan_batch_size), not + O(journal): the newest settled batch seeds the seen-set and the + cursor, and everything older is excluded by the cursor's keyset bound + instead of by enumerating it (issue #183). Rows younger than the + settle watermark are deliberately NOT seeded — they are delivered + once settled, so events that raced a restart are not lost; a row the + pre-restart process already delivered is suppressed by the durable + enqueue's idempotent primary key, not re-POSTed. """ try: events = self._fetch_pipeline_events(newest_first=True, limit=self.scan_batch_size) @@ -218,15 +245,15 @@ def mark_existing_events_seen(self) -> None: self.seen_event_ids.add(_seen_event_key(event)) if events: # newest_first — seed the cursor from the newest row whose - # processed_at parses. If the very newest row's timestamp is - # malformed, fall back to the next parseable row rather than - # leaving the cursor None: a None cursor makes the next scan fetch - # with min_processed_at=None (from the oldest journal row), which + # processed_at parses AND that carries an event_id (both halves of + # the keyset). If the very newest row is unusable, fall back to the + # next usable row rather than leaving the cursor None: a None cursor + # makes the next scan fetch from the oldest journal row, which # re-delivers the whole batch we just seeded (audit #185, defensive # seed-edge). All rows in this batch are already in the seen-set, so - # advancing to any of their timestamps drops nothing. + # advancing to any of their keys drops nothing. for event in events: - seeded = _cursor_timestamp(event) + seeded = _cursor_key(event) if seeded is not None: self._scan_cursor = seeded break @@ -243,17 +270,23 @@ async def dispatch_new_events(self) -> None: # Delivery stays tenant-scoped below. Metric-cache invalidation is # owned by MetricCacheController (S7) and does not require this loop. # - # The scan is incremental (issue #183): a bounded batch at/after the - # cursor, ASC. The cursor advances over the contiguous prefix of rows - # that end this pass seen (previously or newly), and freezes at the - # first row left unseen (an enqueue failure) so that row is re-fetched - # and retried next pass — the retry-forever semantics the full scan - # provided, without materializing the whole journal every 2 s. + # The scan is incremental (issue #183): a bounded batch strictly after + # the (processed_at, event_id) keyset cursor, ASC. The cursor advances + # over the contiguous prefix of rows that end this pass seen (previously + # or newly), and freezes at the first row left unseen (an enqueue + # failure) so that row is re-fetched and retried next pass — the + # retry-forever semantics the full scan provided, without materializing + # the whole journal every 2 s. Because the cursor carries event_id, the + # frontier moves even inside a single second that holds more than one + # batch of rows (audit 2026-07-17 #1). newly_seen = 0 - advance: str | None = None + advance: tuple[str, str] | None = None frozen = False + cursor = self._scan_cursor for event in self._fetch_pipeline_events( - limit=self.scan_batch_size, min_processed_at=self._scan_cursor + limit=self.scan_batch_size, + min_processed_at=cursor[0] if cursor is not None else None, + min_event_id=cursor[1] if cursor is not None else None, ): event_id = str(event.get("event_id") or "") seen_key = _seen_event_key(event) @@ -268,7 +301,7 @@ async def dispatch_new_events(self) -> None: # the cursor move over it so a frontier of already-seen rows # cannot pin the scan window. if not frozen: - advance = _cursor_timestamp(event) or advance + advance = _cursor_key(event) or advance continue tenant = str(event.get("tenant_id") or "default") @@ -320,7 +353,7 @@ async def dispatch_new_events(self) -> None: self.seen_event_ids.add(seen_key) newly_seen += 1 if not frozen: - advance = _cursor_timestamp(event) or advance + advance = _cursor_key(event) or advance else: frozen = True @@ -463,16 +496,22 @@ def _fetch_pipeline_events( limit: int | None = None, newest_first: bool = False, min_processed_at: str | None = None, + min_event_id: str | 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 # matter arrive from an out-of-process writer, and this scan is what # drives both webhook delivery and metric-cache invalidation. + # Single chokepoint for the settle watermark: both the poll scan and + # the startup seeding go through here, so neither can advance the + # frontier into a second writers may still stamp. events: list[dict] = self.app.state.query_engine.fetch_pipeline_events( tenant_id=tenant, limit=limit, newest_first=newest_first, min_processed_at=min_processed_at, + min_event_id=min_event_id, + settle_seconds=self.settle_seconds, ) return events @@ -579,22 +618,52 @@ def _seen_event_key(event: dict) -> str: def _cursor_timestamp(event: dict) -> str | None: """Normalize a journal row's ``processed_at`` into a scan-cursor string. - Returns ``YYYY-MM-DD HH:MM:SS`` (second precision — the journal's own - granularity on ClickHouse) or ``None`` when the value is missing or does - not round-trip through strict parsing, so a malformed row can degrade one - cursor advance but never poison the next scan's SQL. + Returns ``YYYY-MM-DD HH:MM:SS[.ffffff]`` or ``None`` when the value is + missing or does not round-trip through strict parsing, so a malformed row + can degrade one cursor advance but never poison the next scan's SQL. + + Sub-second precision is *preserved* when present: the composite keyset the + cursor feeds compares ``processed_at`` exactly, and flooring a DuckDB + microsecond timestamp to its whole second would collapse a saturated + second's rows into one key and let the scan re-wedge there (audit + 2026-07-17 #1). On ClickHouse ``processed_at`` is second-granular, so the + string is a whole second either way — the same literal shape as before. """ value = event.get("processed_at") if value is None: return None if isinstance(value, datetime): + if value.microsecond: + return value.strftime("%Y-%m-%d %H:%M:%S.%f") return value.strftime("%Y-%m-%d %H:%M:%S") - text = str(value).strip().replace("T", " ").split(".", 1)[0] - try: - datetime.strptime(text, "%Y-%m-%d %H:%M:%S") - except ValueError: + text = str(value).strip().replace("T", " ") + for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"): + try: + parsed = datetime.strptime(text, fmt) + except ValueError: + continue + if parsed.microsecond: + return parsed.strftime("%Y-%m-%d %H:%M:%S.%f") + return parsed.strftime("%Y-%m-%d %H:%M:%S") + return None + + +def _cursor_key(event: dict) -> tuple[str, str] | None: + """Composite keyset cursor ``(processed_at, event_id)`` for a journal row. + + Returns ``None`` when the row cannot anchor a cursor — a missing/malformed + ``processed_at`` or a missing ``event_id`` — so the caller keeps its prior + cursor rather than advancing onto an unusable key. Both halves are required: + the keyset only defeats the same-second cohort wedge when it can order rows + that share a ``processed_at`` by ``event_id``. + """ + ts = _cursor_timestamp(event) + if ts is None: + return None + event_id = str(event.get("event_id") or "") + if not event_id: return None - return text + return (ts, event_id) def _event_type_matches(event_type: str, requested: str) -> bool: diff --git a/src/serving/control_plane/postgres.py b/src/serving/control_plane/postgres.py index ef8d50a7..861a8d44 100644 --- a/src/serving/control_plane/postgres.py +++ b/src/serving/control_plane/postgres.py @@ -594,54 +594,69 @@ def record_webhook_delivery_outcome( max_attempts: int, backoff_seconds: Sequence[float], ) -> None: - with self._connect() as conn: - 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() - WHERE webhook_id = %s AND event_id = %s - """, - (status_code, 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, - lease_expires_at = NULL, updated_at = now() - WHERE webhook_id = %s AND event_id = %s - """, - (attempts, status_code, error, 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 = %s, last_status_code = %s, - last_error = %s, next_attempt_at = %s, - lease_expires_at = NULL, updated_at = now() - WHERE webhook_id = %s AND event_id = %s - """, - ( - attempts, - status_code, - error, - datetime.now(UTC) + timedelta(seconds=delay), - webhook_id, - event_id, - ), - ) + # 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). + last_error: Exception | None = None + for attempt in range(3): + try: + with self._connect() as conn: + 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() + WHERE webhook_id = %s AND event_id = %s + """, + (status_code, 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, + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + (attempts, status_code, error, 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 = %s, last_status_code = %s, + last_error = %s, + next_attempt_at = now() + make_interval(secs => %s), + lease_expires_at = NULL, updated_at = now() + WHERE webhook_id = %s AND event_id = %s + """, + ( + attempts, + status_code, + error, + delay, + webhook_id, + event_id, + ), + ) + return + except _TRANSIENT_ERRORS as exc: + last_error = exc + time.sleep(0.01 * (attempt + 1)) + assert last_error is not None + raise last_error def park_webhook_delivery(self, *, webhook_id: str, event_id: str, error: str) -> None: with self._connect() as conn: @@ -974,21 +989,20 @@ def schedule_outbox_retry( ) 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 - ) - if retry_count >= max_retries: + is_failed = retry_count >= max_retries + if is_failed: status = "failed" - next_attempt_at = None with self._connect() as conn: conn.execute( """ UPDATE outbox - SET status = %s, retry_count = %s, next_attempt_at = %s, + SET status = %s, retry_count = %s, + next_attempt_at = CASE WHEN %s THEN NULL + ELSE now() + make_interval(secs => %s) END, last_error = %s, lease_expires_at = NULL WHERE id = %s """, - (status, retry_count, next_attempt_at, error_message, outbox_id), + (status, retry_count, is_failed, retry_delay_seconds, error_message, outbox_id), ) if status == "failed": conn.execute( diff --git a/src/serving/semantic_layer/query/engine.py b/src/serving/semantic_layer/query/engine.py index 9a6c3c45..155c2822 100644 --- a/src/serving/semantic_layer/query/engine.py +++ b/src/serving/semantic_layer/query/engine.py @@ -23,16 +23,24 @@ from .sql_builder import SQLBuilderMixin -def _coerce_journal_timestamp(value: datetime | str) -> datetime: - """Parse a journal-scan cursor into a naive, second-precision datetime. +def _coerce_journal_timestamp(value: datetime | str, *, floor_seconds: bool = True) -> datetime: + """Parse a journal-scan cursor into a naive datetime. Accepts what the journal itself hands back: ``datetime`` objects (DuckDB rows) or ``YYYY-MM-DD HH:MM:SS[.ffffff]`` strings, ``T``-separated or not (ClickHouse JSON transport). Anything else raises ``ValueError`` — the cursor is interpolated into SQL, so it must never pass through as free - text. Sub-second precision is floored: journal timestamps are - second-granular on ClickHouse, and an inclusive ``>=`` re-fetch of the - cursor second is what the callers' seen-sets are for. + text. + + ``floor_seconds`` (default) truncates to whole seconds: the inclusive + ``>=`` re-fetch of the cursor second is what that path's seen-set is for, + and ClickHouse journal timestamps are second-granular anyway. The + **composite keyset** path passes ``floor_seconds=False`` and keeps + sub-second precision on purpose: with a second floored away, every row of a + saturated second collapses to one comparison key and the ``(processed_at, + event_id)`` keyset can no longer advance *within* that second — which is the + exact cohort-wedge this cursor exists to prevent (audit 2026-07-17 #1). The + value is still strict-parsed, so it can never reach SQL as free text. Timezone-aware input is rejected rather than converted: journal timestamps are stored naive (N2 — UTC on ClickHouse, local on DuckDB), @@ -42,10 +50,15 @@ def _coerce_journal_timestamp(value: datetime | str) -> datetime: if isinstance(value, datetime): if value.tzinfo is not None: raise ValueError("journal-scan cursor must be naive (store-local) time") - return value.replace(microsecond=0) + return value.replace(microsecond=0) if floor_seconds else value text = str(value).strip().replace("T", " ") - base = text.split(".", 1)[0] - return datetime.strptime(base, "%Y-%m-%d %H:%M:%S") + if floor_seconds: + base = text.split(".", 1)[0] + return datetime.strptime(base, "%Y-%m-%d %H:%M:%S") + try: + return datetime.strptime(text, "%Y-%m-%d %H:%M:%S.%f") + except ValueError: + return datetime.strptime(text, "%Y-%m-%d %H:%M:%S") class QueryEngine( @@ -166,6 +179,8 @@ def fetch_pipeline_events( validated_only: bool = False, newest_first: bool = False, min_processed_at: datetime | str | None = None, + min_event_id: str | None = None, + settle_seconds: int | None = None, ) -> list[dict]: """Read the ``pipeline_events`` journal through the serving backend. @@ -194,13 +209,48 @@ def fetch_pipeline_events( ``event_type``/``validated_only``, usable without an ``entity_id`` for a bulk scan across many entities in one query. - ``min_processed_at`` is the incremental-scan cursor (issue #183): an - inclusive lower bound on the journal time column, accepting the - ``datetime``/string forms the journal itself returns (strictly parsed - — see ``_coerce_journal_timestamp``). Pair it with ``limit`` so a - poller re-reads only the frontier of a large journal instead of - materializing all of it on every pass. Ignored when the journal has - no time column (the scan then stays bounded by ``limit`` alone). + ``min_processed_at`` is the incremental-scan cursor (issue #183), + accepting the ``datetime``/string forms the journal itself returns + (strictly parsed — see ``_coerce_journal_timestamp``). Pair it with + ``limit`` so a poller re-reads only the frontier of a large journal + instead of materializing all of it on every pass. Ignored when the + journal has no time column (the scan then stays bounded by ``limit`` + alone). It has two modes: + + * alone, it is an **inclusive** lower bound (``processed_at >= cursor``, + second-floored) — the caller's seen-set dedups the re-fetched second. + * paired with ``min_event_id``, it becomes a **composite keyset**: + ``processed_at > ts OR (processed_at = ts AND event_id > id)`` — + strictly after the ``(processed_at, event_id)`` of the last row the + caller consumed, keeping the same ``ORDER BY processed_at, event_id``. + This is what lets the scan advance *within* a single second that holds + more than ``limit`` rows; the inclusive bound alone pins there forever + and silently drops every later event (audit 2026-07-17 #1). The + predicate is written as the OR-decomposition rather than a row-value + tuple ``(a, b) > (x, y)`` because the tuple form does not transpile to + ClickHouse, while the decomposition round-trips through sqlglot on + both backends. ``min_event_id`` alone (no ``min_processed_at``) is + ignored — a keyset needs both halves. + + ``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 <= + CAST(now() AS TIMESTAMP) - INTERVAL 'N' SECOND``, evaluated on the + *database* clock. A strict keyset cursor is only lossless over + seconds that no writer will stamp again: on ClickHouse + ``processed_at`` is second-granular and ``event_id`` is a UUID (not + monotonic), so a cursor that advances into the still-open wall-clock + second permanently excludes any same-second row that becomes visible + later with a lower ``event_id``. The watermark keeps the frontier out + of open seconds entirely. Both timestamp frames are self-consistent + per backend: DuckDB stores tz-aware stamps in session-local wall time + and ``CAST(now() AS TIMESTAMP)`` yields the same frame; ClickHouse + stores and compares in server time (UTC in every shipped deployment). + Invariant the caller owns: ``settle_seconds`` must exceed the + writers' stamp-to-visibility lag plus writer↔DB clock skew, or a + late-visible row can still land behind the frontier. ``None`` (the + default) adds no bound — re-fetching callers (search refresh, SSE, + metric cache) need none. ``tenant_id=None`` is an explicit invariant, not incidental behaviour (n4, G2 audit): it means "no tenant filter" — an unscoped, cross- @@ -305,12 +355,47 @@ def render(value: str) -> str: return [] where_clauses.append(f"entity_id = {render(entity_id)}") if min_processed_at is not None and time_column is not None: - cursor = _coerce_journal_timestamp(min_processed_at) - rendered = render(cursor.strftime("%Y-%m-%d %H:%M:%S")) # CAST keeps the comparison typed on both engines (DuckDB binds a # VARCHAR param; the ClickHouse transpile maps TIMESTAMP to # DateTime). - where_clauses.append(f"{time_column} >= CAST({rendered} AS TIMESTAMP)") + if min_event_id is not None: + # Composite keyset (audit 2026-07-17 #1): strictly past + # (processed_at, event_id) so a second holding >= `limit` rows + # cannot pin the window. Sub-second precision is preserved + # (floor_seconds=False) so a saturated DuckDB second stays + # discriminable; on ClickHouse processed_at is second-granular + # so the literal is a whole second either way. The row-value + # tuple form does not transpile — this OR-decomposition does + # (verified duckdb + clickhouse). `render` is called once per + # placeholder so the DuckDB param list matches the SQL. + cursor = _coerce_journal_timestamp(min_processed_at, floor_seconds=False) + if cursor.microsecond: + cursor_text = cursor.strftime("%Y-%m-%d %H:%M:%S.%f") + else: + cursor_text = cursor.strftime("%Y-%m-%d %H:%M:%S") + gt_ts = render(cursor_text) + eq_ts = render(cursor_text) + cursor_id = render(str(min_event_id)) + where_clauses.append( + f"({time_column} > CAST({gt_ts} AS TIMESTAMP) " + f"OR ({time_column} = CAST({eq_ts} AS TIMESTAMP) " + f"AND event_id > {cursor_id}))" + ) + else: + 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 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 + # clause at all — review follow-up: `<= now()` would still withhold + # future-stamped rows under forward writer skew). int() keeps the + # literal allowlisted; both dialects execute this expression as + # written (verified live on DuckDB and ClickHouse 25.3). + where_clauses.append( + f"{time_column} <= CAST(now() AS TIMESTAMP) " + f"- INTERVAL '{int(settle_seconds)}' SECOND" + ) # column names come from the schema allowlist above sql = f"SELECT {', '.join(select_columns)} FROM pipeline_events" # nosec B608 diff --git a/tests/integration/test_control_plane_postgres_live.py b/tests/integration/test_control_plane_postgres_live.py index c456b880..ee4c211d 100644 --- a/tests/integration/test_control_plane_postgres_live.py +++ b/tests/integration/test_control_plane_postgres_live.py @@ -87,6 +87,23 @@ def _queue_row(webhook_id: str, event_id: str) -> tuple | None: ).fetchone() +def _release_enqueue_lease(*event_ids: str) -> None: + """Hand freshly enqueued rows over to the claim path. + + The enqueue winner stamps a claim lease on insert (it inline-delivers; + see ``test_enqueue_stamps_lease_so_redrive_cannot_steal_inline``), so a + fresh row is invisible to ``claim_due_webhook_deliveries`` until an + outcome clears the lease or it expires. The claim-semantics probes below + are about the *claim* path, not the inline race — release the lease the + way expiry would, without disturbing status/attempts/next_attempt_at. + """ + with psycopg.connect(PG_DSN) as conn: + conn.execute( + "UPDATE webhook_delivery_queue SET lease_expires_at = NULL WHERE event_id = ANY(%s)", + (list(event_ids),), + ) + + def _stagger_created_at(table: str, key_column: str, key: str, offset_seconds: int) -> None: with psycopg.connect(PG_DSN) as conn: conn.execute( @@ -178,6 +195,7 @@ def test_parallel_claims_never_hand_the_same_row_to_two_workers( ) -> None: for index in range(10): _enqueue(store, "wh-1", f"e{index}") + _release_enqueue_lease(*(f"e{index}" for index in range(10))) workers = 4 barrier = threading.Barrier(workers) @@ -197,6 +215,7 @@ def test_claimed_rows_are_invisible_until_their_lease_expires( store: PostgresControlPlaneStore, ) -> None: _enqueue(store, "wh-1", "e1") + _release_enqueue_lease("e1") assert [row.event_id for row in store.claim_due_webhook_deliveries(limit=10)] == ["e1"] # Still pending (the claim is a lease, not a state flip), but leased — # a second worker sees nothing. @@ -213,6 +232,7 @@ def test_expired_lease_makes_the_row_due_again(store: PostgresControlPlaneStore) short_lease = PostgresControlPlaneStore(PG_DSN, claim_lease_seconds=0.4) try: _enqueue(short_lease, "wh-1", "e1") + _release_enqueue_lease("e1") assert [row.event_id for row in short_lease.claim_due_webhook_deliveries(limit=10)] == [ "e1" ] @@ -256,6 +276,7 @@ def test_pending_delivery_survives_a_new_store_instance( store: PostgresControlPlaneStore, ) -> None: _enqueue(store, "wh-1", "e1") + _release_enqueue_lease("e1") # the winner's inline attempt died with the process del store # simulate process exit; only the PostgreSQL rows remain reborn = PostgresControlPlaneStore(PG_DSN) @@ -329,6 +350,7 @@ def test_claims_come_back_oldest_first_within_the_limit( for index in range(3): _enqueue(store, "wh-1", f"e{index}") _stagger_created_at("webhook_delivery_queue", "event_id", f"e{index}", index) + _release_enqueue_lease("e0", "e1", "e2") claimed = store.claim_due_webhook_deliveries(limit=2) diff --git a/tests/integration/test_webhook_keyset_clickhouse_live.py b/tests/integration/test_webhook_keyset_clickhouse_live.py new file mode 100644 index 00000000..064ef85b --- /dev/null +++ b/tests/integration/test_webhook_keyset_clickhouse_live.py @@ -0,0 +1,250 @@ +"""Live-ClickHouse regression for the webhook cohort-wedge (audit 2026-07-17 #1). + +``test_webhook_dispatcher_unit.py`` proves the composite keyset drains a +saturated second on DuckDB, and ``test_pipeline_events_scan.py`` proves the +keyset predicate transpiles to ClickHouse — but nothing ran the dispatcher's +scan loop against a real ClickHouse server, and ClickHouse is exactly where +the wedge premise lives: ``processed_at`` is second-granular there, so a burst +second's rows all share one timestamp and only the ``event_id`` half of the +keyset can order them. These tests close that gap end-to-end: real transpiled +SQL, real JSON transport (timestamps come back as strings, not datetimes), +real ``ORDER BY``/``LIMIT`` pagination inside one second. + +Gated on ``CLICKHOUSE_LIVE_HOST`` like the other live suites: CI provides a +`clickhouse` service container on the test-integration job; locally, point it +at any disposable server. The suite provisions and drops its own database so +the shared demo seed and the tenant-isolation store are never touched. +""" + +from __future__ import annotations + +import os +from collections.abc import Iterator +from pathlib import Path +from types import SimpleNamespace + +import duckdb +import pytest + +from src.serving.api.webhook_dispatcher import ( + WebhookDispatcher, + WebhookFilters, + create_webhook, +) +from src.serving.backends.clickhouse_backend import ClickHouseBackend +from src.serving.backends.duckdb_backend import DuckDBBackend +from src.serving.semantic_layer.query import QueryEngine + +LIVE_HOST = os.getenv("CLICKHOUSE_LIVE_HOST") + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not LIVE_HOST, + reason="CLICKHOUSE_LIVE_HOST not configured (live ClickHouse required)", + ), +] + +# Same shape as the DuckDB regression: a single second holding MORE rows than +# one scan batch, then rows strictly after it — the events the pre-fix cursor +# hid forever. Fixed timestamps keep the journal deterministic across runs. +BATCH = 5 +COHORT = 12 +SATURATED = "2026-07-10 10:00:00" +COHORT_IDS = [f"EVT-KS-{i:03d}" for i in range(COHORT)] +AFTER = [("Z-AFTER-1", "2026-07-10 10:00:01"), ("Z-AFTER-2", "2026-07-10 10:00:02")] + + +def _live_database() -> str: + """A database of this suite's own — see the tenant-isolation suite for why + sharing the demo-seeded ``agentflow`` store is not an option.""" + return f"{os.getenv('CLICKHOUSE_LIVE_DATABASE', 'agentflow')}_keyset_live" + + +def _backend(database: str) -> ClickHouseBackend: + return ClickHouseBackend( + host=LIVE_HOST or "localhost", + port=int(os.getenv("CLICKHOUSE_LIVE_PORT", "8123")), + user=os.getenv("CLICKHOUSE_LIVE_USER", "agentflow"), + password=os.getenv("CLICKHOUSE_LIVE_PASSWORD", "agentflow"), + database=database, + ) + + +def _ddl(backend: ClickHouseBackend, sql: str) -> None: + backend._request(sql, expect_json=False, translate=False, use_database=False) # noqa: SLF001 + + +@pytest.fixture(scope="module") +def live_clickhouse() -> Iterator[ClickHouseBackend]: + database = _live_database() + backend = _backend(database) + _ddl(backend, f"DROP DATABASE IF EXISTS {database}") + backend.ensure_schema() + backend.insert_rows( + "pipeline_events", + [ + { + "event_id": event_id, + "topic": "orders.raw", + "tenant_id": "acme", + "entity_id": f"ORD-{event_id}", + "event_type": "order.created", + "latency_ms": 100, + "processed_at": processed_at, + } + for event_id, processed_at in [ + *((event_id, SATURATED) for event_id in COHORT_IDS), + *AFTER, + ] + ], + ) + yield backend + _ddl(backend, f"DROP DATABASE IF EXISTS {database}") + + +@pytest.fixture +def clickhouse_engine(live_clickhouse: ClickHouseBackend) -> Iterator[QueryEngine]: + """A QueryEngine whose journal scan hits live ClickHouse. + + Built via ``__new__`` exactly like the dispatcher unit suite's engine stub: + the serving backend (journal) is the live server, while the embedded DuckDB + connection only hosts the durable delivery queue — the same split a real + ClickHouse-backed API process runs with. + """ + conn = duckdb.connect(":memory:") + engine = QueryEngine.__new__(QueryEngine) + engine._backend = live_clickhouse + engine._backend_name = live_clickhouse.name + engine._duckdb_backend = DuckDBBackend(db_path=":memory:", connection=conn) + engine._conn = conn + try: + yield engine + finally: + conn.close() + + +def test_inclusive_bound_alone_cannot_pass_a_saturated_second( + clickhouse_engine: QueryEngine, +) -> None: + """The wedge premise, demonstrated on the real server: with only the + inclusive ``>=`` cursor (no ``min_event_id``), a full batch from the + saturated second returns the *same* lowest-``event_id`` rows every time — + the window cannot progress. This is the live behavior the keyset exists to + escape; if ClickHouse ever stops behaving this way the fix's premise (and + this suite) should be revisited. + """ + first = clickhouse_engine.fetch_pipeline_events(min_processed_at=SATURATED, limit=BATCH) + second = clickhouse_engine.fetch_pipeline_events(min_processed_at=SATURATED, limit=BATCH) + + assert [row["event_id"] for row in first] == COHORT_IDS[:BATCH] + assert [row["event_id"] for row in second] == COHORT_IDS[:BATCH] + + +def test_keyset_advances_within_the_saturated_second_on_live_clickhouse( + clickhouse_engine: QueryEngine, +) -> None: + """The transpiled OR-decomposition keyset paginates *through* the saturated + second on the real server — each page strictly after the last row's + ``(processed_at, event_id)``, no overlap, no wedge, tail rows reached.""" + seen: list[str] = [] + cursor: tuple[str, str] | None = None + for _ in range(10): # 14 rows / batch 5 -> 3 pages; headroom, never a spin + rows = clickhouse_engine.fetch_pipeline_events( + limit=BATCH, + min_processed_at=cursor[0] if cursor else SATURATED, + min_event_id=cursor[1] if cursor else None, + ) + if not rows: + break + seen.extend(str(row["event_id"]) for row in rows) + last = rows[-1] + cursor = (str(last["processed_at"]), str(last["event_id"])) + + assert seen == COHORT_IDS + [event_id for event_id, _ in AFTER] + assert cursor == ("2026-07-10 10:00:02", "Z-AFTER-2") + + +async def test_dispatcher_drains_a_saturated_second_on_live_clickhouse( + clickhouse_engine: QueryEngine, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The full dispatcher loop against the live journal: every event in and + after the saturated second is delivered exactly once and the scan cursor + climbs past the second — the end-to-end property audit #1 said no test + proved outside DuckDB.""" + app = SimpleNamespace( + state=SimpleNamespace( + query_engine=clickhouse_engine, + webhook_config_path=tmp_path / "webhooks.yaml", + ) + ) + create_webhook(app, url="https://keyset.test/hook", tenant="acme", filters=WebhookFilters()) + dispatcher = WebhookDispatcher(app, scan_batch_size=BATCH) + + delivered: list[str] = [] + + async def _deliver(webhook: object, event: dict) -> dict: + delivered.append(str(event["event_id"])) + return {"success": True, "status_code": 200, "event_id": event["event_id"]} + + monkeypatch.setattr(dispatcher, "deliver", _deliver) + + # Drive the poll loop by hand, exactly like the DuckDB regression: the + # keyset advances strictly each pass, so a bounded number of passes drains + # the journal; a re-wedge surfaces as a failed assertion, never a spin. + for _ in range(50): + before = dispatcher._scan_cursor + await dispatcher.dispatch_new_events() + if dispatcher._scan_cursor == before: + break + + expected = set(COHORT_IDS) | {event_id for event_id, _ in AFTER} + assert set(delivered) == expected + assert len(delivered) == len(expected), "idempotent enqueue + strict keyset: no duplicates" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:02", "Z-AFTER-2") + + +def test_settle_watermark_excludes_open_second_on_live_clickhouse( + clickhouse_engine: QueryEngine, live_clickhouse: ClickHouseBackend +) -> None: + """The DB-clock settle watermark on the real server (review follow-up to + audit #1): a freshly stamped row is invisible to a settled fetch and + visible to an unbounded one — the transpiled ``CAST(now() AS TIMESTAMP) - + INTERVAL`` expression executes and compares in ClickHouse's own clock + frame. Runs last in this module and deletes its row so the earlier tests' + exact-set assertions stay valid on any re-run ordering. + """ + from datetime import UTC, datetime + + fresh_id = "W-FRESH-NOW" + live_clickhouse.insert_rows( + "pipeline_events", + [ + { + "event_id": fresh_id, + "topic": "orders.raw", + "tenant_id": "acme", + "entity_id": "ORD-W-FRESH", + "event_type": "order.created", + "latency_ms": 100, + # UTC wall time — the same frame the ClickHouse server (UTC in + # CI and every shipped deployment) stamps and compares in. + "processed_at": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), + } + ], + ) + try: + settled = clickhouse_engine.fetch_pipeline_events(settle_seconds=30, limit=100) + unbounded = clickhouse_engine.fetch_pipeline_events(limit=100) + + assert fresh_id not in {row["event_id"] for row in settled} + assert fresh_id in {row["event_id"] for row in unbounded} + finally: + live_clickhouse._request( # noqa: SLF001 + f"ALTER TABLE pipeline_events DELETE WHERE event_id = '{fresh_id}' " + "SETTINGS mutations_sync = 1", + expect_json=False, + translate=False, + ) diff --git a/tests/integration/test_webhooks.py b/tests/integration/test_webhooks.py index df0ef0ed..e4a28398 100644 --- a/tests/integration/test_webhooks.py +++ b/tests/integration/test_webhooks.py @@ -141,7 +141,7 @@ def _prepare_pipeline_events(client: TestClient) -> None: ( "evt-order-1", "events.validated", - "3 seconds", + "33 seconds", "order.created", "ORD-1", 15, @@ -151,7 +151,7 @@ def _prepare_pipeline_events(client: TestClient) -> None: ( "evt-payment-1", "events.validated", - "2 seconds", + "32 seconds", "payment.initiated", "ORD-1", 20, @@ -161,7 +161,7 @@ def _prepare_pipeline_events(client: TestClient) -> None: ( "evt-order-2", "events.validated", - "1 seconds", + "31 seconds", "order.created", "ORD-2", 12, @@ -199,7 +199,7 @@ def _prepare_tenant_pipeline_events(client: TestClient) -> None: ( "evt-acme-order", "events.validated", - "2 seconds", + "32 seconds", "order.created", "ORD-ACME", 12, @@ -209,7 +209,7 @@ def _prepare_tenant_pipeline_events(client: TestClient) -> None: ( "evt-beta-order", "events.validated", - "1 seconds", + "31 seconds", "order.created", "ORD-BETA", 15, diff --git a/tests/unit/test_cache_invalidation.py b/tests/unit/test_cache_invalidation.py index af043d6f..cc9381bf 100644 --- a/tests/unit/test_cache_invalidation.py +++ b/tests/unit/test_cache_invalidation.py @@ -219,7 +219,9 @@ async def test_webhook_listener_hook_fires_on_new_events(): engine._backend_name = backend.name engine._conn = conn app = SimpleNamespace(state=SimpleNamespace(query_engine=engine)) - dispatcher = WebhookDispatcher(app) + # settle_seconds=0: the probe is the listener hook, and its single row is + # stamped NOW() — held back by the default settle watermark. + dispatcher = WebhookDispatcher(app, settle_seconds=0) hits: list[int] = [] async def listener() -> None: diff --git a/tests/unit/test_helm_values_contract.py b/tests/unit/test_helm_values_contract.py index 57a131bb..ea1c04cc 100644 --- a/tests/unit/test_helm_values_contract.py +++ b/tests/unit/test_helm_values_contract.py @@ -403,11 +403,7 @@ def test_serviceaccount_is_pre_hook_before_provision_job(): # Multi-doc YAML: find SA and provision Job hook metadata. docs = list(yaml.safe_load_all(rendered.stdout)) - sa_docs = [ - d - for d in docs - if d and d.get("kind") == "ServiceAccount" - ] + sa_docs = [d for d in docs if d and d.get("kind") == "ServiceAccount"] job_docs = [ d for d in docs @@ -425,9 +421,10 @@ def test_serviceaccount_is_pre_hook_before_provision_job(): assert int(sa_ann["helm.sh/hook-weight"]) < int(job_ann["helm.sh/hook-weight"]) assert sa_ann["helm.sh/hook-delete-policy"] == "before-hook-creation" # Job must still bind the chart SA (not default). - assert job_docs[0]["spec"]["template"]["spec"]["serviceAccountName"] == sa_docs[0][ - "metadata" - ]["name"] + assert ( + job_docs[0]["spec"]["template"]["spec"]["serviceAccountName"] + == sa_docs[0]["metadata"]["name"] + ) def test_serving_clickhouse_tls_render_is_first_class(): diff --git a/tests/unit/test_k8s_replica_correctness_verify.py b/tests/unit/test_k8s_replica_correctness_verify.py index e607cbc0..c651345e 100644 --- a/tests/unit/test_k8s_replica_correctness_verify.py +++ b/tests/unit/test_k8s_replica_correctness_verify.py @@ -29,7 +29,9 @@ def test_replica_correctness_script_automates_checks_1_through_4() -> None: assert "delivery_id" in text # POST target must accept POST with 2xx (example.com → 405 confuses exactly-one). assert "httpbin.org/post" in text or "WEBHOOK_URL" in text - assert "example.com/agentflow-replica-verify" not in text.split("WEBHOOK_URL=")[1].splitlines()[0] + assert ( + "example.com/agentflow-replica-verify" not in text.split("WEBHOOK_URL=")[1].splitlines()[0] + ) assert "[Check 4]" in text assert "/v1/alerts" in text diff --git a/tests/unit/test_pipeline_events_scan.py b/tests/unit/test_pipeline_events_scan.py index 2984f413..89621ded 100644 --- a/tests/unit/test_pipeline_events_scan.py +++ b/tests/unit/test_pipeline_events_scan.py @@ -152,3 +152,143 @@ def test_min_processed_at_ignored_when_journal_has_no_time_column() -> None: ((sql, _),) = backend.calls assert "CAST" not in sql, "no time column to bound on — the scan stays limit-bounded" assert sql.endswith("LIMIT 50") + + +# --- composite keyset cursor (audit 2026-07-17 #1) ----------------------------- + + +def test_min_event_id_renders_composite_keyset_not_inclusive_bound() -> None: + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events( + min_processed_at="2026-07-10 10:00:00", min_event_id="e0004", limit=1000 + ) + + ((sql, params),) = backend.calls + # Strict keyset PAST (processed_at, event_id), written as the portable + # OR-decomposition (row-value tuples do not transpile to ClickHouse). This is + # what lets the scan advance WITHIN a second holding more than one batch of + # rows; the inclusive `>=` bound alone pins there forever (the cohort-wedge). + assert params is None + assert ( + "(processed_at > CAST('2026-07-10 10:00:00' AS TIMESTAMP) " + "OR (processed_at = CAST('2026-07-10 10:00:00' AS TIMESTAMP) " + "AND event_id > 'e0004'))" + ) in sql + assert ">=" not in sql, "keyset must not fall back to the wedge-prone inclusive bound" + assert sql.endswith("ORDER BY processed_at ASC, event_id ASC LIMIT 1000") + + +def test_min_event_id_preserves_sub_second_precision_in_the_keyset() -> None: + # A DuckDB journal timestamp can carry microseconds; the keyset must compare + # at full precision, otherwise a saturated second's sub-second rows collapse + # to one key and re-wedge. (On ClickHouse processed_at is second-granular, so + # this branch simply never produces a fractional literal there.) + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events(min_processed_at="2026-07-10T10:00:00.123456", min_event_id="e1") + + ((sql, _),) = backend.calls + assert "CAST('2026-07-10 10:00:00.123456' AS TIMESTAMP)" in sql + + +def test_min_event_id_without_min_processed_at_is_ignored() -> None: + # A keyset needs both halves; an event id alone is not a cursor. + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events(min_event_id="e1", limit=10) + + ((sql, _),) = backend.calls + assert "event_id >" not in sql + assert "CAST" not in sql + + +# --- settle watermark (audit 2026-07-17 #1 follow-up) -------------------------- + + +def test_settle_seconds_renders_db_clock_watermark() -> None: + # The watermark is evaluated on the DATABASE clock — no app-clock literal + # may appear (the whole point: writer stamps and the bound live in the + # same clock frame per backend). + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events(settle_seconds=3, limit=100) + + ((sql, _),) = backend.calls + assert "processed_at <= CAST(now() AS TIMESTAMP) - INTERVAL '3' SECOND" in sql + assert sql.endswith("LIMIT 100") + + +def test_settle_seconds_absent_by_default() -> None: + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events(limit=10) + + ((sql, _),) = backend.calls + assert "INTERVAL" not in sql, "re-fetching callers must keep the unbounded scan" + + +def test_settle_seconds_zero_is_a_true_opt_out() -> None: + # 0 must emit NO bound at all — `<= now()` would still withhold rows from + # a forward-skewed writer, which is not what "opt out" means. + backend = FakeExternalBackend() + engine = _engine(backend) + + engine.fetch_pipeline_events(settle_seconds=0, limit=10) + + ((sql, _),) = backend.calls + assert "INTERVAL" not in sql + assert "now()" not in sql + + +def test_settle_seconds_ignored_without_time_column() -> None: + backend = FakeExternalBackend(columns={"event_id", "topic"}) + engine = _engine(backend) + + engine.fetch_pipeline_events(settle_seconds=3, limit=10) + + ((sql, _),) = backend.calls + assert "INTERVAL" not in sql + + +def test_keyset_predicate_transpiles_to_clickhouse() -> None: + # Two-backend portability guard (audit 2026-07-17 #1): the keyset predicate + # the external path emits must survive the exact ClickHouseBackend transpile + # (parse duckdb -> generate clickhouse -> re-parse clickhouse) with table + # references preserved — the backend's own fail-closed invariant + # (`_assert_scope_preserved`). A row-value tuple `(a, b) > (x, y)` would not + # round-trip; this OR-decomposition does. Pins the property in CI so a + # sqlglot regression can never silently reintroduce the ClickHouse-only wedge. + import sqlglot + from sqlglot import exp + + backend = FakeExternalBackend() + engine = _engine(backend) + engine.fetch_pipeline_events( + min_processed_at="2026-07-10 10:00:00", min_event_id="e0004", settle_seconds=3 + ) + ((sql, _),) = backend.calls + + statements = [s for s in sqlglot.parse(sql, read="duckdb") if s is not None] + assert len(statements) == 1 + translated = statements[0].sql(dialect="clickhouse") + reparsed = sqlglot.parse_one(translated, read="clickhouse") # must not raise + + def _refs(node: exp.Expr) -> list[tuple[str, str, str]]: + return sorted( + ((t.catalog or "").lower(), (t.db or "").lower(), (t.name or "").lower()) + for t in node.find_all(exp.Table) + ) + + assert _refs(statements[0]) == _refs(reparsed), "transpile changed table references" + # the keyset survived as an OR of two typed comparisons over event_id + assert "event_id" in translated + assert " OR " in translated.upper() + # ...and the DB-clock settle watermark survived alongside it + assert "now()" in translated.lower() + assert "interval" in translated.lower() diff --git a/tests/unit/test_postgres_enqueue_lease_contract.py b/tests/unit/test_postgres_enqueue_lease_contract.py index 1e05afa5..4b75706e 100644 --- a/tests/unit/test_postgres_enqueue_lease_contract.py +++ b/tests/unit/test_postgres_enqueue_lease_contract.py @@ -7,9 +7,7 @@ from pathlib import Path -POSTGRES = ( - Path(__file__).resolve().parents[2] / "src" / "serving" / "control_plane" / "postgres.py" -) +POSTGRES = Path(__file__).resolve().parents[2] / "src" / "serving" / "control_plane" / "postgres.py" def test_enqueue_webhook_delivery_stamps_lease_on_insert() -> None: diff --git a/tests/unit/test_webhook_dispatcher_unit.py b/tests/unit/test_webhook_dispatcher_unit.py index 76320253..45081510 100644 --- a/tests/unit/test_webhook_dispatcher_unit.py +++ b/tests/unit/test_webhook_dispatcher_unit.py @@ -237,7 +237,10 @@ def test_fetch_pipeline_events_no_rows_returns_empty() -> None: def test_mark_existing_events_seen_populates_seen_ids( pipeline_conn: duckdb.DuckDBPyConnection, ) -> None: - dispatcher = WebhookDispatcher(_stub_app(pipeline_conn)) + # settle_seconds=0: this probe is about seeding mechanics, and its fixture + # stamps e3 with NOW() — under the default settle watermark an open-second + # row is deliberately not seeded (pinned separately below). + dispatcher = WebhookDispatcher(_stub_app(pipeline_conn), settle_seconds=0) dispatcher.mark_existing_events_seen() @@ -247,8 +250,11 @@ def test_mark_existing_events_seen_populates_seen_ids( # --- bounded incremental journal scan (issue #183) ---------------------------- -def _journal_conn(rows: list[tuple[str, str, str, str]]) -> duckdb.DuckDBPyConnection: - """pipeline_events with explicit timestamps: (event_id, tenant, event_type, ts).""" +def _journal_conn(rows: list[tuple[str, str, str, str | datetime]]) -> duckdb.DuckDBPyConnection: + """pipeline_events with explicit timestamps: (event_id, tenant, event_type, ts). + + ``ts`` may be a string literal (fixed-date regressions) or an aware + datetime (binds exactly like production journal writes).""" conn = duckdb.connect(":memory:") conn.execute( """ @@ -281,8 +287,9 @@ def test_mark_existing_events_seen_is_bounded_and_sets_cursor() -> None: # O(batch), not O(journal): only the newest batch seeds the set... assert dispatcher.seen_event_ids == {"acme:mid", "acme:new"} - # ...and the cursor sits at the journal tail, excluding older rows. - assert dispatcher._scan_cursor == "2026-07-10 10:00:10" + # ...and the cursor sits at the journal tail (composite keyset), excluding + # older rows. + assert dispatcher._scan_cursor == ("2026-07-10 10:00:10", "new") finally: conn.close() @@ -315,7 +322,9 @@ def spy_fetch(**kwargs: object) -> list[dict]: (kwargs,) = captured assert kwargs["limit"] == dispatcher.scan_batch_size + # Composite keyset: both halves of the frontier are handed to the scan. assert kwargs["min_processed_at"] == "2026-07-10 10:00:10" + assert kwargs["min_event_id"] == "e-new" finally: conn.close() @@ -327,7 +336,7 @@ async def test_dispatch_advances_cursor_over_newly_seen_events() -> None: dispatcher = WebhookDispatcher(_stub_app(conn)) await dispatcher.dispatch_new_events() - assert dispatcher._scan_cursor == "2026-07-10 10:00:00" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:00", "e1") conn.execute( "INSERT INTO pipeline_events VALUES " @@ -335,7 +344,7 @@ async def test_dispatch_advances_cursor_over_newly_seen_events() -> None: ) await dispatcher.dispatch_new_events() - assert dispatcher._scan_cursor == "2026-07-10 10:00:07" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:07", "e2") assert "acme:e2" in dispatcher.seen_event_ids finally: conn.close() @@ -345,6 +354,8 @@ async def test_dispatch_advances_cursor_over_newly_seen_events() -> None: async def test_dispatch_advances_cursor_over_an_all_seen_full_batch() -> None: # Livelock guard: a full batch of already-seen rows must still move the # cursor, otherwise a seen frontier wider than one batch pins the window. + # With the composite keyset the frontier is strict (the boundary row is not + # re-fetched), so two passes clear e1..e4 rather than overlapping on e2/e3. conn = _journal_conn( [ (f"e{i}", "acme", "order.created", f"2026-07-10 10:00:0{i}") @@ -356,11 +367,167 @@ async def test_dispatch_advances_cursor_over_an_all_seen_full_batch() -> None: for i in range(1, 5): dispatcher.seen_event_ids.add(f"acme:e{i}") - await dispatcher.dispatch_new_events() # window [start, e2] — all seen - assert dispatcher._scan_cursor == "2026-07-10 10:00:02" + await dispatcher.dispatch_new_events() # window (start, e2] — all seen + assert dispatcher._scan_cursor == ("2026-07-10 10:00:02", "e2") + + await dispatcher.dispatch_new_events() # window (e2, e4] — all seen + assert dispatcher._scan_cursor == ("2026-07-10 10:00:04", "e4") + finally: + conn.close() + + +@pytest.mark.asyncio +async def test_dispatch_drains_a_second_holding_more_than_a_batch( + config_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Regression for audit_2026-07-17 #1 (cohort-wedge). A SINGLE second that + # holds MORE than scan_batch_size journal rows must not pin the scan. With a + # second-granular cursor the bounded batch fills with that second's + # lowest-event_id rows, `advance` is set to that SAME second, the next fetch + # (`WHERE processed_at >= cursor`) returns the identical rows, and every + # webhook for every event at/after that second is silently, permanently + # undelivered. The composite (processed_at, event_id) keyset advances WITHIN + # the saturated second, so the scan drains it and moves past. + # + # PROVING PROPERTY: on the pre-fix second-granular cursor this test FAILS + # (only the first `batch` events ever deliver; the tail and everything after + # the second are lost) and TERMINATES rather than hanging — confirmed by + # temporarily reverting the fix. + batch = 5 + cohort = 12 # > batch: the wedge trigger + saturated = "2026-07-10 10:00:00" + rows = [(f"e{i:03d}", "acme", "order.created", saturated) for i in range(cohort)] + # ...plus rows strictly AFTER the saturated second — exactly what the wedge + # hides forever behind the pinned cursor. + rows.append(("z-after-1", "acme", "order.created", "2026-07-10 10:00:01")) + rows.append(("z-after-2", "acme", "order.created", "2026-07-10 10:00:02")) + conn = _journal_conn(rows) + try: + app = _stub_app(conn) + app.state.webhook_config_path = config_path + create_webhook(app, url="https://a.test/h", tenant="acme", filters=WebhookFilters()) + dispatcher = WebhookDispatcher(app, scan_batch_size=batch) + + delivered: list[str] = [] + + async def _deliver(webhook: object, event: dict) -> dict: + delivered.append(str(event["event_id"])) + return {"success": True, "status_code": 200, "event_id": event["event_id"]} + + monkeypatch.setattr(dispatcher, "deliver", _deliver) + + # Drive the poll loop by hand. Each pass advances the keyset strictly, so + # a bounded number of passes drains the journal; bound the loop so a + # regression that RE-wedges surfaces as a failed assertion, never a spin. + for _ in range(50): + before = dispatcher._scan_cursor + await dispatcher.dispatch_new_events() + if dispatcher._scan_cursor == before: + break # no forward progress -> drained (or, pre-fix, wedged) + + expected = {f"e{i:03d}" for i in range(cohort)} | {"z-after-1", "z-after-2"} + # Every event — the high-event_id tail of the saturated second AND + # everything after it — is delivered exactly once (idempotent enqueue + + # strict keyset: no duplicate POSTs). + assert set(delivered) == expected + assert len(delivered) == len(expected) + # The cursor climbed PAST the saturated second to the true journal tail. + assert dispatcher._scan_cursor == ("2026-07-10 10:00:02", "z-after-2") + finally: + conn.close() + + +@pytest.mark.asyncio +async def test_settle_watermark_holds_back_open_second_rows_then_delivers( + config_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Audit 2026-07-17 #1 follow-up (review finding): a STRICT keyset cursor + # advanced into the still-open second permanently drops any same-second row + # that becomes visible later with a lower event_id (ClickHouse processed_at + # is second-granular, event ids are UUIDs — not monotonic). The settle + # watermark keeps every fetch out of unsettled seconds, so the frontier + # only crosses seconds no writer will stamp again. + # + # Rows are stamped exactly the way production DuckDB journal writes are — + # binding an AWARE datetime.now(UTC) (local_pipeline binds the same) — so + # this test also pins the frame equivalence the watermark relies on: the + # aware param and CAST(now() AS TIMESTAMP) both land in the session-local + # frame, whatever the host zone. + now_utc = datetime.now(UTC).replace(microsecond=123456) + stored = now_utc.astimezone().replace(tzinfo=None) # DuckDB's stored (session-local) frame + open_ts = stored.strftime("%Y-%m-%d %H:%M:%S.%f") + conn = _journal_conn( + [ + ("e-settled", "acme", "order.created", now_utc - timedelta(seconds=30)), + ("zz-open-high", "acme", "order.created", now_utc), + ] + ) + try: + app = _stub_app(conn) + app.state.webhook_config_path = config_path + create_webhook(app, url="https://a.test/h", tenant="acme", filters=WebhookFilters()) + dispatcher = WebhookDispatcher(app, settle_seconds=5) + + delivered: list[str] = [] + + async def _deliver(webhook: object, event: dict) -> dict: + delivered.append(str(event["event_id"])) + return {"success": True, "status_code": 200, "event_id": event["event_id"]} + + monkeypatch.setattr(dispatcher, "deliver", _deliver) + + await dispatcher.dispatch_new_events() + # Only the settled row is visible; the cursor must NOT enter the open + # second. + assert delivered == ["e-settled"] + settled_ts = (stored - timedelta(seconds=30)).strftime("%Y-%m-%d %H:%M:%S.%f") + assert dispatcher._scan_cursor == (settled_ts, "e-settled") + + # The drop scenario the watermark exists for: a same-second row with a + # LOWER event_id becomes visible after the high one already did. Behind + # a strict frontier it would be excluded forever; behind the watermark + # it is simply not visible yet. + conn.execute( + "INSERT INTO pipeline_events VALUES (?, 'orders.raw', ?, ?, ?)", + ["aa-open-low", "acme", "order.created", now_utc], + ) + await dispatcher.dispatch_new_events() + assert delivered == ["e-settled"] # still held back — and not lost + + # Time passes (simulated by dropping the watermark): BOTH open-second + # rows deliver, including the late lower-id one, exactly once each. + dispatcher.settle_seconds = 0 + await dispatcher.dispatch_new_events() + assert sorted(delivered) == ["aa-open-low", "e-settled", "zz-open-high"] + assert dispatcher._scan_cursor == (open_ts, "zz-open-high") + finally: + conn.close() - await dispatcher.dispatch_new_events() # window [e2, e3] — all seen - assert dispatcher._scan_cursor == "2026-07-10 10:00:03" + +def test_mark_existing_does_not_seed_unsettled_rows(config_path: Path) -> None: + # Startup: rows younger than the settle watermark are NOT marked seen — + # they deliver once settled (an event that raced a restart is not lost); + # a row the pre-restart process already delivered is suppressed by the + # durable enqueue's idempotent key, not re-POSTed. Stamps bind aware + # datetime.now(UTC), exactly like production journal writes. + now_utc = datetime.now(UTC).replace(microsecond=123456) + conn = _journal_conn( + [ + ("e-old", "acme", "order.created", now_utc - timedelta(seconds=30)), + ("e-fresh", "acme", "order.created", now_utc), + ] + ) + try: + app = _stub_app(conn) + app.state.webhook_config_path = config_path + dispatcher = WebhookDispatcher(app, settle_seconds=5) + + dispatcher.mark_existing_events_seen() + + assert "acme:e-old" in dispatcher.seen_event_ids + assert "acme:e-fresh" not in dispatcher.seen_event_ids + assert dispatcher._scan_cursor is not None + assert dispatcher._scan_cursor[1] == "e-old" finally: conn.close() @@ -415,7 +582,7 @@ def flaky_enqueue(webhook: object, event: dict) -> bool: await dispatcher.dispatch_new_events() assert "acme:e-fail" in dispatcher.seen_event_ids - assert dispatcher._scan_cursor == "2026-07-10 10:00:02" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:02", "e-ok") # e-ok's durable row already existed (idempotent enqueue), so the # retry pass delivered exactly the failed event — no duplicate POST. assert delivered == ["e-ok", "e-fail"] @@ -440,7 +607,7 @@ async def test_seen_set_stays_bounded_across_scans() -> None: # The cursor, not the seen-set, is what keeps old rows out of the # window — it reached the journal tail even though early ids were # evicted. - assert dispatcher._scan_cursor == "2026-07-10 10:00:07" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:07", "e7") finally: conn.close() @@ -466,7 +633,7 @@ def test_mark_existing_seeds_cursor_past_a_malformed_newest_timestamp() -> None: dispatcher.mark_existing_events_seen() # newest row's ts is malformed -> cursor falls back to the next parseable row - assert dispatcher._scan_cursor == "2026-07-10 10:00:05" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:05", "mid") # every row is marked seen regardless of ts validity (id-keyed, not ts-keyed) assert dispatcher.seen_event_ids == {"acme:new", "acme:mid", "acme:old"} @@ -485,7 +652,7 @@ async def test_dispatch_does_not_dedup_same_event_id_across_tenants() -> None: # 'other:e1' is processed (marked seen) despite 'acme:e1' being seen assert "other:e1" in dispatcher.seen_event_ids - assert dispatcher._scan_cursor == "2026-07-10 10:00:00" + assert dispatcher._scan_cursor == ("2026-07-10 10:00:00", "e1") finally: conn.close() @@ -589,7 +756,9 @@ async def test_dispatch_isolates_webhook_failure_and_enqueues_all( app.state.webhook_config_path = config_path wh1 = create_webhook(app, url="https://a.test/h1", tenant="acme", filters=WebhookFilters()) wh2 = create_webhook(app, url="https://b.test/h2", tenant="acme", filters=WebhookFilters()) - dispatcher = WebhookDispatcher(app) + # settle_seconds=0: the probe is failure isolation, and its single row + # is stamped NOW() — held back by the default settle watermark. + dispatcher = WebhookDispatcher(app, settle_seconds=0) async def _deliver(webhook: object, event: dict) -> dict: if getattr(webhook, "id", None) == wh1.id: