diff --git a/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java b/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java index ec7c5665e3..6cba18023f 100644 --- a/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java +++ b/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java @@ -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() + "."); } diff --git a/app/common/src/main/java/stirling/software/common/cluster/HostPort.java b/app/common/src/main/java/stirling/software/common/cluster/HostPort.java index 54367e9aa2..a5ffe0f6a9 100644 --- a/app/common/src/main/java/stirling/software/common/cluster/HostPort.java +++ b/app/common/src/main/java/stirling/software/common/cluster/HostPort.java @@ -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} */ diff --git a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java index b6621934ce..7ec39164cb 100644 --- a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java +++ b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java @@ -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; diff --git a/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java b/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java index 5a401cc68b..8e99a3ca45 100644 --- a/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java +++ b/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java @@ -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); diff --git a/app/core/src/main/resources/settings.yml.template b/app/core/src/main/resources/settings.yml.template index 6e3d0d1dd8..5089336c47 100644 --- a/app/core/src/main/resources/settings.yml.template +++ b/app/core/src/main/resources/settings.yml.template @@ -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 ...'. 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 ...'. 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 diff --git a/app/proprietary/build.gradle b/app/proprietary/build.gradle index 4ae6053a51..08aa79ce58 100644 --- a/app/proprietary/build.gradle +++ b/app/proprietary/build.gradle @@ -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}" diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfiguration.java b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfiguration.java index 1d22a96360..b1d19ac5a3 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfiguration.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfiguration.java @@ -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. */ diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStore.java b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStore.java index 7e24c238c5..6cefa68703 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStore.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStore.java @@ -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 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 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 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 args = new ArrayList<>(1 + fields.size() * 2); - args.add(Long.toString(ttlMs)); + List fieldArgs = new ArrayList<>(fields.size() * 2); for (Map.Entry f : fields.entrySet()) { - args.add(f.getKey()); - args.add(f.getValue()); + fieldArgs.add(f.getKey()); + fieldArgs.add(f.getValue()); } + if (!isClusterAware()) { + List keys = new ArrayList<>(1 + indexKeys.size()); + keys.add(key); + keys.addAll(indexKeys); + List 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 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) { diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/security/InitialSecuritySetup.java b/app/proprietary/src/main/java/stirling/software/proprietary/security/InitialSecuritySetup.java index 55a3c89cdc..c521a2f320 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/security/InitialSecuritySetup.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/security/InitialSecuritySetup.java @@ -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.", diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeySentinelModeTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveExternalSentinelTest.java similarity index 95% rename from app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeySentinelModeTest.java rename to app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveExternalSentinelTest.java index e7ab0dd70c..672f852a45 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeySentinelModeTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveExternalSentinelTest.java @@ -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); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeyPoolingTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeyPoolingTest.java index 9bb4e05fb1..952f1851da 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeyPoolingTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeyPoolingTest.java @@ -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-, 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 clients = diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfigurationTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfigurationTest.java index 476d2b9c69..9aec7bf2c5 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfigurationTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfigurationTest.java @@ -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 diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStoreTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStoreTest.java index 12b793fb62..67f5589395 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStoreTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyJobStoreTest.java @@ -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 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 writtenKeys = new ArrayList<>(); + List 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 setKey = ArgumentCaptor.forClass(String.class); + ArgumentCaptor setValue = ArgumentCaptor.forClass(String.class); + ArgumentCaptor 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()); } } diff --git a/testing/compose/docker-compose-multinode.betterdb.yml b/testing/compose/docker-compose-multinode.betterdb.yml index 62486d79b1..1f749a76a3 100644 --- a/testing/compose/docker-compose-multinode.betterdb.yml +++ b/testing/compose/docker-compose-multinode.betterdb.yml @@ -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 diff --git a/testing/compose/docker-compose-multinode.yml b/testing/compose/docker-compose-multinode.yml index 6826ed1bcf..8b300973e5 100644 --- a/testing/compose/docker-compose-multinode.yml +++ b/testing/compose/docker-compose-multinode.yml @@ -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 ----------- diff --git a/testing/compose/multinode/loadgen.py b/testing/compose/multinode/loadgen.py index b55c47376d..4f1cebab84 100644 --- a/testing/compose/multinode/loadgen.py +++ b/testing/compose/multinode/loadgen.py @@ -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 diff --git a/testing/compose/multinode/seed.sh b/testing/compose/multinode/seed.sh index 9ccbff0520..1f77387db3 100644 --- a/testing/compose/multinode/seed.sh +++ b/testing/compose/multinode/seed.sh @@ -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' 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 </dev/null) -log "S3 connection id: ${conn_id:-}" +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:-}" + [ -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:-}" + 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:-}" + [ -n "${src_id:-}" ] || fail_note "S3 source was not created" + fi if [ -n "${src_id:-}" ]; then pol_body=$(cat </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' /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" diff --git a/testing/compose/run-multinode-regression.sh b/testing/compose/run-multinode-regression.sh index 3056a7d97d..6f48cc9b58 100644 --- a/testing/compose/run-multinode-regression.sh +++ b/testing/compose/run-multinode-regression.sh @@ -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 diff --git a/testing/compose/start-multinode-test.sh b/testing/compose/start-multinode-test.sh index c1dca19e3b..c2a54fa260 100644 --- a/testing/compose/start-multinode-test.sh +++ b/testing/compose/start-multinode-test.sh @@ -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 diff --git a/testing/compose/validate-multinode-test.sh b/testing/compose/validate-multinode-test.sh index 80501a0c08..200680c9a7 100644 --- a/testing/compose/validate-multinode-test.sh +++ b/testing/compose/validate-multinode-test.sh @@ -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<>endobj\n2 0 obj<>endobj\n3 0 obj<>endobj\ntrailer<>\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 diff --git a/testing/cucumber/features/steps/multinode_step_definitions.py b/testing/cucumber/features/steps/multinode_step_definitions.py index ddaf99248d..6a6f54ef1d 100644 --- a/testing/cucumber/features/steps/multinode_step_definitions.py +++ b/testing/cucumber/features/steps/multinode_step_definitions.py @@ -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: 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