Fix Valkey backplane regressions and multi-node test tooling

2026-09-03 05:10:16 +03:00 · 2026-08-25 10:09:00 +01:00
parent f860eb725a
commit 5d5bd20d7f
21 changed files with 663 additions and 199 deletions
@@ -143,6 +143,13 @@ public class ClusterConfig {
                            + valkey.getMaxRedirects()
                            + ".");
        }
+        // Lettuce rejects a non-positive refresh period with an opaque assertion at boot.
+        if (valkey.getTopologyRefreshMs() <= 0) {
+            throw new IllegalStateException(
+                    "cluster.valkey.topologyRefreshMs must be > 0 in cluster mode; got "
+                            + valkey.getTopologyRefreshMs()
+                            + ".");
+        }
    }

    private static void validateCommon(
@@ -159,8 +166,9 @@ public class ClusterConfig {
        }
        if (pool.isEnabled() && pool.getMaxWaitMillis() <= 0) {
            throw new IllegalStateException(
-                    "cluster.valkey.pool.maxWaitMillis must be > 0 (a non-positive value means"
-                            + " block forever, which defeats cluster.valkey.commandTimeoutMs); got "
+                    "cluster.valkey.pool.maxWaitMillis must be > 0 (a negative value blocks"
+                            + " forever, which defeats cluster.valkey.commandTimeoutMs, and 0"
+                            + " fails the borrow instantly once the pool is exhausted); got "
                            + pool.getMaxWaitMillis()
                            + ".");
        }
@@ -7,8 +7,8 @@ package stirling.software.common.cluster;
 public record HostPort(String host, int port) {

    /**
-     * The port is always explicit: a bare host would silently take a default and connect somewhere
-     * the operator never named. IPv6 literals must be bracketed ({@code [::1]:6379}).
+     * Port is always explicit: a bare host would silently connect somewhere the operator never
+     * named. IPv6 literals must be bracketed ({@code [::1]:6379}).
     *
     * @throws IllegalStateException naming {@code propertyName} and echoing {@code entry}
     */
@@ -790,14 +790,14 @@ public class ApplicationProperties {
            @Data
            public static class Pool {
                /**
-                 * Connection pooling for dedicated connections. On by default: without a pool every
-                 * blocking/transactional call opens and tears down a TCP connection.
+                 * Pooling for dedicated connections. Backplane traffic multiplexes over the shared
+                 * native connection, so the pool backs only that one connection today.
                 */
                private boolean enabled = true;

                /**
                 * Max pooled connections; at least 2, one is held by the shared native connection.
-                 * A hard ceiling - borrows past it fail after {@link #maxWaitMillis}, not queue.
+                 * Headroom for a future dedicated path - raising it changes no current throughput.
                 */
                private int maxActive = 16;

@@ -811,8 +811,8 @@ public class ApplicationProperties {
                private int minIdle = 0;

                /**
-                 * Max wait for a pooled connection. Never set 0/negative: commons-pool2 treats that
-                 * as block-forever, which would defeat commandTimeoutMs.
+                 * Max wait for a pooled connection. Never 0/negative: negative blocks forever and
+                 * defeats commandTimeoutMs, 0 fails the borrow instantly once the pool is drained.
                 */
                private long maxWaitMillis = 2000;

@@ -165,7 +165,7 @@ class ClusterConfigValidationTest {
        }

        @Test
-        @DisplayName("V9: pool.maxWaitMillis <= 0 is rejected (it would block forever)")
+        @DisplayName("V9: pool.maxWaitMillis <= 0 is rejected (0 fails borrows, negative blocks)")
        void poolMaxWaitMustBePositive() {
            Valkey v = validStandalone();
            v.getPool().setMaxWaitMillis(0);
@@ -443,47 +443,58 @@ mcp:
    requireExistingAccount: true # Reject tokens whose subject has no enabled Stirling account (recommended)
  engineCapabilityRefreshMinutes: 5 # How often to refresh the AI capabilities manifest from the engine

-# Cluster configuration. Multi-node deployments only - leave 'enabled: false' for a single instance.
+# Multi-node Stirling PDF. Two different things are called "nodes" here: cluster.node.* is THIS
+# Stirling PDF instance, cluster.valkey.* is the shared Valkey - so 'mode: cluster' is Valkey's own.
 cluster:
-  enabled: false # Master switch. 'false' (default) wires the in-process backplane and skips all cluster checks. Single-instance installs do not need to change anything here.
-  backplane: inprocess # Backplane implementation: 'inprocess' (single JVM only) or 'valkey' (multi-node via Valkey/Redis)
-  artifactStore: local # Transient cluster job-artifact backend: 'local' (per-node disk; single-node only) or 's3' (shared object store; required for multi-node). Distinct from 'storage.provider' which controls persistent user uploads - when both are 's3' they share the storage.s3.* credentials block. Multi-node deployments MUST set this to 's3'.
+  enabled: false # Run more than one Stirling PDF node. 'false' (default) = single instance; nothing below applies.
+  backplane: inprocess # How Stirling PDF nodes share state: 'inprocess' (one JVM) or 'valkey'.
+  artifactStore: local # Job artifacts: 'local' (per-node disk) or 's3'. Multi-node MUST use 's3'.
  s3:
-    keyPrefix: transient/ # Bucket key prefix used by the cluster artifact store when artifactStore=s3. Trailing slash recommended. Lets a single bucket host both persistent uploads (storage.s3.*) and transient job artifacts under separate prefixes.
-  valkey:
-    mode: "" # Topology: 'standalone' (single server, uses url), 'sentinel' (HA primary/replica watched by sentinels), or 'cluster' (sharded). Blank (default) infers: sentinel.master set = sentinel, nodes set = cluster, otherwise standalone.
-    # SECURITY - in sentinel/cluster mode the url below is ignored ENTIRELY, including its 'rediss://'
-    # scheme and any embedded credentials. A 'rediss://' url with tls.enabled=false refuses to boot.
-    url: "" # Valkey/Redis URL, e.g. 'redis://valkey:6379' or 'rediss://...' for TLS. Required when enabled=true and mode=standalone. Read ONLY in standalone mode.
-    username: "" # Data-node ACL username. Overrides any user in 'url'. Blank = default user.
-    password: "" # Data-node password. Overrides any password in 'url'. Never logged.
-    nodes: [] # Valkey Cluster seed nodes, e.g. ['valkey-1:6379','valkey-2:6379','valkey-3:6379']. Required when mode=cluster.
-    maxRedirects: 3 # Cluster mode only: max MOVED/ASK redirects followed before a command fails.
-    topologyRefreshMs: 30000 # Cluster mode only: periodic topology refresh backstop (ms). Adaptive refresh on MOVED/ASK/reconnect is always enabled.
-    clientName: "" # CLIENT SETNAME applied to every connection so Valkey monitoring can attribute load per node. Blank (default) = 'stirling-' + cluster.node.id, falling back to the hostname (the container name under Docker/Kubernetes) and only then to the per-boot UUID. Set to 'off' to send no CLIENT SETNAME at all, for locked-down ACL users denied the CLIENT command.
-    commandTimeoutMs: 2000 # Per-command timeout (ms). Bounds every backplane call so a slow Valkey cannot stall request threads.
-    sentinel:
-      master: "" # Monitored primary name, i.e. the name in 'sentinel monitor <name> ...'. Required when mode=sentinel.
-      nodes: [] # Sentinel endpoints, e.g. ['sentinel-1:26379','sentinel-2:26379','sentinel-3:26379']. Required when mode=sentinel.
-      username: "" # Username used when connecting to the SENTINELS (separate from the data-node username).
-      password: "" # Password used when connecting to the SENTINELS. Separate from 'password' - setting only that one does NOT authenticate to sentinels.
-    tls:
-      enabled: false # The ONLY way to get TLS in sentinel/cluster mode - the url and its 'rediss://' scheme are ignored there. In standalone mode it is OR-ed with the url scheme: 'true' forces TLS even on a 'redis://' url, and 'false' still honours 'rediss://'.
-      skipCertVerification: false # set to 'true' to skip TLS certificate verification on Valkey connections (dev/test only)
-    pool:
-      enabled: true # Pool dedicated connections. Turning this off makes every transactional/blocking call open and tear down a TCP connection.
-      maxActive: 16 # Max pooled connections. Must be >= 2 - one is permanently held by the shared native connection. This is a HARD ceiling; dedicated connections were unbounded before pooling existed, so a borrow past it now fails after maxWaitMillis with "Could not get a resource from the pool". Raise it (or set enabled: false) if that appears under load.
-      maxIdle: 16 # Max idle pooled connections. Keep equal to maxActive so idle connections are not churned.
-      minIdle: 0 # Connections kept warm. 0 (default): backplane traffic runs on the shared native connection, so warm pooled sockets would sit idle on every node. Raise it only to pre-pay the connect cost for bursty job traffic.
-      maxWaitMillis: 2000 # Max wait for a pooled connection (ms). Must be > 0 - a non-positive value blocks forever.
-      timeBetweenEvictionRunsMillis: 30000 # Idle-evictor interval (ms). minIdle is only honoured while the evictor runs.
-      testOnBorrow: true # Validate a pooled connection on borrow. Local isOpen() check only, no round trip: it catches explicitly closed connections, NOT one that a failover dropped (Lettuce still reports those open while it auto-reconnects).
+    keyPrefix: transient/ # Prefix for job artifacts, so one bucket can also hold storage.s3.* uploads.
+
+  # This Stirling PDF instance's own identity.
  node:
-    id: "" # Optional explicit node id. Blank = auto-generated UUID at startup.
-    role: both # 'web' (serves HTTP), 'worker' (runs jobs), or 'both' (default)
-    internalAddress: "" # host:port advertised in the instance registry for peer-to-peer cluster traffic. Blank = derived at startup.
-    scheme: http # 'http' or 'https' - scheme peers use to call this node's /internal/cluster/** endpoints
-    heartbeatIntervalMs: 5000 # Heartbeat publish interval for the instance registry (ms)
+    id: "" # Blank = fresh UUID each boot. Set it for a stable name in monitoring.
+    role: both # 'web' (serves HTTP), 'worker' (runs jobs), or 'both'.
+    internalAddress: "" # host:port other Stirling PDF nodes reach this one on. Blank = derived at startup.
+    scheme: http # Scheme peers use for this node's /internal/cluster/** endpoints.
+    heartbeatIntervalMs: 5000 # How often this node publishes its heartbeat.
+
+  # The Valkey/Redis server every Stirling PDF node shares.
+  valkey:
+    mode: "" # 'standalone' | 'sentinel' | 'cluster'. Blank = inferred from whichever block below is filled in.
+    url: "" # e.g. 'redis://valkey:6379'. STANDALONE ONLY - sentinel and cluster ignore it, so a 'rediss://' url there refuses to boot unless tls.enabled is true.
+    username: "" # Valkey ACL user. Overrides any user in 'url'.
+    password: "" # Valkey password. Overrides any password in 'url'. Never logged.
+    commandTimeoutMs: 2000 # Per-command timeout, so a slow Valkey cannot stall request threads.
+    clientName: "" # CLIENT SETNAME, so monitoring can tell the Stirling PDF nodes apart. Blank = 'stirling-' + node id; 'off' sends none.
+
+    # mode=sentinel: one Valkey primary plus replicas, with sentinels failing over between them.
+    sentinel:
+      master: "" # The name from 'sentinel monitor <name> ...'. Required for mode=sentinel.
+      nodes: [] # e.g. ['sentinel-1:26379','sentinel-2:26379']. Required for mode=sentinel.
+      username: "" # Sentinel ACL user - separate from the Valkey user above.
+      password: "" # Sentinel password - separate from the Valkey password; setting only that one does NOT authenticate here.
+
+    # mode=cluster: several Valkey nodes sharding the keyspace between them.
+    nodes: [] # e.g. ['valkey-1:6379','valkey-2:6379','valkey-3:6379']. Required for mode=cluster.
+    maxRedirects: 3 # Max MOVED/ASK redirects before a command fails.
+    topologyRefreshMs: 30000 # Periodic topology re-read; refresh on MOVED/ASK/reconnect is always on.
+
+    tls:
+      enabled: false # The only way to get TLS in sentinel/cluster mode. Standalone also accepts a 'rediss://' url.
+      skipCertVerification: false # Skip TLS chain and hostname checks. Dev only.
+
+    # Backplane traffic rides one shared connection, so the pool holds only that today.
+    # These are headroom for a future dedicated-connection path.
+    pool:
+      enabled: true
+      maxActive: 16 # Must be >= 2. Raising it does not change current throughput.
+      maxIdle: 16 # Keep equal to maxActive so idle connections are not churned.
+      minIdle: 0 # Connections kept warm. 0 because nothing borrows from the pool today.
+      maxWaitMillis: 2000 # Must be > 0: negative blocks forever, 0 fails the borrow instantly.
+      timeBetweenEvictionRunsMillis: 30000 # Idle-evictor interval; minIdle only applies while it runs.
+      testOnBorrow: true # Local isOpen() check only, no round trip.

 pdfEditor:
  fallback-font: classpath:/static/fonts/NotoSans-Regular.ttf # Override to point at a custom fallback font
@@ -27,8 +27,8 @@ dependencies {
    api 'org.springframework.boot:spring-boot-starter-cache'
    api 'com.github.ben-manes.caffeine:caffeine'
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'
-    // Pools DEDICATED Lettuce connections (MULTI/EXEC, blocking, scripts). Without it each such
-    // call is a fresh TCP connect + teardown: measured 50 new connections/sec at ~50 req/s.
+    // Supplies GenericObjectPoolConfig for the Lettuce pool, which today backs only the single
+    // shared native connection; headroom for a future dedicated-connection path.
    implementation 'org.apache.commons:commons-pool2'
    api 'io.swagger.core.v3:swagger-core-jakarta:2.2.53'
    implementation "com.bucket4j:bucket4j_jdk17-core:${bucket4jVersion}"
@@ -48,6 +48,12 @@ public class ValkeyConnectionConfiguration {
    private static final java.util.regex.Pattern NOPERM_COMMAND =
            java.util.regex.Pattern.compile("run the '([^']+)'");

+    /** Valkey rejects a client name outside printable ASCII, space included, during HELLO. */
+    private static final java.util.regex.Pattern UNSAFE_CLIENT_NAME =
+            java.util.regex.Pattern.compile("[^\\x21-\\x7e]");
+
+    private static final int BOOT_PROBE_ATTEMPTS = 10;
+
    private final ApplicationProperties applicationProperties;

    @Bean(destroyMethod = "destroy")
@@ -314,16 +320,31 @@ public class ValkeyConnectionConfiguration {
    }

    /**
-     * Blank = {@code stirling-} + node name. {@code off}/{@code none}/{@code disabled} returns
-     * null: Lettuce sends CLIENT SETNAME in the handshake, so a NOPERM there refuses all conns.
+     * Blank = {@code stirling-} + node name; off/none/disabled = null. A missing SETNAME ACL
+     * refuses nothing (RESP3 folds it into HELLO, RESP2 swallows it); a name Valkey rejects does.
     */
    static String resolveClientName(Cluster cluster) {
        String configured = cluster.getValkey().getClientName();
        if (!isSet(configured)) {
-            return "stirling-" + cluster.resolvedNodeName();
+            return sanitiseClientName("stirling-" + cluster.resolvedNodeName(), "cluster.node.id");
        }
        String trimmed = configured.trim();
-        return isClientNameOptOut(trimmed) ? null : trimmed;
+        return isClientNameOptOut(trimmed)
+                ? null
+                : sanitiseClientName(trimmed, "cluster.valkey.clientName");
+    }
+
+    private static String sanitiseClientName(String name, String source) {
+        String safe = UNSAFE_CLIENT_NAME.matcher(name).replaceAll("-");
+        if (!safe.equals(name)) {
+            log.warn(
+                    "Valkey client name from {} contains characters Valkey refuses in the"
+                            + " handshake; using '{}' instead of '{}'",
+                    source,
+                    safe,
+                    name);
+        }
+        return safe;
    }

    private static boolean isClientNameOptOut(String value) {
@@ -370,7 +391,9 @@ public class ValkeyConnectionConfiguration {
                ("stirling:health:boot:" + (clientName == null ? "unnamed" : clientName))
                        .getBytes(StandardCharsets.UTF_8);
        RuntimeException last = null;
-        for (int attempt = 1; attempt <= 10; attempt++) {
+        int attempt = 0;
+        while (attempt < BOOT_PROBE_ATTEMPTS) {
+            attempt++;
            try {
                RedisConnection conn = factory.getConnection();
                try {
@@ -394,28 +417,38 @@ public class ValkeyConnectionConfiguration {
                }
                last = ex;
                log.warn(
-                        "Valkey probe attempt {}/10 failed ({}, tls={}): {}",
+                        "Valkey probe attempt {}/{} failed ({}, tls={}): {}",
                        attempt,
+                        BOOT_PROBE_ATTEMPTS,
                        target,
                        tls,
                        ex.getMessage());
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException ie) {
+                    // Destroy first: an armed interrupt aborts Lettuce's shutdown await and
+                    // leaks the client resources.
+                    factory.destroy();
                    Thread.currentThread().interrupt();
-                    break;
+                    throw new IllegalStateException(
+                            unreachableMessage(attempt, target, tls, last), last);
                }
            }
        }
        factory.destroy();
-        throw new IllegalStateException(
-                "Valkey unreachable at boot after 10 attempts ("
-                        + target
-                        + ", tls="
-                        + tls
-                        + "): "
-                        + (last == null ? "no detail" : last.getMessage()),
-                last);
+        throw new IllegalStateException(unreachableMessage(attempt, target, tls, last), last);
+    }
+
+    private static String unreachableMessage(
+            int attempts, String target, boolean tls, RuntimeException last) {
+        return "Valkey unreachable at boot after "
+                + attempts
+                + " attempts ("
+                + target
+                + ", tls="
+                + tls
+                + "): "
+                + (last == null ? "no detail" : last.getMessage());
    }

    /**
@@ -457,9 +490,8 @@ public class ValkeyConnectionConfiguration {
                + ". The credentials were accepted; this ACL user is missing a command or key"
                + " permission"
                + (command == null ? "" : " - grant '+" + command + "'")
-                + ". The backplane needs the boot probe (EXISTS), read/write access to the"
-                + " 'stirling:*' keyspace, and 'client|setname' unless"
-                + " cluster.valkey.client-name is set to 'off'.";
+                + ". The backplane needs the boot probe (EXISTS) and read/write access to the"
+                + " 'stirling:*' keyspace.";
    }

    /** NOPERM only - a permitted-command/key problem, distinct from bad credentials. */
@@ -10,6 +10,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;

+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -28,8 +30,8 @@ import stirling.software.common.cluster.JobStore;
 import stirling.software.common.cluster.JobStoreEntry;

 /**
- * put() is NOT atomic across keys: hash+TTL is one Lua script, each index row a SET PX, so a torn
- * put() leaves stale index rows. Lua, not MULTI/WATCH: the cluster client rejects both.
+ * One hash per job plus a fileId to jobId index. Atomic on standalone/sentinel via one Lua script;
+ * on cluster those keys are cross-slot, so writes are separate and hash-first to keep tears benign.
 */
@Component
@RequiredArgsConstructor
@@ -58,8 +60,43 @@ public class ValkeyJobStore implements JobStore {
                            + " else return 0 end",
                    Long.class);

+    // Non-cluster put(): hash, TTL and index rows in one round trip, so no torn write.
+    // KEYS[1]=hash, KEYS[2..]=index rows; ARGV[1]=ttlMs, ARGV[2]=jobId, ARGV[3..]=fields.
+    private static final RedisScript<Long> PUT_ATOMIC =
+            new DefaultRedisScript<>(
+                    "redis.call('HSET', KEYS[1], unpack(ARGV, 3));"
+                            + " redis.call('PEXPIRE', KEYS[1], ARGV[1]);"
+                            + " for i = 2, #KEYS do"
+                            + " redis.call('SET', KEYS[i], ARGV[2], 'PX', ARGV[1]) end;"
+                            + " return 1",
+                    Long.class);
+
+    // Non-cluster delete(): hash read, DEL and the value-guarded index deletes in one round trip.
+    // ARGV[1]=jobId, ARGV[2]=index prefix; malformed fileIds JSON must not error the script.
+    private static final RedisScript<Long> DELETE_JOB_AND_INDEX =
+            new DefaultRedisScript<>(
+                    "local ids = redis.call('HGET', KEYS[1], 'fileIds');"
+                            + " redis.call('DEL', KEYS[1]);"
+                            + " if not ids then return 0 end;"
+                            + " local ok, decoded = pcall(cjson.decode, ids);"
+                            + " if not ok or type(decoded) ~= 'table' then return 0 end;"
+                            + " local n = 0;"
+                            + " for i = 1, #decoded do"
+                            + " if type(decoded[i]) == 'string' then"
+                            + " local k = ARGV[2] .. decoded[i];"
+                            + " if redis.call('GET', k) == ARGV[1] then"
+                            + " n = n + redis.call('DEL', k) end end end;"
+                            + " return n",
+                    Long.class);
+
    private final StringRedisTemplate template;

+    // Cluster rejects a script spanning the job key and its index keys: they hash to other slots.
+    private boolean isClusterAware() {
+        RedisConnectionFactory factory = template.getConnectionFactory();
+        return factory instanceof LettuceConnectionFactory lettuce && lettuce.isClusterAware();
+    }
+
    @Override
    public void put(JobStoreEntry entry, Duration ttl) {
        String key = JOB_PREFIX + entry.jobId();
@@ -95,20 +132,37 @@ public class ValkeyJobStore implements JobStore {
                "resultMeta",
                writeJson(entry.resultMeta() == null ? Map.of() : entry.resultMeta()));

-        // Index entries first: the surviving crash window then leaves "index points at a job not
-        // yet visible", which every caller already handles (findJobIdByFileId returns Optional).
+        List<String> indexKeys = new ArrayList<>();
        if (entry.fileIds() != null) {
            for (String fileId : entry.fileIds()) {
-                template.opsForValue().set(FILE_INDEX_PREFIX + fileId, entry.jobId(), ttl);
+                indexKeys.add(FILE_INDEX_PREFIX + fileId);
            }
        }
-        List<String> args = new ArrayList<>(1 + fields.size() * 2);
-        args.add(Long.toString(ttlMs));
+        List<String> fieldArgs = new ArrayList<>(fields.size() * 2);
        for (Map.Entry<String, String> f : fields.entrySet()) {
-            args.add(f.getKey());
-            args.add(f.getValue());
+            fieldArgs.add(f.getKey());
+            fieldArgs.add(f.getValue());
        }
+        if (!isClusterAware()) {
+            List<String> keys = new ArrayList<>(1 + indexKeys.size());
+            keys.add(key);
+            keys.addAll(indexKeys);
+            List<String> args = new ArrayList<>(2 + fieldArgs.size());
+            args.add(Long.toString(ttlMs));
+            args.add(entry.jobId());
+            args.addAll(fieldArgs);
+            template.execute(PUT_ATOMIC, keys, args.toArray());
+            return;
+        }
+        // Cluster only: hash first, so a torn write leaves an unindexed job rather than an index
+        // row pointing at a hash that does not exist.
+        List<String> args = new ArrayList<>(1 + fieldArgs.size());
+        args.add(Long.toString(ttlMs));
+        args.addAll(fieldArgs);
        template.execute(HSET_WITH_TTL, List.of(key), args.toArray());
+        for (String indexKey : indexKeys) {
+            template.opsForValue().set(indexKey, entry.jobId(), ttl);
+        }
    }

    @Override
@@ -117,12 +171,16 @@ public class ValkeyJobStore implements JobStore {
    }

    /**
-     * Not atomic: a concurrent put() for the same jobId can resurrect the hash after the DEL and
-     * leave its fileIds unindexed. Tolerated - dead path on Valkey (shouldRunLocalCleanup=false).
+     * Live path: /api/v1/general/jobs/cleanup and /api/v1/admin/job/cleanup?force=true, neither
+     * gated by shouldRunLocalCleanup(). One script except on cluster, where a put() can interleave.
     */
    @Override
    public void delete(String jobId) {
        String jobKey = JOB_PREFIX + jobId;
+        if (!isClusterAware()) {
+            template.execute(DELETE_JOB_AND_INDEX, List.of(jobKey), jobId, FILE_INDEX_PREFIX);
+            return;
+        }
        String fileIdsJson = (String) template.opsForHash().get(jobKey, "fileIds");
        template.delete(jobKey);
        if (fileIdsJson == null) {
@@ -8,7 +8,7 @@ import java.util.UUID;

 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.core.env.Environment;
-import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DataAccessException;
 import org.springframework.stereotype.Component;

 import jakarta.annotation.PostConstruct;
@@ -59,11 +59,12 @@ public class InitialSecuritySetup {
    @PostConstruct
    public void init() {
        try {
+            boolean restoredFromBackup = importBackupIfNeeded();
            for (int attempt = 1; ; attempt++) {
                try {
-                    runBootstrap();
+                    runBootstrap(restoredFromBackup);
                    return;
-                } catch (DataIntegrityViolationException e) {
+                } catch (DataAccessException e) {
                    if (attempt >= BOOTSTRAP_RACE_ATTEMPTS) {
                        throw e;
                    }
@@ -77,20 +78,27 @@ public class InitialSecuritySetup {
        } catch (IllegalArgumentException
                | SQLException
                | UnsupportedProviderException
-                | DataIntegrityViolationException e) {
+                | DataAccessException e) {
+            // Widened for diagnosis, not recovery: unrecoverable cases such as duplicate team rows
+            // (tracked separately) still just exhaust the attempts and exit here.
            log.error("Failed to initialize security setup.", e);
            System.exit(1);
        }
    }

-    private void runBootstrap()
+    // Restoring a backup replays the whole schema, so it must run outside the retry loop.
+    private boolean importBackupIfNeeded() {
+        if (userService.hasUsers() || !databaseService.hasBackup()) {
+            return false;
+        }
+        databaseService.importDatabase();
+        return true;
+    }
+
+    private void runBootstrap(boolean restoredFromBackup)
            throws IllegalArgumentException, SQLException, UnsupportedProviderException {
-        if (!userService.hasUsers()) {
-            if (databaseService.hasBackup()) {
-                databaseService.importDatabase();
-            } else {
-                initializeAdminUser();
-            }
+        if (!restoredFromBackup && !userService.hasUsers()) {
+            initializeAdminUser();
        }

        configureJWTSettings();
@@ -139,8 +147,10 @@ public class InitialSecuritySetup {
            }
        }

-        userService.saveAll(usersWithoutTeam); // batch save
+        // A null team_id is the retry guard, so commit it last: syncMembership is idempotent and
+        // only needs the already-persisted team id, so a half-done pass is re-found and finished.
        usersWithoutTeam.forEach(teamMembershipService::syncMembership);
+        userService.saveAll(usersWithoutTeam); // batch save
        if (usersWithoutTeam != null && !usersWithoutTeam.isEmpty()) {
            log.info(
                    "Assigned {} user(s) without a team to the default team.",
@@ -24,10 +24,12 @@ import stirling.software.common.cluster.ClusterNode;
 import stirling.software.common.cluster.JobStoreEntry;
 import stirling.software.common.model.ApplicationProperties;

-// Opt-in, needs an external sentinel deployment. Env vars, prefix STIRLING_TEST_VALKEY_:
-// SENTINEL_NODES (required), SENTINEL_MASTER (default mymaster), SENTINEL_PASSWORD, PASSWORD.
+/**
+ * Opt-in, never runs in CI: needs STIRLING_TEST_VALKEY_SENTINEL_NODES (comma-separated host:port).
+ * Optional STIRLING_TEST_VALKEY_: SENTINEL_MASTER (default mymaster), SENTINEL_PASSWORD, PASSWORD.
+ */
@EnabledIfEnvironmentVariable(named = "STIRLING_TEST_VALKEY_SENTINEL_NODES", matches = ".+")
-class LiveValkeySentinelModeTest {
+class LiveExternalSentinelTest {

    private static final String NODE_ID = "sentinel-node";
    private static final String RUN = UUID.randomUUID().toString().substring(0, 8);
@@ -30,7 +30,7 @@ import org.testcontainers.utility.DockerImageName;
 import stirling.software.common.cluster.JobStoreEntry;
 import stirling.software.common.model.ApplicationProperties;

-// Real server. Unpooled writes churned ~50 new TCP connections/sec at 50 req/s, and
+// Real server. Backplane commands multiplex over one shared native connection, and
 // connections without CLIENT SETNAME show as name="" in every Valkey monitor.
@Testcontainers
@EnabledIf("isDockerAvailable")
@@ -79,12 +79,14 @@ class LiveValkeyPoolingTest {
    void factoryIsPooled() {
        assertTrue(
                factory.getClientConfiguration() instanceof LettucePoolingClientConfiguration,
-                "pool.enabled=true must reach the factory, or dedicated connections churn TCP");
+                "pool.enabled=true must reach the factory; nothing else pins the pool wiring");
    }

    @Test
-    @DisplayName(WRITES + " job writes do not churn connections and do not grow the client count")
-    void writesDoNotChurnConnections() {
+    @DisplayName(WRITES + " job writes multiplex over the single shared native connection")
+    void writesMultiplexOverTheSharedNativeConnection() {
+        // Not a pooling proof: shareNativeConnection defaults true and no command queues, so
+        // nothing borrows from the pool. This pins the multiplexing, which pooling cannot change.
        long before = statLong("stats", "total_connections_received");

        ValkeyJobStore store = new ValkeyJobStore(template);
@@ -105,7 +107,7 @@ class LiveValkeyPoolingTest {
        long delta = statLong("stats", "total_connections_received") - before;
        assertTrue(
                delta < 20,
-                "pooled writes must reuse connections; "
+                "backplane writes must reuse the shared connection; "
                        + WRITES
                        + " writes opened "
                        + delta
@@ -113,14 +115,14 @@ class LiveValkeyPoolingTest {
        long connected = statLong("clients", "connected_clients");
        assertTrue(
                connected <= 6,
-                "connection count must stay bounded by the pool; connected_clients=" + connected);
+                "connection count must stay flat under write load; connected_clients=" + connected);
    }

    @Test
    @DisplayName(
            "every connection is attributable: CLIENT LIST shows stirling-<nodeId>, never \"\"")
    void everyConnectionIsNamed() {
-        // Force at least one write so the pool is warm before the CLIENT LIST snapshot.
+        // Force at least one write so the connection is live before the CLIENT LIST snapshot.
        template.opsForValue().set("pool:probe", "v", Duration.ofSeconds(30));

        List<RedisClientInfo> clients =
@@ -702,6 +702,110 @@ class ValkeyConnectionConfigurationTest {
            assertEquals(
                    "stirling-web-a", ValkeyConnectionConfiguration.resolveClientName(cluster));
        }
+
+        @Test
+        @DisplayName("off/none/disabled opt out of CLIENT SETNAME entirely")
+        void optOutValuesReturnNull() {
+            for (String optOut : List.of("off", "none", "disabled", "OFF", "  None  ")) {
+                Cluster cluster = new ApplicationProperties().getCluster();
+                cluster.getValkey().setClientName(optOut);
+                assertNull(
+                        ValkeyConnectionConfiguration.resolveClientName(cluster),
+                        "'" + optOut + "' must send no CLIENT SETNAME at all");
+            }
+        }
+
+        @Test
+        @DisplayName("a node id Valkey would reject in HELLO is sanitised, not passed through")
+        void unsafeNodeIdIsSanitised() {
+            // Valkey answers -ERR to a name with spaces and aborts the RESP3 handshake, so an
+            // unsanitised name refuses every connection against a healthy server.
+            Cluster cluster = new ApplicationProperties().getCluster();
+            cluster.getNode().setId("node 7\nweb\ttab");
+            assertEquals(
+                    "stirling-node-7-web-tab",
+                    ValkeyConnectionConfiguration.resolveClientName(cluster));
+        }
+
+        @Test
+        @DisplayName("an explicit clientName is sanitised the same way")
+        void unsafeExplicitNameIsSanitised() {
+            Cluster cluster = new ApplicationProperties().getCluster();
+            cluster.getValkey().setClientName("stirling web a");
+            assertEquals(
+                    "stirling-web-a", ValkeyConnectionConfiguration.resolveClientName(cluster));
+        }
+
+        @Test
+        @DisplayName("a safe name is returned byte-identical")
+        void safeNameIsUntouched() {
+            Cluster cluster = new ApplicationProperties().getCluster();
+            cluster.getValkey().setClientName("stirling-web_a.1:2");
+            assertEquals(
+                    "stirling-web_a.1:2", ValkeyConnectionConfiguration.resolveClientName(cluster));
+        }
+    }
+
+    @Nested
+    @DisplayName("guardIgnoredUrl()")
+    class GuardIgnoredUrl {
+
+        private Valkey withUrl(String url) {
+            Valkey v = new Valkey();
+            v.setUrl(url);
+            return v;
+        }
+
+        @Test
+        @DisplayName(
+                "a rediss:// url ignored by sentinel mode refuses boot rather than downgrading")
+        void redissIgnoredWithoutTlsThrows() {
+            IllegalStateException ex =
+                    assertThrows(
+                            IllegalStateException.class,
+                            () ->
+                                    ValkeyConnectionConfiguration.guardIgnoredUrl(
+                                            withUrl("rediss://valkey:6379"),
+                                            ValkeyMode.SENTINEL,
+                                            false));
+            assertTrue(
+                    ex.getMessage().contains("rediss://"),
+                    "the operator must be told which setting silently dropped TLS");
+        }
+
+        @Test
+        @DisplayName("the same url is allowed once tls.enabled restores TLS")
+        void redissIgnoredWithTlsIsAllowed() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("rediss://valkey:6379"), ValkeyMode.CLUSTER, true);
+        }
+
+        @Test
+        @DisplayName("standalone reads the url, so it is never guarded")
+        void standaloneIsNeverGuarded() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("rediss://valkey:6379"), ValkeyMode.STANDALONE, false);
+        }
+
+        @Test
+        @DisplayName("a plaintext url carrying credentials warns but still boots")
+        void userInfoOnlyWarns() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("redis://user:pw@valkey:6379"), ValkeyMode.SENTINEL, false);
+        }
+
+        @Test
+        @DisplayName("an unparseable ignored url warns rather than failing boot")
+        void malformedUrlDoesNotThrow() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("redis://valkey:6379 with spaces"), ValkeyMode.CLUSTER, false);
+        }
+
+        @Test
+        @DisplayName("a blank url is nothing to guard")
+        void blankUrlIsIgnored() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(withUrl(""), ValkeyMode.CLUSTER, false);
+        }
    }

    @Nested
@@ -3,19 +3,23 @@ package stirling.software.proprietary.cluster.valkey;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockingDetails;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;

 import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;

 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.Invocation;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.ValueOperations;
 import org.springframework.data.redis.core.script.RedisScript;
@@ -26,6 +30,7 @@ import stirling.software.common.cluster.JobStoreEntry.JobState;
 class ValkeyJobStoreTest {

    private static final String INDEX_KEY = "stirling:file2job:file-a";
+    private static final String JOB_KEY = "stirling:job:job-1";

    private static JobStoreEntry entry() {
        return new JobStoreEntry(
@@ -65,8 +70,8 @@ class ValkeyJobStoreTest {
    }

    @Test
-    @DisplayName("a live TTL writes the index row with that TTL and deletes nothing")
-    void liveTtlWritesTheIndexRow() {
+    @DisplayName("a live TTL writes the job hash and the index row with that TTL, deleting nothing")
+    void liveTtlWritesTheJobHashAndIndexRow() {
        StringRedisTemplate template = mock(StringRedisTemplate.class);
        @SuppressWarnings("unchecked")
        ValueOperations<String, String> values = mock(ValueOperations.class);
@@ -74,7 +79,40 @@ class ValkeyJobStoreTest {

        new ValkeyJobStore(template).put(entry(), Duration.ofMinutes(5));

-        verify(values).set(INDEX_KEY, "job-1", Duration.ofMinutes(5));
+        // Shape-agnostic: a row may be written by its own command or by one multi-key script.
+        List<String> writtenKeys = new ArrayList<>();
+        List<String> writtenArgs = new ArrayList<>();
+        for (Invocation invocation : mockingDetails(template).getInvocations()) {
+            Object[] raw = invocation.getRawArguments();
+            if (!"execute".equals(invocation.getMethod().getName()) || raw.length < 3) {
+                continue;
+            }
+            if (raw[1] instanceof List<?> keys && raw[2] instanceof Object[] args) {
+                keys.forEach(key -> writtenKeys.add(String.valueOf(key)));
+                for (Object arg : args) {
+                    writtenArgs.add(String.valueOf(arg));
+                }
+            }
+        }
+
+        String ttlMillis = Long.toString(Duration.ofMinutes(5).toMillis());
+        assertTrue(writtenKeys.contains(JOB_KEY), "put() must write the job hash, not just index");
+        assertTrue(writtenArgs.contains(ttlMillis), "the job hash must be written with its TTL");
+        assertTrue(
+                writtenArgs.containsAll(List.of("jobId", "job-1", "state", "COMPLETE")),
+                "the job hash must carry the entry's fields");
+
+        ArgumentCaptor<String> setKey = ArgumentCaptor.forClass(String.class);
+        ArgumentCaptor<String> setValue = ArgumentCaptor.forClass(String.class);
+        ArgumentCaptor<Duration> setTtl = ArgumentCaptor.forClass(Duration.class);
+        verify(values, atLeast(0)).set(setKey.capture(), setValue.capture(), setTtl.capture());
+        boolean indexSetPx =
+                setKey.getAllValues().contains(INDEX_KEY)
+                        && setValue.getAllValues().contains("job-1")
+                        && setTtl.getAllValues().contains(Duration.ofMinutes(5));
+        assertTrue(
+                indexSetPx || writtenKeys.contains(INDEX_KEY),
+                "put() must write the file index row");
        verify(template, never()).delete(anyString());
    }
 }
@@ -21,7 +21,8 @@ services:
      - --maxmemory-policy
      - "noeviction"
    ports:
-      - "6379:6379" # host access for valkey-cli inspection
+      # Loopback-only: this Valkey has no auth, so never expose it on all interfaces.
+      - "127.0.0.1:6379:6379" # host access for valkey-cli inspection

  # One-shot: BetterDB runs as UID 1001 and cannot write a root-owned volume.
  betterdb-init:
@@ -53,7 +54,8 @@ services:
      - stirling-multinode

  betterdb:
-    image: betterdb/monitor:latest
+    # Digest-pinned: this third-party image holds the stack's Postgres credentials, so no floating :latest.
+    image: betterdb/monitor:0.39.0-no-ai@sha256:630cf435a129f285ca2d877ea1686d9788a7c2cb346cb6e38a3d227034d294f6
    container_name: multinode-betterdb
    restart: unless-stopped
    depends_on:
@@ -82,7 +84,7 @@ services:
      # No phoning home from a local test stack.
      BETTERDB_TELEMETRY: "false"
    ports:
-      - "3001:3001"
+      - "127.0.0.1:3001:3001"
    volumes:
      - betterdb-data:/app/data
    # The image's own healthcheck probes "localhost", which resolves to ::1 while the server
@@ -69,11 +69,11 @@ x-stirling-node: &stirling-node
    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.
+    # Pinned to the shipped settings.yml.template defaults so the test stack exercises what users run.
    CLUSTER_VALKEY_POOL_ENABLED: "true"
    CLUSTER_VALKEY_POOL_MAXACTIVE: "16"
    CLUSTER_VALKEY_POOL_MAXIDLE: "16"
-    CLUSTER_VALKEY_POOL_MINIDLE: "4"
+    CLUSTER_VALKEY_POOL_MINIDLE: "0"
    CLUSTER_VALKEY_POOL_MAXWAITMILLIS: "2000"
    # SPRING_DATA_REDIS_REPOSITORIES_ENABLED is not needed: DataRedisRepositoriesAutoConfiguration is excluded (see application.properties, multinode/README.md).

@@ -174,6 +174,8 @@ services:
    environment:
      <<: *stirling-env
      SYSTEM_NODEID: "node-1"
+      # Pinned instead of the per-boot UUID default so tests can assert on registered node ids.
+      CLUSTER_NODE_ID: "multinode-1"
      CLUSTER_VALKEY_CLIENTNAME: "stirling-node-1"

  # ---- App node 2 -----------------------------------------------------------
@@ -184,6 +186,8 @@ services:
    environment:
      <<: *stirling-env
      SYSTEM_NODEID: "node-2"
+      # Pinned instead of the per-boot UUID default so tests can assert on registered node ids.
+      CLUSTER_NODE_ID: "multinode-2"
      CLUSTER_VALKEY_CLIENTNAME: "stirling-node-2"

  # ---- One-shot seeder: teams, ~40 users, S3 connection, policies -----------
@@ -5,15 +5,15 @@ mixed sync/async so the Valkey backplane sees job, lock, rate-limit and cache tr
 import json
 import os
 import random
-import ssl
 import sys
 import threading
 import time
+import traceback
 import urllib.error
 import urllib.request
 import uuid
 from collections import defaultdict
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed

 BASE_URL = os.environ.get("BASE_URL", "http://nginx:8080").rstrip("/")
 DURATION = int(os.environ.get("DURATION_SECONDS", "300"))
@@ -24,14 +24,19 @@ 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"))
+REQUEST_TIMEOUT = int(os.environ.get("REQUEST_TIMEOUT_SECONDS", "120"))
+# Above this share of failed requests the run exits non-zero, so CI cannot green on a broken stack.
+MAX_ERROR_RATE = float(os.environ.get("MAX_ERROR_RATE", "0.05"))
+# Grace for in-flight async polls after the clock runs out; bounds how far a run can overrun.
+POLL_GRACE_SECONDS = int(os.environ.get("POLL_GRACE_SECONDS", "30"))

-SSL_CTX = ssl.create_default_context()
-SSL_CTX.check_hostname = False
-SSL_CTX.verify_mode = ssl.CERT_NONE
+# Cheap plumbing calls: excluded from the PDF-operation rate so the headline number is comparable.
+NON_PDF_OPS = {"info/status", "job-poll"}

+run_started = 0.0
 stop_at = 0.0
 stats_lock = threading.Lock()
-stats = defaultdict(lambda: {"ok": 0, "err": 0, "ms": 0.0, "bytes": 0})
+stats = defaultdict(lambda: {"ok": 0, "err": 0, "lat": [], "bytes": 0})
 nodes_seen = defaultdict(int)
 errors = defaultdict(int)
 job_stats = {"submitted": 0, "completed": 0, "failed": 0, "sticky_410": 0}
@@ -68,7 +73,8 @@ def make_pdf(pages: int, filler_lines: int) -> bytes:
    offsets = []
    for idx, body in enumerate(objs, start=1):
        offsets.append(len(out))
-        out += f"{idx} 0 obj".encode() + body + b"endobj\n"
+        # Newlines around the body keep 'endstream' and 'endobj' from fusing into one token.
+        out += f"{idx} 0 obj\n".encode() + body + b"\nendobj\n"

    xref_at = len(out)
    out += f"xref\n0 {len(objs) + 1}\n".encode()
@@ -112,7 +118,7 @@ def encode_multipart(fields, files):
    return bytes(body), f"multipart/form-data; boundary={boundary}"


-def request(method, path, token=None, body=None, content_type=None, timeout=180):
+def request(method, path, token=None, body=None, content_type=None, timeout=REQUEST_TIMEOUT):
    url = path if path.startswith("http") else BASE_URL + path
    req = urllib.request.Request(url, data=body, method=method)
    if token:
@@ -120,7 +126,7 @@ def request(method, path, token=None, body=None, content_type=None, timeout=180)
    if content_type:
        req.add_header("Content-Type", content_type)
    try:
-        with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, resp.read(), dict(resp.headers)
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read(), dict(exc.headers)
@@ -128,25 +134,42 @@ def request(method, path, token=None, body=None, content_type=None, timeout=180)
        return 0, str(exc).encode(), {}


-def record(op, status, elapsed_ms, size, headers):
+def budget(default):
+    """Clamp a timeout to what is left of the run so one hung call cannot outlive DURATION."""
+    return max(1.0, min(default, (stop_at + POLL_GRACE_SECONDS) - time.time()))
+
+
+def record(op, status, elapsed_ms, size, headers, ok=None):
+    # ok overrides the 2xx check for expected non-2xx replies (async 410 re-routes).
+    if ok is None:
+        ok = 200 <= status < 300
    with stats_lock:
        entry = stats[op]
-        if 200 <= status < 300:
+        if ok:
            entry["ok"] += 1
            entry["bytes"] += size
        else:
            entry["err"] += 1
            errors[f"{op} -> {status}"] += 1
-        entry["ms"] += elapsed_ms
+        entry["lat"].append(elapsed_ms)
        served = headers.get("X-Served-By")
        if served:
            nodes_seen[served] += 1


+def percentile(values, pct):
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    idx = min(len(ordered) - 1, int(round(pct / 100.0 * (len(ordered) - 1))))
+    return ordered[idx]
+
+
 # ---------------------------------------------------------------- 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")
+    status, data, _ = request("POST", "/api/v1/auth/login", body=body,
+                              content_type="application/json", timeout=30)
    if status != 200:
        return None
    try:
@@ -241,7 +264,12 @@ 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)
+        started = time.time()
+        status, data, headers = request("GET", f"/api/v1/general/job/{job_id}", token=token,
+                                        timeout=budget(30))
+        # A 410 is an expected cross-node re-route, counted in job_stats rather than as an error.
+        record("job-poll", status, (time.time() - started) * 1000, len(data), headers,
+               ok=(status == 410 or 200 <= status < 300))
        if status == 410:
            with stats_lock:
                job_stats["sticky_410"] += 1
@@ -263,7 +291,7 @@ def poll_job(job_id, token, deadline):
 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))
+    time.sleep(max(0.0, min(rng.uniform(0, RAMP_SECONDS), stop_at - time.time())))
    while time.time() < stop_at:
        _, token = rng.choice(tokens)
        _, name, builder = rng.choices(WORKLOAD, weights=WEIGHTS, k=1)[0]
@@ -274,7 +302,8 @@ def worker(worker_id, corpus, tokens):
        body, content_type = encode_multipart(fields, files)

        started = time.time()
-        status, data, headers = request("POST", path, token=token, body=body, content_type=content_type)
+        status, data, headers = request("POST", path, token=token, body=body,
+                                        content_type=content_type, timeout=budget(REQUEST_TIMEOUT))
        elapsed = (time.time() - started) * 1000
        label = f"{name}{' (async)' if use_async else ''}"
        record(label, status, elapsed, len(data), headers)
@@ -287,7 +316,8 @@ def worker(worker_id, corpus, tokens):
            if job_id:
                with stats_lock:
                    job_stats["submitted"] += 1
-                if poll_job(job_id, token, time.time() + 240):
+                deadline = min(time.time() + 240, stop_at + POLL_GRACE_SECONDS)
+                if poll_job(job_id, token, deadline):
                    with stats_lock:
                        job_stats["completed"] += 1
                else:
@@ -295,48 +325,62 @@ def worker(worker_id, corpus, tokens):
                        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)
+        if time.time() < stop_at and rng.random() < 0.3:
+            started = time.time()
+            s, d, h = request("GET", "/api/v1/info/status", token=token, timeout=budget(20))
+            record("info/status", s, (time.time() - started) * 1000, len(d), h)


 def progress_printer():
-    last = 0
+    last_ops = 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())
+            ops = sum(v["ok"] + v["err"] for op, v in stats.items() if op not in NON_PDF_OPS)
            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
+        rate = (ops - last_ops) / 15.0
+        last_ops = ops
        print(
-            f"    [{remaining:4d}s left] {total:6d} reqs  {ok:6d} ok  {rate:5.1f} req/s  "
+            f"    [{remaining:4d}s left] {total:6d} reqs  {ok:6d} ok  {rate:5.1f} pdf-op/s  "
            f"{mb:7.1f} MB down  async {jobs['completed']}/{jobs['submitted']}",
            flush=True,
        )


 def report():
-    print("\n" + "=" * 78)
+    """Print per-endpoint counts and latency percentiles. Returns (ok, err) totals."""
+    elapsed = max(0.001, time.time() - run_started)
+    with stats_lock:
+        snapshot = {op: dict(entry, lat=list(entry["lat"])) for op, entry in stats.items()}
+    print("\n" + "=" * 84)
    print(" LOAD TEST SUMMARY")
-    print("=" * 78)
-    print(f"{'operation':24s} {'ok':>7s} {'err':>6s} {'avg ms':>9s} {'MB down':>9s}")
-    print("-" * 78)
+    print("=" * 84)
+    print(f"{'operation':22s} {'ok':>7s} {'err':>6s} {'p50':>7s} {'p95':>7s} {'p99':>7s} "
+          f"{'req/s':>7s} {'MB':>7s}")
+    print("-" * 84)
    total_ok = total_err = 0
-    for op in sorted(stats):
-        entry = stats[op]
+    for op in sorted(snapshot):
+        entry = snapshot[op]
+        lat = entry["lat"]
        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}"
+            f"{op:22s} {entry['ok']:7d} {entry['err']:6d} {percentile(lat, 50):7.0f} "
+            f"{percentile(lat, 95):7.0f} {percentile(lat, 99):7.0f} {calls / elapsed:7.2f} "
+            f"{entry['bytes'] / 1024 / 1024:7.1f}"
        )
-    print("-" * 78)
-    print(f"{'TOTAL':24s} {total_ok:7d} {total_err:6d}")
+    print("-" * 84)
+    total = total_ok + total_err
+    print(f"{'TOTAL':22s} {total_ok:7d} {total_err:6d} {'':7s} {'':7s} {'':7s} "
+          f"{total / elapsed:7.2f}")
+    pdf_calls = sum(e["ok"] + e["err"] for op, e in snapshot.items() if op not in NON_PDF_OPS)
+    print(f"\n PDF operations only: {pdf_calls} calls, {pdf_calls / elapsed:.2f} op/s over "
+          f"{elapsed:.0f}s (health probes and job polls excluded).")

    print("\n Load-balancer spread (X-Served-By):")
    for node, count in sorted(nodes_seen.items(), key=lambda kv: -kv[1]):
@@ -350,7 +394,8 @@ def report():
        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)
+    print("=" * 84, flush=True)
+    return total_ok, total_err


 def wait_for_app():
@@ -365,7 +410,7 @@ def wait_for_app():


 def main():
-    global stop_at
+    global stop_at, run_started
    if not wait_for_app():
        print("app never came up", file=sys.stderr)
        return 1
@@ -376,7 +421,8 @@ def main():
        print("no logins succeeded - cannot generate authenticated load", file=sys.stderr)
        return 1

-    stop_at = time.time() + DURATION
+    run_started = time.time()
+    stop_at = run_started + DURATION
    print(
        f"\n==> Driving load for {DURATION}s: {CONCURRENCY} workers, "
        f"{len(tokens)} users, {int(ASYNC_RATIO * 100)}% async\n",
@@ -384,10 +430,29 @@ def main():
    )
    ticker = threading.Thread(target=progress_printer, daemon=True)
    ticker.start()
+    crashed = 0
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
-        for i in range(CONCURRENCY):
-            pool.submit(worker, i, corpus, tokens)
-    report()
+        futures = [pool.submit(worker, i, corpus, tokens) for i in range(CONCURRENCY)]
+        for future in as_completed(futures):
+            try:
+                future.result()
+            except Exception:
+                crashed += 1
+                traceback.print_exc()
+    total_ok, total_err = report()
+
+    total = total_ok + total_err
+    if crashed:
+        print(f"\n{crashed} worker(s) crashed - see the tracebacks above", file=sys.stderr)
+        return 1
+    if not total:
+        print("\nno requests were issued", file=sys.stderr)
+        return 1
+    err_rate = total_err / total
+    if err_rate > MAX_ERROR_RATE:
+        print(f"\nerror rate {err_rate:.1%} exceeds MAX_ERROR_RATE {MAX_ERROR_RATE:.1%}",
+              file=sys.stderr)
+        return 1
    return 0


@@ -1,5 +1,5 @@
 #!/bin/sh
-# Seeds a running stack (teams, users, S3 connection + policy). Best-effort and idempotent-ish.
+# Seeds a running stack (teams, users, S3 connection + policy). Idempotent; exits non-zero if a create fails.
 # Auth uses the Bearer JWT from the login body, not a cookie - the global API key can't create teams.
 set -u

@@ -40,11 +40,23 @@ TOKEN=$(jq -r '.session.access_token' </tmp/login.json)

 auth() { curl -sS -H "Authorization: Bearer $TOKEN" "$@"; }

+errors=0
+fail_note() { errors=$((errors+1)); log "ERROR: $*"; }
+
+# The regression runner re-seeds an already-seeded stack, so every create is check-then-create by name.
+find_id_by_name() { # $1=list path  $2=name -> the existing id, or empty
+  auth "$BASE_URL$1" 2>/dev/null \
+    | jq -r --arg n "$2" '[.. | objects | select(.name? == $n) | .id?] | map(select(. != null)) | first // empty' 2>/dev/null
+}
+
 # --- teams -------------------------------------------------------------------
 for t in $TEAMS; do
  code=$(auth -o /dev/null -w '%{http_code}' -X POST "$BASE_URL/api/v1/team/create" \
    --data-urlencode "name=$t")
-  log "team '$t': HTTP $code"
+  case "$code" in
+    2*|409) log "team '$t': HTTP $code" ;;
+    *)      fail_note "team '$t': HTTP $code" ;;
+  esac
 done

 # Resolve team ids from the DB (no admin list endpoint self-hosted).
@@ -82,6 +94,7 @@ while [ "$n" -le "$USER_COUNT" ]; do
  n=$((n+1))
 done
 log "users created: $created (failed: $failed, requested: $USER_COUNT)"
+[ "$failed" -eq 0 ] || fail_note "$failed of $USER_COUNT user creates failed"

 # --- S3 connection -> the in-cluster MinIO 'policy-data' bucket ---------------
 conn_body=$(cat <<JSON
@@ -89,9 +102,15 @@ conn_body=$(cat <<JSON
 "config":{"bucket":"policy-data","region":"us-east-1","endpoint":"http://minio:9000","accessKeyId":"minioadmin","secretAccessKey":"minioadmin","pathStyleAccess":true}}
 JSON
 )
-conn_id=$(auth -X POST "$BASE_URL/api/v1/integrations" -H 'Content-Type: application/json' -d "$conn_body" \
-  | jq -r '.id // empty' 2>/dev/null)
-log "S3 connection id: ${conn_id:-<none>}"
+conn_id=$(find_id_by_name "/api/v1/integrations" "MinIO policy bucket")
+if [ -n "${conn_id:-}" ]; then
+  log "S3 connection already exists: id $conn_id"
+else
+  conn_id=$(auth -X POST "$BASE_URL/api/v1/integrations" -H 'Content-Type: application/json' -d "$conn_body" \
+    | jq -r '.id // empty' 2>/dev/null)
+  log "S3 connection id: ${conn_id:-<none>}"
+  [ -n "${conn_id:-}" ] || fail_note "S3 connection was not created"
+fi

 # --- a scheduled S3 -> compress -> S3 policy ---------------------------------
 if [ -n "${conn_id:-}" ]; then
@@ -100,9 +119,15 @@ if [ -n "${conn_id:-}" ]; then
 "options":{"connectionId":$conn_id,"prefix":"incoming/","mode":"consume"}}
 JSON
 )
-  src_id=$(auth -X POST "$BASE_URL/api/v1/sources" -H 'Content-Type: application/json' -d "$src_body" \
-    | jq -r '.id // empty' 2>/dev/null)
-  log "S3 source id: ${src_id:-<none>}"
+  src_id=$(find_id_by_name "/api/v1/sources" "Incoming S3")
+  if [ -n "${src_id:-}" ]; then
+    log "S3 source already exists: id $src_id"
+  else
+    src_id=$(auth -X POST "$BASE_URL/api/v1/sources" -H 'Content-Type: application/json' -d "$src_body" \
+      | jq -r '.id // empty' 2>/dev/null)
+    log "S3 source id: ${src_id:-<none>}"
+    [ -n "${src_id:-}" ] || fail_note "S3 source was not created"
+  fi

  if [ -n "${src_id:-}" ]; then
    pol_body=$(cat <<JSON
@@ -113,9 +138,18 @@ JSON
 "output":{"type":"s3","options":{"connectionId":$conn_id,"prefix":"processed/"}}}
 JSON
 )
-    code=$(auth -o /tmp/pol.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/policies" \
-      -H 'Content-Type: application/json' -d "$pol_body")
-    log "policy create: HTTP $code $( [ "$code" != 200 ] && head -c 160 /tmp/pol.json )"
+    pol_id=$(find_id_by_name "/api/v1/policies" "Compress incoming PDFs")
+    if [ -n "${pol_id:-}" ]; then
+      log "policy already exists: id $pol_id"
+    else
+      code=$(auth -o /tmp/pol.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/policies" \
+        -H 'Content-Type: application/json' -d "$pol_body")
+      log "policy create: HTTP $code $( [ "$code" != 200 ] && head -c 160 /tmp/pol.json )"
+      case "$code" in
+        2*|409) ;;
+        *)      fail_note "policy create: HTTP $code" ;;
+      esac
+    fi
  fi

  # --- webhook source + policy (only if this build has the webhook type) -----
@@ -124,16 +158,27 @@ JSON
 "options":{"connectionId":$conn_id,"mode":"consume"}}
 JSON
 )
-  wh=$(auth -o /tmp/wh.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/sources" \
-    -H 'Content-Type: application/json' -d "$wh_body")
-  if [ "$wh" = "200" ] || [ "$wh" = "201" ]; then
-    wh_url=$(jq -r '.options.webhookId // empty' </tmp/wh.json 2>/dev/null)
-    log "webhook source created (deliver to /api/v1/webhooks/$wh_url)"
+  wh_id=$(find_id_by_name "/api/v1/sources" "Partner webhook")
+  if [ -n "${wh_id:-}" ]; then
+    log "webhook source already exists: id $wh_id"
  else
-    log "webhook source not created (HTTP $wh) - expected on builds without the webhook branch"
+    wh=$(auth -o /tmp/wh.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/sources" \
+      -H 'Content-Type: application/json' -d "$wh_body")
+    if [ "$wh" = "200" ] || [ "$wh" = "201" ]; then
+      wh_url=$(jq -r '.options.webhookId // empty' </tmp/wh.json 2>/dev/null)
+      log "webhook source created (deliver to /api/v1/webhooks/$wh_url)"
+    else
+      # Not counted as an error: builds without the webhook branch legitimately reject this type.
+      log "webhook source not created (HTTP $wh) - expected on builds without the webhook branch"
+    fi
  fi
 fi

+if [ "$errors" -gt 0 ]; then
+  log "seed FAILED with $errors error(s)."
+  exit 1
+fi
+
 log "seed complete."
 log "  login: $ADMIN_USER / $ADMIN_PASS at $BASE_URL"
 log "  users: user01..$(printf '%02d' "$USER_COUNT")@stirling.test / $USER_PASS"
@@ -14,7 +14,8 @@ 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)      [ "$#" -ge 2 ] || { echo "--valkey needs a value (standalone|sentinel|cluster)"; exit 2; }
+                   VALKEY_TOPOLOGY="$2"; shift 2 ;;
    --valkey=*)    VALKEY_TOPOLOGY="${1#*=}"; shift ;;
    *) echo "Unknown argument '$1'"; exit 2 ;;
  esac
@@ -25,12 +26,18 @@ done
 COMPOSE=$(compose_cmd_for_topology "$VALKEY_TOPOLOGY") \
  || { echo "Unknown --valkey topology '$VALKEY_TOPOLOGY' (expected standalone|sentinel|cluster)"; exit 2; }

+SEED_LOG="$(pwd)/multinode/seed.log"
+
 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
+# Exact match: "unhealthy" contains "healthy", so a substring test reads a broken node as up.
+node_health=$(docker inspect -f '{{.State.Health.Status}}' multinode-stirling-1 2>/dev/null | tr -d '\r')
+if [ "$node_health" != "healthy" ]; then
  ./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)"
+  if ! $COMPOSE --profile seed run --rm seed 2>&1 | tee "$SEED_LOG"; then
+    echo "    (seed reported issues, continuing; full output in $SEED_LOG)"
+  fi
 fi

 echo "==> Checking Python + behave..."
@@ -55,12 +62,24 @@ run_behave "~@destructive" "core" || rc=1
 if [ "$RUN_FAILOVER" = 1 ]; then
  run_behave "@destructive" "failover" || rc=1
  echo "==> Restoring any killed nodes..."
-  $COMPOSE up -d >/dev/null 2>&1
+  if ! $COMPOSE up -d --remove-orphans; then
+    echo "    ERROR: could not restore the stack ('up -d' failed)"
+    rc=1
+  fi
  for n in multinode-stirling-1 multinode-stirling-2; do
+    restored=0
    for i in $(seq 1 24); do
-      [ "$(docker inspect -f '{{.State.Health.Status}}' "$n" 2>/dev/null)" = "healthy" ] && break
+      [ "$(docker inspect -f '{{.State.Health.Status}}' "$n" 2>/dev/null | tr -d '\r')" = "healthy" ] \
+        && { restored=1; break; }
      sleep 5
    done
+    if [ "$restored" = 1 ]; then
+      echo "    $n: healthy"
+    else
+      echo "    ERROR: $n did not return to healthy after failover; last 40 log lines:"
+      docker logs --tail 40 "$n" 2>&1 | sed 's/^/      /'
+      rc=1
+    fi
  done
 fi

@@ -11,7 +11,8 @@ DOWN=0
 SEED=1
 while [ "$#" -gt 0 ]; do
  case "$1" in
-    --valkey)   VALKEY_TOPOLOGY="${2:-}"; shift 2 ;;
+    --valkey)   [ "$#" -ge 2 ] || { echo "--valkey needs a value (standalone|sentinel|cluster)"; exit 2; }
+                VALKEY_TOPOLOGY="$2"; shift 2 ;;
    --valkey=*) VALKEY_TOPOLOGY="${1#*=}"; shift ;;
    --down)     DOWN=1; shift ;;
    --no-seed)  SEED=0; shift ;;
