Add Valkey cluster backplane and sticky-410 ownership (clusters) (#6472)

2026-07-16 11:23:10 +02:00 · 2026-06-02 14:59:20 +01:00
parent de9d6ad3f5
commit b355ccec9e
30 changed files with 3858 additions and 213 deletions
@@ -0,0 +1,40 @@
+package stirling.software.proprietary.cluster;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Configuration;
+
+import jakarta.annotation.PostConstruct;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Runtime license gate for cluster mode. Cluster mode requires a SERVER or ENTERPRISE license; the
+ * SaaS flavor bypasses (no {@code runningProOrHigher} bean is published). The Valkey connection
+ * config {@code @DependsOn} this bean, so it runs before any Valkey bean is constructed.
+ */
+@Configuration
+@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
+@Slf4j
+public class ClusterLicenseGate {
+
+    @Autowired(required = false)
+    @Qualifier("runningProOrHigher")
+    private Boolean runningProOrHigher;
+
+    @PostConstruct
+    void verifyLicense() {
+        if (runningProOrHigher == null) {
+            return; // saas flavor - licensed via Stripe elsewhere
+        }
+        if (!runningProOrHigher) {
+            throw new IllegalStateException(
+                    "Cluster mode (cluster.enabled=true) requires a SERVER or"
+                            + " ENTERPRISE license. Configure stirling.premium.key with a valid"
+                            + " license key (contact [email protected] to obtain one), or set"
+                            + " cluster.enabled=false.");
+        }
+        log.info("Cluster license gate: SERVER/ENTERPRISE license verified, cluster mode allowed.");
+    }
+}
@@ -0,0 +1,113 @@
+package stirling.software.proprietary.cluster;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+
+import stirling.software.common.cluster.StickyMissRecorder;
+import stirling.software.common.model.ApplicationProperties;
+
+/**
+ * Cluster operation metrics exposed via {@code /actuator/prometheus}. Registered only when cluster
+ * mode is on.
+ */
+@Component
+@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
+public class ClusterMetrics implements StickyMissRecorder {
+
+    private final MeterRegistry registry;
+    private final ApplicationProperties applicationProperties;
+
+    private final Counter stickyMissTotal;
+    private final Counter rateLimitRejected;
+    private final Timer backplaneLatency;
+    private final Timer jobWaitSeconds;
+
+    // Per-lane queue depth gauges. Lanes are a fixed enum (FAST, SLOW, AI), so we register all
+    // three eagerly so dashboards never have a missing series.
+    private static final List<String> KNOWN_LANES = List.of("FAST", "SLOW", "AI");
+    private final ConcurrentHashMap<String, AtomicLong> queueDepth = new ConcurrentHashMap<>();
+
+    private final AtomicLong jobsInflight = new AtomicLong();
+
+    public ClusterMetrics(MeterRegistry registry, ApplicationProperties applicationProperties) {
+        this.registry = registry;
+        this.applicationProperties = applicationProperties;
+        this.stickyMissTotal =
+                Counter.builder("stirling_cluster_sticky_miss_total")
+                        .description(
+                                "Sticky-session misses: a download for a job whose result lives on"
+                                        + " a peer node landed on this node. High sustained value means"
+                                        + " LB affinity is broken.")
+                        .register(registry);
+        this.rateLimitRejected =
+                Counter.builder("stirling_cluster_ratelimit_rejected_total")
+                        .description("Cluster-wide rate limit rejections")
+                        .register(registry);
+        this.backplaneLatency =
+                Timer.builder("stirling_cluster_backplane_latency_seconds")
+                        .description("Backplane round-trip latency")
+                        .register(registry);
+        this.jobWaitSeconds =
+                Timer.builder("stirling_cluster_job_wait_seconds")
+                        .description("Time jobs spend queued before execution")
+                        .register(registry);
+        Gauge.builder("stirling_cluster_jobs_inflight", jobsInflight, AtomicLong::doubleValue)
+                .description("Jobs currently in flight on this node")
+                .tag("node", applicationProperties.getCluster().resolvedNodeId())
+                .register(registry);
+        for (String lane : KNOWN_LANES) {
+            ensureLaneGauge(lane);
+        }
+    }
+
+    @Override
+    public void recordStickyMiss() {
+        stickyMissTotal.increment();
+    }
+
+    public void recordRateLimitReject() {
+        rateLimitRejected.increment();
+    }
+
+    public Timer backplaneLatency() {
+        return backplaneLatency;
+    }
+
+    public Timer jobWaitSeconds() {
+        return jobWaitSeconds;
+    }
+
+    public void incrementInflight() {
+        jobsInflight.incrementAndGet();
+    }
+
+    public void decrementInflight() {
+        jobsInflight.decrementAndGet();
+    }
+
+    public void setQueueDepth(String lane, long depth) {
+        ensureLaneGauge(lane).set(depth);
+    }
+
+    private AtomicLong ensureLaneGauge(String lane) {
+        return queueDepth.computeIfAbsent(
+                lane,
+                l -> {
+                    AtomicLong holder = new AtomicLong();
+                    Gauge.builder("stirling_cluster_queue_depth", holder, AtomicLong::doubleValue)
+                            .description("Pending items in a job queue lane")
+                            .tag("lane", l)
+                            .register(registry);
+                    return holder;
+                });
+    }
+}
@@ -0,0 +1,201 @@
+package stirling.software.proprietary.cluster;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Locale;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.ClusterNode;
+import stirling.software.common.cluster.InstanceRegistry;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.ApplicationProperties.Cluster;
+
+/**
+ * Registers the local node with {@link InstanceRegistry} on startup, refreshes the entry at 1/3 of
+ * the TTL, and deregisters cleanly on shutdown.
+ *
+ * <p>Implements {@link SmartLifecycle} with {@code getPhase() == Integer.MAX_VALUE} so Spring tears
+ * this bean down before {@code LettuceConnectionFactory} - deregister therefore runs while the
+ * Valkey connection is still alive.
+ */
+@Component
+@Slf4j
+@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
+public class ClusterNodeBootstrap implements SmartLifecycle {
+
+    private final Duration heartbeatTtl;
+
+    private final ApplicationProperties applicationProperties;
+    private final InstanceRegistry instanceRegistry;
+
+    @Value("${server.port:8080}")
+    private int serverPort;
+
+    private volatile String nodeId;
+    private volatile String internalAddress;
+    private volatile boolean running = false;
+
+    public ClusterNodeBootstrap(
+            ApplicationProperties applicationProperties, InstanceRegistry instanceRegistry) {
+        this.applicationProperties = applicationProperties;
+        this.instanceRegistry = instanceRegistry;
+        Cluster cluster = applicationProperties.getCluster();
+        // Default must match the @Scheduled fallback below AND the model default
+        // (ApplicationProperties.Cluster.Node.heartbeatIntervalMs = 5000); otherwise the TTL is
+        // computed from a different interval than the scheduler runs at and the 3x margin breaks.
+        long heartbeatMs =
+                cluster.getNode() == null ? 5000L : cluster.getNode().getHeartbeatIntervalMs();
+        // TTL = 3x heartbeat: tolerate two missed ticks before the node drops out of the registry.
+        this.heartbeatTtl = Duration.ofMillis(heartbeatMs * 3);
+    }
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void registerOnStartup() {
+        nodeId = applicationProperties.getCluster().resolvedNodeId();
+        internalAddress = resolveInternalAddress();
+        registerSelf("register");
+    }
+
+    @Scheduled(fixedDelayString = "${cluster.node.heartbeat-interval-ms:5000}")
+    public void heartbeat() {
+        // Heartbeat-after-stop race: SmartLifecycle.stop() deregisters, but the @Scheduled
+        // tick keeps firing during a slow drain. Without this guard, the next tick re-registers
+        // the dead node and the entry resurfaces in the registry until TTL expiry.
+        if (!running) {
+            return;
+        }
+        if (nodeId == null) {
+            return; // not yet registered (startup race)
+        }
+        // Self-healing: register() is idempotent and re-populates every field, so a wiped
+        // Valkey (FLUSHALL, hash eviction) recovers on the next tick without operator action.
+        registerSelf("heartbeat");
+    }
+
+    private void registerSelf(String reason) {
+        try {
+            instanceRegistry.register(
+                    new ClusterNode(nodeId, internalAddress, Instant.now(), role()), heartbeatTtl);
+            if ("register".equals(reason)) {
+                log.info(
+                        "Cluster node registered: nodeId={}, internalAddress={}, role={}, ttl={}s",
+                        nodeId,
+                        internalAddress,
+                        role(),
+                        heartbeatTtl.toSeconds());
+            }
+        } catch (RuntimeException e) {
+            log.debug("Cluster {} failed for {}", reason, nodeId, e);
+        }
+    }
+
+    @Override
+    public void start() {
+        running = true;
+    }
+
+    @Override
+    public void stop() {
+        running = false;
+        if (nodeId == null) {
+            return;
+        }
+        try {
+            instanceRegistry.deregister(nodeId);
+            log.info("Cluster node deregistered: {}", nodeId);
+        } catch (RuntimeException e) {
+            // Registry entry will TTL-expire within heartbeatTtl anyway.
+            log.warn(
+                    "Cluster deregister failed for {} (will TTL-expire within {}s): {}",
+                    nodeId,
+                    heartbeatTtl.toSeconds(),
+                    e.getMessage());
+        }
+    }
+
+    @Override
+    public boolean isRunning() {
+        return running;
+    }
+
+    @Override
+    public int getPhase() {
+        return Integer.MAX_VALUE;
+    }
+
+    @Override
+    public boolean isAutoStartup() {
+        return true;
+    }
+
+    /**
+     * Resolve the address peers should hit. Order: explicit config -> {@code POD_IP} env (K8s
+     * downward API) -> JDK hostname -> fail loud (never silently fall back to a loopback).
+     *
+     * <p>Scheme is taken from {@code cluster.node.scheme} (default {@code http}). Set to {@code
+     * https} when nodes terminate TLS themselves; leave as {@code http} when an upstream LB
+     * terminates TLS and intra-cluster traffic is plain HTTP.
+     */
+    private String resolveInternalAddress() {
+        Cluster cluster = applicationProperties.getCluster();
+        String configured =
+                cluster.getNode() == null ? null : cluster.getNode().getInternalAddress();
+        if (configured != null && !configured.isBlank()) {
+            return ensurePort(configured);
+        }
+        String podIp = System.getenv("POD_IP");
+        if (podIp != null && !podIp.isBlank()) {
+            return scheme() + "://" + podIp + ":" + serverPort;
+        }
+        try {
+            return scheme()
+                    + "://"
+                    + InetAddress.getLocalHost().getHostAddress()
+                    + ":"
+                    + serverPort;
+        } catch (UnknownHostException e) {
+            throw new IllegalStateException(
+                    "Could not resolve this host's address for cluster registration; set"
+                            + " cluster.node.internal-address explicitly (or set POD_IP"
+                            + " in the Kubernetes downward API).",
+                    e);
+        }
+    }
+
+    private String ensurePort(String addr) {
+        if (addr.startsWith("http://") || addr.startsWith("https://")) {
+            return addr;
+        }
+        if (addr.contains(":")) {
+            return scheme() + "://" + addr;
+        }
+        return scheme() + "://" + addr + ":" + serverPort;
+    }
+
+    private String scheme() {
+        Cluster cluster = applicationProperties.getCluster();
+        if (cluster.getNode() == null
+                || cluster.getNode().getScheme() == null
+                || cluster.getNode().getScheme().isBlank()) {
+            return "http";
+        }
+        String s = cluster.getNode().getScheme().trim().toLowerCase(Locale.ROOT);
+        return "https".equals(s) ? "https" : "http";
+    }
+
+    private String role() {
+        Cluster.NodeRole r = applicationProperties.getCluster().resolvedRole();
+        return r == null ? "BOTH" : r.name();
+    }
+}
@@ -0,0 +1,19 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+
+/**
+ * Composite condition: matches only when cluster.enabled=true AND cluster.backplane=valkey. Both
+ * checks are required (enabled alone may select the in-process backplane, which must not load
+ * Valkey beans); a single {@code @ConditionalOnExpression} keeps the guard in one place.
+ */
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@ConditionalOnExpression(
+        "${cluster.enabled:false} and '${cluster.backplane:inprocess}'.equals('valkey')")
+public @interface ConditionalOnValkeyBackplane {}
@@ -0,0 +1,55 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.model.ApplicationProperties;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+public class ValkeyClusterBackplane implements ClusterBackplane {
+
+    private final ApplicationProperties applicationProperties;
+    private final StringRedisTemplate template;
+
+    @Override
+    public boolean isHealthy() {
+        try {
+            // template.execute() borrows from the pool and returns the connection in a finally
+            // block - critical because isHealthy() is hit on every k8s liveness/readiness probe
+            // tick. Calling getConnectionFactory().getConnection() directly leaks the connection
+            // and exhausts the pool under monitoring load.
+            String pong = template.execute((RedisCallback<String>) connection -> connection.ping());
+            return "PONG".equalsIgnoreCase(pong);
+        } catch (RuntimeException ex) {
+            log.warn("Valkey backplane health check failed: {}", ex.getMessage());
+            return false;
+        }
+    }
+
+    @Override
+    public String backplaneType() {
+        return "valkey";
+    }
+
+    @Override
+    public String localNodeId() {
+        return applicationProperties.getCluster().resolvedNodeId();
+    }
+
+    /**
+     * Valkey TTL evicts job entries; local cleanup loop is redundant and would race with cluster
+     * state.
+     */
+    @Override
+    public boolean shouldRunLocalCleanup() {
+        return false;
+    }
+}
@@ -0,0 +1,265 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.time.Duration;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisPassword;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import io.lettuce.core.RedisCommandExecutionException;
+import io.lettuce.core.SslVerifyMode;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.ApplicationProperties.Cluster;
+
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
+@DependsOn("clusterLicenseGate")
+public class ValkeyConnectionConfiguration {
+
+    private final ApplicationProperties applicationProperties;
+
+    @Bean(destroyMethod = "destroy")
+    @ConditionalOnProperty(name = "cluster.backplane", havingValue = "valkey")
+    public LettuceConnectionFactory valkeyConnectionFactory() {
+        Cluster cluster = applicationProperties.getCluster();
+        Endpoint endpoint = parseUrl(cluster.getValkey().getUrl());
+        RedisStandaloneConfiguration cfg =
+                new RedisStandaloneConfiguration(endpoint.host(), endpoint.port());
+        if (endpoint.username() != null) {
+            cfg.setUsername(endpoint.username());
+        }
+        if (endpoint.password() != null) {
+            cfg.setPassword(RedisPassword.of(endpoint.password()));
+        }
+        boolean skipCertVerification =
+                cluster.getValkey().getTls() != null
+                        && cluster.getValkey().getTls().isSkipCertVerification();
+        LettuceClientConfiguration clientConfig =
+                buildClientConfiguration(endpoint.tls(), skipCertVerification);
+        LettuceConnectionFactory factory = new LettuceConnectionFactory(cfg, clientConfig);
+        factory.afterPropertiesSet();
+        // Eager handshake with retry tolerates docker-compose DNS races; fails boot loudly
+        // if Valkey is genuinely unreachable.
+        eagerHandshake(factory, endpoint.host(), endpoint.port(), endpoint.tls());
+        log.info(
+                "Valkey connection configured: {}:{} tls={} verifyPeer={}",
+                endpoint.host(),
+                endpoint.port(),
+                endpoint.tls(),
+                endpoint.tls() ? clientConfig.getVerifyMode() : "n/a");
+        return factory;
+    }
+
+    /** Parsed connection endpoint; username/password are null when absent. */
+    record Endpoint(String host, int port, boolean tls, String username, String password) {}
+
+    /**
+     * Parses {@code redis://[user:password@]host[:port]} (or {@code rediss://} for TLS) into an
+     * {@link Endpoint}. Package-private and side-effect-free so URL handling is unit-testable.
+     *
+     * <ul>
+     *   <li>Missing port defaults to 6379.
+     *   <li>{@code rediss} scheme selects TLS.
+     *   <li>Userinfo {@code :password@} (empty user) is treated as password-only auth against the
+     *       default user, not a login with an empty username.
+     *   <li>Reserved characters in the password ({@code @ : / # ?}) must be percent-encoded; {@link
+     *       URI} parses them structurally otherwise (e.g. {@code #} starts the fragment).
+     * </ul>
+     *
+     * @throws IllegalStateException if the URL is blank, syntactically invalid, or has no host
+     */
+    static Endpoint parseUrl(String url) {
+        if (url == null || url.isBlank()) {
+            throw new IllegalStateException("cluster.valkey.url must be set when backplane=valkey");
+        }
+        URI uri;
+        try {
+            uri = new URI(url);
+        } catch (URISyntaxException ex) {
+            throw new IllegalStateException(
+                    "cluster.valkey.url is not a valid URI: " + url + " (" + ex.getMessage() + ")",
+                    ex);
+        }
+        String host = uri.getHost();
+        if (host == null || host.isBlank()) {
+            throw new IllegalStateException(
+                    "cluster.valkey.url has no host: "
+                            + url
+                            + " (expected redis://[user:password@]host[:port])");
+        }
+        boolean tls = "rediss".equalsIgnoreCase(uri.getScheme());
+        int port = uri.getPort() <= 0 ? 6379 : uri.getPort();
+        String username = null;
+        String password = null;
+        String userInfo = uri.getUserInfo();
+        if (userInfo != null) {
+            String[] parts = userInfo.split(":", 2);
+            if (parts.length == 2) {
+                username = parts[0].isEmpty() ? null : parts[0];
+                password = parts[1];
+            } else if (!parts[0].isBlank()) {
+                password = parts[0];
+            }
+        }
+        return new Endpoint(host, port, tls, username, password);
+    }
+
+    /**
+     * Package-private for testing. verifyPeer(FULL) is pinned explicitly so a Spring Data Redis
+     * default change cannot silently weaken our TLS handshake. skipCertVerification is dev-only.
+     */
+    static LettuceClientConfiguration buildClientConfiguration(
+            boolean tls, boolean skipCertVerification) {
+        LettuceClientConfiguration.LettuceClientConfigurationBuilder clientBuilder =
+                LettuceClientConfiguration.builder();
+        // Bound every backplane command. Lettuce defaults to 60s; without this a partitioned or
+        // slow Valkey would stall hot-path calls (e.g. JobController.guardNonOwner -> jobStore.get
+        // on each request) for up to a minute, exhausting request threads. All backplane ops are
+        // non-blocking single commands, so a short timeout is safe.
+        clientBuilder.commandTimeout(Duration.ofSeconds(2));
+        if (tls) {
+            clientBuilder
+                    .useSsl()
+                    .verifyPeer(skipCertVerification ? SslVerifyMode.NONE : SslVerifyMode.FULL);
+            if (skipCertVerification) {
+                log.warn(
+                        "Valkey TLS hostname/chain verification DISABLED via"
+                                + " cluster.valkey.tls.skip-cert-verification=true"
+                                + " - insecure, dev-only");
+            }
+        }
+        return clientBuilder.build();
+    }
+
+    /**
+     * 10 x 3s = 30s boot-time retry. Auth failures (WRONGPASS/NOAUTH/NOPERM) short-circuit
+     * immediately; only transport errors get the loop. Package-private for testing.
+     */
+    static void eagerHandshake(
+            LettuceConnectionFactory factory, String host, int port, boolean tls) {
+        RuntimeException last = null;
+        for (int attempt = 1; attempt <= 10; attempt++) {
+            try {
+                String pong;
+                RedisConnection conn = factory.getConnection();
+                try {
+                    pong = conn.ping();
+                } finally {
+                    conn.close();
+                }
+                if (!"PONG".equalsIgnoreCase(pong)) {
+                    throw new IllegalStateException(
+                            "Valkey PING returned '" + pong + "' (expected PONG)");
+                }
+                if (attempt > 1) {
+                    log.info("Valkey reachable after {} attempts", attempt);
+                }
+                return;
+            } catch (RuntimeException ex) {
+                if (isAuthFailure(ex)) {
+                    factory.destroy();
+                    throw new IllegalStateException(
+                            "Valkey authentication failed for "
+                                    + host
+                                    + ":"
+                                    + port
+                                    + " (tls="
+                                    + tls
+                                    + "): "
+                                    + rootAuthMessage(ex)
+                                    + ". Check cluster.valkey.url credentials"
+                                    + " (user/password and ACL permissions).",
+                            ex);
+                }
+                last = ex;
+                log.warn(
+                        "Valkey PING attempt {}/10 failed ({}:{}, tls={}): {}",
+                        attempt,
+                        host,
+                        port,
+                        tls,
+                        ex.getMessage());
+                try {
+                    Thread.sleep(3000);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
+        factory.destroy();
+        throw new IllegalStateException(
+                "Valkey unreachable at boot after 10 attempts ("
+                        + host
+                        + ":"
+                        + port
+                        + ", tls="
+                        + tls
+                        + "): "
+                        + (last == null ? "no detail" : last.getMessage()),
+                last);
+    }
+
+    /**
+     * Walks the cause chain for WRONGPASS/NOAUTH/NOPERM replies. Spring Data Redis wraps Lettuce's
+     * RedisCommandExecutionException in RedisSystemException, so the auth signal may be one level
+     * down. No typed auth exception exists in spring-data-redis 4.0.5 / Lettuce 6.8.2.
+     */
+    static boolean isAuthFailure(Throwable t) {
+        for (Throwable cur = t; cur != null; cur = cur.getCause()) {
+            if (cur instanceof RedisCommandExecutionException && hasAuthPrefix(cur.getMessage())) {
+                return true;
+            }
+            if (hasAuthPrefix(cur.getMessage())) {
+                return true;
+            }
+            if (cur.getCause() == cur) {
+                break;
+            }
+        }
+        return false;
+    }
+
+    private static boolean hasAuthPrefix(String message) {
+        if (message == null) {
+            return false;
+        }
+        String upper = message.toUpperCase(java.util.Locale.ROOT).stripLeading();
+        return upper.startsWith("WRONGPASS")
+                || upper.startsWith("NOAUTH")
+                || upper.startsWith("NOPERM");
+    }
+
+    private static String rootAuthMessage(Throwable t) {
+        for (Throwable cur = t; cur != null; cur = cur.getCause()) {
+            if (cur instanceof RedisCommandExecutionException && cur.getMessage() != null) {
+                return cur.getMessage();
+            }
+            if (cur.getCause() == cur) {
+                break;
+            }
+        }
+        return t.getMessage();
+    }
+
+    @Bean
+    @ConditionalOnProperty(name = "cluster.backplane", havingValue = "valkey")
+    public StringRedisTemplate valkeyTemplate(LettuceConnectionFactory factory) {
+        return new StringRedisTemplate(factory);
+    }
+}
@@ -0,0 +1,102 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.DistributedLock;
+
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+@Slf4j
+public class ValkeyDistributedLock implements DistributedLock {
+
+    private static final String PREFIX = "stirling:lock:";
+
+    private static final RedisScript<Long> RELEASE_SCRIPT =
+            new DefaultRedisScript<>(
+                    "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
+                    Long.class);
+
+    private static final RedisScript<Long> RENEW_SCRIPT =
+            new DefaultRedisScript<>(
+                    "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], ARGV[2]) else return 0 end",
+                    Long.class);
+
+    private final StringRedisTemplate template;
+
+    @Override
+    public Optional<LockHandle> tryAcquire(String lockKey, Duration leaseTime) {
+        String key = PREFIX + lockKey;
+        String value = UUID.randomUUID().toString();
+        Boolean ok = template.opsForValue().setIfAbsent(key, value, leaseTime);
+        if (Boolean.TRUE.equals(ok)) {
+            return Optional.of(new ValkeyHandle(template, key, value));
+        }
+        return Optional.empty();
+    }
+
+    private static final class ValkeyHandle implements LockHandle {
+        private final StringRedisTemplate template;
+        private final String key;
+        private final String value;
+        private boolean released;
+
+        ValkeyHandle(StringRedisTemplate template, String key, String value) {
+            this.template = template;
+            this.key = key;
+            this.value = value;
+        }
+
+        @Override
+        public synchronized void release() {
+            if (released) {
+                return;
+            }
+            released = true;
+            // Swallow + log: LockHandle is AutoCloseable, so release() runs from close() inside
+            // try-with-resources. An uncaught Valkey error here would mask the body's exception.
+            // The lease TTL-expires anyway, so a failed explicit release is safe.
+            try {
+                template.execute(RELEASE_SCRIPT, Collections.singletonList(key), value);
+            } catch (RuntimeException ex) {
+                log.warn(
+                        "Lock release failed for {} (lease will TTL-expire): {}",
+                        key,
+                        ex.getMessage());
+            }
+        }
+
+        @Override
+        public synchronized boolean renew(Duration leaseTime) {
+            if (released) {
+                return false;
+            }
+            try {
+                Long result =
+                        template.execute(
+                                RENEW_SCRIPT,
+                                Collections.singletonList(key),
+                                value,
+                                Long.toString(leaseTime.toMillis()));
+                return result != null && result == 1L;
+            } catch (RuntimeException ex) {
+                log.warn(
+                        "Lock renew failed for {} (treated as lost lease): {}",
+                        key,
+                        ex.getMessage());
+                return false;
+            }
+        }
+    }
+}
@@ -0,0 +1,115 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.common.cluster.ClusterNode;
+import stirling.software.common.cluster.InstanceRegistry;
+
+/**
+ * Valkey-backed {@link InstanceRegistry}. Each node is stored as a hash with a TTL equal to the
+ * configured heartbeat TTL; the heartbeat re-arms the TTL.
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+public class ValkeyInstanceRegistry implements InstanceRegistry {
+
+    private static final String PREFIX = "stirling:nodes:";
+
+    private final StringRedisTemplate template;
+
+    @Override
+    public void register(ClusterNode node, Duration heartbeatTtl) {
+        String key = PREFIX + node.nodeId();
+        long ttlMs = heartbeatTtl.toMillis();
+        Map<String, String> fields = new LinkedHashMap<>();
+        fields.put("nodeId", node.nodeId());
+        fields.put("internalAddress", node.internalAddress());
+        fields.put("role", node.role());
+        fields.put("lastHeartbeat", node.lastHeartbeat().toString());
+
+        // MULTI/EXEC so the hash fields and the TTL commit together. Without this, a crash
+        // between HSET and EXPIRE leaves the hash with no TTL: it never expires, masks the
+        // dead node as alive, and only a subsequent successful register() would re-arm it.
+        template.execute(
+                (RedisCallback<Object>)
+                        connection -> {
+                            connection.multi();
+                            byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
+                            Map<byte[], byte[]> hashBytes = new LinkedHashMap<>();
+                            for (Map.Entry<String, String> f : fields.entrySet()) {
+                                hashBytes.put(
+                                        f.getKey().getBytes(StandardCharsets.UTF_8),
+                                        f.getValue().getBytes(StandardCharsets.UTF_8));
+                            }
+                            connection.hashCommands().hMSet(keyBytes, hashBytes);
+                            connection.keyCommands().pExpire(keyBytes, ttlMs);
+                            connection.exec();
+                            return null;
+                        });
+    }
+
+    @Override
+    public Optional<ClusterNode> lookup(String nodeId) {
+        return readNode(PREFIX + nodeId);
+    }
+
+    @Override
+    public Collection<ClusterNode> activeNodes() {
+        ScanOptions options = ScanOptions.scanOptions().match(PREFIX + "*").count(256).build();
+        List<ClusterNode> nodes = new ArrayList<>();
+        try (Cursor<String> cursor = template.scan(options)) {
+            while (cursor.hasNext()) {
+                readNode(cursor.next()).ifPresent(nodes::add);
+            }
+        }
+        return nodes;
+    }
+
+    @Override
+    public void deregister(String nodeId) {
+        template.delete(PREFIX + nodeId);
+    }
+
+    private Optional<ClusterNode> readNode(String key) {
+        Map<Object, Object> entries = template.opsForHash().entries(key);
+        if (entries == null || entries.isEmpty()) {
+            return Optional.empty();
+        }
+        Object nodeId = entries.get("nodeId");
+        if (nodeId == null) {
+            return Optional.empty();
+        }
+        Instant heartbeat = Instant.now();
+        Object hb = entries.get("lastHeartbeat");
+        if (hb != null) {
+            try {
+                heartbeat = Instant.parse(hb.toString());
+            } catch (RuntimeException ignored) {
+                // keep default
+            }
+        }
+        return Optional.of(
+                new ClusterNode(
+                        nodeId.toString(),
+                        String.valueOf(entries.getOrDefault("internalAddress", "")),
+                        heartbeat,
+                        String.valueOf(entries.getOrDefault("role", "BOTH"))));
+    }
+}
@@ -0,0 +1,297 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.JobStoreEntry;
+
+/**
+ * Valkey-backed {@link JobStore}. Each job is one hash; a reverse index maps fileId to jobId.
+ *
+ * <p><b>put() atomicity:</b> the hash fields, the per-job TTL, and the reverse-index entries are
+ * issued inside a single pipelined Redis transaction (MULTI/EXEC). A partial failure cannot leave
+ * the hash without a TTL or with half the file→job index entries written.
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+@Slf4j
+public class ValkeyJobStore implements JobStore {
+
+    private static final String JOB_PREFIX = "stirling:job:";
+    private static final String FILE_INDEX_PREFIX = "stirling:file2job:";
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final TypeReference<List<String>> LIST_STRING = new TypeReference<>() {};
+    private static final TypeReference<Map<String, String>> MAP_STRING = new TypeReference<>() {};
+
+    private final StringRedisTemplate template;
+
+    @Override
+    public void put(JobStoreEntry entry, Duration ttl) {
+        String key = JOB_PREFIX + entry.jobId();
+        long ttlMs = ttl.toMillis();
+        Map<String, String> fields = new LinkedHashMap<>();
+        fields.put("jobId", entry.jobId());
+        fields.put("state", entry.state().name());
+        fields.put("owningNodeId", entry.owningNodeId() == null ? "" : entry.owningNodeId());
+        if (entry.createdAt() != null) {
+            fields.put("createdAt", entry.createdAt().toString());
+        }
+        if (entry.completedAt() != null) {
+            fields.put("completedAt", entry.completedAt().toString());
+        }
+        if (entry.error() != null) {
+            fields.put("error", entry.error());
+        }
+        fields.put("fileIds", writeJson(entry.fileIds() == null ? List.of() : entry.fileIds()));
+        fields.put(
+                "resultMeta",
+                writeJson(entry.resultMeta() == null ? Map.of() : entry.resultMeta()));
+
+        // Build pipelined MULTI/EXEC so the hash, its TTL, and every reverse-index entry
+        // commit atomically.
+        template.execute(
+                (RedisCallback<Object>)
+                        connection -> {
+                            connection.multi();
+                            byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
+                            Map<byte[], byte[]> hashBytes = new LinkedHashMap<>();
+                            for (Map.Entry<String, String> f : fields.entrySet()) {
+                                hashBytes.put(
+                                        f.getKey().getBytes(StandardCharsets.UTF_8),
+                                        f.getValue().getBytes(StandardCharsets.UTF_8));
+                            }
+                            connection.hashCommands().hMSet(keyBytes, hashBytes);
+                            connection.keyCommands().pExpire(keyBytes, ttlMs);
+                            if (entry.fileIds() != null) {
+                                for (String fileId : entry.fileIds()) {
+                                    byte[] idxKey =
+                                            (FILE_INDEX_PREFIX + fileId)
+                                                    .getBytes(StandardCharsets.UTF_8);
+                                    connection
+                                            .stringCommands()
+                                            .set(
+                                                    idxKey,
+                                                    entry.jobId().getBytes(StandardCharsets.UTF_8));
+                                    connection.keyCommands().pExpire(idxKey, ttlMs);
+                                }
+                            }
+                            connection.exec();
+                            return null;
+                        });
+    }
+
+    @Override
+    public Optional<JobStoreEntry> get(String jobId) {
+        return readEntry(JOB_PREFIX + jobId);
+    }
+
+    @Override
+    public void delete(String jobId) {
+        // WATCH/MULTI/EXEC: read fileIds INSIDE the watched scope so a concurrent put() that
+        // adds new fileIds between our read and EXEC aborts the transaction. Without this guard,
+        // an interleaved put() that grows fileIds would leave orphaned reverse-index entries
+        // pointing at the deleted jobId until their TTL expires. One retry handles the common
+        // case; further contention falls through to lazy TTL cleanup (acceptable - this is an
+        // eviction path, not a correctness primitive).
+        String jobKey = JOB_PREFIX + jobId;
+        byte[] jobKeyBytes = jobKey.getBytes(StandardCharsets.UTF_8);
+        for (int attempt = 0; attempt < 2; attempt++) {
+            Boolean committed =
+                    template.execute(
+                            (RedisCallback<Boolean>)
+                                    connection -> {
+                                        connection.watch(jobKeyBytes);
+                                        // Read the single fileIds field with hGet rather than
+                                        // hGetAll + map.get: hGetAll returns a Map<byte[],byte[]>
+                                        // whose keys compare by identity, so a fresh
+                                        // "fileIds".getBytes() lookup never matches and the reverse
+                                        // index would be left orphaned. hGet resolves the field
+                                        // server-side.
+                                        byte[] fileIdsBytes =
+                                                connection
+                                                        .hashCommands()
+                                                        .hGet(
+                                                                jobKeyBytes,
+                                                                "fileIds"
+                                                                        .getBytes(
+                                                                                StandardCharsets
+                                                                                        .UTF_8));
+                                        List<byte[]> keysToDelete = new ArrayList<>();
+                                        keysToDelete.add(jobKeyBytes);
+                                        if (fileIdsBytes != null) {
+                                            List<String> fileIds =
+                                                    readJsonList(
+                                                            new String(
+                                                                    fileIdsBytes,
+                                                                    StandardCharsets.UTF_8),
+                                                            jobKey);
+                                            for (String fileId : fileIds) {
+                                                keysToDelete.add(
+                                                        (FILE_INDEX_PREFIX + fileId)
+                                                                .getBytes(StandardCharsets.UTF_8));
+                                            }
+                                        }
+                                        connection.multi();
+                                        for (byte[] key : keysToDelete) {
+                                            connection.keyCommands().del(key);
+                                        }
+                                        List<Object> results = connection.exec();
+                                        // exec() returns null when WATCH detected a concurrent
+                                        // write; spring-data-redis surfaces this as either null
+                                        // or empty depending on the driver path.
+                                        return results != null && !results.isEmpty();
+                                    });
+            if (Boolean.TRUE.equals(committed)) {
+                return;
+            }
+        }
+        log.warn(
+                "JobStore.delete({}) lost two WATCH races to concurrent put(); reverse-index"
+                        + " entries may linger until TTL expiry",
+                jobId);
+    }
+
+    @Override
+    public boolean exists(String jobId) {
+        Boolean exists = template.hasKey(JOB_PREFIX + jobId);
+        return Boolean.TRUE.equals(exists);
+    }
+
+    @Override
+    public Optional<String> findJobIdByFileId(String fileId) {
+        return Optional.ofNullable(template.opsForValue().get(FILE_INDEX_PREFIX + fileId));
+    }
+
+    @Override
+    public Collection<JobStoreEntry> all() {
+        // SCAN, not KEYS - KEYS blocks the Valkey server for the duration of the walk.
+        ScanOptions options = ScanOptions.scanOptions().match(JOB_PREFIX + "*").count(256).build();
+        List<JobStoreEntry> result = new ArrayList<>();
+        try (Cursor<String> cursor = template.scan(options)) {
+            while (cursor.hasNext()) {
+                readEntry(cursor.next()).ifPresent(result::add);
+            }
+        }
+        return result;
+    }
+
+    private Optional<JobStoreEntry> readEntry(String key) {
+        Map<Object, Object> entries = template.opsForHash().entries(key);
+        if (entries == null || entries.isEmpty()) {
+            return Optional.empty();
+        }
+        Object jobId = entries.get("jobId");
+        if (jobId == null) {
+            return Optional.empty();
+        }
+        Instant createdAt = parseInstant(entries.get("createdAt"), key, "createdAt");
+        Instant completedAt = parseInstant(entries.get("completedAt"), key, "completedAt");
+        List<String> fileIds = parseList(entries.get("fileIds"), key);
+        Map<String, String> resultMeta = parseMap(entries.get("resultMeta"), key);
+        String stateName =
+                String.valueOf(
+                        entries.getOrDefault("state", JobStoreEntry.JobState.PENDING.name()));
+        JobStoreEntry.JobState state;
+        try {
+            state = JobStoreEntry.JobState.valueOf(stateName);
+        } catch (IllegalArgumentException ex) {
+            log.warn("Unrecognised job state '{}' in {}, defaulting to PENDING", stateName, key);
+            state = JobStoreEntry.JobState.PENDING;
+        }
+        String owningNodeId = String.valueOf(entries.getOrDefault("owningNodeId", ""));
+        String error = entries.get("error") == null ? null : entries.get("error").toString();
+        return Optional.of(
+                new JobStoreEntry(
+                        jobId.toString(),
+                        state,
+                        owningNodeId,
+                        createdAt,
+                        completedAt,
+                        error,
+                        fileIds,
+                        resultMeta));
+    }
+
+    private Instant parseInstant(Object v, String key, String field) {
+        if (v == null) {
+            return null;
+        }
+        try {
+            return Instant.parse(v.toString());
+        } catch (RuntimeException e) {
+            log.warn(
+                    "JobStore {} field '{}' has malformed timestamp '{}' - treating as missing",
+                    key,
+                    field,
+                    v);
+            return null;
+        }
+    }
+
+    private List<String> parseList(Object v, String key) {
+        if (v == null) {
+            return new ArrayList<>();
+        }
+        return readJsonList(v.toString(), key);
+    }
+
+    private Map<String, String> parseMap(Object v, String key) {
+        if (v == null) {
+            return new HashMap<>();
+        }
+        try {
+            return MAPPER.readValue(v.toString(), MAP_STRING);
+        } catch (JsonProcessingException e) {
+            log.warn(
+                    "JobStore {} field 'resultMeta' is not valid JSON '{}' - treating as empty",
+                    key,
+                    v);
+            return new HashMap<>();
+        }
+    }
+
+    private static String writeJson(Object value) {
+        try {
+            return MAPPER.writeValueAsString(value);
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("Failed to JSON-serialize JobStore field", e);
+        }
+    }
+
+    private List<String> readJsonList(String json, String key) {
+        try {
+            List<String> parsed = MAPPER.readValue(json, LIST_STRING);
+            return parsed == null ? new ArrayList<>() : parsed;
+        } catch (JsonProcessingException e) {
+            log.warn(
+                    "JobStore {} field 'fileIds' is not valid JSON '{}' - treating as empty",
+                    key,
+                    json);
+            return new ArrayList<>();
+        }
+    }
+}
@@ -0,0 +1,61 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.common.cluster.KeyValueCache;
+
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+public class ValkeyKeyValueCache implements KeyValueCache {
+
+    private static final String PREFIX = "stirling:kv:";
+
+    private final StringRedisTemplate template;
+
+    @Override
+    public void put(String namespace, String key, String value, Duration ttl) {
+        template.opsForValue()
+                .set(buildKey(namespace, key), value, ttl.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public Optional<String> get(String namespace, String key) {
+        return Optional.ofNullable(template.opsForValue().get(buildKey(namespace, key)));
+    }
+
+    @Override
+    public void evict(String namespace, String key) {
+        template.delete(buildKey(namespace, key));
+    }
+
+    @Override
+    public void evictNamespace(String namespace) {
+        ScanOptions options =
+                ScanOptions.scanOptions().match(PREFIX + namespace + ":*").count(256).build();
+        List<String> keys = new ArrayList<>();
+        try (Cursor<String> cursor = template.scan(options)) {
+            while (cursor.hasNext()) {
+                keys.add(cursor.next());
+            }
+        }
+        if (!keys.isEmpty()) {
+            template.delete(keys);
+        }
+    }
+
+    private String buildKey(String namespace, String key) {
+        return PREFIX + namespace + ":" + key;
+    }
+}
@@ -0,0 +1,83 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.stereotype.Component;
+
+import io.github.bucket4j.BucketConfiguration;
+import io.github.bucket4j.ConsumptionProbe;
+import io.github.bucket4j.distributed.BucketProxy;
+import io.github.bucket4j.distributed.ExpirationAfterWriteStrategy;
+import io.github.bucket4j.distributed.proxy.ProxyManager;
+import io.github.bucket4j.redis.lettuce.Bucket4jLettuce;
+import io.lettuce.core.AbstractRedisClient;
+import io.lettuce.core.RedisClient;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+
+import stirling.software.common.cluster.RateLimitStore;
+
+/**
+ * Valkey-backed token-bucket rate limiting via Bucket4j's Lettuce ProxyManager. The token bucket
+ * refills continuously and enforces one global limit across nodes, with the same semantics as the
+ * in-process {@code InProcessRateLimitStore} (which also uses Bucket4j).
+ */
+@Component
+@ConditionalOnValkeyBackplane
+public class ValkeyRateLimitStore implements RateLimitStore {
+
+    private static final String PREFIX = "stirling:rl:";
+
+    private final LettuceConnectionFactory connectionFactory;
+    private ProxyManager<byte[]> proxyManager;
+
+    public ValkeyRateLimitStore(LettuceConnectionFactory connectionFactory) {
+        this.connectionFactory = connectionFactory;
+    }
+
+    @PostConstruct
+    void initProxyManager() {
+        AbstractRedisClient client = connectionFactory.getNativeClient();
+        if (!(client instanceof RedisClient redisClient)) {
+            throw new IllegalStateException(
+                    "ValkeyRateLimitStore requires a standalone Lettuce RedisClient; got "
+                            + (client == null ? "null" : client.getClass().getName())
+                            + " (cluster client not supported by this rate limit impl)");
+        }
+        // Expire idle bucket keys so they do not accumulate forever in Valkey (one key per
+        // user / API-key / IP). TTL tracks the time to refill the bucket from empty, capped at
+        // 25h to cover the longest (daily) rate-limit window; an idle bucket evicts after that.
+        this.proxyManager =
+                Bucket4jLettuce.casBasedBuilder(redisClient)
+                        .expirationAfterWrite(
+                                ExpirationAfterWriteStrategy.basedOnTimeForRefillingBucketUpToMax(
+                                        Duration.ofHours(25)))
+                        .build();
+    }
+
+    @PreDestroy
+    void shutdown() {
+        proxyManager = null;
+    }
+
+    @Override
+    public RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod) {
+        byte[] key = (PREFIX + bucketKey).getBytes(StandardCharsets.UTF_8);
+        BucketConfiguration cfg =
+                BucketConfiguration.builder()
+                        .addLimit(
+                                stage ->
+                                        stage.capacity(capacity)
+                                                .refillGreedy(capacity, refillPeriod))
+                        .build();
+        BucketProxy bucket = proxyManager.builder().build(key, () -> cfg);
+        ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1);
+        if (probe.isConsumed()) {
+            return new RateLimitDecision(true, probe.getRemainingTokens(), 0L);
+        }
+        return new RateLimitDecision(false, 0L, probe.getNanosToWaitForRefill());
+    }
+}