diff --git a/docker/docker-compose-nats.yml b/docker/docker-compose-nats.yml index 4141230..c5eff97 100644 --- a/docker/docker-compose-nats.yml +++ b/docker/docker-compose-nats.yml @@ -114,9 +114,14 @@ services: # The Bento materializers attach with `bind: true`, so their durable consumers must # exist before they start — mirrors the chart's nats-bootstrap hook. Without these # the local stack fails with "nats: consumer not found". - for c in "TF_RAW thingsflow-greptimedb-durable tf.ingest.> tf.deliver.greptimedb thingsflow-nats-greptimedb" \ - "TF_RAW thingsflow-latest-kv-durable tf.ingest.> tf.deliver.latest-kv thingsflow-nats-latest-kv" \ - "TF_RAW thingsflow-alarms-durable tf.ingest.> tf.deliver.alarms thingsflow-nats-alarms"; do + # 6th field is max-pending. It is per-consumer, not shared: the latest-kv + # durable must be 1 -- that is the doc-merge's serialization mechanism, and + # Bento asserts its own value on bind, so a mismatch here fails the bind + # with "configuration requests max ack pending to be N, but consumer's + # value is M" and the materializer never attaches. + for c in "TF_RAW thingsflow-greptimedb-durable tf.ingest.> tf.deliver.greptimedb thingsflow-nats-greptimedb 1024" \ + "TF_RAW thingsflow-latest-kv-durable tf.ingest.> tf.deliver.latest-kv thingsflow-nats-latest-kv 1" \ + "TF_RAW thingsflow-alarms-durable tf.ingest.> tf.deliver.alarms thingsflow-nats-alarms 1024"; do set -- $$c nats --server nats://nats:4222 consumer add "$1" "$2" \ --filter "$3" \ @@ -127,7 +132,7 @@ services: --replay instant \ --max-deliver 500 \ --wait 30s \ - --max-pending 1024 \ + --max-pending "$6" \ --defaults || true done depends_on: @@ -313,6 +318,11 @@ services: NATS_DURABLE: thingsflow-latest-kv-durable NATS_FILTER_SUBJECT: "tf.ingest.>" NATS_KV_BUCKET: twin_state + # Must equal the durable's max-pending above (1). Bento's config declares + # this as a required env var with no default, so omitting it does not fall + # back -- it fails the lint and the container never starts, which is how + # the fresh-install smoke lost its latest-values pipeline entirely. + LATEST_KV_MAX_ACK_PENDING: "1" DEFAULT_TENANT_ID: aaaaaaaa-1dd2-11b2-8080-808080808080 METRICS_PORT: "4197" ports: diff --git a/docs/DIGITAL_TWIN.md b/docs/DIGITAL_TWIN.md index aac7ee7..e62d575 100644 --- a/docs/DIGITAL_TWIN.md +++ b/docs/DIGITAL_TWIN.md @@ -762,6 +762,28 @@ FLOW_TEST_PG_DSN='postgres://postgres:postgres@localhost:5432/flowtest_twin?sslm go test ./internal/twin ``` +The throwaway database is belt-and-braces, not the mechanism. Every database +harness routes its DSN through `internal/testdb.Scoped`, which creates a private +schema per test and returns a DSN carrying `search_path=`, so the +`DROP TABLE ... CASCADE` statements the harnesses run resolve inside that schema +and cannot reach `public`. Two properties are load-bearing and easy to undo by +accident: + +- **`search_path` is a connection parameter, never a `SET` statement.** + `sql.DB` is a pool; `SET search_path` binds only the connection that served + it, so under concurrency a later query lands on a connection still pointed at + `public` — the confinement failing open, silently, exactly when load makes it + matter. +- **Catalogue queries must pin `current_schema()`.** + `information_schema` and `pg_constraint` span the whole database, so an + assertion like `WHERE table_name='policy'` counts every concurrent test + schema. That one made 2 of 5 parallel runs fail while passing every time under + CI's `-p 1`. + +`tools/python/test_db_test_isolation.py` enforces all three rules. Because the +harnesses are confined, `go test ./...` now runs packages in parallel — the +default — instead of needing `-p 1`. + Full release gate: ```bash diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 3d70519..adeaa5b 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -596,6 +596,44 @@ Go/no-go criteria: - logs contain no bearer tokens, MQTT passwords, OIDC secrets, or raw payload dumps. +## Upgrading: never `--reuse-values` + +`--reuse-values` reuses the **computed** values of the previous release, not the +user-supplied ones. Every default the chart has changed since that release is +silently shadowed by the old computed value, so an upgrade that exists precisely +to ship a new default ships the old one and reports success. + +This is not hypothetical. `natsDataPlane.latestKv.maxAckPending` was changed +from `1024` to `1` because it became the doc-merge consumer's serialization +mechanism — with `--reuse-values`, a production dry-run still rendered `1024`, +which NATS would have rejected at bind time (`configuration requests max ack +pending to be 1024, but consumer's value is 1`), leaving the consumer unable to +attach at all. + +Upgrade by passing your overlay explicitly instead (start from one of the +tracked `values-*.example.yaml` templates): + +```bash +helm upgrade thingsflow ./k8s/helm/thingsflow -n thingsflow -f my-overlay.yaml +``` + +If the release carries overrides that live only in the cluster, capture them +first rather than reusing them blindly — this keeps the operator's intent and +still picks up new chart defaults: + +```bash +helm -n thingsflow get values thingsflow | tail -n +2 > /tmp/live-values.yaml +helm upgrade thingsflow ./k8s/helm/thingsflow -n thingsflow -f /tmp/live-values.yaml +``` + +Always confirm what actually rendered before trusting the upgrade: + +```bash +helm -n thingsflow get values thingsflow --all | grep -A3 latestKv +kubectl -n thingsflow get deploy thingsflow-nats-latest-kv \ + -o jsonpath='{.spec.template.spec.containers[0].env[?(@.name=="LATEST_KV_MAX_ACK_PENDING")].value}{"\n"}' +``` + ## Routine Checks Daily pilot checks: @@ -649,6 +687,7 @@ All commands in this playbook assume the canonical install (`helm upgrade |---|---|---|---|---| | `greptimedb-ttl-guard` | GreptimeDB DB/table TTL present and healthy | `0 * * * *` (hourly) | `retention.greptimedb.schedule` | `thingsflow_greptimedb_ttl_ok`, `thingsflow_greptimedb_ttl_drift_repaired` | | `greptimedb-freshness-guard` | telemetry still landing in GreptimeDB | `*/5 * * * *` | `monitoring.freshnessGuard.schedule` | `thingsflow_greptimedb_write_stale` | +| `nats-consumer-guard` | every JetStream durable is bound and draining | `*/5 * * * *` | `monitoring.consumerGuard.schedule` | `thingsflow_nats_consumer_progress_ok` | | `device-silence-guard` | individual devices that stopped reporting | `*/5 * * * *` (guard off by default) | `alarms.deviceSilence.schedule` | none — emits `DeviceSilent` alarm intents; a failed Job means the guard itself broke | | NATS stream drift-verify | live JetStream stream/consumer config vs chart intent, plus the PVC budget | every `helm install`/`upgrade` (hook, not a CronJob) | n/a — runs with each upgrade | `thingsflow_nats_stream_ok` | | `postgres-alarm-retention` | sweep of cleared+acked alarms past the window | `0 3 * * 0` (Sun 03:00) | `retention.alarm.schedule` | `thingsflow_postgres_alarm_deleted_total` | @@ -780,6 +819,63 @@ it too if entity rows are also stale.) If NATS itself lost its streams, follow If GreptimeDB is unreachable, fix that first — the guard cannot distinguish further until it can query the store. +### nats-consumer-guard + +**The failure it exists for.** A rolling node maintenance restarted NATS and +left three Bento consumers (`latest-kv`, `alarms`, `entity-greptimedb`) holding +dead subscriptions. All three reported `Running 1/1` for hours. Kubernetes could +not see it: every data-plane Deployment sets `livenessProbe: /ping` and +`readinessProbe: /ready`, and both stay green while the pod spins on +`nats: connection closed` — `/ready` is satisfied by a `nats.Conn` handle that is +dead at the *subscription* level, and `/ping` only pings Bento's HTTP server. + +This is also the failure `greptimedb-freshness-guard` structurally cannot catch: +that guard asks whether the *store* is receiving anything, and it stays green +while `latest-kv` is dead because a different consumer keeps writing rows. + +**What it checks, and why not backlog.** Queue depth is the wrong signal. +`latest-kv` runs `max_ack_pending: 1` as a deliberate single-writer +serialization mechanism, so under load it holds a large and often *growing* +backlog while working perfectly — measured at a sustained backlog of 6,729 while +its ack floor advanced 61,179 → 100,526. A depth threshold alerts there. The +guard uses two signals instead: + +1. `push_bound` — the server-side boolean saying whether anything is subscribed + to the durable's deliver subject right now. This is what the nats CLI renders + as `Active Interest: No interest`. Necessary but not sufficient: with more + than one replica in a deliver group, one dead pod still leaves it `true`. +2. `ack_floor.consumer_seq` movement across two samples — what separates + "serialized but progressing" from "dead". (`consumer_seq`, not `stream_seq`: + TF_RAW is discard-old, so retention deleting messages under a stalled + consumer drags the stream-side floor forward and reads as false progress.) + +An alert requires the fault in **both** samples, so a `helm upgrade` rolling the +Bento pods does not trip it. + +**Reading a failure.** + +```bash +kubectl -n thingsflow logs -l app=nats-consumer-guard --tail=40 +``` + +| Line | Meaning | Remedy | +|---|---|---| +| `no subscriber bound to deliver subject` | the classic silent stall — pod alive, subscription dead | `kubectl -n thingsflow rollout restart deploy/thingsflow-` | +| `ack floor is frozen ... bound but not draining` | subscribed but not acking: a wedged pipeline, a poison message against `maxDeliver`, or one dead pod in a multi-replica deliver group | check that consumer's pod logs before restarting | +| `consumer could not be read (absent, or NATS refused)` | a durable the chart expects does not exist | re-run `helm upgrade` so the nats-bootstrap hook recreates it | +| `NATS unreachable` | not evidence of health, deliberately alerts | fix NATS first | + +**Scope.** The guard covers every JetStream durable, including the +`alarm-materializer` one (created by the Go client, and the only consumer with +no Kubernetes probes at all). It cannot see the WebSocket journal fan-out +(`flow-core/internal/ws/journal.go`) — that is a *core* NATS subscription with +its own resubscribe loop and creates no JetStream consumer. + +An orphaned durable (one whose component was disabled but whose consumer was +left behind) will alert forever, correctly: on a capped, discard-old stream an +unattended consumer is not inert. Remove it at the source rather than adding it +to `monitoring.consumerGuard.ignore`. + ### device-silence-guard **Off by default** (`alarms.deviceSilence.enabled: false`): on a fleet still diff --git a/docs/adr/0003-nats-consumer-progress-guard.md b/docs/adr/0003-nats-consumer-progress-guard.md new file mode 100644 index 0000000..baedef2 --- /dev/null +++ b/docs/adr/0003-nats-consumer-progress-guard.md @@ -0,0 +1,184 @@ +# ADR 0003 — NATS consumer liveness: progress, not backlog + +- **Status:** Accepted +- **Date:** 2026-08-20 +- **Context owners:** data plane (NATS/Bento), platform ops +- **Relates to:** `k8s/helm/thingsflow/templates/nats-consumer-guard{,-scripts}.yaml`, + `greptimedb-freshness-guard.yaml`, `nats-stream-guard-scripts.yaml`, D-07 + +## Context + +On 2026-08-19 a rolling kubelet restart across all four nodes of the production +cluster restarted NATS. Three Bento consumers — `latest-kv`, `alarms` and +`entity-greptimedb` — were left holding dead subscriptions. All three kept +reporting `Running 1/1`. The halt was found only because a human ran +`nats consumer info` by hand; `entity-greptimedb` had been stopped for roughly +seven hours by then. + +This is a repeat, not a novelty. `benchmarks/FINDING-twin-state-localization.md` +records the same shape after an earlier NATS OOM-restart: four consumer pods +spinning on `level=error msg="Failed to read message: nats: connection closed"` +**with zero reconnection attempts**, a backlog of 100,921 messages, and +`Active Interest: No interest`. + +### Why nothing caught it + +**Kubernetes could not.** Every data-plane Deployment already sets +`livenessProbe: /ping` and `readinessProbe: /ready` +(`nats-data-plane-bento.yaml`). Both stayed green. `/ready` is satisfied by a +`nats.Conn` handle that is dead at the *subscription* level; `/ping` only pings +Bento's own HTTP server. Bento 1.8.1 cannot self-report this failure — that is +an observation from the field evidence above, not an assumption: if the input +had surfaced the dead subscription, the reader would have reconnected and +`/ready` would have flipped the pods to `0/1`. Neither happened. + +**The existing guard could not.** `greptimedb-freshness-guard` already contains +the right instinct — *"what separates the two is whether there is WORK +WAITING"* — but it probes only the greptimedb durable (the one consumer that did +**not** stall) and uses backlog as a *secondary* signal qualifying a GreptimeDB +freshness query. When `latest-kv` dies, GreptimeDB keeps receiving rows from a +different consumer, so that guard exits 0, green, correctly, and uselessly. + +## Decision + +Add a dedicated `nats-consumer-guard` CronJob covering **every** JetStream +durable, following the established convention (D-07: the failed Job **is** the +alert; no Prometheus, no ServiceMonitor, no push-gateway). + +### Backlog depth is the wrong signal — this is the load-bearing decision + +The obvious design is "alert when backlog exceeds a threshold, or grows between +two samples." It is wrong, and it fails in the direction that makes a guard +worthless. + +`latest-kv` runs `max_ack_pending: 1` as a deliberate single-writer +serialization mechanism, not a throughput knob: the doc-merge must complete a +whole `cache get → merge → cache set` before the next delivery. Under load it +therefore holds a large — and frequently *growing* — backlog **while working +perfectly**. Measured against a real NATS 2.10.26 server: a sustained backlog of +6,729 messages while the ack floor advanced 61,179 → 100,526 in the same window. +A depth-based rule reports that healthy consumer as dead on every tick. A +`>=`-based "not shrinking" rule is worse still: a saturated-but-healthy consumer +whose arrival rate equals its drain rate holds a *flat* non-zero backlog, which +is the definition of a working queue at capacity. + +Two signals are used instead: + +1. **`push_bound`** — all durables are push consumers created with an explicit + `--target` and `--deliver-group`, so the server tracks whether anything is + subscribed to the deliver subject *right now*. It flips within milliseconds + of the subscription dying, needs one sample, and cannot false-positive on a + serialized consumer. This is the `Active Interest: No interest` from the + incident. It is **necessary but not sufficient**: with more than one replica + in a deliver group, one dead pod still leaves it `true` — and + `thingsflow-nats-greptimedb` runs two replicas in production. + +2. **`ack_floor.consumer_seq` movement** across two samples — what actually + separates "serialized but progressing" from "dead". Deliberately + `consumer_seq`, **not** `stream_seq`: TF_RAW is `discard old` with a + `max_bytes` cap, so retention deleting messages out from under a *stalled* + consumer drags its stream-side floor forward and would read as false + progress. + +Two samples are required rather than preferred: NATS 2.10.26 exposes no +`ack_floor.last_active` field (verified against the pinned server — `ack_floor` +carries only `consumer_seq` and `stream_seq`), so there is no stateless +"seconds since last ack" available. + +An alert requires the fault in **both** samples. A `helm upgrade` rolls the +Bento pods and leaves a brief window with no subscriber; firing on that would +alert on every deployment, which is how operators learn to ignore a guard. + +### Alert only — no auto-remediation + +The chart contains no RBAC objects at all. Granting a scheduled Job standing +patch authority over the data plane to restart Deployments is a far larger +change than the guard, and it would be reviewed as one by anyone deploying this. +Beyond that, a guard that fixes the problem and exits 0 destroys the alert +surface D-07 is built on, and during a real NATS outage a self-remediating tick +would restart every consumer every five minutes — a stateless CronJob has no +memory with which to rate-limit itself, so a recoverable outage becomes a +self-inflicted one. + +### Which consumers are checked + +Live enumeration (`stream ls` → `consumer ls` → `consumer info`) **union** a +render-injected expected set. Live enumeration cannot drift from reality but is +blind by construction to a consumer that *vanished*; the expected set covers +exactly that. A generic Helm `range` over `.Values.natsDataPlane` would be +wrong — that map mixes scalars with the consumer sub-maps, and real enablement +depends on `historyStore`, which the sub-maps do not carry — so the expected set +mirrors the same gates the Bento Deployments use, plus the `alarm-materializer` +durable, which is created by the Go client, lives under a different values key, +and whose Deployment has no probes at all. + +## Consequences + +- An orphaned durable (component disabled, consumer left behind) now alerts + forever, correctly: on a capped, discard-old stream an unattended consumer is + not inert. The `thingsflow-questdb-durable` orphan documented in `values.yaml` + (grew to 106k pending, never draining) is therefore removed at the source — + `templates/nats.yaml` gained an `else` branch that `consumer rm`s it when + `questdb.enabled` is false — rather than being added to an ignore list. +- `monitoring.consumerGuard.ignore` exists as an explicit, reviewable escape + hatch, never a silent skip. +- The guard cannot see the WebSocket journal fan-out + (`flow-core/internal/ws/journal.go`): that is a *core* NATS subscription with + its own resubscribe loop and creates no JetStream consumer. Stated here so + "every consumer is guarded" is not read as "every subscription is guarded." +- **Every overlay shipped `latestKv.replicas: 2`**, contradicting the + single-writer invariant documented in `values.yaml` and + `files/bento-nats-latest-kv.yaml`. This was not one stale file: the cluster, + pilot and demo overlays all carried it, including the + `values-*.example.yaml` templates new deployments copy from, so every fresh + cluster inherited it. Production happened to be running 1, which is why it was + a landmine for the next clean upgrade rather than a live fault. All are now 1, + pinned by a test over every tracked values file. + + Whether 2 would in fact be safe is **not settled** — `max_ack_pending` is a + property of the shared consumer, so the server withholds message N+1 until N + is acked regardless of which pod acked it, and Bento acks after the cache set. + That argument is plausible and untested; scaling out belongs to the + sharded-merge work already open as a milestone criterion, with a measurement. + +## Verification + +Against a real NATS 2.10.26 + nats-box 0.16.0, using the script extracted from +the **rendered** chart, all four decision branches: + +| Scenario | Observed | Verdict | +|---|---|---| +| No subscriber, backlog 93,000 | `push_bound false` | ALERT | +| Subscriber alive, never acking (wedged / dead replica in a group) | `push_bound true`, backlog 4,000, ack floor frozen at 0 | ALERT | +| Expected durable absent | unreadable in both samples | ALERT | +| Serialized `max_ack_pending: 1` under sustained load | backlog 10,974, ack floor 117,109 → 149,520 | **OK** | +| Whole platform healthy under load | all five bound and progressing | OK, exit 0 | + +A silent defect was found this way and fixed before shipping: the list-splitting +helper used `printf '%s'` without a trailing newline, so `while read` dropped the +**last** element of every comma-separated list — and the last element of the +rendered expected set is the `alarm-materializer` durable, the one consumer with +no Kubernetes probes of any kind. + +### In-cluster reproduction + +A **single clean NATS pod restart did not reproduce the stall**: every consumer +reconnected and the guard correctly reported health. That is itself worth +recording — it is why the failure looked intermittent and hard to pin down. + +Scaling the NATS StatefulSet to zero for four minutes — long enough to exhaust +the client's reconnect attempts, which is what a node-level disruption produces +and a pod restart does not — reproduced it exactly: + +- All five consumers came back `push_bound false`. +- Every Bento pod stayed `Running 1/1` with its restart count **unchanged**, so + Kubernetes reported the data plane as healthy throughout. +- The guard failed its Job and named all five durables. + +`kubectl rollout restart` on the five Deployments restored every subscription +and the next tick returned to `exit 0`, confirming the remedy in the operations +playbook. + +The reproduction recipe matters as much as the guard: it is the only known way +to exercise this failure on demand, and a pod restart — the obvious thing to +try — does not do it. diff --git a/flow-core/assets_entityviews_post_test.go b/flow-core/assets_entityviews_post_test.go index 1aeb4be..44dc439 100644 --- a/flow-core/assets_entityviews_post_test.go +++ b/flow-core/assets_entityviews_post_test.go @@ -13,6 +13,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002, @@ -26,7 +27,7 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } @@ -100,7 +101,15 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) { }) t.Run("POST /api/entityViews creates a real row", func(t *testing.T) { - body, _ := json.Marshal(map[string]interface{}{"name": "view-1", "type": "default"}) + body, _ := json.Marshal(map[string]interface{}{ + "name": "view-1", "type": "default", + // The real UI always names a source entity; a view without one is + // meaningless. Exercise that path rather than the degenerate body. + "entityId": map[string]interface{}{ + "entityType": "DEVICE", + "id": "88888888-8888-8888-8888-888888888888", + }, + }) req := httptest.NewRequest(http.MethodPost, "/api/entityViews", bytes.NewReader(body)) req.Header.Set("X-Authorization", "Bearer "+tok) req.Header.Set("Content-Type", "application/json") @@ -117,5 +126,43 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) { if count != 1 { t.Fatalf("entity_view rows named view-1: got %d, want 1 (a prior version silently re-listed instead of creating)", count) } + var storedEntity sql.NullString + if err := db.QueryRow( + "SELECT entity_id FROM entity_view WHERE tenant_id = $1 AND name = 'view-1'", tenantID, + ).Scan(&storedEntity); err != nil { + t.Fatalf("read back entity_id: %v", err) + } + if !storedEntity.Valid || storedEntity.String != "88888888-8888-8888-8888-888888888888" { + t.Errorf("entity_id did not round-trip: %+v", storedEntity) + } + }) + + // entity_id is a uuid column and ExtractEntityID yields "" for an absent + // field, so a body without entityId used to hand the driver an empty string + // and fail with `invalid input syntax for type uuid` -- reported as a 500, + // i.e. a malformed request answered as a server fault. It must store NULL. + t.Run("POST /api/entityViews without entityId does not 500", func(t *testing.T) { + body, _ := json.Marshal(map[string]interface{}{"name": "view-no-entity", "type": "default"}) + req := httptest.NewRequest(http.MethodPost, "/api/entityViews", bytes.NewReader(body)) + req.Header.Set("X-Authorization", "Bearer "+tok) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code == http.StatusInternalServerError { + t.Fatalf("got 500 for a body with no entityId: %s", rec.Body.String()) + } + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, body=%s", rec.Code, rec.Body.String()) + } + var stored sql.NullString + if err := db.QueryRow( + "SELECT entity_id FROM entity_view WHERE tenant_id = $1 AND name = 'view-no-entity'", tenantID, + ).Scan(&stored); err != nil { + t.Fatalf("verify: %v", err) + } + if stored.Valid { + t.Errorf("entity_id should be NULL when omitted, got %q", stored.String) + } }) } diff --git a/flow-core/attributes_roundtrip_test.go b/flow-core/attributes_roundtrip_test.go index 4889093..3472c7e 100644 --- a/flow-core/attributes_roundtrip_test.go +++ b/flow-core/attributes_roundtrip_test.go @@ -23,6 +23,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" "flow-core/internal/tenant" + "flow-core/internal/testdb" "flow-core/internal/transport" ) @@ -43,7 +44,7 @@ func newRootAttrDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/alarmmaterializer/postgres_insert_test.go b/flow-core/internal/alarmmaterializer/postgres_insert_test.go index 30341a8..fc41b35 100644 --- a/flow-core/internal/alarmmaterializer/postgres_insert_test.go +++ b/flow-core/internal/alarmmaterializer/postgres_insert_test.go @@ -1,6 +1,8 @@ package alarmmaterializer import ( + "flow-core/internal/testdb" + "context" "database/sql" "errors" @@ -26,7 +28,7 @@ func insertTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/asset/asset_test.go b/flow-core/internal/asset/asset_test.go index 02bf0cc..94dc104 100644 --- a/flow-core/internal/asset/asset_test.go +++ b/flow-core/internal/asset/asset_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) func newTestDB(t *testing.T) *sql.DB { @@ -24,7 +25,7 @@ func newTestDB(t *testing.T) *sql.DB { // per-test Pool swap below (see internal/device/device_test.go for the // full rationale — first Write fixes the mode for the whole binary). t.Setenv("AUDIT_LOG_QUEUE_SIZE", "0") - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/audit/audit.go b/flow-core/internal/audit/audit.go index d07cbc7..bb67b83 100644 --- a/flow-core/internal/audit/audit.go +++ b/flow-core/internal/audit/audit.go @@ -152,6 +152,17 @@ func Write(e Event) { } func doWrite(e Event) { + // The writer is a goroutine draining a queue, so an event can reach here + // after the pool it needs is gone -- during shutdown, or in tests where the + // harness resets the pool at cleanup while a write is still in flight. + // Dereferencing a nil pool panics the whole process, and killing a server to + // avoid losing one audit row is the wrong trade: the stores are the source of + // truth and this journal is secondary. Drop the event and count it. + if dbpkg.Pool == nil { + drops.Add(1) + metricDropped.Inc() + return + } now := time.Now().UnixMilli() _, err := dbpkg.Pool.Exec(` INSERT INTO audit_log ( diff --git a/flow-core/internal/audit/audit_test.go b/flow-core/internal/audit/audit_test.go index ed3c448..e2f1ee7 100644 --- a/flow-core/internal/audit/audit_test.go +++ b/flow-core/internal/audit/audit_test.go @@ -8,6 +8,7 @@ import ( "time" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" _ "github.com/lib/pq" ) @@ -20,7 +21,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("sql.Open: %v", err) } diff --git a/flow-core/internal/customer/customer_test.go b/flow-core/internal/customer/customer_test.go index 3405fd9..d8d896c 100644 --- a/flow-core/internal/customer/customer_test.go +++ b/flow-core/internal/customer/customer_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) const tenantA = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" @@ -23,7 +24,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/dashboard/dashboard_test.go b/flow-core/internal/dashboard/dashboard_test.go index 60afe60..6499f73 100644 --- a/flow-core/internal/dashboard/dashboard_test.go +++ b/flow-core/internal/dashboard/dashboard_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) const tenantA = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" @@ -23,7 +24,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/device/device_test.go b/flow-core/internal/device/device_test.go index 2d1677b..ac84881 100644 --- a/flow-core/internal/device/device_test.go +++ b/flow-core/internal/device/device_test.go @@ -13,6 +13,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) func newTestDB(t *testing.T) *sql.DB { @@ -29,7 +30,7 @@ func newTestDB(t *testing.T) *sql.DB { // EVERY test of the package: audit's writer starts under a sync.Once, so // the first Write decides the mode for the whole test binary. t.Setenv("AUDIT_LOG_QUEUE_SIZE", "0") - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/entityview/entityview.go b/flow-core/internal/entityview/entityview.go index 50fcb71..f8a2d2e 100644 --- a/flow-core/internal/entityview/entityview.go +++ b/flow-core/internal/entityview/entityview.go @@ -36,6 +36,12 @@ func Save(w http.ResponseWriter, r *http.Request) { if viewType == "" { viewType = "default" } + // entity_id is a uuid column, and ExtractEntityID returns "" when the field + // is absent or malformed. Passing that empty string straight to the driver + // made a request without an entityId fail with `invalid input syntax for + // type uuid` and surface as a 500 -- a malformed body answered as a server + // fault. NullUUID stores NULL instead, matching how customerId below has + // always been handled. entityId := httputil.ExtractEntityID(body, "entityId") entityType := "" if v, ok := body["entityId"].(map[string]interface{}); ok { @@ -69,7 +75,7 @@ func Save(w http.ResponseWriter, r *http.Request) { UPDATE entity_view SET name = $1, type = $2, entity_id = $3, entity_type = $4, keys = $5, customer_id = $6, start_ts = $7, end_ts = $8, version = COALESCE(version, 1) + 1 WHERE id = $9`, - name, viewType, entityId, entityType, keys, dbutil.NullUUID(customerId), startTs, endTs, id) + name, viewType, dbutil.NullUUID(entityId), entityType, keys, dbutil.NullUUID(customerId), startTs, endTs, id) if err != nil { httputil.WriteError(w, http.StatusInternalServerError, "Failed to update entity view") return @@ -83,7 +89,7 @@ func Save(w http.ResponseWriter, r *http.Request) { INSERT INTO entity_view (id, created_time, name, type, entity_id, entity_type, tenant_id, customer_id, keys, start_ts, end_ts, version) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 1)`, - id, now, name, viewType, entityId, entityType, tenantId, + id, now, name, viewType, dbutil.NullUUID(entityId), entityType, tenantId, dbutil.NullUUID(customerId), keys, startTs, endTs) if err != nil { httputil.WriteError(w, http.StatusInternalServerError, "Failed to create entity view") diff --git a/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql b/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql index 6d8819a..65bdb27 100644 --- a/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql +++ b/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql @@ -1,3 +1,55 @@ +-- Rolling this back means removing a column three views depend on. +-- +-- device_info_active_attribute_view, device_info_active_ts_view and +-- device_info_view are all `SELECT d.*` over `device` (installed by the baseline +-- schema, k8s/helm/thingsflow/files/sql/02_schema-views.sql), so they pick up +-- every column the table has -- including this one. Postgres therefore refuses +-- the DROP COLUMN outright and this migration could never roll back: +-- +-- pq: cannot drop column security_status of table device because other +-- objects depend on it, view device_info_active_attribute_view depends on +-- column security_status of table device +-- +-- `DROP COLUMN ... CASCADE` would "succeed" by silently deleting all three +-- views and leaving the schema three objects short, with nothing in the up +-- direction that recreates them -- a rollback that quietly breaks every device +-- list query is worse than one that refuses. +-- +-- So: drop the views, drop the column, rebuild the views from the same +-- definitions the baseline installs. Rebuilt after the column is gone, `SELECT +-- d.*` yields exactly the pre-migration shape, which is the point. +-- +-- Keep these definitions in step with 02_schema-views.sql. They are duplicated +-- deliberately: a down migration has to restore the state it disturbed, and it +-- cannot reach into the baseline to do it. + +DROP VIEW IF EXISTS device_info_view CASCADE; +DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE; +DROP VIEW IF EXISTS device_info_active_ts_view CASCADE; + DROP INDEX IF EXISTS idx_device_tenant_security_status; ALTER TABLE device DROP CONSTRAINT IF EXISTS device_security_status_chk; ALTER TABLE device DROP COLUMN IF EXISTS security_status; + +CREATE OR REPLACE VIEW device_info_active_attribute_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(da.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary where key = 'active'); + +CREATE OR REPLACE VIEW device_info_active_ts_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(dt.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active'); + +-- Depends on the attribute view above, so it is rebuilt last. +CREATE OR REPLACE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view; diff --git a/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql b/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql index 883bdf6..8134efc 100644 --- a/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql +++ b/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql @@ -20,3 +20,41 @@ END $$; CREATE INDEX IF NOT EXISTS idx_device_tenant_security_status ON device (tenant_id, security_status); + +-- Rebuild the three device_info views so they pick up the new column. +-- +-- They are `SELECT d.*` over `device`, so their column list is fixed at +-- creation time: a view built before this migration keeps the OLD shape and +-- never shows security_status. Without this, up -> down -> up leaves a schema +-- that differs from the one it started with -- the views come back one column +-- short, silently, and only a view-based query would ever notice. +-- +-- CREATE OR REPLACE cannot be used: it refuses to change a view's column list. +-- Definitions mirror k8s/helm/thingsflow/files/sql/02_schema-views.sql; keep +-- the two in step. +DROP VIEW IF EXISTS device_info_view CASCADE; +DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE; +DROP VIEW IF EXISTS device_info_active_ts_view CASCADE; + +CREATE VIEW device_info_active_attribute_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(da.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary where key = 'active'); + +CREATE VIEW device_info_active_ts_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(dt.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active'); + +-- Depends on the attribute view above, so it is rebuilt last. +CREATE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view; diff --git a/flow-core/internal/migrations/migrations_0014_test.go b/flow-core/internal/migrations/migrations_0014_test.go index 9c1a7ef..cd1c565 100644 --- a/flow-core/internal/migrations/migrations_0014_test.go +++ b/flow-core/internal/migrations/migrations_0014_test.go @@ -257,9 +257,14 @@ func assert0014Rollback(t *testing.T, db *sql.DB) { func assertConstraintContains0014(t *testing.T, db *sql.DB, table, name, fragment string) { t.Helper() var definition string + // Scoped to current_schema(): relname+conname are unique per schema, not per + // database, so a concurrent test schema holding the same table makes this + // return multiple rows and QueryRow silently takes whichever comes first. if err := db.QueryRow(`SELECT pg_get_constraintdef(c.oid) - FROM pg_constraint c JOIN pg_class r ON r.oid=c.conrelid - WHERE r.relname=$1 AND c.conname=$2`, table, name).Scan(&definition); err != nil { + FROM pg_constraint c + JOIN pg_class r ON r.oid=c.conrelid + JOIN pg_namespace n ON n.oid=r.relnamespace + WHERE n.nspname=current_schema() AND r.relname=$1 AND c.conname=$2`, table, name).Scan(&definition); err != nil { t.Fatalf("read constraint %s: %v", name, err) } normalized := strings.ToLower(strings.ReplaceAll(definition, "\"", "")) diff --git a/flow-core/internal/migrations/migrations_0015_test.go b/flow-core/internal/migrations/migrations_0015_test.go index 9a0530d..2235915 100644 --- a/flow-core/internal/migrations/migrations_0015_test.go +++ b/flow-core/internal/migrations/migrations_0015_test.go @@ -83,15 +83,24 @@ func TestPolicyCatalog0015(t *testing.T) { // The kind column and the version CHECK backstop exist on the real table // (mirroring twin_model). An out-of-band non-canonical version is rejected. var kindColumn int + // table_schema=current_schema(): the catalogue spans the WHOLE database, so + // without it this counts every concurrent test schema that also has a + // `policy` table and reports "found 2" for a perfectly correct migration. + // It only shows up when packages run in parallel, which is why it survived. if err := db.QueryRow(`SELECT count(*) FROM information_schema.columns - WHERE table_name='policy' AND column_name='kind'`).Scan(&kindColumn); err != nil { + WHERE table_schema=current_schema() + AND table_name='policy' AND column_name='kind'`).Scan(&kindColumn); err != nil { t.Fatalf("check kind column: %v", err) } if kindColumn != 1 { t.Fatalf("expected policy.kind column, found %d", kindColumn) } var checkCount int - if err := db.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conname='policy_version_chk'`).Scan(&checkCount); err != nil { + // Same blindness: constraint names are unique per schema, not per database. + if err := db.QueryRow(`SELECT count(*) FROM pg_constraint c + JOIN pg_class r ON r.oid=c.conrelid + JOIN pg_namespace n ON n.oid=r.relnamespace + WHERE n.nspname=current_schema() AND c.conname='policy_version_chk'`).Scan(&checkCount); err != nil { t.Fatalf("check version constraint: %v", err) } if checkCount != 1 { diff --git a/flow-core/internal/provisioning/provisioning_db_test.go b/flow-core/internal/provisioning/provisioning_db_test.go index 488ed54..2609799 100644 --- a/flow-core/internal/provisioning/provisioning_db_test.go +++ b/flow-core/internal/provisioning/provisioning_db_test.go @@ -6,6 +6,7 @@ import ( "testing" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) const provTestTenant = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" @@ -17,7 +18,7 @@ func newProvisioningTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/relations/relations_test.go b/flow-core/internal/relations/relations_test.go index 36762fa..406a6c0 100644 --- a/flow-core/internal/relations/relations_test.go +++ b/flow-core/internal/relations/relations_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" "flow-core/internal/topology" ) @@ -29,7 +30,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/resource/image_import_test.go b/flow-core/internal/resource/image_import_test.go index 2450fee..ac7ddfb 100644 --- a/flow-core/internal/resource/image_import_test.go +++ b/flow-core/internal/resource/image_import_test.go @@ -13,6 +13,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002, @@ -25,7 +26,7 @@ func TestImageImport_ReturnsFullFieldSet(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/admin_settings_authz_test.go b/flow-core/internal/system/admin_settings_authz_test.go index 7ca37fa..957cb1b 100644 --- a/flow-core/internal/system/admin_settings_authz_test.go +++ b/flow-core/internal/system/admin_settings_authz_test.go @@ -13,6 +13,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Coverage for the admin-settings leak fix: the jwt/mail/security keys are @@ -26,7 +27,7 @@ func newAdminSettingsDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/asset_bulk_import_test.go b/flow-core/internal/system/asset_bulk_import_test.go index 02d4054..397650e 100644 --- a/flow-core/internal/system/asset_bulk_import_test.go +++ b/flow-core/internal/system/asset_bulk_import_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) const bulkImportTenant = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" @@ -22,7 +23,7 @@ func newAssetBulkImportDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/oauth2_infos_test.go b/flow-core/internal/system/oauth2_infos_test.go index 8237a00..d784da9 100644 --- a/flow-core/internal/system/oauth2_infos_test.go +++ b/flow-core/internal/system/oauth2_infos_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // fakeSystemJWT mints a TENANT_ADMIN token for tenantID, shared across this @@ -42,7 +43,7 @@ func TestOAuth2ClientInfos_ReadsRealTable(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } @@ -89,7 +90,7 @@ func TestOAuth2ConfigTemplate_ReadsRealTable(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/queues_write_test.go b/flow-core/internal/system/queues_write_test.go new file mode 100644 index 0000000..6fce29a --- /dev/null +++ b/flow-core/internal/system/queues_write_test.go @@ -0,0 +1,89 @@ +package system + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + authpkg "flow-core/internal/auth" +) + +// Regression coverage for the ui-contract data-fidelity audit +// (docs/adr/0002, docs/UI_CONTRACT_DATA_FIDELITY.md): /api/queues was +// registered method-agnostic, so a POST fell through to the SELECT and +// answered 200 with the queue LIST. The UI's "save queue" therefore reported +// success for a queue that was never created — the same shape as the +// widgetType DELETE that returned the row it had not deleted. +// +// The audit left this one open as a product decision rather than a missed +// wire-up, and the decision is not to implement the write: on this platform +// queues and their consumers are Helm/k8s-owned (the durables are created by +// the nats-bootstrap hook from chart values) and nothing in the data plane +// reads the `queue` table. A row written through the API would configure +// nothing, which is a more expensive lie than a refusal — so the endpoint +// answers honestly, like the other capabilities the platform deliberately +// does not have. +// +// Both cases reject before touching the database, so no Postgres is needed. + +func queuesJWT(t *testing.T) string { + t.Helper() + t.Setenv("JWT_TOKEN_SIGNING_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=") + authpkg.InitConfig() + tok, err := authpkg.GenerateAccess(authpkg.Subject{ + UserID: "00000000-0000-0000-0000-000000000001", + Email: "x@x.org", + Authority: "TENANT_ADMIN", + TenantID: "11111111-1111-1111-1111-111111111111", + }, "test") + if err != nil { + t.Fatalf("jwt: %v", err) + } + return tok +} + +func TestQueuesWriteIsAnsweredHonestly(t *testing.T) { + tok := queuesJWT(t) + + for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} { + t.Run(method, func(t *testing.T) { + body := bytes.NewReader([]byte(`{"name":"Main","topic":"tb_rule_engine.main"}`)) + req := httptest.NewRequest(method, "/api/queues", body) + req.Header.Set("X-Authorization", "Bearer "+tok) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + + HandleQueues(rec, req) + + if rec.Code == http.StatusOK { + t.Fatalf("%s /api/queues returned 200 — a prior version answered "+ + "with the queue list, reporting success for a write that never "+ + "happened; body=%s", method, rec.Body.String()) + } + if rec.Code != http.StatusNotImplemented { + t.Errorf("got %d, want %d", rec.Code, http.StatusNotImplemented) + } + // The message has to say WHERE queues actually come from, or the + // next operator reads 501 as "unfinished" and files a bug. + if !bytes.Contains(rec.Body.Bytes(), []byte("Helm")) { + t.Errorf("the refusal does not point at the real source of queue "+ + "configuration: %s", rec.Body.String()) + } + }) + } +} + +func TestQueuesReadStillRequiresAuth(t *testing.T) { + // The honest-refusal branch must not run before authentication — an + // unauthenticated caller should still get 401, not a 501 that leaks that + // the endpoint exists in this shape. + req := httptest.NewRequest(http.MethodPost, "/api/queues", nil) + rec := httptest.NewRecorder() + + HandleQueues(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("got %d, want 401", rec.Code) + } +} diff --git a/flow-core/internal/system/stubs_handlers.go b/flow-core/internal/system/stubs_handlers.go index 324b8e8..afab778 100644 --- a/flow-core/internal/system/stubs_handlers.go +++ b/flow-core/internal/system/stubs_handlers.go @@ -412,6 +412,23 @@ func HandleQueues(w http.ResponseWriter, r *http.Request) { httputil.WriteError(w, http.StatusUnauthorized, "Unauthorized") return } + // Non-GET fell through to the SELECT below and answered 200 with the queue + // LIST, so the UI's "save queue" reported success for a queue that was + // never created -- the same shape as the widgetType DELETE that returned + // the row it had not deleted. + // + // The fix is not to implement the write. On this platform queues and their + // consumers are Helm/k8s-owned: the durables are created by the + // nats-bootstrap hook from chart values, and nothing in the data plane + // reads the `queue` table. A row written here would configure nothing, + // which is a more expensive lie than a refusal. So this is answered + // honestly, the same way the platform answers the other capabilities it + // deliberately does not have. + if r.Method != http.MethodGet { + httputil.WriteError(w, http.StatusNotImplemented, + "Queue configuration is managed by the platform deployment (Helm values), not through the API") + return + } tenantId, _ := claims["tenantId"].(string) rows, err := dbpkg.Pool.Query( `SELECT id, created_time, name, topic, poll_interval, partitions, consumer_per_partition, diff --git a/flow-core/internal/system/tenant_dashboard_home_test.go b/flow-core/internal/system/tenant_dashboard_home_test.go index 0a8ae82..16d3479 100644 --- a/flow-core/internal/system/tenant_dashboard_home_test.go +++ b/flow-core/internal/system/tenant_dashboard_home_test.go @@ -12,6 +12,7 @@ import ( _ "github.com/lib/pq" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002, @@ -26,7 +27,7 @@ func TestTenantDashboardHomeInfo_RoundTrips(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/usage_test.go b/flow-core/internal/system/usage_test.go index 97a28dc..e99016f 100644 --- a/flow-core/internal/system/usage_test.go +++ b/flow-core/internal/system/usage_test.go @@ -12,6 +12,7 @@ import ( dbpkg "flow-core/internal/db" "flow-core/internal/quotas" + "flow-core/internal/testdb" ) // Regression coverage for the ui-contract data-fidelity audit @@ -34,7 +35,7 @@ func TestHandleUsage_ReadsRealQuotaLimits(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/system/user_settings_test.go b/flow-core/internal/system/user_settings_test.go index c2e6b94..4531704 100644 --- a/flow-core/internal/system/user_settings_test.go +++ b/flow-core/internal/system/user_settings_test.go @@ -9,6 +9,7 @@ import ( _ "github.com/lib/pq" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Coverage for the sidebar-menu fix: the TB UI reads @@ -23,7 +24,7 @@ func newUserSettingsDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/telemetry/questdb_integration_test.go b/flow-core/internal/telemetry/questdb_integration_test.go index 8d2d605..ff20c2f 100644 --- a/flow-core/internal/telemetry/questdb_integration_test.go +++ b/flow-core/internal/telemetry/questdb_integration_test.go @@ -32,6 +32,11 @@ func newQuestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_QUESTDB_DSN not set") } + // NOT testdb.Scoped: this DSN points at QuestDB, which speaks the + // Postgres wire protocol but has no schemas -- CREATE SCHEMA and + // search_path both fail there. Confinement is neither available nor + // needed: the target is a dedicated QuestDB instance, not a shared + // Postgres a human might also be using. db, err := sql.Open("postgres", dsn) if err != nil { t.Fatalf("open: %v", err) diff --git a/flow-core/internal/telemetry/reader_entity_tenant_test.go b/flow-core/internal/telemetry/reader_entity_tenant_test.go index d16f59c..99f01be 100644 --- a/flow-core/internal/telemetry/reader_entity_tenant_test.go +++ b/flow-core/internal/telemetry/reader_entity_tenant_test.go @@ -14,6 +14,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // These tests close the non-DEVICE cross-tenant read gap (adversarial review @@ -38,7 +39,7 @@ func newEntityTenantTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/telemetry/reader_tenant_test.go b/flow-core/internal/telemetry/reader_tenant_test.go index 59f88ff..e120944 100644 --- a/flow-core/internal/telemetry/reader_tenant_test.go +++ b/flow-core/internal/telemetry/reader_tenant_test.go @@ -13,6 +13,7 @@ import ( _ "github.com/lib/pq" authpkg "flow-core/internal/auth" + "flow-core/internal/testdb" ) // These tests verify the tenant predicate added to the device telemetry read @@ -32,7 +33,7 @@ func newTelemetryTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/tenant/tenant_handler_authz_test.go b/flow-core/internal/tenant/tenant_handler_authz_test.go index 016d507..0758ed5 100644 --- a/flow-core/internal/tenant/tenant_handler_authz_test.go +++ b/flow-core/internal/tenant/tenant_handler_authz_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Cross-tenant IDOR coverage for the GET-by-id handlers: a TENANT_ADMIN of @@ -33,7 +34,7 @@ func newAuthzDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/testdb/testdb.go b/flow-core/internal/testdb/testdb.go new file mode 100644 index 0000000..2a1df51 --- /dev/null +++ b/flow-core/internal/testdb/testdb.go @@ -0,0 +1,105 @@ +// Package testdb hands tests a Postgres connection confined to a private, +// throwaway schema. +// +// WHY. The database harnesses in this repo open FLOW_TEST_PG_DSN and then run +// `DROP TABLE IF EXISTS asset CASCADE` (and a dozen siblings) to get a clean +// slate. Against the disposable Postgres that CI starts as a service container +// that is fine. Against any database a human might plausibly point the variable +// at — a local dev stack, a shared scratch database, a pilot — it is a data-loss +// bug waiting for one careless export. The harnesses also collide with each +// other: two packages dropping and creating `asset` at the same moment leave +// each other with half a schema. CI hides that today by running `go test -p 1`, +// so the failure only appears for whoever runs `go test ./...` locally, where +// packages run in parallel by default. +// +// HOW. `Scoped` creates a schema named for the test and returns a DSN carrying +// `search_path=`. Every unqualified DDL statement the harness already +// runs then resolves inside that schema: `DROP TABLE asset` drops the copy the +// test made, never `public.asset`. Nothing in the harness bodies has to change. +// +// search_path is passed as a CONNECTION PARAMETER, not as a `SET` statement. +// sql.DB is a pool: `SET search_path` applies to whichever pooled connection +// happened to serve it, so under any concurrency the next query can land on a +// connection still pointed at `public` — which is precisely the confinement +// failing open, silently, exactly when it matters. As a DSN parameter the +// server applies it at connection setup, to every connection the pool opens. +// Verified against postgres:16 with lib/pq: 40 concurrent queries, 0 wrong. +package testdb + +import ( + "database/sql" + "fmt" + "regexp" + "strings" + "sync/atomic" + "testing" + "time" +) + +// seq disambiguates two schemas created inside the same nanosecond, which does +// happen for subtests that set up back to back. +var seq atomic.Uint64 + +var unsafeChars = regexp.MustCompile(`[^a-z0-9_]+`) + +// Scoped creates a throwaway schema and returns a DSN bound to it. The schema +// is dropped when the test ends. +// +// Call it in place of the raw DSN: +// +// pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) +// +// Accepts testing.TB so benchmarks get the same confinement as tests -- +// a benchmark that drops real tables is no less destructive. +func Scoped(t testing.TB, dsn string) string { + t.Helper() + + admin, err := sql.Open("postgres", dsn) + if err != nil { + t.Fatalf("testdb: open admin connection: %v", err) + } + defer admin.Close() + if err := admin.Ping(); err != nil { + t.Fatalf("testdb: ping: %v", err) + } + + schema := schemaName(t.Name()) + if _, err := admin.Exec(`CREATE SCHEMA ` + schema); err != nil { + t.Fatalf("testdb: create schema %s: %v", schema, err) + } + + t.Cleanup(func() { + // A fresh connection: the test's own pool is closed by its cleanup, + // and cleanup order between the two is not guaranteed. + cleaner, err := sql.Open("postgres", dsn) + if err != nil { + t.Logf("testdb: could not reopen to drop schema %s: %v", schema, err) + return + } + defer cleaner.Close() + if _, err := cleaner.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`); err != nil { + t.Logf("testdb: drop schema %s: %v", schema, err) + } + }) + + sep := "?" + if strings.Contains(dsn, "?") { + sep = "&" + } + return dsn + sep + "search_path=" + schema +} + +// schemaName derives a readable, unique, injection-safe identifier. The test +// name is carried through so a leaked schema names its own culprit. +func schemaName(testName string) string { + base := unsafeChars.ReplaceAllString(strings.ToLower(testName), "_") + base = strings.Trim(base, "_") + if len(base) > 24 { + base = base[:24] + } + if base == "" { + base = "test" + } + // Postgres truncates identifiers at 63 bytes; this stays well inside. + return fmt.Sprintf("t_%s_%d_%d", base, time.Now().UnixNano(), seq.Add(1)) +} diff --git a/flow-core/internal/testdb/testdb_test.go b/flow-core/internal/testdb/testdb_test.go new file mode 100644 index 0000000..6f6eb4c --- /dev/null +++ b/flow-core/internal/testdb/testdb_test.go @@ -0,0 +1,150 @@ +package testdb + +import ( + "database/sql" + "os" + "strings" + "sync" + "testing" + + _ "github.com/lib/pq" +) + +// These tests pin the two properties the whole isolation scheme rests on. Both +// were verified against postgres:16 before the 29 harnesses were migrated, and +// both fail silently if someone "simplifies" Scoped into a `SET search_path`. + +func dsn(t *testing.T) string { + t.Helper() + d := os.Getenv("FLOW_TEST_PG_DSN") + if d == "" { + t.Skip("FLOW_TEST_PG_DSN not set") + } + return d +} + +// An unqualified DROP TABLE — which is what every migrated harness runs — must +// not reach a same-named table outside the throwaway schema. This is the whole +// point: the harnesses keep their destructive DDL, confinement makes it safe. +func TestScopedDropCannotReachRealTables(t *testing.T) { + base := dsn(t) + + // Unique per run. A fixed name in `public` couples one run to the next -- + // the very defect this package exists to remove, and it duly showed up as a + // stale row the first time the suite ran in parallel. + canary := "testdb_canary_" + strings.Map(func(r rune) rune { + if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' { + return r + } + return '_' + }, strings.ToLower(t.Name())) + "_" + schemaName("c") + + admin, err := sql.Open("postgres", base) + if err != nil { + t.Fatalf("open: %v", err) + } + defer admin.Close() + if _, err := admin.Exec(`CREATE TABLE public.` + canary + ` (x int)`); err != nil { + t.Fatalf("seed canary: %v", err) + } + t.Cleanup(func() { _, _ = admin.Exec(`DROP TABLE IF EXISTS public.` + canary) }) + if _, err := admin.Exec(`INSERT INTO public.` + canary + ` VALUES (1)`); err != nil { + t.Fatalf("seed row: %v", err) + } + + scoped, err := sql.Open("postgres", Scoped(t, base)) + if err != nil { + t.Fatalf("open scoped: %v", err) + } + defer scoped.Close() + + // Exactly what a harness does: drop, then recreate its own copy. + if _, err := scoped.Exec(`DROP TABLE IF EXISTS ` + canary + ` CASCADE`); err != nil { + t.Fatalf("scoped drop: %v", err) + } + if _, err := scoped.Exec(`CREATE TABLE ` + canary + ` (y text)`); err != nil { + t.Fatalf("scoped create: %v", err) + } + + var n int + if err := admin.QueryRow(`SELECT count(*) FROM public.` + canary).Scan(&n); err != nil { + t.Fatalf("the scoped harness destroyed a table outside its schema: %v", err) + } + if n != 1 { + t.Errorf("public.%s lost rows: got %d, want 1", canary, n) + } +} + +// search_path must be a CONNECTION parameter, not a SET. sql.DB is a pool: a +// `SET search_path` binds only the connection that served it, so under +// concurrency later queries land on connections still pointed at public — the +// confinement failing open exactly when load makes it matter. +func TestScopedAppliesToEveryPooledConnection(t *testing.T) { + base := dsn(t) + + scopedDSN := Scoped(t, base) + if !strings.Contains(scopedDSN, "search_path=") { + t.Fatalf("Scoped did not bind search_path in the DSN: %q", scopedDSN) + } + + db, err := sql.Open("postgres", scopedDSN) + if err != nil { + t.Fatalf("open: %v", err) + } + defer db.Close() + db.SetMaxOpenConns(8) + + var wg sync.WaitGroup + wrong := make(chan string, 64) + for i := 0; i < 64; i++ { + wg.Add(1) + go func() { + defer wg.Done() + var sp string + if err := db.QueryRow("SHOW search_path").Scan(&sp); err != nil { + wrong <- "query failed: " + err.Error() + return + } + if !strings.Contains(sp, "t_") { + wrong <- sp + } + }() + } + wg.Wait() + close(wrong) + + if n := len(wrong); n != 0 { + t.Errorf("%d/64 pooled connections were not confined (first: %q)", n, <-wrong) + } +} + +// Two Scoped calls must never collide, including within the same nanosecond — +// which is what subtests setting up back to back actually do. +func TestScopedSchemasAreUnique(t *testing.T) { + seen := map[string]bool{} + for i := 0; i < 200; i++ { + name := schemaName("TestSomething/case") + if seen[name] { + t.Fatalf("duplicate schema name generated: %s", name) + } + seen[name] = true + } +} + +func TestSchemaNameIsSafeAndReadable(t *testing.T) { + // Test names carry slashes, spaces and case; all of that has to become a + // legal identifier, and the origin must stay legible so a leaked schema + // names its own culprit. + got := schemaName("TestFoo/sub case #2") + for _, bad := range []string{"/", " ", "#", "'", `"`, ";"} { + if strings.Contains(got, bad) { + t.Errorf("schema name %q contains unsafe %q", got, bad) + } + } + if !strings.HasPrefix(got, "t_testfoo_sub_case") { + t.Errorf("schema name lost its origin: %q", got) + } + if len(got) > 63 { + t.Errorf("schema name exceeds the Postgres identifier limit: %d", len(got)) + } +} diff --git a/flow-core/internal/topology/topology_benchmark_test.go b/flow-core/internal/topology/topology_benchmark_test.go index 29aa753..8927d80 100644 --- a/flow-core/internal/topology/topology_benchmark_test.go +++ b/flow-core/internal/topology/topology_benchmark_test.go @@ -1,6 +1,8 @@ package topology import ( + "flow-core/internal/testdb" + "database/sql" "fmt" "os" @@ -20,7 +22,7 @@ func BenchmarkTopologyNeighborsDepth5(b *testing.B) { if dsn == "" { b.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(b, dsn)) if err != nil { b.Fatalf("open: %v", err) } @@ -182,7 +184,7 @@ func benchmarkExpandCTE(b *testing.B, depth int) { if dsn == "" { b.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(b, dsn)) if err != nil { b.Fatalf("open: %v", err) } diff --git a/flow-core/internal/topology/topology_test.go b/flow-core/internal/topology/topology_test.go index 18c2f61..a63acc5 100644 --- a/flow-core/internal/topology/topology_test.go +++ b/flow-core/internal/topology/topology_test.go @@ -15,6 +15,7 @@ import ( "unicode/utf8" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) const ( @@ -34,7 +35,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/transport/transport_test.go b/flow-core/internal/transport/transport_test.go index a7aba8b..623462b 100644 --- a/flow-core/internal/transport/transport_test.go +++ b/flow-core/internal/transport/transport_test.go @@ -7,6 +7,7 @@ import ( "time" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) func newTestDB(t *testing.T) *sql.DB { @@ -15,7 +16,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/twin/twin_test.go b/flow-core/internal/twin/twin_test.go index 40b39ff..90b3794 100644 --- a/flow-core/internal/twin/twin_test.go +++ b/flow-core/internal/twin/twin_test.go @@ -12,6 +12,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" "flow-core/internal/twinstore" _ "github.com/lib/pq" @@ -31,7 +32,7 @@ func newTwinTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } @@ -52,6 +53,11 @@ func setupTwinTables(t *testing.T, db *sql.DB) { `DROP TABLE IF EXISTS twin_registry CASCADE`, `DROP TABLE IF EXISTS twin_model CASCADE`, `DROP TABLE IF EXISTS topology_edge CASCADE`, + `DROP TABLE IF EXISTS customer CASCADE`, + `DROP TABLE IF EXISTS entity_view CASCADE`, + `DROP TABLE IF EXISTS dashboard CASCADE`, + `DROP TABLE IF EXISTS device_profile CASCADE`, + `DROP TABLE IF EXISTS asset_profile CASCADE`, `DROP TABLE IF EXISTS asset CASCADE`, `DROP TABLE IF EXISTS device CASCADE`, `DROP TABLE IF EXISTS ts_kv_latest CASCADE`, @@ -65,6 +71,21 @@ func setupTwinTables(t *testing.T, db *sql.DB) { id uuid PRIMARY KEY, created_time bigint, tenant_id uuid, customer_id uuid, name text, type text, label text, additional_info text, version bigint default 1)`, + // The expand CTE resolves a relation's endpoints against every entity + // table it can point at, so this package needs them present even though its + // own fixtures only use assets and devices. They were never declared here: + // the tests passed because other packages (internal/customer, + // internal/dashboard, ...) had created them in the shared `public` schema + // earlier in the run. Once each package got a private schema that borrowed + // state vanished and the traversal 500'd -- the isolation did not break + // these tests, it revealed what they had been leaning on. + // + // Only id and tenant_id are read by the join, so the shapes stay minimal. + `CREATE TABLE customer (id uuid PRIMARY KEY, tenant_id uuid)`, + `CREATE TABLE entity_view (id uuid PRIMARY KEY, tenant_id uuid)`, + `CREATE TABLE dashboard (id uuid PRIMARY KEY, tenant_id uuid)`, + `CREATE TABLE device_profile (id uuid PRIMARY KEY, tenant_id uuid)`, + `CREATE TABLE asset_profile (id uuid PRIMARY KEY, tenant_id uuid)`, `CREATE TABLE topology_edge ( tenant_id uuid not null, from_id uuid not null, from_type text not null, to_id uuid not null, to_type text not null, relation_type text not null, diff --git a/flow-core/internal/twinmodel/normalization_sql_test.go b/flow-core/internal/twinmodel/normalization_sql_test.go index 37a29b1..c7b9e1a 100644 --- a/flow-core/internal/twinmodel/normalization_sql_test.go +++ b/flow-core/internal/twinmodel/normalization_sql_test.go @@ -1,6 +1,8 @@ package twinmodel import ( + "flow-core/internal/testdb" + "database/sql" "os" "strings" @@ -14,7 +16,7 @@ func TestSQLNormalizationMatchesGo(t *testing.T) { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open postgres: %v", err) } diff --git a/flow-core/internal/user/auth_handler_test.go b/flow-core/internal/user/auth_handler_test.go index afb9b79..c3c07f1 100644 --- a/flow-core/internal/user/auth_handler_test.go +++ b/flow-core/internal/user/auth_handler_test.go @@ -15,6 +15,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // newTestDB pulls the postgres pointed at by FLOW_TEST_PG_DSN. Tests @@ -26,7 +27,7 @@ func newTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/user/crud_user_authz_test.go b/flow-core/internal/user/crud_user_authz_test.go index 2ac7a1c..dad3b55 100644 --- a/flow-core/internal/user/crud_user_authz_test.go +++ b/flow-core/internal/user/crud_user_authz_test.go @@ -14,6 +14,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Privilege-escalation coverage for POST/PUT /api/user @@ -35,7 +36,7 @@ func newUserAuthzDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - db, err := sql.Open("postgres", dsn) + db, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/flow-core/internal/ws/ws_tenant_test.go b/flow-core/internal/ws/ws_tenant_test.go index 5142c7c..e4a3b66 100644 --- a/flow-core/internal/ws/ws_tenant_test.go +++ b/flow-core/internal/ws/ws_tenant_test.go @@ -15,6 +15,7 @@ import ( authpkg "flow-core/internal/auth" dbpkg "flow-core/internal/db" + "flow-core/internal/testdb" ) // Phase 5b — the WebSocket plane was never covered by the Phase 2 HTTP tenant @@ -48,7 +49,7 @@ func newWSTenantTestDB(t *testing.T) *sql.DB { if dsn == "" { t.Skip("FLOW_TEST_PG_DSN not set") } - pool, err := sql.Open("postgres", dsn) + pool, err := sql.Open("postgres", testdb.Scoped(t, dsn)) if err != nil { t.Fatalf("open: %v", err) } diff --git a/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml b/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml new file mode 100644 index 0000000..376bfc3 --- /dev/null +++ b/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml @@ -0,0 +1,226 @@ +{{- $cg := (.Values.monitoring | default dict).consumerGuard | default dict }} +{{- if and (.Values.nats.enabled | default true) ($cg.enabled | default true) }} +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard-scripts + namespace: {{ .Release.Namespace | default "thingsflow" }} + labels: + app: nats-consumer-guard +data: + # check.sh -- liveness guard for every JetStream durable consumer on the platform. + # + # WHY THIS EXISTS. A rolling node maintenance restarted NATS and left three + # Bento consumers (latest-kv, alarms, entity-greptimedb) with dead + # subscriptions. All three kept reporting Running 1/1 for hours; the halt was + # found only when a human ran `nats consumer info` by hand. Bento cannot + # self-report this: every data-plane Deployment already sets livenessProbe + # /ping and readinessProbe /ready (nats-data-plane-bento.yaml), and BOTH stayed + # green while the pods span on `nats: connection closed` with zero reconnection + # attempts (benchmarks/FINDING-twin-state-localization.md). + # + # WHY NOT BACKLOG DEPTH -- read this before "simplifying" the check. Queue depth + # is the WRONG signal and a threshold on it would make this guard useless: + # latest-kv runs max_ack_pending=1 as a deliberate single-writer serialization + # mechanism (values.yaml), so under load it holds a large and often GROWING + # backlog while working perfectly. Measured in a NATS 2.10.26 sandbox: backlog + # 6,729 sustained while the ack floor advanced 61,179 -> 100,526. A depth-based + # guard alerts there. This one does not. + # + # THE TWO SIGNALS, in order: + # 1. push_bound -- all durables are push consumers with an explicit --target + # and --deliver-group (nats.yaml), so the server tracks whether ANYTHING is + # subscribed to the deliver subject right now. It flips within ms of the + # subscription dying. This is the `Active Interest: No interest` seen in the + # incident. Necessary but NOT sufficient: with >1 replica in one deliver + # group, one dead pod still leaves push_bound true. + # 2. ack_floor.consumer_seq movement -- what separates "serialized but + # progressing" from "dead". Requires two samples; NATS 2.10.26 has no + # last_active field on ack_floor (verified against the pinned server), so a + # one-shot version of this check is not available. + # + # consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes cap, + # so retention deleting messages out from under a STALLED consumer drags its + # stream-side floor forward and reads as false progress. + # + # BOTH-SAMPLE RULE. An alert requires the fault in BOTH samples. A helm upgrade + # rolls the Bento pods and leaves a brief window with no subscriber; firing on + # that would alert on every deployment, which is how operators learn to ignore a + # guard (the lesson greptimedb-freshness-guard.yaml already documents). + # + # HEALTH SIGNAL (failed Job IS the alert -- D-07, no Prometheus): + # thingsflow_nats_consumer_progress_ok{stream=,durable=} 1|0 per consumer + # thingsflow_nats_consumer_progress_ok 1|0 aggregate + # Exit: 0 only when every checked consumer is bound and progressing; else 1. + # + # NO SCRAPE INFRA (Pitfall 1): there is no Prometheus scrape target for a Job + # stdout line. Do NOT add a ServiceMonitor, a push-gateway, or a scrape + # annotation -- the metric is consumed from Job logs / exit status. + # + # SCOPE -- say this out loud so nobody reads "everything is guarded" into it: + # the WS journal fan-out (flow-core/internal/ws/journal.go) is a CORE NATS + # subscription with its own resubscribe loop and creates no JetStream consumer + # at all. No consumer guard can see it. + # + # ENV CONTRACT -- the CronJob MUST inject: + # NATS_URL internal NATS URL (nats://[user:pass@]-nats:4222) + # GUARD_EXPECTED comma-separated "stream|durable" pairs the chart intends to exist + # GUARD_IGNORE comma-separated "stream|durable" pairs to skip (explicit, reviewable) + # GUARD_SAMPLE_SLEEP seconds between the two samples + # GUARD_NATS_TIMEOUT per-request timeout for every nats invocation + # + # POSIX /bin/sh only -- no bashisms (no [[ ]], no arrays, no `local`). + check.sh: | + # POSIX /bin/sh only -- no bashisms (no [[ ]], no arrays, no `local`). + set -u + + SLEEP="${GUARD_SAMPLE_SLEEP:-60}" + TMO="${GUARD_NATS_TIMEOUT:-5s}" + EXPECTED="${GUARD_EXPECTED:-}" + IGNORE="${GUARD_IGNORE:-}" + STATE="${GUARD_STATE_DIR:-/tmp/consumer-guard}" + + mkdir -p "$STATE" || exit 2 + ok=1 + + n() { nats --server "$NATS_URL" --timeout "$TMO" "$@" 2>/dev/null; } + key() { printf '%s' "$1_$2" | tr -c 'A-Za-z0-9_' '_'; } + # printf '%s\n', NOT '%s': without the trailing newline `while read` drops the + # LAST element of every list -- silently, and the last element of the rendered + # GUARD_EXPECTED is the alarm-materializer durable, the one consumer with no + # Kubernetes probes of any kind. Caught in the sandbox, not in production. + listify() { printf '%s\n' "$1" | tr ',' '\n' | sed '/^[[:space:]]*$/d'; } + + # Reads one consumer's state as TSV: kind push_bound backlog ack_floor delivered redelivered + probe() { + n consumer info "$1" "$2" --json | jq -r ' + [ (if ((.config.deliver_subject // "") | length) > 0 then "push" else "pull" end), + (.push_bound // false), + ((.num_pending // 0) + (.num_ack_pending // 0)), + (.ack_floor.consumer_seq // 0), + (.delivered.consumer_seq // 0), + (.num_redelivered // 0) + ] | @tsv' 2>/dev/null + } + + # --- fail-safe: unreachable NATS is ONE clear alert, not six confusing ones --- + attempts=0 + until n server check connection >/dev/null 2>&1; do + attempts=$((attempts + 1)) + [ "$attempts" -ge 3 ] && { + echo "thingsflow_nats_consumer_progress_ok 0" + echo "[consumer-guard] ALERT: NATS unreachable at ${NATS_URL} after ${attempts} attempts -- consumer liveness cannot be established (not evidence of health)" + exit 1 + } + sleep 2 + done + + # --- build the check set: live enumeration UNION expected ------------------ + # Live enumeration cannot drift from reality, but is blind by construction to a + # consumer that VANISHED. The render-injected expected set covers exactly that. + : > "$STATE/set" + for s in $(n stream ls --json | jq -r '(. // []) | .[]'); do + for c in $(n consumer ls "$s" --json | jq -r '(. // []) | .[]'); do + echo "${s}|${c}" >> "$STATE/set" + done + done + + listify "$EXPECTED" | while read -r e; do + grep -qxF "$e" "$STATE/set" 2>/dev/null || echo "$e" >> "$STATE/set" + done + + # Ignore list is explicit and reviewable -- never a silent skip. + listify "$IGNORE" | while read -r i; do + [ -n "$i" ] && echo "[consumer-guard] ignoring ${i} (GUARD_IGNORE)" + done + + sort -u "$STATE/set" -o "$STATE/set" + + # --- pass 1 ---------------------------------------------------------------- + # Printed BEFORE the sleep so a Job killed by activeDeadlineSeconds still + # leaves diagnostics rather than an empty log. + echo "[consumer-guard] pass 1 (t=0):" + while IFS= read -r entry; do + st="${entry%%|*}"; du="${entry##*|}" + listify "$IGNORE" | grep -qxF "$entry" && continue + k=$(key "$st" "$du") + probe "$st" "$du" > "$STATE/$k.p1" + if [ ! -s "$STATE/$k.p1" ]; then + echo " ${st}/${du}: UNREADABLE" + else + echo " ${st}/${du}: $(cat "$STATE/$k.p1" | tr '\t' ' ')" + fi + done < "$STATE/set" + + sleep "$SLEEP" + + # --- pass 2 + verdict ------------------------------------------------------ + echo "[consumer-guard] pass 2 (t=${SLEEP}s):" + while IFS= read -r entry; do + st="${entry%%|*}"; du="${entry##*|}" + listify "$IGNORE" | grep -qxF "$entry" && continue + k=$(key "$st" "$du") + probe "$st" "$du" > "$STATE/$k.p2" + + if [ ! -s "$STATE/$k.p2" ] && [ ! -s "$STATE/$k.p1" ]; then + # Deliberately alerts. Not being able to read a consumer is not evidence of + # health, and swallowing it reintroduces the silent blind spot this exists for. + # Requires BOTH samples to fail: a single blip is a round-trip, not a halt. + ok=0 + echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0" + echo " ALERT ${st}/${du}: consumer could not be read (absent, or NATS refused) -- a halt cannot be ruled out" + continue + fi + + # If only ONE sample was readable, fall back to it for both -- a single failed + # round-trip must not be reported as a stall, nor as health. + [ -s "$STATE/$k.p2" ] || cp "$STATE/$k.p1" "$STATE/$k.p2" + [ -s "$STATE/$k.p1" ] || cp "$STATE/$k.p2" "$STATE/$k.p1" + + IFS=' ' read -r kind bound backlog floor delivered redeliv < "$STATE/$k.p2" + IFS=' ' read -r _k1 bound1 _bk1 floor1 _d1 _r1 < "$STATE/$k.p1" + + # Layer 1 -- push interest. One sample. Flips within ms of the subscription + # dying and cannot false-positive on a deliberately serialized consumer. + # Requires the interest to be gone in BOTH samples. A helm upgrade rolls the + # Bento pods and leaves a brief window with no subscriber; a guard that fired + # on that would alert on every deployment, which is how operators learn to + # ignore it. A real stall lasts hours and is unbound in both. + if [ "$kind" = "push" ] && [ "$bound" != "true" ] && [ "$bound1" != "true" ]; then + ok=0 + echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0" + echo " ALERT ${st}/${du}: no subscriber bound to deliver subject in either sample over ${SLEEP}s (backlog ${backlog}) -- the pod is Running but its subscription is dead" + continue + fi + + # Caught up / idle. Nothing waiting means nothing is stuck. + if [ "$backlog" = "0" ]; then + echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 1" + echo " OK ${st}/${du}: caught up (backlog 0)" + continue + fi + + # Layer 2 -- ack floor progress. Queue DEPTH is not the signal: latest-kv runs + # max_ack_pending=1 as a deliberate serialization mechanism and holds a large, + # often growing backlog while working perfectly. What separates "serialized but + # progressing" from "dead" is that the ack floor MOVES. + # + # consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes cap, so + # retention deleting messages under a stalled consumer drags the stream-side + # floor forward and would read as false progress. + if [ "$floor" -gt "$floor1" ] 2>/dev/null; then + echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 1" + echo " OK ${st}/${du}: progressing, ack floor ${floor1} -> ${floor} (backlog ${backlog} is irrelevant)" + continue + fi + + ok=0 + echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0" + echo " ALERT ${st}/${du}: ${backlog} message(s) waiting and the ack floor is frozen at ${floor} over ${SLEEP}s (delivered ${delivered}, redelivered ${redeliv}) -- bound but not draining" + done < "$STATE/set" + + echo "thingsflow_nats_consumer_progress_ok ${ok}" + [ "$ok" -eq 1 ] || exit 1 + echo "[consumer-guard] OK: every consumer is bound and progressing" + +{{- end }} diff --git a/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml b/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml new file mode 100644 index 0000000..4e6955c --- /dev/null +++ b/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml @@ -0,0 +1,138 @@ +{{- $cg := (.Values.monitoring | default dict).consumerGuard | default dict }} +{{- if and (.Values.nats.enabled | default true) ($cg.enabled | default true) }} +{{- $dp := .Values.natsDataPlane | default dict }} +{{- $historyStore := $dp.historyStore | default (.Values.timeseries.store | default "greptimedb") }} +{{- $lkv := $dp.latestKv | default dict }} +{{- $gdb := $dp.greptimedb | default dict }} +{{- $egdb := $dp.entityGreptimedb | default dict }} +{{- $qdb := $dp.questdb | default dict }} +{{- $alm := $dp.alarms | default dict }} +{{- $am := .Values.alarmMaterializer | default dict }} +{{- $sleep := $cg.sampleSeconds | default 60 }} +{{/* + Expected set -- mirrors the SAME gates the Bento Deployments use + (nats-data-plane-bento.yaml) so "expected but absent" is accurate. Live + enumeration cannot drift from reality but is blind by construction to a + consumer that VANISHED; this list is what covers that case. + A generic `range` over .Values.natsDataPlane would be wrong: that map mixes + scalars (enabled, image, natsUrl, rawSubject) with the consumer sub-maps, and + real enablement depends on historyStore, which the sub-maps do not carry. +*/}} +{{- $expected := list }} +{{- if $dp.enabled }} +{{- if ($lkv.enabled | default true) }} +{{- $expected = append $expected (printf "%s|%s" ($lkv.stream | default "TF_RAW") ($lkv.durable | default "thingsflow-latest-kv-durable")) }} +{{- end }} +{{- if and (eq $historyStore "greptimedb") ($gdb.enabled | default true) }} +{{- $expected = append $expected (printf "%s|%s" ($gdb.stream | default "TF_RAW") ($gdb.durable | default "thingsflow-greptimedb-durable")) }} +{{- end }} +{{- if and (eq $historyStore "greptimedb") ($egdb.enabled | default true) }} +{{- $expected = append $expected (printf "%s|%s" ($egdb.stream | default "TF_ENTITY") ($egdb.durable | default "thingsflow-entity-greptimedb-durable")) }} +{{- end }} +{{- if and (eq $historyStore "questdb") ($qdb.enabled | default false) }} +{{- $expected = append $expected (printf "%s|%s" ($qdb.stream | default "TF_RAW") ($qdb.durable | default "thingsflow-questdb-durable")) }} +{{- end }} +{{- if ($alm.enabled | default true) }} +{{- $expected = append $expected (printf "%s|%s" ($alm.stream | default "TF_RAW") ($alm.durable | default "thingsflow-alarms-durable")) }} +{{- end }} +{{- end }} +{{- if ($am.enabled | default true) }} +{{/* + The alarm-materializer durable is created by the Go client's QueueSubscribe + (cmd/alarm-materializer/main.go), NOT by the nats-bootstrap hook, and it lives + under .Values.alarmMaterializer rather than natsDataPlane. Its Deployment has + no probes at all, so it is the same nats.go silent-death class as the Bento + consumers with even less covering it. +*/}} +{{- $expected = append $expected (printf "%s|%s" ($am.intentStream | default "TF_ALARMS") ($am.durable | default "thingsflow-alarm-materializer")) }} +{{- end }} +# +# NATS consumer liveness guard. +# +# Distinct from greptimedb-freshness-guard, which asks whether the STORE is +# receiving anything at all. That guard stays green while latest-kv is dead, +# because a different consumer keeps writing rows to GreptimeDB -- which is +# exactly how three stalled consumers went unnoticed for hours. +# +# Distinct from nats-stream-guard's verify.sh, which owns CONFIG drift +# (deliver_policy / max_deliver / ack_wait / max_ack_pending / deliver_subject). +# This guard owns LIVENESS only. One guard per concern. +# +# Alert-only, deliberately: remediating would mean granting a scheduled Job +# standing patch authority over the data plane (the chart has no RBAC objects at +# all today), and a guard that fixes the problem and exits 0 destroys the very +# alert surface D-07 is built on. During a real NATS outage a self-remediating +# tick would also restart every consumer every 5 minutes, turning a recoverable +# outage into a self-inflicted one -- a stateless CronJob has no memory with +# which to rate-limit itself. +apiVersion: batch/v1 +kind: CronJob +metadata: + name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard + namespace: {{ .Release.Namespace | default "thingsflow" }} + labels: + app: nats-consumer-guard +spec: + schedule: {{ $cg.schedule | default "*/5 * * * *" | quote }} + # Declared EXPLICITLY, not omitted -- a CronJob without `suspend` in the + # manifest can be stopped by hand and no `helm upgrade` will re-enable it + # (the freshness guard was silently suspended for 47 hours). Recovering a + # guard already suspended by `kubectl patch` needs `helm upgrade + # --force-conflicts` ONCE, not `--force`. + suspend: false + concurrencyPolicy: Forbid + startingDeadlineSeconds: 120 + successfulJobsHistoryLimit: 3 + # Failed Jobs ARE the alert surface, so keep more of them than successes. + failedJobsHistoryLimit: 8 + jobTemplate: + spec: + # A failed tick STAYS failed: retrying would paper over the alert. + backoffLimit: 0 + # Must exceed the two-sample window with room for image pull and round + # trips. It CANNOT stay at the freshness guard's 90: a DeadlineExceeded + # Job is indistinguishable from a real alert but carries no metric line. + activeDeadlineSeconds: {{ add $sleep 180 }} + template: + metadata: + labels: + app: nats-consumer-guard + spec: + restartPolicy: Never + containers: + - name: consumer-guard + image: {{ .Values.nats.cliImage | quote }} + imagePullPolicy: {{ .Values.images.pullPolicy }} + env: + # Credentials FIRST: thingsflow.natsURL emits $(NATS_USER)/$(NATS_PASSWORD) + # for Kubernetes to expand, and the vars must already be declared. + # natsURLLiteral is deliberately NOT used -- it renders NO credentials + # when existingSecret is set, which is exactly production. +{{ include "thingsflow.natsAuthEnv" . | indent 12 }} + - name: NATS_URL + value: {{ include "thingsflow.natsURL" . | quote }} + - name: GUARD_EXPECTED + value: {{ join "," $expected | quote }} + - name: GUARD_IGNORE + value: {{ join "," ($cg.ignore | default (list)) | quote }} + - name: GUARD_SAMPLE_SLEEP + value: {{ $sleep | quote }} + - name: GUARD_NATS_TIMEOUT + value: {{ $cg.natsTimeout | default "5s" | quote }} + command: ["/bin/sh", "-c", "sh /scripts/check.sh"] + volumeMounts: + - name: scripts + mountPath: /scripts + resources: + limits: + cpu: "200m" + memory: "128Mi" + requests: + cpu: "20m" + memory: "48Mi" + volumes: + - name: scripts + configMap: + name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard-scripts + defaultMode: 0755 +{{- end }} diff --git a/k8s/helm/thingsflow/templates/nats.yaml b/k8s/helm/thingsflow/templates/nats.yaml index cf3a053..cc46a01 100644 --- a/k8s/helm/thingsflow/templates/nats.yaml +++ b/k8s/helm/thingsflow/templates/nats.yaml @@ -436,6 +436,15 @@ spec: # the cluster overlay where questdb.enabled=false). Shares TF_RAW but owns its own ack floor. converge_consumer {{ $qdb.stream | default "TF_RAW" | quote }} {{ $qdb.durable | default "thingsflow-questdb-durable" | quote }} {{ $qdb.filterSubject | default "tf.ingest.>" | quote }} \ {{ $qdb.deliverGroup | default ($qdb.queueGroup | default "thingsflow-nats-questdb") | quote }} {{ $qdb.maxDeliver | default 500 | quote }} {{ $qdb.ackWait | default "30s" | quote }} {{ $qdb.maxAckPending | default 1024 | quote }} {{ $qdb.deliverSubject | default "tf.deliver.questdb" | quote }} + {{- else }} + {{- $qdbOff := (.Values.natsDataPlane | default dict).questdb | default dict }} + # Disabling questdb removed its Deployment but NOT its durable, which then sat on + # TF_RAW with no subscriber and its own ack floor, growing monotonically to 106k + # pending and never draining (values.yaml documents the live occurrence). On a + # capped, discard-old stream an unattended consumer is not inert: it is a + # permanently unbound durable that the consumer-liveness guard must legitimately + # alert on. Remove it at the source instead of teaching the guard to ignore it. + nats --server "$NATS_URL" consumer rm {{ $qdbOff.stream | default "TF_RAW" | quote }} {{ $qdbOff.durable | default "thingsflow-questdb-durable" | quote }} -f 2>/dev/null || true {{- end }} # TF_LATEST / TF_ALARMS: idempotent create, then converge the PVC cap # (max-bytes/max-age) onto the LIVE stream so no file-backed stream is diff --git a/k8s/helm/thingsflow/values-cluster.example.yaml b/k8s/helm/thingsflow/values-cluster.example.yaml index 7118cf5..8c08cc2 100644 --- a/k8s/helm/thingsflow/values-cluster.example.yaml +++ b/k8s/helm/thingsflow/values-cluster.example.yaml @@ -207,7 +207,11 @@ natsDataPlane: enabled: true latestKv: enabled: true - replicas: 2 + # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge -> + # set per message), serialized by the durable's max_ack_pending=1. See + # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the + # sharded-merge work, not a replica bump. + replicas: 1 resources: limits: cpu: "1500m" diff --git a/k8s/helm/thingsflow/values-demo.yaml b/k8s/helm/thingsflow/values-demo.yaml index c90acdc..44aca22 100644 --- a/k8s/helm/thingsflow/values-demo.yaml +++ b/k8s/helm/thingsflow/values-demo.yaml @@ -23,7 +23,11 @@ rmqttEdge: natsDataPlane: latestKv: - replicas: 2 + # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge -> + # set per message), serialized by the durable's max_ack_pending=1. See + # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the + # sharded-merge work, not a replica bump. + replicas: 1 questdb: replicas: 2 alarms: diff --git a/k8s/helm/thingsflow/values-pilot.example.yaml b/k8s/helm/thingsflow/values-pilot.example.yaml index d2146da..176eedc 100644 --- a/k8s/helm/thingsflow/values-pilot.example.yaml +++ b/k8s/helm/thingsflow/values-pilot.example.yaml @@ -155,7 +155,11 @@ natsDataPlane: historyStore: "greptimedb" latestKv: enabled: true - replicas: 2 + # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge -> + # set per message), serialized by the durable's max_ack_pending=1. See + # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the + # sharded-merge work, not a replica bump. + replicas: 1 resources: limits: cpu: "1500m" diff --git a/k8s/helm/thingsflow/values.yaml b/k8s/helm/thingsflow/values.yaml index 49ae739..45ef86b 100644 --- a/k8s/helm/thingsflow/values.yaml +++ b/k8s/helm/thingsflow/values.yaml @@ -255,6 +255,27 @@ monitoring: schedule: "*/5 * * * *" staleMinutes: 15 + # Consumer liveness guard: fails a CronJob (the alert surface) when any + # JetStream durable is unbound or not draining. This catches the failure the + # freshness guard structurally cannot -- a dead Bento subscription while the + # pod stays Running 1/1 and OTHER consumers keep GreptimeDB fresh. It happened + # to latest-kv, alarms and entity-greptimedb in a single node maintenance. + consumerGuard: + enabled: true + schedule: "*/5 * * * *" + # Seconds between the two samples. The check needs two: the ack-floor + # progress signal has no stateless equivalent on NATS 2.10.26 (there is no + # ack_floor.last_active field on the pinned server). Raising this makes the + # guard more patient with slow consumers; activeDeadlineSeconds tracks it + # automatically (sampleSeconds + 180). + sampleSeconds: 60 + natsTimeout: "5s" + # Explicit, reviewable skip list of "STREAM|durable" pairs -- never a silent + # skip. Prefer REMOVING an orphaned consumer at the source (templates/nats.yaml) + # over ignoring it here: an unattended durable accumulating backlog on a + # capped stream is a real problem, not noise. + ignore: [] + retention: greptimedb: enabled: true diff --git a/mkdocs.yml b/mkdocs.yml index 7b166f4..96e4221 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -103,3 +103,4 @@ nav: - Decisions: - "ADR-0001: Edge store-and-forward backfill": adr/0001-edge-store-and-forward-backfill.md - "ADR-0002: TB UI contract data-fidelity audit": adr/0002-ui-contract-data-fidelity-audit.md + - "ADR-0003: NATS consumer liveness guard": adr/0003-nats-consumer-progress-guard.md diff --git a/tools/python/test_bento_env_contract.py b/tools/python/test_bento_env_contract.py new file mode 100644 index 0000000..f903769 --- /dev/null +++ b/tools/python/test_bento_env_contract.py @@ -0,0 +1,96 @@ +"""Every env var a Bento config requires must be supplied by BOTH deploy paths. + +Why this exists: the latest-kv config gained `max_ack_pending: +${LATEST_KV_MAX_ACK_PENDING}` when that consumer became the single-writer +doc-merge. The Helm template was updated; docker-compose was not. Bento treats a +`${VAR}` with no default as REQUIRED and refuses to start on it -- not with a +warning, and not by falling back: + + Config lint error lint="(1,1) required environment variables were not set: + [LATEST_KV_MAX_ACK_PENDING]" + Shutting down due to linter errors + +So the container never came up at all, the twin-state KV was never populated, +and the fresh-install smoke failed on a symptom three steps downstream ("NATS KV +twin_state did not hold both keys within 60s"). Nothing pointed at the missing +variable. + +The chart and docker-compose mount the SAME config files, so any required +variable has to be satisfied twice. This test makes that contract explicit +rather than leaving it to whoever remembers both call sites. +""" +import pathlib +import re +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[2] +FILES = ROOT / "k8s/helm/thingsflow/files" +COMPOSE = ROOT / "docker/docker-compose-nats.yml" +BENTO_TEMPLATE = ROOT / "k8s/helm/thingsflow/templates/nats-data-plane-bento.yaml" +INGEST_TEMPLATE = ROOT / "k8s/helm/thingsflow/templates/http-ingest.yaml" + +# `${VAR}` is required; `${VAR:-default}` and `${VAR:default}` are not. +REQUIRED = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)\}") + +# Supplied by the runtime rather than by either deploy path. +RUNTIME_PROVIDED = {"BENTO_HTTP_BIND_ADDRESS", "BENTO_HTTP_PORT"} + + +def required_vars(path): + return {v for v in REQUIRED.findall(path.read_text())} - RUNTIME_PROVIDED + + +class BentoEnvContractTest(unittest.TestCase): + def test_every_required_var_is_set_in_both_deploy_paths(self): + compose = COMPOSE.read_text() + charts = BENTO_TEMPLATE.read_text() + if INGEST_TEMPLATE.exists(): + charts += INGEST_TEMPLATE.read_text() + + configs = sorted(FILES.glob("bento-*.yaml")) + self.assertTrue(configs, "no bento configs found — did the path move?") + + for config in configs: + # Only assert against compose for configs compose actually MOUNTS. + # The questdb path is the legacy backend and has no compose service; + # demanding its vars there would be a standing false alarm. If anyone + # adds the service, the mount appears and this starts covering it. + mounted_by_compose = config.name in compose + + for var in sorted(required_vars(config)): + with self.subTest(config=config.name, var=var): + self.assertIn( + var, + charts, + f"{config.name} requires {var}, absent from the chart " + f"templates — the pod will fail Bento's lint and never start", + ) + if mounted_by_compose: + self.assertIn( + var, + compose, + f"{config.name} requires {var}, absent from " + f"docker-compose-nats.yml — the local stack and the " + f"fresh-install smoke will fail Bento's lint and never start", + ) + + def test_latest_kv_serialization_matches_across_both_paths(self): + """The doc-merge value must agree in three places or the bind fails. + + Bento asserts its own max_ack_pending against the durable's, so a + mismatch is not a performance difference — NATS refuses the bind with + "configuration requests max ack pending to be N, but consumer's value is + M" and the consumer never attaches. + """ + compose = COMPOSE.read_text() + self.assertIn('LATEST_KV_MAX_ACK_PENDING: "1"', compose) + # The compose bootstrap must create that durable with a matching 1, so + # max-pending cannot be a single shared value across the consumer loop. + self.assertIn("thingsflow-latest-kv-durable", compose) + self.assertNotIn("--max-pending 1024", compose) + self.assertIn('--max-pending "$6"', compose) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/python/test_consumer_guard_progress.py b/tools/python/test_consumer_guard_progress.py new file mode 100644 index 0000000..ad044ba --- /dev/null +++ b/tools/python/test_consumer_guard_progress.py @@ -0,0 +1,203 @@ +"""The consumer guard must judge progress, not queue depth. + +Why this exists: a rolling node maintenance restarted NATS and left three Bento +consumers (latest-kv, alarms, entity-greptimedb) holding dead subscriptions. +All three reported Running 1/1 for hours -- neither /ping nor /ready detects a +subscription that is dead underneath a live connection -- and the halt was found +only when a human ran `nats consumer info` by hand. + +The obvious detector is a backlog threshold, and it is wrong in the direction +that makes a guard worthless. latest-kv runs max_ack_pending=1 as a deliberate +single-writer serialization mechanism, so under load it holds a large and often +GROWING backlog while working perfectly: measured against a real NATS 2.10.26 +server at a sustained backlog of 6,729 while its ack floor advanced 61,179 -> +100,526. A depth rule alerts there, every tick, forever. + +So the guard checks push_bound (is anything subscribed to the deliver subject) +and whether ack_floor.consumer_seq MOVES between two samples. These tests pin +the properties that are easy to "simplify" away later and that no rendering +error would surface on its own. + +Verified in a NATS 2.10.26 sandbox across all four branches before this test was +written: unbound with 93,000 waiting exits 1; bound but never acking (4,000 +waiting, ack floor frozen) exits 1; an expected-but-absent durable exits 1; and a +serialized consumer with 10,974 waiting and an advancing ack floor exits 0. +""" +import pathlib +import subprocess +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[2] +CHART = ROOT / "k8s/helm/thingsflow" +SCRIPTS = "templates/nats-consumer-guard-scripts.yaml" +CRONJOB = "templates/nats-consumer-guard.yaml" + + +def render(template, *set_args, values=None): + cmd = ["helm", "template", "tf", str(CHART), "--show-only", template] + if values: + cmd += ["-f", str(CHART / values)] + for arg in set_args: + cmd += ["--set", arg] + out = subprocess.run(cmd, capture_output=True, text=True, timeout=180) + if out.returncode != 0: + raise AssertionError(f"helm template failed: {out.stderr}") + return out.stdout + + +class ConsumerGuardScriptTest(unittest.TestCase): + def setUp(self): + self.script = render(SCRIPTS) + + def test_progress_is_measured_on_the_consumer_ack_floor(self): + # consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes + # cap, so retention deleting messages out from under a STALLED consumer + # drags its stream-side floor forward and reads as false progress. + self.assertIn("ack_floor.consumer_seq", self.script) + self.assertNotIn("ack_floor.stream_seq", self.script) + + def test_backlog_alone_never_decides(self): + # Backlog may only be used to short-circuit to HEALTHY (nothing waiting + # means nothing is stuck). It must never be compared against a threshold + # or against its own earlier value -- that is the rule that would report + # a serialized latest-kv under load as dead. + self.assertIn('[ "$backlog" = "0" ]', self.script) + self.assertNotIn("$backlog -gt", self.script) + self.assertNotIn("$backlog1", self.script) + self.assertNotIn("FLOOR", self.script) + + def test_absent_push_bound_is_read_as_false(self): + # push_bound is omitempty: the server OMITS it when no subscriber is + # bound, so it arrives as null. Without the `// false` default a naive + # test would treat the stalled case as healthy -- the exact blind spot. + self.assertIn(".push_bound // false", self.script) + + def test_an_alert_requires_the_fault_in_both_samples(self): + # A helm upgrade rolls the Bento pods and leaves a brief window with no + # subscriber. Firing on that would alert on every deployment. + self.assertIn('[ "$bound" != "true" ] && [ "$bound1" != "true" ]', self.script) + + def test_lists_keep_their_last_element(self): + # printf '%s' without a trailing newline makes `while read` silently drop + # the LAST element of every comma-separated list -- and the last element + # of the rendered expected set is the alarm-materializer durable, the one + # consumer with no Kubernetes probes at all. + self.assertIn("printf '%s\\n' \"$1\" | tr ',' '\\n'", self.script) + + def test_unreadable_consumers_alert_rather_than_pass(self): + # Not being able to read a consumer is not evidence of health. + self.assertIn("a halt cannot be ruled out", self.script) + + def test_unreachable_nats_is_one_alert_not_a_storm(self): + self.assertIn("NATS unreachable", self.script) + self.assertIn("thingsflow_nats_consumer_progress_ok 0", self.script) + + +class ConsumerGuardCronJobTest(unittest.TestCase): + def test_expected_set_tracks_the_history_store(self): + # The expected set must mirror the same gates the Bento Deployments use. + # A generic range over natsDataPlane cannot: that map mixes scalars with + # the consumer sub-maps, and enablement depends on historyStore, which + # the sub-maps do not carry. + gt = render(CRONJOB, "natsDataPlane.historyStore=greptimedb") + self.assertIn("thingsflow-greptimedb-durable", gt) + self.assertIn("thingsflow-entity-greptimedb-durable", gt) + + qdb = render(CRONJOB, "natsDataPlane.historyStore=questdb") + self.assertNotIn("thingsflow-greptimedb-durable", qdb) + self.assertNotIn("thingsflow-entity-greptimedb-durable", qdb) + + def test_the_probeless_alarm_materializer_durable_is_covered(self): + # Created by the Go client, not the bootstrap hook, and living under a + # different values key -- so a values-driven guard would miss it. Its + # Deployment has no probes of any kind. + rendered = render(CRONJOB) + self.assertIn("TF_ALARMS|thingsflow-alarm-materializer", rendered) + + def test_credentials_are_declared_before_the_url_that_expands_them(self): + # thingsflow.natsURL emits $(NATS_USER)/$(NATS_PASSWORD) for Kubernetes + # to expand; the vars must already be declared at that point. + rendered = render(CRONJOB, "nats.auth.enabled=true", "nats.auth.password=x") + self.assertLess( + rendered.index("name: NATS_USER"), rendered.index("name: NATS_URL") + ) + # natsURLLiteral renders NO credentials under existingSecret, which is + # exactly production, so the env-expanding form must be the one used. + self.assertIn("$(NATS_USER)", rendered) + + def test_deadline_outlives_the_two_sample_window(self): + # The freshness guard's 90s would be exceeded by the sleep alone, and a + # DeadlineExceeded Job is indistinguishable from a real alert but carries + # no metric line. + rendered = render(CRONJOB, "monitoring.consumerGuard.sampleSeconds=120") + self.assertIn("activeDeadlineSeconds: 300", rendered) + + def test_a_failed_tick_stays_failed_and_is_kept(self): + rendered = render(CRONJOB) + self.assertIn("backoffLimit: 0", rendered) + self.assertIn("suspend: false", rendered) + self.assertIn("concurrencyPolicy: Forbid", rendered) + + def test_no_scrape_infrastructure_is_introduced(self): + # D-07: the failed Job IS the alert. There is no scrape target for a Job + # stdout line. + rendered = render(CRONJOB) + render(SCRIPTS) + # Narrow assertions: the templates' own comments say "do NOT add a + # ServiceMonitor", so a bare substring check matches the warning itself. + self.assertNotIn("kind: ServiceMonitor", rendered) + self.assertNotIn("prometheus.io/scrape:", rendered) + + +class OrphanDurableTest(unittest.TestCase): + def test_disabling_questdb_removes_its_durable(self): + # An orphaned durable is not inert on a capped, discard-old stream: the + # questdb one grew to 106k pending and never drained. The guard would + # alert on it forever and correctly, so it is removed at the source + # rather than added to an ignore list. + rendered = render("templates/nats.yaml", "natsDataPlane.questdb.enabled=false") + self.assertIn('consumer rm "TF_RAW" "thingsflow-questdb-durable"', rendered) + + +class LatestKvSingleWriterTest(unittest.TestCase): + """latest-kv must never ship with more than one replica. + + The overlays carried `replicas: 2` against the invariant documented in + values.yaml and files/bento-nats-latest-kv.yaml -- in the example files new + deployments copy from, so every fresh cluster inherited it. Production was + running 1, which is why it was a landmine rather than a live fault. + """ + + def test_every_shipped_values_file_keeps_the_single_writer(self): + import yaml + + # values-cluster.yaml is gitignored (the tracked template is + # values-cluster.example.yaml), so it is deliberately not listed here -- + # a test may only depend on files the repo actually ships. + for name in sorted(CHART.glob("values*.yaml")): + with self.subTest(values=name.name): + doc = yaml.safe_load(name.read_text()) or {} + latest = (doc.get("natsDataPlane") or {}).get("latestKv") or {} + if "replicas" not in latest: + continue + self.assertEqual( + latest["replicas"], + 1, + f"{name.name} scales the single-writer doc-merge consumer", + ) + + def test_the_rendered_deployment_is_a_single_writer(self): + import yaml + + rendered = render("templates/nats-data-plane-bento.yaml") + deploys = { + d["metadata"]["name"]: d + for d in yaml.safe_load_all(rendered) + if d and d.get("kind") == "Deployment" + } + latest = next(n for n in deploys if n.endswith("nats-latest-kv")) + self.assertEqual(deploys[latest]["spec"]["replicas"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/python/test_db_test_isolation.py b/tools/python/test_db_test_isolation.py new file mode 100644 index 0000000..38afaa0 --- /dev/null +++ b/tools/python/test_db_test_isolation.py @@ -0,0 +1,162 @@ +"""Database test harnesses must be confined to a throwaway schema. + +Why this exists: 29 harnesses opened FLOW_TEST_PG_DSN and ran +`DROP TABLE IF EXISTS asset CASCADE` (and a dozen siblings) to get a clean +slate. Against the disposable Postgres CI starts as a service container that is +fine. Against any database a human might point that variable at -- a local dev +stack, a shared scratch database, a pilot -- it is a data-loss bug waiting for +one careless export. They also collided with each other; CI hid that by running +`go test -p 1`, so the breakage only reached whoever ran `go test ./...` +locally, where packages run in parallel by default. + +`internal/testdb.Scoped` fixes it by handing back a DSN bound to a private +schema, so the existing destructive DDL resolves inside that schema and can +never reach `public`. The rule this test enforces is the one that is easy to +forget: a NEW harness that opens the raw DSN gets none of that protection, and +nothing about it looks wrong in review. + +Migrating those harnesses also surfaced real hidden coupling: internal/twin's +tests only passed because internal/customer, internal/dashboard and friends had +created `customer`, `dashboard`, `entity_view`, `device_profile` and +`asset_profile` in the shared `public` schema earlier in the run. The isolation +did not break those tests; it revealed what they had been leaning on. +""" +import pathlib +import re +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[2] +FLOW_CORE = ROOT / "flow-core" +HELPER = "internal/testdb" + +# `sql.Open("postgres", )` where the something is a bare dsn-ish +# variable rather than a testdb.Scoped(...) call. +RAW_OPEN = re.compile(r'sql\.Open\(\s*"postgres"\s*,\s*(?!testdb\.Scoped)([A-Za-z_][\w.]*)\s*\)') + + +def code_of(path): + """Source with `//` comment lines removed. + + A substring assertion over raw source matches the very comment that explains + why something must not be done, so every check here reads code only. + """ + return "\n".join( + l for l in path.read_text().splitlines() if not l.lstrip().startswith("//") + ) + + +def test_files(): + for p in sorted(FLOW_CORE.rglob("*_test.go")): + yield p + + +class DBTestIsolationTest(unittest.TestCase): + def test_no_harness_opens_an_unconfined_connection(self): + offenders = [] + for path in test_files(): + text = path.read_text() + if "FLOW_TEST_PG_DSN" not in text: + continue + # The helper's own tests must open an UNCONFINED connection: that is + # how they place a canary table in `public` and prove the confined + # side cannot reach it. Exempting them is not a loophole — without + # the raw connection there is nothing to prove. + if path.parent.name == "testdb": + continue + # A harness that builds its own schema by hand predates the helper + # and is already confined; it just does it the long way. + if "CREATE SCHEMA" in text: + continue + for m in RAW_OPEN.finditer(text): + offenders.append(f"{path.relative_to(ROOT)} -> sql.Open(\"postgres\", {m.group(1)})") + + self.assertEqual( + [], + offenders, + "these harnesses open FLOW_TEST_PG_DSN without confinement, so their " + "DROP TABLE statements reach whatever database the variable points at:\n " + + "\n ".join(offenders) + + f"\n\nWrap the DSN: sql.Open(\"postgres\", testdb.Scoped(t, dsn)) " + f"(package flow-core/{HELPER}).", + ) + + def test_non_postgres_backends_are_not_wrapped(self): + """QuestDB speaks the Postgres wire protocol but has no schemas. + + `CREATE SCHEMA` and `search_path` both fail against it, so wrapping its + DSN in Scoped turns a working test into a failing one. This is not + hypothetical: the mechanical migration did exactly that, and it stayed + invisible locally because FLOW_TEST_QUESTDB_DSN is normally unset and the + test skips — it only surfaced in CI, which does set it. + + Confinement is also unnecessary there: the target is a dedicated QuestDB + instance, not a shared Postgres someone might also be using. + """ + offenders = [] + for path in test_files(): + text = code_of(path) + if "testdb.Scoped(" not in text: + continue + for foreign in ("FLOW_TEST_QUESTDB_DSN", "GREPTIME"): + if foreign in text: + offenders.append(f"{path.relative_to(ROOT)} ({foreign})") + self.assertEqual( + [], + offenders, + "these tests wrap a non-Postgres DSN in testdb.Scoped, which needs " + "CREATE SCHEMA and search_path:\n " + "\n ".join(sorted(set(offenders))), + ) + + def test_the_helper_binds_search_path_as_a_connection_parameter(self): + """Not as a `SET`. + + sql.DB is a pool. `SET search_path` applies only to the connection that + served it, so under any concurrency a later query lands on a connection + still pointed at `public` — the confinement failing open, silently, and + exactly when load makes it matter. As a DSN parameter the server applies + it at connection setup, to every connection the pool opens. + """ + path = FLOW_CORE / HELPER / "testdb.go" + self.assertIn('"search_path="', path.read_text()) + self.assertNotIn("SET search_path", code_of(path)) + + def test_catalogue_queries_are_scoped_to_the_current_schema(self): + """A catalogue query spans the whole database, not the test's schema. + + `SELECT count(*) FROM information_schema.columns WHERE table_name='policy'` + counts that table in EVERY schema, so once packages run concurrently and + each holds its own copy the assertion reports "found 2" for a perfectly + correct migration. `pg_constraint` has the same shape: constraint names + are unique per schema, not per database, and QueryRow silently takes + whichever row comes first. + + Measured before the fix: 2 of 5 parallel runs failed. It never showed up + under CI's `-p 1`, which is exactly why it survived. + """ + offenders = [] + for path in test_files(): + text = path.read_text() + for catalogue in ("information_schema.", "pg_constraint", "pg_indexes"): + if catalogue not in text: + continue + # Every query touching a catalogue must constrain the namespace. + if "current_schema()" not in text: + offenders.append(f"{path.relative_to(ROOT)} ({catalogue})") + self.assertEqual( + [], + offenders, + "these tests query a database-wide catalogue without pinning it to " + "current_schema(), so they see other packages' concurrent schemas:\n " + + "\n ".join(sorted(set(offenders))), + ) + + def test_the_helper_drops_what_it_creates(self): + helper = (FLOW_CORE / HELPER / "testdb.go").read_text() + self.assertIn("CREATE SCHEMA", helper) + self.assertIn("DROP SCHEMA IF EXISTS", helper) + self.assertIn("t.Cleanup", helper) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/smoke/fresh-install-smoke.sh b/tools/smoke/fresh-install-smoke.sh index f376e22..ac79cef 100755 --- a/tools/smoke/fresh-install-smoke.sh +++ b/tools/smoke/fresh-install-smoke.sh @@ -186,19 +186,31 @@ done log "history (GreptimeDB) read OK: $(printf '%s' "$HIST_BODY" | jq -c .)" log "polling NATS KV twin_state bucket directly for both keys" -kv_get() { +# ONE document per device, not one entry per key. The latest-kv consumer became +# a doc-merge (KV get -> merge -> set) writing the whole device state under +# DEVICE..; the old DEVICE...telemetry. +# fan-out no longer exists. Asserting the old shape here failed with "did not +# hold both keys", which reads as a dead pipeline rather than a stale assertion. +kv_doc() { docker compose -f "$COMPOSE_FILE" run --rm nats-box -c \ - "nats --server nats://nats:4222 kv get twin_state 'DEVICE.$TENANT_UUID.$DEVICE_ID.telemetry.$1' --raw" 2>/dev/null + "nats --server nats://nats:4222 kv get twin_state 'DEVICE.$TENANT_UUID.$DEVICE_ID' --raw" 2>/dev/null } kv_found=0 +KV_BODY="" for _ in $(seq 1 $((ROW_TIMEOUT_SECS / 2))); do - if kv_get smoke_http | grep -q "$EPOCH" && kv_get smoke_mqtt | grep -q "$EPOCH"; then + KV_BODY="$(kv_doc)" + # Both keys must be present in the SAME document with the published value — + # that is what proves the merge, not just that something was written. + if printf '%s' "$KV_BODY" | jq -e \ + --arg v "$EPOCH" \ + '(.telemetry.smoke_http.value | tostring | contains($v)) and + (.telemetry.smoke_mqtt.value | tostring | contains($v))' >/dev/null 2>&1; then kv_found=1; break fi sleep 2 done -[ "$kv_found" -eq 1 ] || die "NATS KV twin_state did not hold both keys within ${ROW_TIMEOUT_SECS}s (latest-values pipeline dead?)" -log "latest (NATS KV) bucket holds both keys" +[ "$kv_found" -eq 1 ] || die "NATS KV twin_state doc DEVICE.$TENANT_UUID.$DEVICE_ID did not hold both keys within ${ROW_TIMEOUT_SECS}s (latest-values pipeline dead?); last doc: ${KV_BODY:-}" +log "latest (NATS KV) device doc holds both keys" # --- h. verdict (cleanup runs via the EXIT trap) ----------------------------- log "PASS — fresh install serves login, device identity, HTTP edge, MQTT edge, GreptimeDB history, and the NATS KV latest pipeline"