Skip to content

Commit a9164ed

Browse files
Merge pull request #187 from brownjuly2003-code/e4-replica-topology-proof
E4 replica-correctness topology proof + serving audit fixes (#184-186)
2 parents 9935bdc + b5b950d commit a9164ed

6 files changed

Lines changed: 153 additions & 17 deletions

File tree

‎docs/clickhouse-cutover-plan.md‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -130,13 +130,16 @@ executed accidentally.
130130
unless `controlPlane.store=postgres` and `serving.backend=clickhouse` —
131131
see ADR 0010. Slice 6 (2026-07-04) released the enum ratchet to
132132
`[embedded, postgres]`.)*
133-
- [ ] **LIVE verify (kind, Docker — attempted on the Mac 2026-07-06, genuinely
134-
incomplete):** `scripts/k8s_staging_up.sh` on kind with the scale
135-
overlay, `k8s_smoke_test.sh` green, `replicaCount: 2` schedules without
136-
PVC contention; **plus the ADR 0010 replica-correctness checks** —
137-
exactly one delivery per (webhook, event) across two pods, one alert
138-
page per incident, a webhook registered via either pod visible to both.
139-
See the recipe below.
133+
- [x] **LIVE verify (kind, Docker) — DONE 2026-07-11.** The two-real-pods
134+
replica-correctness proof passes on a freed VM: `replicaCount: 2` scheduled
135+
without PVC contention, both pods on the postgres store, and a webhook
136+
registered via the Service was visible on all 8 round-robin reads across the
137+
two pods (no split-brain) — `scripts/k8s_replica_correctness_verify.sh`
138+
Checks 1–2, exit 0. Canon: `docs/perf/e4-replica-topology-2026-07-11.md`.
139+
Exactly-one delivery / one-alert-per-incident are covered at the store level
140+
(31/31, slice-5 PG probe); their two-pod emission harness is the remaining
141+
optional layer. First attempted 2026-07-06 and blocked by host resource
142+
contention — kept below as honest history.
140143
*(Chart-side render verified locally 2026-07-04 via `helm template`/
141144
`lint`, and re-verified live-adjacent on the Mac 2026-07-06: `helm lint`
142145
clean; `helm template --set replicaCount=2 --set persistence.enabled=false`
Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
# E4 — two-real-pods replica-correctness topology proof (kind)
2+
3+
> Executed: `2026-07-11` on `deproject-mac` (kind + Docker on Colima vz, 6 GiB / 4 CPU,
4+
> macOS 13.7.8 Intel). Code: `main` @ `9935bdc`.
5+
> Closes the cutover-plan Phase 3 item that was **attempted-but-blocked on 2026-07-06**
6+
> (shared-host resource crisis — kube control-plane crash-loop, build stall). This run
7+
> freed the VM first and completed the proof.
8+
9+
## Result
10+
11+
`scripts/k8s_replica_correctness_verify.sh` → **exit 0, both checks PASS**:
12+
13+
- **[Check 1] ≥2 ready pods on the postgres store** — `deployment/agentflow` 2/2 Ready,
14+
every pod carries `AGENTFLOW_CONTROLPLANE_STORE=postgres`.
15+
- **[Check 2] cross-pod registration visibility** — one webhook registered through the
16+
Service (`webhook_id=4a4709a0-0bdc-42bc-803a-2d49c1fb8f04`) was visible on **all 8
17+
round-robin reads across the two pods**. On the embedded per-pod store a read served by
18+
the pod that did not register would miss it (the sharpest split-brain, ADR 0010 class 5);
19+
on PostgreSQL the single registration table makes every read see it.
20+
21+
Checks 2–3 in the recipe (exactly-one delivery per (webhook,event); one alert page per
22+
incident) need an event/alert emitter + capture sink and are not part of this automated
23+
script. Their **store-level guarantee** (idempotent enqueue insert-win, single-flight
24+
`claim_alert_tick`, outbox↔dead-letter atomicity) is already live-verified 31/31 by the
25+
slice-5 standalone-PostgreSQL probe suite (`docs/perf/control-plane-pg-verify-2026-07-03.md`).
26+
Phase 3 adds only the two-real-pods topology layer on top — done here.
27+
28+
## Topology stood up
29+
30+
- kind cluster `agentflow-staging` (`kindest/node:v1.31.0`), `k8s/kind-config.yaml`
31+
(NodePort 30080 → host 8080).
32+
- Chart: `helm/agentflow` with `-f values-staging.yaml -f values-staging-scale.yaml.example`
33+
`--set serving.clickhouse.user=agentflow` →
34+
`replicaCount=2`, `persistence.enabled=false` (stateless request path),
35+
`serving.backend=clickhouse`, `controlPlane.store=postgres`.
36+
- In-cluster deps (the chart ships neither PG nor CH): single-replica **PostgreSQL 16**
37+
(`agentflow-postgres`, control-plane store), **ClickHouse 24.8** (`agentflow-clickhouse`,
38+
serving backend), **Redis 7.4** (`agentflow-redis`, rate limit) + the two operator secrets
39+
`agentflow-controlplane-pg` (keyword-form DSN) and `agentflow-clickhouse`.
40+
- Pods reach Ready without Kafka/Flink: `/v1/health` returns 200 with degraded components
41+
(each `HealthCollector` check is defensive), so readiness is process-up, not dependency-up.
42+
43+
## Environment gotchas found this run (Mac stand)
44+
45+
- The docker data disk (`/dev/vdb1`, ~59 G) is **separate from the VM root** (`/`, ~19 G):
46+
the first build died `[Errno 28] No space left on device` at the *pip install* step with
47+
the docker disk at 96 %, while `df /` still showed 18 G free. Freed it by removing the
48+
orphaned old DE-compose **named** volumes (`agentflow-docker-check_{clickhouse,kafka,minio}-data`,
49+
`…_flink-checkpoints`) — never `docker volume prune`/`system prune`, which would also
50+
delete co-tenant data (`rag_support_assistant_pgdata`, anonymous datalens/auto_bi volumes)
51+
and the stopped co-tenant containers themselves.
52+
- The rendered Service takes an auto-assigned nodePort; patch it to **30080** so kind's
53+
fixed host mapping (`30080 → 127.0.0.1:8080`) reaches it, then `BASE_URL=http://127.0.0.1:8080`.
54+
- Right after the nodePort patch, `curl` can time out for a few seconds until kube-proxy
55+
rewires the mapping — the verify script (run a moment later) succeeds.
56+
- kind + a 2-replica agentflow + PG + CH + Redis fits in 6 GiB **only with the co-tenant
57+
containers stopped** (DataLens ×7 + auto_bi). The 2026-07-06 failure was exactly this
58+
contention; free the VM first (stop co-tenants, prune orphaned DE volumes/images), and
59+
return the co-tenants on teardown.
60+
61+
## Teardown
62+
63+
`kind delete cluster --name agentflow-staging`; co-tenant containers restarted from the
64+
snapshot (`/tmp/agentflow_foreign_stopped.txt`). VM returned to ~5.3 GiB available.

‎docs/serving-bridge.md‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ Metric-cache drops are **push-driven**, not hostage to the webhook loop.
157157
|---|---|---|
158158
| **Push** | After a successful apply the bridge publishes on Redis channel `agentflow:cache:metrics_invalidate` (payload `{"event_ids":[…]}`). Every API pod's `MetricCacheController` is subscribed and runs `QueryCache.invalidate_metrics()`. | Primary path for the bridge (standalone and in-process). |
159159
| **In-process callback** | `ServingBridge(..., on_batch_applied=callback)` — the API schedules a local invalidate on the event loop so the DuckDB arm does not wait for the pub/sub round-trip. | Same-process only. |
160-
| **Journal scan fallback** | `MetricCacheController` polls `QueryEngine.fetch_pipeline_events` independently of `WebhookDispatcher` — a `newest_first` window of 200 rows (`journal_scan_fetch`), so detection stays O(window) and correct however large the journal grows. | Writers that do not push (node-ingest, seed); also covers `webhook_dispatcher_autostart=False`. |
160+
| **Journal scan fallback** | `MetricCacheController` polls `QueryEngine.fetch_pipeline_events` independently of `WebhookDispatcher` — a `newest_first` window of 2000 rows (`journal_scan_fetch`), sized to hold the tail a non-pushing writer can produce between two passes at the ≥100 eps target (~2 physical rows/event pre-merge), so detection stays O(window) and correct however large the journal grows. | Writers that do not push (node-ingest, seed); also covers `webhook_dispatcher_autostart=False`. |
161161

162162
The historical monkey-patch that wrapped `WebhookDispatcher.dispatch_new_events`
163163
in `main.py` is gone. Webhooks keep their own scan for delivery; cache
@@ -188,10 +188,11 @@ now bounded:
188188
inserted rows) and a redundant metric-cache invalidate just repopulates on
189189
the next read.
190190
- **Cache scan fallback** — the lifespan used to wire an *ascending* limited
191-
scan: the oldest 200 rows, a window that stops changing once the journal
191+
scan: the oldest rows, a window that stops changing once the journal
192192
outgrows it, which silently disabled scan-driven invalidation on grown
193193
journals (push kept the soak honest). `journal_scan_fetch` reads the tail
194-
window instead.
194+
window instead, and it is sized (`DEFAULT_SCAN_WINDOW_ROWS = 2000`) to hold a
195+
full ≥100 eps burst between two passes rather than clip it.
195196
- **SSE stream** (`/v1/stream/events`) — the per-connection dedup cache is a
196197
`BoundedSeenSet` too (was a bare set growing one entry per distinct event
197198
for the connection's lifetime). Eviction cannot re-emit: the scan window is

‎src/serving/api/webhook_dispatcher.py‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -217,8 +217,19 @@ def mark_existing_events_seen(self) -> None:
217217
if event_id:
218218
self.seen_event_ids.add(_seen_event_key(event))
219219
if events:
220-
# newest_first — the first row is the journal tail.
221-
self._scan_cursor = _cursor_timestamp(events[0]) or self._scan_cursor
220+
# newest_first — seed the cursor from the newest row whose
221+
# processed_at parses. If the very newest row's timestamp is
222+
# malformed, fall back to the next parseable row rather than
223+
# leaving the cursor None: a None cursor makes the next scan fetch
224+
# with min_processed_at=None (from the oldest journal row), which
225+
# re-delivers the whole batch we just seeded (audit #185, defensive
226+
# seed-edge). All rows in this batch are already in the seen-set, so
227+
# advancing to any of their timestamps drops nothing.
228+
for event in events:
229+
seeded = _cursor_timestamp(event)
230+
if seeded is not None:
231+
self._scan_cursor = seeded
232+
break
222233

223234
async def dispatch_new_events(self) -> None:
224235
webhooks = [webhook for webhook in load_webhooks(self.app) if webhook.active]
@@ -246,7 +257,13 @@ async def dispatch_new_events(self) -> None:
246257
):
247258
event_id = str(event.get("event_id") or "")
248259
seen_key = _seen_event_key(event)
249-
if not event_id or event_id in self.seen_event_ids or seen_key in self.seen_event_ids:
260+
# Dedup is keyed on ``seen_key`` (``tenant:event_id``) — the only
261+
# shape the seen-set ever stores (see mark_existing_events_seen and
262+
# the add below). A bare-``event_id`` membership test never matched
263+
# and dropped here as dead (audit #184); cross-tenant collisions on
264+
# the same event_id must stay distinct, so the namespaced key is the
265+
# correct and only check.
266+
if not event_id or seen_key in self.seen_event_ids:
250267
# Handled earlier (or unidentifiable — nothing to retry): let
251268
# the cursor move over it so a frontier of already-seen rows
252269
# cannot pin the scan window.

