Add Sentinel and Cluster Valkey test topologies to the multi-node stack and nightly

.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

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:

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"

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 <<EOF
+        port 26379
+        dir /tmp
+        sentinel monitor mymaster $$PRIMARY_IP 6379 2
+        sentinel down-after-milliseconds mymaster 5000
+        sentinel failover-timeout mymaster 10000
+        sentinel parallel-syncs mymaster 1
+        EOF
+        exec valkey-sentinel /tmp/sentinel.conf
+    # ckquorum alone is NOT enough: it reports "OK 3 usable Sentinels" even with ZERO discovered
+    # replicas, i.e. no failover candidate. Assert both.
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - valkey-cli -p 26379 sentinel ckquorum mymaster | grep -q '^OK' && [ "$$(valkey-cli -p 26379 sentinel replicas mymaster | grep -c '^name$$')" -ge 2 ]
+      interval: 3s
+      timeout: 5s
+      retries: 40
+    networks:
+      - stirling-multinode
+
+  sentinel-2:
+    <<: *valkey-sentinel
+    container_name: multinode-valkey-sentinel-2
+
+  sentinel-3:
+    <<: *valkey-sentinel
+    container_name: multinode-valkey-sentinel-3
+
+  # ---- App nodes: point the backplane at the sentinel set ------------------
+  # Only the keys that differ from the base file - compose merges `environment` across -f files.
+  stirling-1:
+    depends_on: *sentinel-deps
+    environment:
+      CLUSTER_VALKEY_MODE: "sentinel"
+      CLUSTER_VALKEY_SENTINEL_MASTER: "mymaster"
+      CLUSTER_VALKEY_SENTINEL_NODES: "sentinel-1:26379,sentinel-2:26379,sentinel-3:26379"
+
+  stirling-2:
+    depends_on: *sentinel-deps
+    environment:
+      CLUSTER_VALKEY_MODE: "sentinel"
+      CLUSTER_VALKEY_SENTINEL_MASTER: "mymaster"
+      CLUSTER_VALKEY_SENTINEL_NODES: "sentinel-1:26379,sentinel-2:26379,sentinel-3:26379"

testing/compose/docker-compose-multinode.yml

@@ -1,5 +1,5 @@
-# Multi-node Stirling-PDF processor test stack: shared Postgres/MinIO/Valkey behind an nginx LB fronting N app nodes, with enterprise features unlocked via a local test licence key.
-# Bring up: ./start-multinode-test.sh   Validate: ./validate-multinode-test.sh   Access: http://localhost:8080 (admin / stirling)
+# Multi-node test stack. Up: ./start-multinode-test.sh [--valkey sentinel|cluster] - default is
+# standalone. Every topology keeps container 'multinode-valkey', so docker-exec steps still work.
 
 x-stirling-node: &stirling-node
   build:
@@ -67,6 +67,14 @@ x-stirling-node: &stirling-node
     CLUSTER_BACKPLANE: "valkey"
     CLUSTER_ARTIFACTSTORE: "s3"
     CLUSTER_VALKEY_URL: "redis://valkey:6379"
+    # Named connections so CLIENT LIST attributes load per node in any Valkey monitor; overridden per node below.
+    CLUSTER_VALKEY_CLIENTNAME: "stirling-node-base"
+    # Pooling is on by default in the app; pinned here so the test stack asserts a known pool size.
+    CLUSTER_VALKEY_POOL_ENABLED: "true"
+    CLUSTER_VALKEY_POOL_MAXACTIVE: "16"
+    CLUSTER_VALKEY_POOL_MAXIDLE: "16"
+    CLUSTER_VALKEY_POOL_MINIDLE: "4"
+    CLUSTER_VALKEY_POOL_MAXWAITMILLIS: "2000"
     # SPRING_DATA_REDIS_REPOSITORIES_ENABLED is not needed: DataRedisRepositoriesAutoConfiguration is excluded (see application.properties, multinode/README.md).
 
     # --- Shared credential-encryption key (REQUIRED in cluster mode) ---
@@ -166,6 +174,7 @@ services:
     environment:
       <<: *stirling-env
       SYSTEM_NODEID: "node-1"
+      CLUSTER_VALKEY_CLIENTNAME: "stirling-node-1"
 
   # ---- App node 2 -----------------------------------------------------------
   # No ordering vs node-1 needed - each node mints its own signing key at boot and publishes the public half to the shared DB, so both verify each other's tokens.
@@ -175,6 +184,7 @@ services:
     environment:
       <<: *stirling-env
       SYSTEM_NODEID: "node-2"
+      CLUSTER_VALKEY_CLIENTNAME: "stirling-node-2"
 
   # ---- One-shot seeder: teams, ~40 users, S3 connection, policies -----------
   # Profile-gated so plain `docker compose up` skips it; the start script runs it once, using postgres:alpine for psql with curl+jq added for the HTTP calls.

testing/compose/multinode/loadgen.py

