diff --git a/.github/workflows/build-enterprise.yml b/.github/workflows/build-enterprise.yml index dfcef8153e..61d1586c88 100644 --- a/.github/workflows/build-enterprise.yml +++ b/.github/workflows/build-enterprise.yml @@ -306,8 +306,8 @@ jobs: path: frontend/playwright-report/ retention-days: 7 - # Multi-node regression: builds + seeds the clustered stack (testing/compose/docker-compose-multinode.yml) - # and runs behave features/multinode. Licence-gated, so it runs after the Playwright job (not in parallel). + # Multi-node regression: builds + seeds the clustered stack once per Valkey topology and runs behave + # features/multinode. Licence-gated, so it runs after the Playwright job (not in parallel). multinode-e2e: needs: [pick, playwright-e2e-enterprise] # Nightly cron + manual dispatch only (heavy build), fork-gated for the licence secret. @@ -316,12 +316,19 @@ jobs: && (github.event_name == 'schedule' || github.event_name == 'workflow_dispatch') runs-on: ${{ needs.pick.outputs.is_fork == 'true' && 'ubuntu-latest' || format('depot-ubuntu-24.04-{0}', inputs.depot_cores || '8') }} timeout-minutes: 60 + strategy: + # One leg per Valkey topology. fail-fast off so a sentinel break still reports cluster. + fail-fast: false + matrix: + valkey: [standalone, sentinel, cluster] env: PREMIUM_KEY: ${{ secrets.PREMIUM_KEY_ENTERPRISE }} PREMIUM_ENABLED: "true" SYSTEM_ENABLEANALYTICS: "false" DEPOT_TOKEN: ${{ secrets.DEPOT_TOKEN }} - MN_COMPOSE: docker-compose-multinode.yml + # Unquoted at every use site so it word-splits into repeated -f flags. The topology overlay + # must come last: it overrides valkey.command and compose REPLACES command. + MN_FILES: -f docker-compose-multinode.yml ${{ matrix.valkey != 'standalone' && format('-f docker-compose-multinode.valkey-{0}.yml', matrix.valkey) || '' }} steps: - name: Harden Runner uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 @@ -341,11 +348,11 @@ jobs: uv sync --project engine --locked --group cucumber - name: Build the multi-node image working-directory: testing/compose - run: docker compose -f "$MN_COMPOSE" build + run: docker compose $MN_FILES build - name: Bring up the cluster and wait for both nodes healthy working-directory: testing/compose run: | - docker compose -f "$MN_COMPOSE" up -d + docker compose $MN_FILES up -d for i in $(seq 1 90); do h1=$(docker inspect -f '{{.State.Health.Status}}' multinode-stirling-1 2>/dev/null || echo starting) h2=$(docker inspect -f '{{.State.Health.Status}}' multinode-stirling-2 2>/dev/null || echo starting) @@ -353,11 +360,11 @@ jobs: sleep 5 done echo "::error::nodes did not become healthy" - docker compose -f "$MN_COMPOSE" logs --tail=200 stirling-1 stirling-2 + docker compose $MN_FILES logs --tail=200 stirling-1 stirling-2 exit 1 - name: Seed the cluster (teams, users, S3 connection, policy) working-directory: testing/compose - run: docker compose -f "$MN_COMPOSE" --profile seed run --rm seed + run: docker compose $MN_FILES --profile seed run --rm seed - name: Run multi-node regression (implemented guarantees) working-directory: testing/cucumber # -e overrides behave.ini's exclusion of features/multinode; ~@known_gap skips any tracked-gap scenarios. @@ -368,8 +375,8 @@ jobs: - name: Dump node logs on failure if: failure() working-directory: testing/compose - run: docker compose -f "$MN_COMPOSE" logs --tail=400 stirling-1 stirling-2 + run: docker compose $MN_FILES logs --tail=400 stirling-1 stirling-2 - name: Tear down if: always() working-directory: testing/compose - run: docker compose -f "$MN_COMPOSE" --profile seed down -v --remove-orphans + run: docker compose $MN_FILES --profile seed down -v --remove-orphans diff --git a/testing/compose/docker-compose-multinode.betterdb.yml b/testing/compose/docker-compose-multinode.betterdb.yml new file mode 100644 index 0000000000..62486d79b1 --- /dev/null +++ b/testing/compose/docker-compose-multinode.betterdb.yml @@ -0,0 +1,117 @@ +# BetterDB observability over the Valkey backplane plus an on-demand load generator. Dashboard: http://localhost:3001 +# Up: docker compose -f docker-compose-multinode.yml -f docker-compose-multinode.betterdb.yml up -d (load: append --profile load run --rm loadgen) + +services: + # Same Valkey, retuned so the observability panels have data on a local box. + valkey: + command: + - valkey-server + - --save + - "" + - --appendonly + - "no" + # Default 10000us never trips locally, so the slowlog stays empty; 500us shows real command spread. + - --slowlog-log-slower-than + - "500" + - --slowlog-max-len + - "1024" + # 0 disables latency monitoring; 50ms populates LATENCY HISTORY/LATEST. + - --latency-monitor-threshold + - "50" + - --maxmemory-policy + - "noeviction" + ports: + - "6379:6379" # host access for valkey-cli inspection + + # One-shot: BetterDB runs as UID 1001 and cannot write a root-owned volume. + betterdb-init: + image: alpine:3.20 + container_name: multinode-betterdb-init + command: ["sh", "-c", "chown -R 1001:1001 /data && echo 'betterdb data dir ready'"] + volumes: + - betterdb-data:/data + networks: + - stirling-multinode + + # One-shot: its own database on the stack's Postgres, so BetterDB history survives restarts. + betterdb-db-init: + image: postgres:17-alpine + container_name: multinode-betterdb-db-init + depends_on: + postgres: + condition: service_healthy + environment: + PGHOST: postgres + PGUSER: stirling + PGPASSWORD: stirling + PGDATABASE: stirling + command: + - sh + - -c + - psql -tAc "SELECT 1 FROM pg_database WHERE datname='betterdb'" | grep -q 1 || psql -c "CREATE DATABASE betterdb" + networks: + - stirling-multinode + + betterdb: + image: betterdb/monitor:latest + container_name: multinode-betterdb + restart: unless-stopped + depends_on: + valkey: + condition: service_healthy + betterdb-init: + condition: service_completed_successfully + betterdb-db-init: + condition: service_completed_successfully + environment: + DB_HOST: valkey + DB_PORT: "6379" + DB_TYPE: valkey + # postgres (not the default memory store) so history survives a restart; sqlite is not + # compiled into the published image despite the docs listing it. + STORAGE_TYPE: postgres + STORAGE_URL: "postgresql://stirling:stirling@postgres:5432/betterdb" + BETTERDB_DATA_DIR: /app/data + # Encrypts stored connection secrets at rest; test-only value. + ENCRYPTION_KEY: "multinode-betterdb-test-encryption-key" + ANOMALY_DETECTION_ENABLED: "true" + ANOMALY_POLL_INTERVAL_MS: "1000" + AUDIT_POLL_INTERVAL_MS: "15000" + CLIENT_ANALYTICS_POLL_INTERVAL_MS: "15000" + KEY_ANALYTICS_INTERVAL_MS: "60000" + # No phoning home from a local test stack. + BETTERDB_TELEMETRY: "false" + ports: + - "3001:3001" + volumes: + - betterdb-data:/app/data + # The image's own healthcheck probes "localhost", which resolves to ::1 while the server + # binds IPv4 only, so it always reports unhealthy. Pin it to 127.0.0.1. + healthcheck: + test: ["CMD-SHELL", "wget -q -O- http://127.0.0.1:3001/api/health >/dev/null || exit 1"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 20s + networks: + - stirling-multinode + + # Profile-gated so `up` skips it; run on demand to drive traffic through the LB. + loadgen: + image: python:3.12-alpine + container_name: multinode-loadgen + profiles: ["load"] + environment: + BASE_URL: "http://nginx:8080" + DURATION_SECONDS: "${DURATION_SECONDS:-300}" + CONCURRENCY: "${CONCURRENCY:-24}" + ASYNC_RATIO: "${ASYNC_RATIO:-0.35}" + USER_COUNT: "${USER_COUNT:-40}" + volumes: + - ./multinode/loadgen.py:/loadgen.py:ro + command: ["python3", "-u", "/loadgen.py"] + networks: + - stirling-multinode + +volumes: + betterdb-data: diff --git a/testing/compose/docker-compose-multinode.valkey-cluster.yml b/testing/compose/docker-compose-multinode.valkey-cluster.yml new file mode 100644 index 0000000000..062529655d --- /dev/null +++ b/testing/compose/docker-compose-multinode.valkey-cluster.yml @@ -0,0 +1,133 @@ +# Must come LAST on the command line: it overrides valkey.command and compose REPLACES command. +# Host tooling cannot follow MOVED on Docker Desktop - run valkey-cli from an in-network container. + +x-valkey-cluster-node: &valkey-cluster-node + image: valkey/valkey:8-alpine + restart: unless-stopped + command: + - valkey-server + - --save + - "" + - --appendonly + - "no" + # nodes.conf in the container layer, not a volume: a persisted membership file plus reshuffled + # container IPs wedges the cluster on recreate. + - --dir + - /tmp + - --cluster-enabled + - "yes" + - --cluster-config-file + - nodes.conf + - --cluster-node-timeout + - "5000" + # No --cluster-announce-ip: a service name is invalid and 127.0.0.1 breaks every in-network + # client with unroutable MOVED targets. Nodes must announce their own container IP. + - --slowlog-log-slower-than + - "500" + - --slowlog-max-len + - "1024" + - --latency-monitor-threshold + - "50" + - --maxmemory-policy + - "noeviction" + # Liveness only. A node answers PING long before it owns any slots, so this cannot gate the app - + # valkey-cluster-init does that. + healthcheck: + test: ["CMD-SHELL", "valkey-cli ping | grep -q PONG"] + interval: 3s + timeout: 5s + retries: 30 + networks: + - stirling-multinode + +services: + # Node 1 keeps the base container_name (multinode-valkey) and the 'valkey' DNS name so existing + # docker exec / `valkey-cli -h valkey` steps still resolve. Note they now see ONE SHARD only. + valkey: + <<: *valkey-cluster-node + + valkey-2: + <<: *valkey-cluster-node + container_name: multinode-valkey-2 + valkey-3: + <<: *valkey-cluster-node + container_name: multinode-valkey-3 + valkey-4: + <<: *valkey-cluster-node + container_name: multinode-valkey-4 + valkey-5: + <<: *valkey-cluster-node + container_name: multinode-valkey-5 + valkey-6: + <<: *valkey-cluster-node + container_name: multinode-valkey-6 + + # Creates against resolved IPs: `--cluster create` bakes whatever it is given into the membership + # table, and IPs are what the nodes gossip anyway. Idempotent, so repeated `up` is safe. + valkey-cluster-init: + image: valkey/valkey:8-alpine + container_name: multinode-valkey-cluster-init + restart: "no" + depends_on: + valkey: {condition: service_healthy} + valkey-2: {condition: service_healthy} + valkey-3: {condition: service_healthy} + valkey-4: {condition: service_healthy} + valkey-5: {condition: service_healthy} + valkey-6: {condition: service_healthy} + entrypoint: + - /bin/sh + - -c + - | + set -e + NODES="valkey valkey-2 valkey-3 valkey-4 valkey-5 valkey-6" + # cluster_state flips to ok before every slot is served, so gate on the slot count too. + formed() { valkey-cli -h "$$1" cluster info 2>/dev/null | tr -d '\r' | grep -q '^cluster_state:ok' \ + && valkey-cli -h "$$1" cluster info 2>/dev/null | tr -d '\r' | grep -q '^cluster_slots_ok:16384'; } + if formed valkey; then + echo "cluster already formed"; valkey-cli -h valkey cluster info | head -3; exit 0 + fi + ADDRS="" + for n in $$NODES; do + IP=$$(getent hosts $$n | awk '{print $$1; exit}') + [ -n "$$IP" ] || { echo "FATAL: cannot resolve $$n"; exit 1; } + echo "$$n -> $$IP" + ADDRS="$$ADDRS $$IP:6379" + done + echo "creating cluster on:$$ADDRS" + valkey-cli --cluster create $$ADDRS --cluster-replicas 1 --cluster-yes + # `--cluster create` returns before every node has the full slot map. Gate on all six + # agreeing, otherwise the first app command can hit CLUSTERDOWN. + for i in $$(seq 1 30); do + ok=1 + for n in $$NODES; do + formed $$n || ok=0 + done + [ "$$ok" = "1" ] && { echo "cluster_state:ok + 16384/16384 slots on all 6 nodes"; exit 0; } + echo "waiting for slot propagation..."; sleep 2 + done + echo "FATAL: cluster did not reach cluster_state:ok with all 16384 slots served"; exit 1 + networks: + - stirling-multinode + + # service_completed_successfully on the init is the real readiness gate; valkey's own healthcheck + # only proves the process is alive. + stirling-1: + depends_on: + valkey-cluster-init: + condition: service_completed_successfully + environment: + CLUSTER_VALKEY_MODE: "cluster" + CLUSTER_VALKEY_NODES: "valkey:6379,valkey-2:6379,valkey-3:6379,valkey-4:6379,valkey-5:6379,valkey-6:6379" + CLUSTER_VALKEY_MAXREDIRECTS: "3" + CLUSTER_VALKEY_TOPOLOGYREFRESHMS: "30000" + + stirling-2: + depends_on: + valkey-cluster-init: + condition: service_completed_successfully + environment: + CLUSTER_VALKEY_MODE: "cluster" + CLUSTER_VALKEY_NODES: "valkey:6379,valkey-2:6379,valkey-3:6379,valkey-4:6379,valkey-5:6379,valkey-6:6379" + CLUSTER_VALKEY_MAXREDIRECTS: "3" + CLUSTER_VALKEY_TOPOLOGYREFRESHMS: "30000" diff --git a/testing/compose/docker-compose-multinode.valkey-sentinel.yml b/testing/compose/docker-compose-multinode.valkey-sentinel.yml new file mode 100644 index 0000000000..7770615efa --- /dev/null +++ b/testing/compose/docker-compose-multinode.valkey-sentinel.yml @@ -0,0 +1,115 @@ +# Valkey Sentinel HA overlay: 1 primary + 2 replicas + 3 sentinels. Up: ./start-multinode-test.sh --valkey sentinel +# Never overrides valkey.command, so the base 'valkey' service survives and this composes with any other overlay in any order. + +x-sentinel-deps: &sentinel-deps + sentinel-1: + condition: service_healthy + sentinel-2: + condition: service_healthy + sentinel-3: + condition: service_healthy + +services: + # No --replica-announce-ip: a hostname makes sentinel discard the replica, and + # 'resolve-hostnames yes' is not a workaround (see the TILT note on the sentinel below). + valkey-replica-1: &valkey-replica + image: valkey/valkey:8-alpine + container_name: multinode-valkey-replica-1 + restart: unless-stopped + depends_on: + valkey: + condition: service_healthy + command: + - valkey-server + - --save + - "" + - --appendonly + - "no" + # Runtime files in the container layer, not the image's anonymous /data volume, so a + # `compose up` recreate always starts from clean state. + - --dir + - /tmp + - --replicaof + - valkey + - "6379" + - --replica-read-only + - "yes" + healthcheck: + test: ["CMD-SHELL", "valkey-cli ping | grep -q PONG"] + interval: 3s + timeout: 5s + retries: 30 + networks: + - stirling-multinode + + valkey-replica-2: + <<: *valkey-replica + container_name: multinode-valkey-replica-2 + + # No host-mounted conf: sentinel rewrites it at runtime, so a read-only mount kills boot and a + # read-write one replays stale known-replica IPs. The entrypoint generates it into /tmp instead. + sentinel-1: &valkey-sentinel + image: valkey/valkey:8-alpine + container_name: multinode-valkey-sentinel-1 + restart: unless-stopped + depends_on: + valkey: + condition: service_healthy + # Monitor a RESOLVED IP, never the name 'valkey': killing the primary drops its DNS record, + # sentinel's synchronous lookup stalls its event loop, and TILT suspends failover. + entrypoint: + - /bin/sh + - -c + - | + set -e + for i in $$(seq 1 30); do + PRIMARY_IP=$$(getent hosts valkey | awk '{print $$1; exit}') + [ -n "$$PRIMARY_IP" ] && break + echo "waiting for DNS: valkey"; sleep 2 + done + [ -n "$$PRIMARY_IP" ] || { echo "FATAL: could not resolve 'valkey'"; exit 1; } + echo "sentinel monitoring primary valkey -> $$PRIMARY_IP" + cat > /tmp/sentinel.conf < bytes: + """Build a valid multi-page PDF from raw syntax so the corpus needs no PDF library.""" + objs = [] # obj number -> body bytes, 1-indexed by position + + font_obj = 3 + pages * 2 # catalog=1, pages=2, then page/content pairs + kids = " ".join(f"{3 + i * 2} 0 R" for i in range(pages)) + objs.append(b"<>") + objs.append(f"<>".encode()) + + for i in range(pages): + content_num = 4 + i * 2 + objs.append( + f"<>>>>>".encode() + ) + lines = [b"BT /F1 14 Tf 54 740 Td (Stirling load-test page " + str(i + 1).encode() + b") Tj ET"] + for n in range(filler_lines): + y = 710 - (n * 18) % 640 + text = f"lorem ipsum dolor sit amet {uuid.uuid4().hex}" + lines.append(f"BT /F1 9 Tf 54 {y} Td ({text}) Tj ET".encode()) + stream = b"\n".join(lines) + objs.append(b"<>stream\n" + stream + b"\nendstream") + + objs.append(b"<>") + + out = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n") + offsets = [] + for idx, body in enumerate(objs, start=1): + offsets.append(len(out)) + out += f"{idx} 0 obj".encode() + body + b"endobj\n" + + xref_at = len(out) + out += f"xref\n0 {len(objs) + 1}\n".encode() + out += b"0000000000 65535 f \n" + for off in offsets: + out += f"{off:010d} 00000 n \n".encode() + out += f"trailer<>\nstartxref\n{xref_at}\n%%EOF\n".encode() + return bytes(out) + + +def build_corpus(): + print("==> Building PDF corpus...", flush=True) + corpus = { + "tiny": make_pdf(2, 6), + "small": make_pdf(8, 14), + "medium": make_pdf(30, 20), + "large": make_pdf(80, 26), + } + for name, data in corpus.items(): + print(f" {name:7s} {len(data) / 1024:8.1f} KB", flush=True) + return corpus + + +# ---------------------------------------------------------------- HTTP plumbing +def encode_multipart(fields, files): + """fields: dict of scalars. files: list of (fieldname, filename, bytes) - repeats allowed.""" + boundary = uuid.uuid4().hex + body = bytearray() + for key, value in fields.items(): + body += f"--{boundary}\r\n".encode() + body += f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode() + body += f"{value}\r\n".encode() + for key, filename, data in files: + body += f"--{boundary}\r\n".encode() + body += ( + f'Content-Disposition: form-data; name="{key}"; filename="{filename}"\r\n' + f"Content-Type: application/pdf\r\n\r\n" + ).encode() + body += data + b"\r\n" + body += f"--{boundary}--\r\n".encode() + return bytes(body), f"multipart/form-data; boundary={boundary}" + + +def request(method, path, token=None, body=None, content_type=None, timeout=180): + url = path if path.startswith("http") else BASE_URL + path + req = urllib.request.Request(url, data=body, method=method) + if token: + req.add_header("Authorization", f"Bearer {token}") + if content_type: + req.add_header("Content-Type", content_type) + try: + with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp: + return resp.status, resp.read(), dict(resp.headers) + except urllib.error.HTTPError as exc: + return exc.code, exc.read(), dict(exc.headers) + except Exception as exc: # connection reset, timeout, DNS + return 0, str(exc).encode(), {} + + +def record(op, status, elapsed_ms, size, headers): + with stats_lock: + entry = stats[op] + if 200 <= status < 300: + entry["ok"] += 1 + entry["bytes"] += size + else: + entry["err"] += 1 + errors[f"{op} -> {status}"] += 1 + entry["ms"] += elapsed_ms + served = headers.get("X-Served-By") + if served: + nodes_seen[served] += 1 + + +# ---------------------------------------------------------------- auth +def login(username, password): + body = json.dumps({"username": username, "password": password}).encode() + status, data, _ = request("POST", "/api/v1/auth/login", body=body, content_type="application/json") + if status != 200: + return None + try: + return json.loads(data).get("session", {}).get("access_token") or json.loads(data).get("access_token") + except Exception: + return None + + +def collect_tokens(): + print("==> Logging in...", flush=True) + tokens = [] + admin = login(ADMIN_USER, ADMIN_PASS) + if admin: + tokens.append(("admin", admin)) + for i in range(1, USER_COUNT + 1): + user = f"user{i:02d}@stirling.test" + tok = login(user, USER_PASSWORD) + if tok: + tokens.append((user, tok)) + print(f" {len(tokens)} authenticated principals", flush=True) + return tokens + + +# ---------------------------------------------------------------- workload +def op_rotate(corpus): + return "/api/v1/general/rotate-pdf", {"angle": random.choice([90, 180, 270])}, [ + ("fileInput", "load.pdf", corpus[random.choice(["tiny", "small", "medium"])]) + ] + + +def op_merge(corpus): + files = [("fileInput", f"m{i}.pdf", corpus[random.choice(["tiny", "small"])]) for i in range(random.randint(2, 4))] + return "/api/v1/general/merge-pdfs", {"sortType": "orderProvided"}, files + + +def op_compress(corpus): + return "/api/v1/misc/compress-pdf", {"optimizeLevel": random.choice([1, 2, 3])}, [ + ("fileInput", "compress.pdf", corpus[random.choice(["medium", "large"])]) + ] + + +def op_remove_pages(corpus): + return "/api/v1/general/remove-pages", {"pageNumbers": "1,3"}, [ + ("fileInput", "rm.pdf", corpus[random.choice(["small", "medium"])]) + ] + + +def op_split(corpus): + return "/api/v1/general/split-pages", {"pageNumbers": "2,4"}, [ + ("fileInput", "split.pdf", corpus[random.choice(["small", "medium"])]) + ] + + +def op_page_numbers(corpus): + return "/api/v1/misc/add-page-numbers", { + "customMargin": "medium", + "position": 8, + "startingNumber": 1, + "pagesToNumber": "all", + "customText": "{n} of {total}", + }, [("fileInput", "num.pdf", corpus[random.choice(["small", "medium"])])] + + +def op_flatten(corpus): + return "/api/v1/misc/flatten", {"flattenOnlyForms": "false"}, [ + ("fileInput", "flat.pdf", corpus[random.choice(["tiny", "small"])]) + ] + + +def op_metadata(corpus): + return "/api/v1/misc/update-metadata", { + "deleteAll": "false", + "author": "load-test", + "title": f"run-{uuid.uuid4().hex[:8]}", + }, [("fileInput", "meta.pdf", corpus[random.choice(["tiny", "small"])])] + + +# (weight, name, builder) - heavier tools are rarer so throughput stays high. +WORKLOAD = [ + (22, "rotate", op_rotate), + (14, "merge", op_merge), + (10, "compress", op_compress), + (14, "remove-pages", op_remove_pages), + (12, "split-pages", op_split), + (12, "add-page-numbers", op_page_numbers), + (8, "flatten", op_flatten), + (8, "update-metadata", op_metadata), +] +WEIGHTS = [w for w, _, _ in WORKLOAD] + + +def poll_job(job_id, token, deadline): + """Poll until the job completes. A 410 means we hit a non-owner node; retry re-routes us.""" + while time.time() < deadline: + status, data, _ = request("GET", f"/api/v1/general/job/{job_id}", token=token, timeout=30) + if status == 410: + with stats_lock: + job_stats["sticky_410"] += 1 + time.sleep(0.3) + continue + if status != 200: + return False + try: + payload = json.loads(data) + except Exception: + return False + result = payload.get("jobResult", payload) + if result.get("complete"): + return result.get("error") is None + time.sleep(0.4) + return False + + +def worker(worker_id, corpus, tokens): + rng = random.Random(worker_id * 7919) + # Stagger startup so all workers do not slam the LB in the same instant. + time.sleep(rng.uniform(0, RAMP_SECONDS)) + while time.time() < stop_at: + _, token = rng.choice(tokens) + _, name, builder = rng.choices(WORKLOAD, weights=WEIGHTS, k=1)[0] + path, fields, files = builder(corpus) + use_async = rng.random() < ASYNC_RATIO + if use_async: + path += "?async=true" + body, content_type = encode_multipart(fields, files) + + started = time.time() + status, data, headers = request("POST", path, token=token, body=body, content_type=content_type) + elapsed = (time.time() - started) * 1000 + label = f"{name}{' (async)' if use_async else ''}" + record(label, status, elapsed, len(data), headers) + + if use_async and 200 <= status < 300: + try: + job_id = json.loads(data).get("jobId") + except Exception: + job_id = None + if job_id: + with stats_lock: + job_stats["submitted"] += 1 + if poll_job(job_id, token, time.time() + 240): + with stats_lock: + job_stats["completed"] += 1 + else: + with stats_lock: + job_stats["failed"] += 1 + + # Cheap reads between jobs: extra request volume and node-registry reads. + if rng.random() < 0.3: + s, d, h = request("GET", "/api/v1/info/status", token=token, timeout=20) + record("info/status", s, 0, len(d), h) + + +def progress_printer(): + last = 0 + while time.time() < stop_at: + time.sleep(15) + with stats_lock: + total = sum(v["ok"] + v["err"] for v in stats.values()) + ok = sum(v["ok"] for v in stats.values()) + mb = sum(v["bytes"] for v in stats.values()) / 1024 / 1024 + jobs = dict(job_stats) + remaining = max(0, int(stop_at - time.time())) + rate = (total - last) / 15.0 + last = total + print( + f" [{remaining:4d}s left] {total:6d} reqs {ok:6d} ok {rate:5.1f} req/s " + f"{mb:7.1f} MB down async {jobs['completed']}/{jobs['submitted']}", + flush=True, + ) + + +def report(): + print("\n" + "=" * 78) + print(" LOAD TEST SUMMARY") + print("=" * 78) + print(f"{'operation':24s} {'ok':>7s} {'err':>6s} {'avg ms':>9s} {'MB down':>9s}") + print("-" * 78) + total_ok = total_err = 0 + for op in sorted(stats): + entry = stats[op] + calls = entry["ok"] + entry["err"] + avg = entry["ms"] / calls if calls else 0 + total_ok += entry["ok"] + total_err += entry["err"] + print( + f"{op:24s} {entry['ok']:7d} {entry['err']:6d} {avg:9.0f} {entry['bytes'] / 1024 / 1024:9.1f}" + ) + print("-" * 78) + print(f"{'TOTAL':24s} {total_ok:7d} {total_err:6d}") + + print("\n Load-balancer spread (X-Served-By):") + for node, count in sorted(nodes_seen.items(), key=lambda kv: -kv[1]): + print(f" {node:24s} {count:7d} responses") + + print("\n Async jobs (these are the Valkey JobStore writes):") + print(f" submitted {job_stats['submitted']} completed {job_stats['completed']}" + f" failed {job_stats['failed']} cross-node 410 re-routes {job_stats['sticky_410']}") + + if errors: + print("\n Top errors:") + for key, count in sorted(errors.items(), key=lambda kv: -kv[1])[:15]: + print(f" {count:6d} {key}") + print("=" * 78, flush=True) + + +def wait_for_app(): + print(f"==> Waiting for {BASE_URL} ...", flush=True) + for _ in range(120): + status, _, _ = request("GET", "/api/v1/info/status", timeout=10) + if status == 200: + print(" app is up", flush=True) + return True + time.sleep(2) + return False + + +def main(): + global stop_at + if not wait_for_app(): + print("app never came up", file=sys.stderr) + return 1 + + corpus = build_corpus() + tokens = collect_tokens() + if not tokens: + print("no logins succeeded - cannot generate authenticated load", file=sys.stderr) + return 1 + + stop_at = time.time() + DURATION + print( + f"\n==> Driving load for {DURATION}s: {CONCURRENCY} workers, " + f"{len(tokens)} users, {int(ASYNC_RATIO * 100)}% async\n", + flush=True, + ) + ticker = threading.Thread(target=progress_printer, daemon=True) + ticker.start() + with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool: + for i in range(CONCURRENCY): + pool.submit(worker, i, corpus, tokens) + report() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/testing/compose/multinode/seed.sh b/testing/compose/multinode/seed.sh index 3fc777b193..9ccbff0520 100644 --- a/testing/compose/multinode/seed.sh +++ b/testing/compose/multinode/seed.sh @@ -1,7 +1,6 @@ #!/bin/sh -# Seeds a running multi-node stack: 4 teams, ~40 users, an S3 connection, a scheduled S3 policy, and a webhook source if the build supports it. -# Auth uses the Bearer JWT from the login response body (not a cookie) since the global API key can't create teams. -# Idempotent-ish: re-running skips existing teams/users; each step is best-effort and logs failures without aborting. +# Seeds a running stack (teams, users, S3 connection + policy). Best-effort and idempotent-ish. +# Auth uses the Bearer JWT from the login body, not a cookie - the global API key can't create teams. set -u BASE_URL="${BASE_URL:-http://localhost:8080}" diff --git a/testing/compose/multinode/valkey-topology.sh b/testing/compose/multinode/valkey-topology.sh new file mode 100644 index 0000000000..eb9c144b53 --- /dev/null +++ b/testing/compose/multinode/valkey-topology.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Sourced by start-multinode-test.sh and run-multinode-regression.sh. + +# Prints the `docker compose -f ...` prefix for a topology; returns 1 on an unknown one. Overlays must +# follow the base file - they override valkey.command / depends_on, and compose REPLACES command. +compose_cmd_for_topology() { + base="docker compose -f docker-compose-multinode.yml" + case "$1" in + standalone) echo "$base" ;; + sentinel) echo "$base -f docker-compose-multinode.valkey-sentinel.yml" ;; + cluster) echo "$base -f docker-compose-multinode.valkey-cluster.yml" ;; + *) return 1 ;; + esac +} diff --git a/testing/compose/run-multinode-regression.sh b/testing/compose/run-multinode-regression.sh index 07767d95dd..3056a7d97d 100644 --- a/testing/compose/run-multinode-regression.sh +++ b/testing/compose/run-multinode-regression.sh @@ -1,24 +1,33 @@ #!/usr/bin/env bash -# Runs the multi-node regression suite (behave features/multinode) against the clustered stack: brings it up if needed, runs non-destructive scenarios then @destructive failover ones, and restores any killed node. -# Usage: ./run-multinode-regression.sh [--no-failover] [--no-seed] -# @known_gap scenarios are expected to fail - they mark work not yet done, so a non-zero exit is fine while those are open. +# Usage: ./run-multinode-regression.sh [--valkey standalone|sentinel|cluster] [--no-failover] [--no-seed] +# @known_gap scenarios are expected to fail, so a non-zero exit is fine while those are open. set -uo pipefail cd "$(dirname "$0")" -COMPOSE="docker compose -f docker-compose-multinode.yml" +. ./multinode/valkey-topology.sh + CUKE_DIR="../cucumber" RUN_FAILOVER=1 SEED=1 -for arg in "$@"; do - case "$arg" in - --no-failover) RUN_FAILOVER=0 ;; - --no-seed) SEED=0 ;; +VALKEY_TOPOLOGY="${VALKEY_TOPOLOGY:-standalone}" +while [ "$#" -gt 0 ]; do + case "$1" in + --no-failover) RUN_FAILOVER=0; shift ;; + --no-seed) SEED=0; shift ;; + --valkey) VALKEY_TOPOLOGY="${2:-}"; shift 2 ;; + --valkey=*) VALKEY_TOPOLOGY="${1#*=}"; shift ;; + *) echo "Unknown argument '$1'"; exit 2 ;; esac done -echo "==> Ensuring the multi-node stack is up..." +# Same overlay set as start-multinode-test.sh, so `up -d` here restores killed nodes without +# silently dropping the sentinel/cluster services back to the base standalone Valkey. +COMPOSE=$(compose_cmd_for_topology "$VALKEY_TOPOLOGY") \ + || { echo "Unknown --valkey topology '$VALKEY_TOPOLOGY' (expected standalone|sentinel|cluster)"; exit 2; } + +echo "==> Ensuring the multi-node stack is up (valkey=$VALKEY_TOPOLOGY)..." if ! docker inspect -f '{{.State.Health.Status}}' multinode-stirling-1 2>/dev/null | grep -q healthy; then - ./start-multinode-test.sh $([ "$SEED" = 0 ] && echo --no-seed) || exit 1 + ./start-multinode-test.sh --valkey "$VALKEY_TOPOLOGY" $([ "$SEED" = 0 ] && echo --no-seed) || exit 1 elif [ "$SEED" = 1 ]; then echo " stack already up; seeding (idempotent)..." $COMPOSE --profile seed run --rm seed >/dev/null 2>&1 || echo " (seed reported issues, continuing)" @@ -58,6 +67,7 @@ fi echo echo "============================================================" echo " Regression run complete. Reports: $REPORT_DIR" +echo " Valkey topology: $VALKEY_TOPOLOGY" echo " Exit $rc (non-zero = at least one scenario failed;" echo " @known_gap scenarios are expected to fail - see the report)." echo " Stack left running: http://localhost:8080 (admin / stirling)" diff --git a/testing/compose/start-multinode-test.sh b/testing/compose/start-multinode-test.sh index d842d68019..c1dca19e3b 100644 --- a/testing/compose/start-multinode-test.sh +++ b/testing/compose/start-multinode-test.sh @@ -1,20 +1,33 @@ #!/usr/bin/env bash # Brings up the multi-node stack (Postgres/Valkey/MinIO/2 app nodes/nginx LB), seeds teams/users/an S3 connection/policies, then leaves it running for manual testing. -# Usage: ./start-multinode-test.sh [--no-seed | --down] +# Usage: ./start-multinode-test.sh [--valkey standalone|sentinel|cluster] [--no-seed | --down] set -euo pipefail cd "$(dirname "$0")" -COMPOSE="docker compose -f docker-compose-multinode.yml" +# Valkey topology, layered as a compose overlay so the default (standalone) stack is unchanged. +. ./multinode/valkey-topology.sh +VALKEY_TOPOLOGY="${VALKEY_TOPOLOGY:-standalone}" +DOWN=0 +SEED=1 +while [ "$#" -gt 0 ]; do + case "$1" in + --valkey) VALKEY_TOPOLOGY="${2:-}"; shift 2 ;; + --valkey=*) VALKEY_TOPOLOGY="${1#*=}"; shift ;; + --down) DOWN=1; shift ;; + --no-seed) SEED=0; shift ;; + *) echo "Unknown argument '$1'. Usage: $0 [--valkey standalone|sentinel|cluster] [--no-seed | --down]"; exit 2 ;; + esac +done -if [ "${1:-}" = "--down" ]; then +COMPOSE=$(compose_cmd_for_topology "$VALKEY_TOPOLOGY") \ + || { echo "Unknown --valkey topology '$VALKEY_TOPOLOGY' (expected standalone|sentinel|cluster)"; exit 2; } + +if [ "$DOWN" = "1" ]; then echo "Tearing down multi-node stack + volumes..." $COMPOSE --profile seed down -v --remove-orphans exit 0 fi -SEED=1 -[ "${1:-}" = "--no-seed" ] && SEED=0 - # Cluster mode is licence-gated. Without a valid key the nodes fail the cluster licence gate at boot. if [ -z "${PREMIUM_KEY:-}" ]; then echo "WARNING: PREMIUM_KEY is not set - cluster mode needs a valid enterprise/pro licence key." @@ -24,7 +37,7 @@ fi echo "==> Building the Stirling image (first run compiles the app; be patient)..." $COMPOSE build -echo "==> Starting Postgres + Valkey + MinIO + 2 app nodes + nginx..." +echo "==> Starting Postgres + Valkey ($VALKEY_TOPOLOGY) + MinIO + 2 app nodes + nginx..." $COMPOSE up -d echo "==> Waiting for both app nodes to report healthy..." @@ -45,7 +58,7 @@ fi cat </dev/null | tr -d '\r' | grep -q '^cluster_enabled:1'; then + echo cluster + elif docker inspect multinode-valkey-sentinel-1 >/dev/null 2>&1; then + echo sentinel + else + echo standalone + fi +} +TOPOLOGY=$(detect_topology) + +# Every stirling:* key in any topology: a cluster shards them so --cluster call fans out, plain KEYS is +# the non-cluster fallback. Must be docker exec - `docker run --entrypoint /bin/sh` is mangled by MSYS. +backplane_keys_raw() { + docker exec multinode-valkey valkey-cli --cluster call --cluster-only-masters 127.0.0.1:6379 keys 'stirling:*' 2>/dev/null \ + || docker exec multinode-valkey valkey-cli keys 'stirling:*' 2>/dev/null +} +# Same, minus the "host:port: " prefix a cluster call prepends to every line. +backplane_keys() { + backplane_keys_raw | tr -d '\r' | sed 's/^[A-Za-z0-9_.-]*:[0-9][0-9]*: //' | grep '^stirling:' +} + +# INFO field off the Valkey the app writes to (the primary in every topology). +valkey_info() { valkey_info_on multinode-valkey "$1"; } +valkey_info_on() { docker exec "$1" valkey-cli info "$2" 2>/dev/null | tr -d '\r'; } + +# Every reachable Valkey DATA container (1 / 3 / 6 by topology); sampling only multinode-valkey would +# cover a sixth of a cluster. Sentinels are the monitoring plane, not the backplane, so they are excluded. +valkey_nodes() { + docker ps --format '{{.Names}}' --filter name=multinode-valkey 2>/dev/null | tr -d '\r' \ + | grep -v -e sentinel -e cluster-init | sort | while read -r c; do + docker exec "$c" valkey-cli ping 2>/dev/null | tr -d '\r' | grep -q '^PONG' && echo "$c" + done +} +VALKEY_NODES=$(valkey_nodes) login() { # -> prints the bearer token curl -s -X POST "$LB/api/v1/auth/login" -H 'Content-Type: application/json' \ @@ -73,9 +112,117 @@ else bad "integration list failed (HTTP $lc) - credential key may not be shared across nodes" fi +echo "== 6. Every Valkey connection is attributable (CLIENT SETNAME) ==" +# Census every data node: one container carries a sixth of the connections in a 6-shard cluster. +# Scoped to lib-name=Lettuce - valkey-cli and monitoring agents legitimately have no name. +sampled=0; anon=0; all_names="" +for c in $VALKEY_NODES; do + sampled=$((sampled+1)) + clients=$(docker exec "$c" valkey-cli client list 2>/dev/null | tr -d '\r' | grep 'lib-name=Lettuce') + anon=$(( anon + $(printf '%s\n' "$clients" | grep -c 'name= ') )) + all_names="$all_names +$(printf '%s\n' "$clients" | grep -o 'name=stirling-[^ ]*')" +done +named=$(printf '%s\n' "$all_names" | grep . | sort -u) +distinct=$(printf '%s\n' "$named" | grep -c .) +echo " sampled $sampled Valkey node(s); app connection names: $(printf '%s' "$named" | paste -sd, -)" +if [ "$sampled" -lt 1 ]; then + bad "no reachable Valkey container found - cannot census client names" +else + [ "${anon:-0}" -eq 0 ] && ok "no unnamed app connections across $sampled node(s) (all carry CLIENT SETNAME)" \ + || bad "$anon Lettuce connection(s) have an empty name= - CLIENT SETNAME is not applied" + [ "${distinct:-0}" -ge 2 ] && ok "$distinct distinct stirling-* client names (load attributable per node)" \ + || bad "only ${distinct:-0} distinct stirling-* client name(s) (expected one per app node)" +fi + +echo "== 7. Connection churn is bounded (pooling is on) ==" +# Job-creating traffic only: reads ride the shared native connection, job-store writes need a dedicated +# one (~3 fresh connects per async job unpooled). Summed over all data nodes - a cluster spreads them. +JOBS=200 +CHURN_BUDGET=50 +conns_total() { + total=0 + for c in $VALKEY_NODES; do + n=$(valkey_info_on "$c" stats | sed -n 's/^total_connections_received:\([0-9]*\).*/\1/p') + total=$(( total + ${n:-0} )) + done + echo "$total" +} +vk_count=$(printf '%s\n' "$VALKEY_NODES" | grep -c .) +if [ -z "${jwt:-}" ]; then + skip "no JWT - cannot drive load to measure churn" +elif [ "${vk_count:-0}" -lt 1 ]; then + bad "no reachable Valkey container found - cannot measure connection churn" +else + probe_pdf=$(mktemp -t mn-probe-XXXXXX.pdf 2>/dev/null || echo /tmp/mn-probe.pdf) + printf '%%PDF-1.4\n1 0 obj<>endobj\n2 0 obj<>endobj\n3 0 obj<>endobj\ntrailer<>\n%%%%EOF\n' > "$probe_pdf" + before=$(conns_total) + submitted=0 + for i in $(seq 1 "$JOBS"); do + hc=$(curl -s -o /dev/null -w '%{http_code}' -H "Authorization: Bearer $jwt" \ + -F "fileInput=@$probe_pdf" -F "angle=90" "$LB/api/v1/general/rotate-pdf?async=true") + [ "$hc" = "200" ] && submitted=$((submitted+1)) + done + after=$(conns_total) + rm -f "$probe_pdf" + delta=$(( ${after:-0} - ${before:-0} )) + echo " total_connections_received over $vk_count Valkey node(s): $before -> $after (delta $delta over $submitted/$JOBS async jobs)" + if [ "$submitted" -lt 1 ]; then + bad "no async jobs were accepted - cannot measure connection churn" + elif [ "$delta" -lt "$CHURN_BUDGET" ]; then + ok "only $delta new connections for $submitted async jobs - connections are pooled" + else + bad "$delta new connections for $submitted async jobs - pooling looks disabled (expected < $CHURN_BUDGET)" + fi +fi + +echo "== 8. Every app node registered a heartbeat in the backplane ==" +nodes_registered=$(backplane_keys | grep -c '^stirling:nodes:') +expected=$(printf '%s\n' $NODES | grep -c .) +echo " stirling:nodes:* heartbeats: ${nodes_registered:-0} (expected >= $expected)" +[ "${nodes_registered:-0}" -ge "$expected" ] && ok "all $expected app nodes registered in Valkey" \ + || bad "only ${nodes_registered:-0} of $expected app nodes registered" + +echo "== 9. Valkey topology is redundant (detected: $TOPOLOGY) ==" +case "$TOPOLOGY" in + sentinel) + reps=$(valkey_info replication | sed -n 's/^connected_slaves:\([0-9]*\).*/\1/p') + [ "${reps:-0}" -ge 2 ] && ok "primary is replicating to ${reps} replicas" \ + || bad "primary has ${reps:-0} connected replicas (expected 2) - no failover target" + quorum=$(docker exec multinode-valkey-sentinel-1 valkey-cli -p 26379 sentinel ckquorum mymaster 2>/dev/null | tr -d '\r') + case "$quorum" in + OK*) ok "sentinels have quorum: $quorum" ;; + *) bad "sentinel quorum check failed: ${quorum:-}" ;; + esac + # ckquorum reports OK even with zero discovered replicas, so assert the failover candidates too. + known=$(docker exec multinode-valkey-sentinel-1 valkey-cli -p 26379 sentinel replicas mymaster 2>/dev/null | tr -d '\r' | grep -c '^name$') + [ "${known:-0}" -ge 2 ] && ok "sentinel-1 knows $known failover candidates" \ + || bad "sentinel-1 knows only ${known:-0} replicas - a failover would have no target" + ;; + cluster) + info=$(docker exec multinode-valkey valkey-cli cluster info 2>/dev/null | tr -d '\r') + printf '%s\n' "$info" | grep -q '^cluster_state:ok' && ok "cluster_state:ok" || bad "cluster_state is not ok" + slots=$(printf '%s\n' "$info" | sed -n 's/^cluster_slots_ok:\([0-9]*\).*/\1/p') + [ "${slots:-0}" = "16384" ] && ok "all 16384 slots served" || bad "only ${slots:-0}/16384 slots served" + # Masters carry '-' in the master-id column; replicas carry their primary's id there. + masters=$(docker exec multinode-valkey valkey-cli cluster nodes 2>/dev/null | tr -d '\r' | grep -c 'master -') + known=$(printf '%s\n' "$info" | sed -n 's/^cluster_known_nodes:\([0-9]*\).*/\1/p') + [ "${masters:-0}" -ge 3 ] && ok "$masters primaries sharding the keyspace (${known:-?} nodes known)" \ + || bad "only ${masters:-0} primaries (expected 3)" + # Keys must actually spread; everything on one shard would mean something pinned them to a slot. + shards=$(backplane_keys_raw | tr -d '\r' | grep -c '^[A-Za-z0-9_.-]*:[0-9][0-9]*: stirling:') + [ "${shards:-0}" -ge 1 ] && ok "backplane keys present on ${shards} shard(s)" \ + || bad "no backplane keys found on any shard" + ;; + *) + skip "standalone Valkey - no replication to verify (use --valkey sentinel|cluster for HA)" + ;; +esac + echo echo "============================================================" echo " Multi-node validation: $pass passed, $fail failed." +echo " Valkey topology: $TOPOLOGY" echo " Stack left running: $LB (admin / stirling)" echo "============================================================" [ "$fail" -eq 0 ] diff --git a/testing/cucumber/features/multinode/cluster_health.feature b/testing/cucumber/features/multinode/cluster_health.feature index def781d895..aa888b7cd1 100644 --- a/testing/cucumber/features/multinode/cluster_health.feature +++ b/testing/cucumber/features/multinode/cluster_health.feature @@ -12,6 +12,11 @@ Feature: Multi-node cluster health Given the multi-node stack is running And both nodes are cluster members using the Valkey backplane + @smoke + Scenario: Every node published a heartbeat to the Valkey backplane + Given the multi-node stack is running + Then every application node should be registered in the backplane + Scenario: The load balancer answers the health endpoint Given the multi-node stack is running When I request "/api/v1/info/status" 4 times through the load balancer diff --git a/testing/cucumber/features/steps/multinode_step_definitions.py b/testing/cucumber/features/steps/multinode_step_definitions.py index e43673dc86..ddaf99248d 100644 --- a/testing/cucumber/features/steps/multinode_step_definitions.py +++ b/testing/cucumber/features/steps/multinode_step_definitions.py @@ -2,6 +2,7 @@ import io import json +import re import subprocess import time import uuid @@ -13,6 +14,8 @@ LB_URL = "http://localhost:8080" NODES = ["multinode-stirling-1", "multinode-stirling-2"] PG = "multinode-postgres" MINIO = "multinode-minio" +# The primary keeps this name in every Valkey topology (standalone, sentinel, cluster). +VALKEY = "multinode-valkey" BUCKET = "policy-data" SOURCE_PREFIX = "incoming/" OUTPUT_PREFIX = "processed/" @@ -397,15 +400,38 @@ def step_run_visible_every(context): # --------------------------------------------------------------------------- rate limiting +# --cluster call prefixes each node's reply with 'host:port: '; a stirling: key never looks like that. +_NODE_PREFIX = re.compile(r"^[A-Za-z0-9_.\-]+:\d+:\s?") + + +def _backplane_keys(): + """Every stirling:* key in the backplane, whatever the Valkey topology.""" + # A sharded cluster splits keys across masters, so KEYS on one node sees only its own slots. + # --cluster call fans out; it errors on a non-cluster server, hence the plain-KEYS fallback. + rc, out, err = _sh(["docker", "exec", VALKEY, "valkey-cli", "--cluster", "call", + "--cluster-only-masters", "127.0.0.1:6379", "keys", "stirling:*"], + timeout=60) + if rc != 0: + rc, out, err = _sh(["docker", "exec", VALKEY, "valkey-cli", "keys", "stirling:*"], timeout=30) + assert rc == 0, f"valkey probe failed: {err.strip() or out.strip()}" + return [k for k in (_NODE_PREFIX.sub("", ln.strip()) for ln in out.splitlines()) + if k.startswith("stirling:")] + + @then("the rate-limit counter should be shared across nodes") def step_ratelimit_shared(context): # In cluster mode the ValkeyRateLimitStore holds counters in Valkey; probe that a key exists. - net = context._net or _network() - rc, out, err = _sh(["docker", "run", "--rm", "--network", net, "--entrypoint", "/bin/sh", - "valkey/valkey:8-alpine", "-c", - "valkey-cli -h valkey keys '*'"], timeout=30) - assert rc == 0, f"valkey probe failed: {err.strip()}" - assert out.strip(), "no keys in Valkey - rate-limit/backplane state is not shared" + keys = _backplane_keys() + assert keys, "no stirling:* keys in Valkey - rate-limit/backplane state is not shared" + + +@then("every application node should be registered in the backplane") +def step_nodes_registered(context): + # One stirling:nodes: heartbeat hash per app node proves every node reached Valkey. + registered = [k for k in _backplane_keys() if k.startswith("stirling:nodes:")] + assert len(registered) >= len(NODES), ( + f"expected >= {len(NODES)} stirling:nodes:* heartbeats, found {len(registered)}: " + f"{sorted(registered)}") # --------------------------------------------------------------------------- failover