‎src/serving/cache_invalidation.py‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,16 @@
3737
# How often the journal fallback re-scans when no push has landed.
3838
DEFAULT_SCAN_INTERVAL_SECONDS = 2.0
3939

40-
# One scan window: enough to always contain the journal tail between two
41-
# 2-second passes with an order of magnitude to spare (87 eps measured × 2 s
42-
# ≈ 175 rows), small enough to stay O(1) against a journal of any size.
43-
DEFAULT_SCAN_WINDOW_ROWS = 200
40+
# One scan window must contain every journal row that lands between two
41+
# 2-second passes, or the newest-first window slides past the middle of a
42+
# burst and a non-pushing writer's events are never marked seen. Sizing at the
43+
# ≥100 eps target, with the journal's pre-merge ReplacingMergeTree emitting
44+
# ~2 physical rows per applied event: 100 eps × 2 rows × 2 s ≈ 400 tail rows,
45+
# so 2000 keeps a ~5× margin (the earlier 200 left none — 87 eps already
46+
# produced ~350 physical rows/pass). Still O(window) against a journal of any
47+
# size. Push feeds (bridge apply) cover writers that publish; this window is
48+
# the fallback that must not silently under-scan the ones that do not.
49+
DEFAULT_SCAN_WINDOW_ROWS = 2000
4450