@@ -0,0 +1,395 @@
+#!/usr/bin/env python3
+"""Load generator for the multi-node stack: real PDF work through the nginx LB as many distinct users,
+mixed sync/async so the Valkey backplane sees job, lock, rate-limit and cache traffic. Stdlib only."""
+
+import json
+import os
+import random
+import ssl
+import sys
+import threading
+import time
+import urllib.error
+import urllib.request
+import uuid
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+
+BASE_URL = os.environ.get("BASE_URL", "http://nginx:8080").rstrip("/")
+DURATION = int(os.environ.get("DURATION_SECONDS", "300"))
+CONCURRENCY = int(os.environ.get("CONCURRENCY", "24"))
+USER_COUNT = int(os.environ.get("USER_COUNT", "40"))
+USER_PASSWORD = os.environ.get("USER_PASSWORD", "Password123!")
+ADMIN_USER = os.environ.get("ADMIN_USER", "admin")
+ADMIN_PASS = os.environ.get("ADMIN_PASS", "stirling")
+ASYNC_RATIO = float(os.environ.get("ASYNC_RATIO", "0.35"))
+RAMP_SECONDS = int(os.environ.get("RAMP_SECONDS", "20"))
+
+SSL_CTX = ssl.create_default_context()
+SSL_CTX.check_hostname = False
+SSL_CTX.verify_mode = ssl.CERT_NONE
+
+stop_at = 0.0
+stats_lock = threading.Lock()
+stats = defaultdict(lambda: {"ok": 0, "err": 0, "ms": 0.0, "bytes": 0})
+nodes_seen = defaultdict(int)
+errors = defaultdict(int)
+job_stats = {"submitted": 0, "completed": 0, "failed": 0, "sticky_410": 0}
+
+
+# ---------------------------------------------------------------- PDF corpus
+def make_pdf(pages: int, filler_lines: int) -> 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"<</Type/Catalog/Pages 2 0 R>>")
+    objs.append(f"<</Type/Pages/Kids[{kids}]/Count {pages}>>".encode())
+
+    for i in range(pages):
+        content_num = 4 + i * 2
+        objs.append(
+            f"<</Type/Page/Parent 2 0 R/MediaBox[0 0 612 792]"
+            f"/Contents {content_num} 0 R"
+            f"/Resources<</Font<</F1 {font_obj} 0 R>>>>>>".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"<</Length " + str(len(stream)).encode() + b">>stream\n" + stream + b"\nendstream")
+
+    objs.append(b"<</Type/Font/Subtype/Type1/BaseFont/Helvetica>>")
+
+    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<</Size {len(objs) + 1}/Root 1 0 R>>\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())

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}"

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
+}

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)"

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 <<EOF
 
 ============================================================================
- Multi-node Stirling is UP.
+ Multi-node Stirling is UP.   Valkey topology: $VALKEY_TOPOLOGY
 
    App (via load balancer): http://localhost:8080     (admin / stirling)
    MinIO console:           http://localhost:9001     (minioadmin / minioadmin)
@@ -57,7 +70,11 @@ cat <<EOF
  Try it:
    ./validate-multinode-test.sh        # multi-node smoke tests (optional)
    $COMPOSE logs -f stirling-1         # tail a node
-   ./start-multinode-test.sh --down    # stop + wipe
+   ./start-multinode-test.sh --valkey $VALKEY_TOPOLOGY --down    # stop + wipe
+
+ Other Valkey topologies (each wipes and rebuilds the backplane):
+   ./start-multinode-test.sh --valkey sentinel   # 1 primary + 2 replicas + 3 sentinels
+   ./start-multinode-test.sh --valkey cluster    # 3 primaries + 3 replicas, sharded
 
  Nodes are reachable directly for cross-node checks:
    docker compose -f docker-compose-multinode.yml exec stirling-1 curl -s localhost:8080/api/v1/info/status

testing/compose/validate-multinode-test.sh

@@ -1,7 +1,6 @@
 #!/usr/bin/env bash
-# Multi-node smoke tests against a running stack (start-multinode-test.sh): load-balancer spread, cross-node JWT validation (signing keys persist in the shared DB), and processor state visible from every node.
-# Auth: extracts the Bearer JWT from the login body via sed (no jq needed host-side) and hits nodes directly with docker exec.
-# Non-destructive - safe to re-run against the stack at http://localhost:8080.
+# Non-destructive smoke tests against a running stack; safe to re-run.
+# JWT is extracted from the login body with sed so no host-side jq is needed.
 set -uo pipefail
 cd "$(dirname "$0")"
 
@@ -13,6 +12,46 @@ PROBE="/api/v1/sources"
 pass=0; fail=0
 ok()   { echo "  PASS - $*"; pass=$((pass+1)); }
 bad()  { echo "  FAIL - $*"; fail=$((fail+1)); }
+skip() { echo "  SKIP - $*"; }
+
+# Which Valkey topology is live, read off the running containers rather than a flag, so this script
+# is correct no matter how the stack was brought up.
+detect_topology() {
+  # cluster_enabled lives in INFO cluster, NOT in CLUSTER INFO (which only reports state/slots).
+  if docker exec multinode-valkey valkey-cli info cluster 2>/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<</Type/Catalog/Pages 2 0 R>>endobj\n2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 612 792]>>endobj\ntrailer<</Root 1 0 R>>\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:-<no reply>}" ;;
+    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 ]

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

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:<id> 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