@@ -22,12 +23,30 @@ done
 COMPOSE=$(compose_cmd_for_topology "$VALKEY_TOPOLOGY") \
  || { echo "Unknown --valkey topology '$VALKEY_TOPOLOGY' (expected standalone|sentinel|cluster)"; exit 2; }

+# Records which topology is live so switching overlays tears the old services down instead of
+# orphaning them (stale containers otherwise confuse validate-multinode-test.sh's detection).
+TOPOLOGY_MARKER="multinode/.active-topology"
+PREV_TOPOLOGY=""
+if [ -f "$TOPOLOGY_MARKER" ]; then
+  PREV_TOPOLOGY=$(tr -d '[:space:]' < "$TOPOLOGY_MARKER" 2>/dev/null || echo "")
+fi
+
 if [ "$DOWN" = "1" ]; then
  echo "Tearing down multi-node stack + volumes..."
  $COMPOSE --profile seed down -v --remove-orphans
+  rm -f "$TOPOLOGY_MARKER"
  exit 0
 fi

+if [ -n "$PREV_TOPOLOGY" ] && [ "$PREV_TOPOLOGY" != "$VALKEY_TOPOLOGY" ]; then
+  echo "==> Topology change ($PREV_TOPOLOGY -> $VALKEY_TOPOLOGY); tearing the old stack down first..."
+  PREV_COMPOSE=$(compose_cmd_for_topology "$PREV_TOPOLOGY") || PREV_COMPOSE=""
+  if [ -n "$PREV_COMPOSE" ]; then
+    $PREV_COMPOSE --profile seed down -v --remove-orphans || true
+  fi
+  rm -f "$TOPOLOGY_MARKER"
+fi
+
 # 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."