4551
# Dedup memory cap. Eviction is harmless here: a re-detected old event only
4652
# causes one redundant invalidate, and the metric cache repopulates on the

‎tests/unit/test_webhook_dispatcher_unit.py‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -445,6 +445,51 @@ async def test_seen_set_stays_bounded_across_scans() -> None:
445445
conn.close()
446446

447447

448+
def test_mark_existing_seeds_cursor_past_a_malformed_newest_timestamp() -> None:
449+
# Defensive seed-edge (audit #185): if the newest journal row's processed_at
450+
# does not parse, the cursor must fall back to the next parseable row rather
451+
# than stay None. A None cursor makes the first dispatch fetch with
452+
# min_processed_at=None (from the oldest journal row) and re-deliver the
453+
# whole batch mark_existing_events_seen just marked seen.
454+
events = [
455+
{"event_id": "new", "tenant_id": "acme", "processed_at": "not-a-timestamp"},
456+
{"event_id": "mid", "tenant_id": "acme", "processed_at": "2026-07-10 10:00:05"},
457+
{"event_id": "old", "tenant_id": "acme", "processed_at": "2026-07-10 10:00:00"},
458+
]
459+
app = SimpleNamespace(
460+
state=SimpleNamespace(
461+
query_engine=SimpleNamespace(fetch_pipeline_events=lambda **kwargs: list(events))
462+
)
463+
)
464+
dispatcher = WebhookDispatcher(app)
465+
466+
dispatcher.mark_existing_events_seen()
467+
468+
# newest row's ts is malformed -> cursor falls back to the next parseable row
469+
assert dispatcher._scan_cursor == "2026-07-10 10:00:05"
470+
# every row is marked seen regardless of ts validity (id-keyed, not ts-keyed)
471+
assert dispatcher.seen_event_ids == {"acme:new", "acme:mid", "acme:old"}
472+
473+
474+
@pytest.mark.asyncio
475+
async def test_dispatch_does_not_dedup_same_event_id_across_tenants() -> None:
476+
# Dedup is strictly on tenant:event_id (audit #184 removed the dead bare
477+
# `event_id in seen` check). The same event_id already seen for one tenant
478+
# must not suppress that id for another tenant — they are distinct events.
479+
conn = _journal_conn([("e1", "other", "order.created", "2026-07-10 10:00:00")])
480+
try:
481+
dispatcher = WebhookDispatcher(_stub_app(conn))
482+
dispatcher.seen_event_ids.add("acme:e1") # same id, different tenant, already seen
483+
484+
await dispatcher.dispatch_new_events()
485+
486+
# 'other:e1' is processed (marked seen) despite 'acme:e1' being seen
487+
assert "other:e1" in dispatcher.seen_event_ids
488+
assert dispatcher._scan_cursor == "2026-07-10 10:00:00"
489+
finally:
490+
conn.close()
491+
492+
448493
def test_delivery_logs_roundtrip() -> None:
449494
conn = duckdb.connect(":memory:")
450495
try:

0 commit comments

Comments
 (0)