@@ -38,7 +57,8 @@ echo "==> Building the Stirling image (first run compiles the app; be patient)..
 $COMPOSE build

 echo "==> Starting Postgres + Valkey ($VALKEY_TOPOLOGY) + MinIO + 2 app nodes + nginx..."
-$COMPOSE up -d
+$COMPOSE up -d --remove-orphans
+printf '%s\n' "$VALKEY_TOPOLOGY" > "$TOPOLOGY_MARKER"

 echo "==> Waiting for both app nodes to report healthy..."
 for node in multinode-stirling-1 multinode-stirling-2; do
@@ -20,7 +20,8 @@ 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
+  # Must be a RUNNING sentinel: docker inspect also succeeds for a stopped leftover from a previous run.
+  elif [ -n "$(docker ps --filter name=multinode-valkey-sentinel-1 --filter status=running -q 2>/dev/null)" ]; then
    echo sentinel
  else
    echo standalone
@@ -115,19 +116,23 @@ 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=""
+sampled=0; anon=0; lettuce=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')
+  lettuce=$(( lettuce + $(printf '%s\n' "$clients" | grep -c '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, -)"
+echo "  sampled $sampled Valkey node(s); $lettuce Lettuce connection(s); app connection names: $(printf '%s' "$named" | paste -sd, -)"
 if [ "$sampled" -lt 1 ]; then
  bad "no reachable Valkey container found - cannot census client names"
+elif [ "${lettuce:-0}" -eq 0 ]; then
+  # Population floor: with zero Lettuce clients the anon count is also zero and would falsely pass.
+  bad "no Lettuce connections found across $sampled node(s) - nothing to census (are the app nodes connected?)"
 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"
@@ -135,9 +140,9 @@ else
                              || 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.
+echo "== 7. Backplane traffic multiplexes over one shared connection =="
+# Job load must not open a stream of new connections: every backplane command rides the shared native
+# connection. This measures multiplexing, NOT whether the Lettuce pool is enabled.
 JOBS=200
 CHURN_BUDGET=50
 conns_total() {
@@ -150,9 +155,9 @@ conns_total() {
 }
 vk_count=$(printf '%s\n' "$VALKEY_NODES" | grep -c .)
 if [ -z "${jwt:-}" ]; then
-  skip "no JWT - cannot drive load to measure churn"
+  skip "no JWT - cannot drive load to measure connection growth"
 elif [ "${vk_count:-0}" -lt 1 ]; then
-  bad "no reachable Valkey container found - cannot measure connection churn"
+  bad "no reachable Valkey container found - cannot measure connection growth"
 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"
@@ -168,11 +173,11 @@ else
  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"
+    bad "no async jobs were accepted - cannot measure connection growth"
  elif [ "$delta" -lt "$CHURN_BUDGET" ]; then
-    ok "only $delta new connections for $submitted async jobs - connections are pooled"
+    ok "only $delta new connections for $submitted async jobs - backplane multiplexes over the shared connection"
  else
-    bad "$delta new connections for $submitted async jobs - pooling looks disabled (expected < $CHURN_BUDGET)"
+    bad "$delta new connections for $submitted async jobs - a connection is opened per operation (expected < $CHURN_BUDGET)"
  fi
 fi

@@ -12,9 +12,12 @@ from behave import given, then, when

 LB_URL = "http://localhost:8080"
 NODES = ["multinode-stirling-1", "multinode-stirling-2"]
+# Must match CLUSTER_NODE_ID in testing/compose/docker-compose-multinode.yml.
+NODE_IDS = ["multinode-1", "multinode-2"]
 PG = "multinode-postgres"
 MINIO = "multinode-minio"
-# The primary keeps this name in every Valkey topology (standalone, sentinel, cluster).
+# The container name is stable but its Valkey role is not (cluster shards, sentinel failover), so
+# every keyspace probe must fan out in cluster mode - see _backplane_keys.
 VALKEY = "multinode-valkey"
 BUCKET = "policy-data"
 SOURCE_PREFIX = "incoming/"
@@ -404,34 +407,70 @@ def step_run_visible_every(context):
 _NODE_PREFIX = re.compile(r"^[A-Za-z0-9_.\-]+:\d+:\s?")


+def _cluster_mode():
+    """True when this Valkey runs in cluster mode, so a plain KEYS would see one shard only."""
+    rc, out, err = _sh(["docker", "exec", VALKEY, "valkey-cli", "info", "cluster"], timeout=30)
+    assert rc == 0, f"valkey INFO cluster failed: {err.strip() or out.strip()}"
+    return "cluster_enabled:1" in out
+
+
 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:
+    if _cluster_mode():
+        # A fan-out failure must be loud: falling back to a single-shard KEYS would silently
+        # report a fraction of the keyspace and every downstream count would be wrong.
+        rc, out, err = _sh(["docker", "exec", VALKEY, "valkey-cli", "--cluster", "call",
+                            "--cluster-only-masters", "127.0.0.1:6379", "keys", "stirling:*"],
+                           timeout=60)
+        assert rc == 0, f"valkey cluster fan-out failed: {err.strip() or out.strip()}"
+    else:
        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:")]


+def _post_through_lb(context, count=3):
+    """POST through the LB; returns True if a node reported the X-Rate-Limit-Remaining header."""
+    if not getattr(context, "jwt_token", None):
+        _lb_login(context)
+    limited = False
+    for _ in range(count):
+        marker = uuid.uuid4().hex[:8]
+        r = requests.post(f"{LB_URL}/api/v1/general/rotate-pdf",
+                          headers={"Authorization": f"Bearer {context.jwt_token}"},
+                          files={"fileInput": (f"rl-{marker}.pdf", _pdf_bytes(marker),
+                                               "application/pdf")},
+                          data={"angle": 90}, timeout=60)
+        limited = limited or "X-Rate-Limit-Remaining" in r.headers
+    return limited
+
+
@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.
-    keys = _backplane_keys()
-    assert keys, "no stirling:* keys in Valkey - rate-limit/backplane state is not shared"
+    # Heartbeat keys are always present, so only a stirling:rl: bucket created by real traffic
+    # proves the counters live in the backplane rather than per node.
+    if not _post_through_lb(context):
+        context.scenario.skip(
+            "rate limiting is not active on this stack (no X-Rate-Limit-Remaining header on a "
+            "POST through the LB), so no stirling:rl: bucket can exist - shared rate limiting is "
+            "unproven here, not proven")
+        return
+    buckets = [k for k in _backplane_keys() if k.startswith("stirling:rl:")]
+    assert buckets, (
+        "POSTs through the LB were rate limited but the backplane holds no stirling:rl: key - "
+        "the counters are per node, 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)}")
+    # Assert the exact node ids: counting keys lets a stale heartbeat from an earlier run stand in
+    # for a node that never registered.
+    keys = set(_backplane_keys())
+    missing = [nid for nid in NODE_IDS if f"stirling:nodes:{nid}" not in keys]
+    assert not missing, (
+        f"no stirling:nodes: heartbeat for {missing}; the backplane holds "
+        f"{sorted(k for k in keys if k.startswith('stirling:nodes:'))}")


 # --------------------------------------------------------------------------- failover