Add Valkey backplane and sticky-410 ownership

This commit is contained in:
Anthony Stirling
2026-05-27 11:57:01 +01:00
parent 4564ed5bec
commit 65493c1919
27 changed files with 3301 additions and 25 deletions
@@ -2,7 +2,15 @@ package stirling.software.common.cluster;
import java.time.Duration;
/** Token-bucket rate limiting backed by the cluster backplane. */
/**
* Token-bucket rate limiting backed by the cluster backplane.
*
* <p>In-process implementations enforce a per-JVM limit (identical to today's behaviour).
* Distributed implementations enforce a single global limit across every node.
*
* <p>Both implementations use a Bucket4j greedy-refill token bucket so semantics match across
* single-node and cluster deployments (no fixed-window boundary doubling).
*/
public interface RateLimitStore {
/**
@@ -112,25 +112,12 @@ public class JobExecutorService {
log.debug("Generated jobId: {} (base: {})", scopedJobKey, baseJobId);
// Store the scoped job ID in the request for potential use by other components
// Store the scoped job ID in the request for potential use by other components.
// Ownership lives in the scoped key itself (userId:jobId) plus the cluster-visible
// JobStore entry, so we no longer mirror it into the HTTP session - that did not
// survive a node hop in cluster mode.
if (request != null) {
request.setAttribute("jobId", scopedJobKey);
// Also track this job ID in the user's session for authorization purposes
// This ensures users can only cancel their own jobs
if (request.getSession() != null) {
@SuppressWarnings("unchecked")
java.util.Set<String> userJobIds =
(java.util.Set<String>) request.getSession().getAttribute("userJobIds");
if (userJobIds == null) {
userJobIds = new java.util.concurrent.ConcurrentSkipListSet<>();
request.getSession().setAttribute("userJobIds", userJobIds);
}
userJobIds.add(scopedJobKey);
log.debug("Added scoped job ID {} to user session", scopedJobKey);
}
}
String jobId = scopedJobKey;
@@ -24,7 +24,9 @@ class InProcessDistributedLockTest {
}
@Test
void reentryFromSameThreadFails() {
void reentryFromSameThreadFails_parityWithValkey() {
// The Valkey impl refuses reentry (SET NX semantics); the in-process impl must match,
// otherwise code working in single-instance silently breaks in cluster mode.
DistributedLock lock = new InProcessDistributedLock();
DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow();
Optional<DistributedLock.LockHandle> reentry = lock.tryAcquire("k", Duration.ofSeconds(30));
@@ -86,6 +86,8 @@ class TaskManagerJobStoreDelegationTest {
@Override
public boolean shouldRunLocalCleanup() {
// Distributed backplanes own job TTL eviction themselves; this mock
// mirrors the real ValkeyClusterBackplane override of the default true.
return false;
}
};
@@ -4,6 +4,7 @@ import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
@@ -22,6 +23,10 @@ import jakarta.servlet.http.HttpServletRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import stirling.software.common.cluster.ClusterBackplane;
import stirling.software.common.cluster.JobStore;
import stirling.software.common.cluster.JobStoreEntry;
import stirling.software.common.cluster.StickyMissRecorder;
import stirling.software.common.model.job.JobResult;
import stirling.software.common.model.job.ResultFile;
import stirling.software.common.service.FileStorage;
@@ -42,10 +47,22 @@ public class JobController {
private final FileStorage fileStorage;
private final JobQueue jobQueue;
private final HttpServletRequest request;
private final ClusterBackplane clusterBackplane;
private final JobStore jobStore;
/**
* Process-local short-TTL cache fronting {@link JobStore#get(String)} on the sticky-410 path.
* Without this every result download / status poll fires a Valkey HGETALL which doubles RTT on
* the hot path when the same client re-requests the same job within seconds.
*/
private final JobOwnershipCache ownershipCache = new JobOwnershipCache();
@Autowired(required = false)
private JobOwnershipService jobOwnershipService;
@Autowired(required = false)
private StickyMissRecorder stickyMissRecorder;
/**
* Get the status of a job
*
@@ -55,6 +72,14 @@ public class JobController {
@GetMapping("/job/{jobId}")
@Operation(summary = "Get job status")
public ResponseEntity<?> getJobStatus(@PathVariable("jobId") String jobId) {
// Sticky-410 must precede user-auth (403): a non-owner node has no way to verify
// ownership for a job it doesn't own, and a 403 here would leak job existence to
// unauthorized callers. Return 410 first so the LB re-routes to the owner.
Optional<ResponseEntity<?>> peerOwned = guardNonOwner(jobId);
if (peerOwned.isPresent()) {
return peerOwned.get();
}
// Validate job ownership
if (!validateJobAccess(jobId)) {
log.warn("Unauthorized attempt to access job status: {}", jobId);
@@ -91,6 +116,14 @@ public class JobController {
@GetMapping("/job/{jobId}/result")
@Operation(summary = "Get job result")
public ResponseEntity<?> getJobResult(@PathVariable("jobId") String jobId) {
// Sticky-410 must precede user-auth (403): a non-owner node has no way to verify
// ownership for a job it doesn't own, and a 403 here would leak job existence to
// unauthorized callers. Return 410 first so the LB re-routes to the owner.
Optional<ResponseEntity<?>> peerOwned = guardNonOwner(jobId);
if (peerOwned.isPresent()) {
return peerOwned.get();
}
// Validate job ownership
if (!validateJobAccess(jobId)) {
log.warn("Unauthorized attempt to access job result: {}", jobId);
@@ -125,11 +158,14 @@ public class JobController {
result.getAllResultFiles()));
}
// Handle single file (download directly)
// Handle single file (download directly). Cross-node ownership was already resolved
// at the top of this method, so reaching here means we ARE the owner (or single-node)
// and the bytes live on our local disk.
if (result.hasFiles() && !result.hasMultipleFiles()) {
try {
List<ResultFile> files = result.getAllResultFiles();
ResultFile singleFile = files.get(0);
byte[] fileContent = fileStorage.retrieveBytes(singleFile.getFileId());
return ResponseEntity.ok()
.header("Content-Type", singleFile.getContentType())
@@ -163,6 +199,15 @@ public class JobController {
public ResponseEntity<?> cancelJob(@PathVariable("jobId") String jobId) {
log.debug("Request to cancel job: {}", jobId);
// Sticky-410 must precede user-auth (403): a non-owner node has no way to verify
// ownership for a job it doesn't own, and a 403 here would leak job existence to
// unauthorized callers. Return 410 first so the LB re-routes to the owner who can
// actually cancel.
Optional<ResponseEntity<?>> peerOwned = guardNonOwner(jobId);
if (peerOwned.isPresent()) {
return peerOwned.get();
}
// Validate job ownership
if (!validateJobAccess(jobId)) {
log.warn("Unauthorized attempt to cancel job: {}", jobId);
@@ -201,7 +246,9 @@ public class JobController {
"queuePosition",
queuePosition >= 0 ? queuePosition : "n/a"));
} else {
// Job not found or already complete
// Job not found or already complete. Cross-node ownership was already resolved at
// the top of this method (sticky-410 precedes user-auth), so any peer-owned case
// has been returned already; reaching here means we ARE the owner (or single-node).
JobResult result = taskManager.getJobResult(jobId);
if (result == null) {
return ResponseEntity.notFound().build();
@@ -224,6 +271,14 @@ public class JobController {
@GetMapping("/job/{jobId}/result/files")
@Operation(summary = "Get job result files")
public ResponseEntity<?> getJobFiles(@PathVariable("jobId") String jobId) {
// Sticky-410 must precede user-auth (403): a non-owner node has no way to verify
// ownership for a job it doesn't own, and a 403 here would leak job existence to
// unauthorized callers. Return 410 first so the LB re-routes to the owner.
Optional<ResponseEntity<?>> peerOwned = guardNonOwner(jobId);
if (peerOwned.isPresent()) {
return peerOwned.get();
}
// Validate job ownership
if (!validateJobAccess(jobId)) {
log.warn("Unauthorized attempt to access job files: {}", jobId);
@@ -267,6 +322,14 @@ public class JobController {
return ResponseEntity.notFound().build();
}
// Sticky-410 must precede user-auth (403): a non-owner node has no way to verify
// ownership for a job it doesn't own, and a 403 here would leak file existence to
// unauthorized callers. Return 410 first so the LB re-routes to the owner.
Optional<ResponseEntity<?>> notOwner = guardNonOwner(jobKey);
if (notOwner.isPresent()) {
return notOwner.get();
}
if (!validateJobAccess(jobKey)) {
log.warn("Unauthorized attempt to access file metadata: {}", fileId);
return ResponseEntity.status(403)
@@ -323,15 +386,21 @@ public class JobController {
return ResponseEntity.notFound().build();
}
// Sticky-410 must precede the user-auth (403) check: a non-owner node has no way to
// verify ownership for a job it doesn't own, and a 403 here would leak file existence
// to unauthorized callers. Return 410 first so the LB re-routes to the owner where
// the real auth check can run.
Optional<ResponseEntity<?>> notOwner = guardNonOwner(jobKey);
if (notOwner.isPresent()) {
return notOwner.get();
}
if (!validateJobAccess(jobKey)) {
log.warn("Unauthorized attempt to download file: {}", fileId);
return ResponseEntity.status(403)
.body(Map.of("message", "You are not authorized to access this file"));
}
// Retrieve file content
byte[] fileContent = fileStorage.retrieveBytes(fileId);
// Find the file metadata from any job that contains this file
// This is for getting the original filename and content type
ResultFile resultFile = taskManager.findResultFileByFileId(fileId);
@@ -342,6 +411,9 @@ public class JobController {
? resultFile.getContentType()
: MediaType.APPLICATION_OCTET_STREAM_VALUE;
// Retrieve file content from local disk
byte[] fileContent = fileStorage.retrieveBytes(fileId);
return ResponseEntity.ok()
.header("Content-Type", contentType)
.header("Content-Disposition", createContentDispositionHeader(fileName))
@@ -356,6 +428,76 @@ public class JobController {
return jobOwnershipService != null;
}
/**
* Returns {@code 410 Gone} with {@code {message, ownedBy, currentNode}} and {@code Retry-After:
* 0} when the job is owned by a peer node. Returns {@link Optional#empty()} when we are the
* owner, when cluster mode is off / JobStore has no entry, or when {@code owningNodeId} is
* blank (caller proceeds with its normal not-found / 200 path).
*
* <p>Wraps the {@link JobStore#get(String)} call in a short-TTL local cache and a defensive
* try/catch so that Valkey RTT cost is not multiplied by every download retry and so that a
* Valkey timeout falls through to the local-disk path instead of surfacing as 500.
*/
private Optional<ResponseEntity<?>> guardNonOwner(String jobId) {
if (clusterBackplane == null || jobStore == null) {
return Optional.empty();
}
Optional<JobStoreEntry> entry;
Optional<Optional<JobStoreEntry>> cached = ownershipCache.get(jobId);
if (cached.isPresent()) {
entry = cached.get();
} else {
try {
entry = jobStore.get(jobId);
} catch (RuntimeException ex) {
// Valkey unavailable / timeout: treat as "no cluster-visible entry" so the request
// can proceed to the local-disk path. Surfacing a 500 here would break every
// download attempt during a brief Valkey blip; the worst case if we miss a real
// peer-owned entry is one wasted round trip + a 404 from the local node.
log.warn(
"JobStore lookup failed for jobId={} - treating as not-found and falling"
+ " through to local path: {}",
jobId,
ex.getMessage());
return Optional.empty();
}
ownershipCache.put(jobId, entry);
}
if (entry.isEmpty()) {
return Optional.empty();
}
String owner = entry.get().owningNodeId();
if (owner == null || owner.isBlank()) {
return Optional.empty();
}
String localId = clusterBackplane.localNodeId();
if (owner.equals(localId)) {
return Optional.empty();
}
log.info(
"Sticky-session miss for jobId={} (owner={}, local={}); returning 410 so client"
+ " retries via LB affinity",
jobId,
owner,
localId);
if (stickyMissRecorder != null) {
stickyMissRecorder.recordStickyMiss();
}
return Optional.of(
ResponseEntity.status(410)
.header("Retry-After", "0")
.body(
Map.of(
"message",
"Result lives on another node. Retry to be routed there"
+ " by the load balancer's sticky-session"
+ " affinity, or re-run the job.",
"ownedBy",
owner,
"currentNode",
localId == null ? "" : localId)));
}
/**
* Create Content-Disposition header with UTF-8 filename support
*
@@ -0,0 +1,51 @@
package stirling.software.common.controller;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import stirling.software.common.cluster.JobStoreEntry;
/**
* Process-local TTL cache for {@link JobStoreEntry} lookups to suppress redundant Valkey HGETALL
* round-trips on the hot result-download path (sticky-410 ownership check).
*
* <p>5 second TTL is short enough that a job's lifecycle transitions (RUNNING -> COMPLETE -> TTL
* expiry) propagate to all nodes within the LB's sticky-session window, and short enough that a
* mistakenly-cached "not found" recovers quickly when an entry actually shows up. Cap the map at
* 2048 entries to bound memory; eviction is best-effort (clear-and-restart) since the cache is
* advisory.
*/
final class JobOwnershipCache {
private static final long TTL_NANOS = 5L * 1_000_000_000L; // 5 s
private static final int MAX_ENTRIES = 2048;
private final ConcurrentMap<String, Entry> entries = new ConcurrentHashMap<>();
Optional<Optional<JobStoreEntry>> get(String jobId) {
Entry e = entries.get(jobId);
if (e == null) {
return Optional.empty();
}
if (System.nanoTime() - e.storedAtNanos > TTL_NANOS) {
entries.remove(jobId, e);
return Optional.empty();
}
return Optional.of(e.value);
}
void put(String jobId, Optional<JobStoreEntry> value) {
if (entries.size() >= MAX_ENTRIES) {
// Best-effort eviction; under burst the cache simply rebuilds.
entries.clear();
}
entries.put(jobId, new Entry(value, System.nanoTime()));
}
void invalidate(String jobId) {
entries.remove(jobId);
}
private record Entry(Optional<JobStoreEntry> value, long storedAtNanos) {}
}
@@ -0,0 +1,502 @@
package stirling.software.common.controller;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.test.util.ReflectionTestUtils;
import jakarta.servlet.http.HttpServletRequest;
import stirling.software.common.cluster.ClusterBackplane;
import stirling.software.common.cluster.JobStore;
import stirling.software.common.cluster.JobStoreEntry;
import stirling.software.common.cluster.StickyMissRecorder;
import stirling.software.common.model.job.JobResult;
import stirling.software.common.service.FileStorage;
import stirling.software.common.service.JobOwnershipService;
import stirling.software.common.service.JobQueue;
import stirling.software.common.service.TaskManager;
/**
* Sticky-session ownership behavior for {@link JobController}.
*
* <p>Result PDFs live on the local disk of whichever node ran the job. When the load balancer's
* cookie/IP affinity *misses* and routes a download to a non-owner node, the controller must return
* {@code 410 Gone} with a structured payload that tells the client to retry (the LB will usually
* route them to the owner on the second attempt).
*
* <p>Contract verified here:
*
* <ul>
* <li>Owner == this node → file is read from local disk (normal 200).
* <li>Owner == another node → {@code 410 Gone} with {@code ownedBy} + {@code currentNode} fields,
* and {@code FileStorage} is <b>never</b> touched.
* <li>JobStore has no entry → behave as single-instance (no 410).
* <li>{@code owningNodeId} blank → behave as single-instance (no 410).
* <li>Single-instance install (no JobStore / no ClusterBackplane) → no 410, no NPE.
* </ul>
*
* <p>Manual mock construction (no {@code MockitoExtension}) so each test can wire its own
* controller with a different {@code ClusterBackplane} / {@code JobStore} combo without setUp stubs
* leaking across cases.
*/
class JobControllerOwnershipTest {
private TaskManager taskManager;
private FileStorage fileStorage;
private JobQueue jobQueue;
private HttpServletRequest request;
private JobOwnershipService jobOwnershipService;
private ClusterBackplane clusterBackplane;
private JobStore jobStore;
private StickyMissRecorder stickyMissRecorder;
private static final String JOB_ID = "job-42";
private static final String FILE_ID = "file-abc";
private static final String LOCAL_NODE = "node-self";
private static final String PEER_NODE = "node-peer";
@BeforeEach
void setUp() {
taskManager = mock(TaskManager.class);
fileStorage = mock(FileStorage.class);
jobQueue = mock(JobQueue.class);
request = mock(HttpServletRequest.class);
jobOwnershipService = mock(JobOwnershipService.class);
clusterBackplane = mock(ClusterBackplane.class);
jobStore = mock(JobStore.class);
stickyMissRecorder = mock(StickyMissRecorder.class);
}
private JobController makeController(ClusterBackplane backplane, JobStore store) {
JobController c =
new JobController(taskManager, fileStorage, jobQueue, request, backplane, store);
// jobOwnershipService is @Autowired(required=false) - field-injected. When non-null,
// validateJobAccess delegates to it. We leave it null by default so the security
// check is a no-op (backwards compat path) and the test focuses on sticky-410.
// stickyMissRecorder is also field-injected; wire by default so the metric assertions
// work without per-test setup.
ReflectionTestUtils.setField(c, "stickyMissRecorder", stickyMissRecorder);
return c;
}
private JobController makeController() {
return makeController(clusterBackplane, jobStore);
}
private JobStoreEntry entryOwnedBy(String ownerNodeId) {
return new JobStoreEntry(
JOB_ID,
JobStoreEntry.JobState.COMPLETE,
ownerNodeId,
Instant.now(),
Instant.now(),
null,
List.of(FILE_ID),
Map.of());
}
private JobResult completedJobWithFile() {
JobResult result = new JobResult();
result.setJobId(JOB_ID);
// completeWithSingleFile populates the resultFiles list, sets complete=true,
// and sets completedAt - all required for the getJobResult single-file branch.
result.completeWithSingleFile(FILE_ID, "out.pdf", "application/pdf", 7L);
return result;
}
/**
* Full sticky-410 contract for {@code downloadFile} when the requested job is owned by a peer.
* Asserts everything in one place (status, Retry-After header, payload shape, no
* implementation-detail leak, metric incremented, storage never touched).
*
* <p>Other tests cover the edge cases independently (locally-owned, no JobStore entry, blank
* owner, etc.) so a single failure here points at exactly one missing or broken contract
* property.
*/
@Test
@DisplayName(
"downloadFile peer-owned → full sticky-410 contract"
+ " (status + Retry-After + payload + metric + storage untouched)")
void downloadFile_peerOwned_fullStickyContract() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
ResponseEntity<?> response = makeController().downloadFile(FILE_ID);
// 1. Status + Retry-After header (immediate-retry hint).
assertEquals(HttpStatus.GONE, response.getStatusCode());
assertEquals("0", response.getHeaders().getFirst("Retry-After"));
// 2. Payload shape: exactly { message, ownedBy, currentNode }, with no leaked secrets.
assertInstanceOf(Map.class, response.getBody());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals(3, body.size(), "exactly: message, ownedBy, currentNode");
assertEquals(PEER_NODE, body.get("ownedBy"));
assertEquals(LOCAL_NODE, body.get("currentNode"));
assertNotNull(body.get("message"));
assertTrue(((String) body.get("message")).toLowerCase().contains("retry"));
assertNull(body.get("internalSecret"));
assertNull(body.get("filePath"));
// 3. Operator-alert metric incremented exactly once.
verify(stickyMissRecorder).recordStickyMiss();
// 4. Storage layer NEVER touched - bytes don't live here, so reading would be wrong.
verify(fileStorage, never()).retrieveBytes(FILE_ID);
}
// --------------------------------------------------------------------------------------
// Happy-path ownership matrix: any non-peer signal (local owner, no entry, blank owner)
// must produce a 200 from FileStorage with NO sticky-miss metric increment.
// --------------------------------------------------------------------------------------
private static Stream<Arguments> downloadHappyPathScenarios() {
return Stream.of(
Arguments.of("locallyOwned", LOCAL_NODE, true),
Arguments.of("noJobStoreEntry", null, false),
Arguments.of("blankOwningNodeId", "", true));
}
@ParameterizedTest(name = "downloadFile {0} -> 200, no sticky-miss")
@MethodSource("downloadHappyPathScenarios")
void downloadFile_happyPath_returnsOkAndNoMetric(
String scenario, String ownerNodeId, boolean entryPresent) throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID))
.thenReturn(
entryPresent ? Optional.of(entryOwnedBy(ownerNodeId)) : Optional.empty());
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
ResponseEntity<?> response = makeController().downloadFile(FILE_ID);
assertEquals(HttpStatus.OK, response.getStatusCode(), scenario);
verify(fileStorage).retrieveBytes(FILE_ID);
verify(stickyMissRecorder, never()).recordStickyMiss();
}
@Test
@DisplayName("getJobResult: locally-owned single-file result → reads from FileStorage, 200 OK")
void getJobResult_singleFile_locallyOwned_readsFromStorage() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.getJobResult(JOB_ID)).thenReturn(completedJobWithFile());
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(LOCAL_NODE)));
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
ResponseEntity<?> response = makeController().getJobResult(JOB_ID);
assertEquals(HttpStatus.OK, response.getStatusCode());
}
// --------------------------------------------------------------------------------------
// Peer-owned 410 matrix across endpoints. Cross-cutting contract:
// - status = 410, Retry-After = 0
// - body.ownedBy = peer node, body.currentNode = local node
// - sticky-miss metric incremented exactly once per request
// The contract test above covers ALL of these properties for downloadFile in detail; this
// parameterized matrix asserts the same status + ownedBy + metric signals on every endpoint.
// --------------------------------------------------------------------------------------
private enum Endpoint {
DOWNLOAD_FILE,
GET_JOB_RESULT,
GET_JOB_STATUS,
CANCEL_JOB
}
private static Stream<Arguments> peerOwned410Scenarios() {
return Stream.of(
Arguments.of(Endpoint.DOWNLOAD_FILE),
Arguments.of(Endpoint.GET_JOB_RESULT),
Arguments.of(Endpoint.GET_JOB_STATUS),
Arguments.of(Endpoint.CANCEL_JOB));
}
@ParameterizedTest(name = "{0} peer-owned -> 410, ownedBy=peer, metric++")
@MethodSource("peerOwned410Scenarios")
void endpoint_peerOwned_returns410(Endpoint endpoint) throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
// Endpoint-specific wiring: the per-endpoint code path needs different mock setup
// before it reaches the sticky-410 guard.
switch (endpoint) {
case DOWNLOAD_FILE -> when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
case GET_JOB_RESULT ->
when(taskManager.getJobResult(JOB_ID)).thenReturn(completedJobWithFile());
case GET_JOB_STATUS -> when(taskManager.getJobResult(JOB_ID)).thenReturn(null);
case CANCEL_JOB -> {
when(jobQueue.isJobQueued(JOB_ID)).thenReturn(false);
when(taskManager.getJobResult(JOB_ID)).thenReturn(null);
}
}
ResponseEntity<?> response =
switch (endpoint) {
case DOWNLOAD_FILE -> makeController().downloadFile(FILE_ID);
case GET_JOB_RESULT -> makeController().getJobResult(JOB_ID);
case GET_JOB_STATUS -> makeController().getJobStatus(JOB_ID);
case CANCEL_JOB -> makeController().cancelJob(JOB_ID);
};
assertEquals(HttpStatus.GONE, response.getStatusCode());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals(PEER_NODE, body.get("ownedBy"));
assertEquals(LOCAL_NODE, body.get("currentNode"));
verify(stickyMissRecorder).recordStickyMiss();
// Storage / mutation must never be touched by a peer-routed request.
verify(fileStorage, never()).retrieveBytes(FILE_ID);
if (endpoint == Endpoint.CANCEL_JOB) {
verify(taskManager, never()).setError(JOB_ID, "Job was cancelled by user");
}
}
// --------------------------------------------------------------------------------------
// Unknown-job 404 matrix: when neither TaskManager nor JobStore knows the jobId, the
// controller must return 404 (not 410) and must NOT count it as a sticky miss.
// --------------------------------------------------------------------------------------
private static Stream<Arguments> unknownJob404Scenarios() {
return Stream.of(Arguments.of(Endpoint.GET_JOB_STATUS), Arguments.of(Endpoint.CANCEL_JOB));
}
@ParameterizedTest(name = "{0} unknown jobId -> 404 (not 410), no metric")
@MethodSource("unknownJob404Scenarios")
void endpoint_unknownJob_returns404(Endpoint endpoint) {
when(taskManager.getJobResult(JOB_ID)).thenReturn(null);
when(jobStore.get(JOB_ID)).thenReturn(Optional.empty());
if (endpoint == Endpoint.CANCEL_JOB) {
when(jobQueue.isJobQueued(JOB_ID)).thenReturn(false);
}
ResponseEntity<?> response =
switch (endpoint) {
case GET_JOB_STATUS -> makeController().getJobStatus(JOB_ID);
case CANCEL_JOB -> makeController().cancelJob(JOB_ID);
default -> throw new IllegalArgumentException(endpoint.name());
};
assertEquals(HttpStatus.NOT_FOUND, response.getStatusCode());
verify(stickyMissRecorder, never()).recordStickyMiss();
}
// --------------------------------------------------------------------------------------
// Single-instance / null-bean wiring: no NPE, no 410, no metric. These test SPECIFIC
// wiring permutations and so stay as discrete tests rather than rows.
// --------------------------------------------------------------------------------------
@Test
@DisplayName("Single-instance install (no ClusterBackplane bean): no 410, no NPE")
void singleInstance_noClusterBackplane_noGoneResponse() throws Exception {
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
ResponseEntity<?> response = makeController(null, jobStore).downloadFile(FILE_ID);
assertEquals(HttpStatus.OK, response.getStatusCode());
verify(fileStorage).retrieveBytes(FILE_ID);
}
@Test
@DisplayName("Single-instance install (no JobStore bean): no 410, no NPE")
void singleInstance_noJobStore_noGoneResponse() throws Exception {
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
ResponseEntity<?> response = makeController(clusterBackplane, null).downloadFile(FILE_ID);
assertEquals(HttpStatus.OK, response.getStatusCode());
verify(fileStorage).retrieveBytes(FILE_ID);
}
@Test
@DisplayName("Single-instance (no StickyMissRecorder bean) → no NPE, still 200 OK")
void noStickyMissRecorder_works() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(LOCAL_NODE)));
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
JobController c = makeController();
ReflectionTestUtils.setField(c, "stickyMissRecorder", null);
ResponseEntity<?> response = c.downloadFile(FILE_ID);
assertEquals(HttpStatus.OK, response.getStatusCode());
}
@Test
@DisplayName(
"cluster-mode but localNodeId is null → no NPE; 410 because owner is set and"
+ " differs from blank")
void clusterBackplanePresent_butLocalNodeIdNull_falsBackGracefully() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(null);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
// We still 410: owner is "node-peer", local is null → they don't match. Rather than
// silently 200-from-wrong-disk (which would serve garbage), we surface the mismatch.
ResponseEntity<?> response = makeController().downloadFile(FILE_ID);
assertEquals(HttpStatus.GONE, response.getStatusCode());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals("", body.get("currentNode"), "blank when localNodeId is null");
assertEquals(PEER_NODE, body.get("ownedBy"));
}
// --------------------------------------------------------------------------------------
// Ownership-service interaction: sticky-410 must take precedence over per-user auth so
// we never 403 on a peer-owned resource (which would leak existence + defeat the redirect).
// --------------------------------------------------------------------------------------
@Test
@DisplayName("Owner returns 410 even when JobOwnershipService allows access (orthogonal)")
void ownershipService_passes_butStickyStillReturns410() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
lenient().when(jobOwnershipService.validateJobAccess(JOB_ID)).thenReturn(true);
JobController c = makeController();
ReflectionTestUtils.setField(c, "jobOwnershipService", jobOwnershipService);
ResponseEntity<?> response = c.downloadFile(FILE_ID);
// OwnershipService is about *user* auth; sticky-410 is about *node* topology.
// Both must pass for a 200, and node-ownership is checked first so a non-owner
// never even runs the user-auth check.
assertEquals(HttpStatus.GONE, response.getStatusCode());
}
@Test
@DisplayName(
"downloadFile: peer-owned + ownership-denied → 410 (NOT 403) so we don't leak"
+ " file existence")
void downloadFile_peerOwned_ownershipDenied_returns410NotForbidden() throws Exception {
// Guard ordering: sticky-410 must run before user-auth. If user-auth ran first, a
// peer-owned-file request on the wrong node would fail user-auth (this node cannot
// verify access to a job it does not own) and return 403, which leaks file existence
// AND defeats the sticky-410 design (the frontend can't retry-with-affinity off a 403).
// Guard first so the user is redirected to the owner where the real auth check happens.
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
lenient().when(jobOwnershipService.validateJobAccess(JOB_ID)).thenReturn(false);
JobController c = makeController();
ReflectionTestUtils.setField(c, "jobOwnershipService", jobOwnershipService);
ResponseEntity<?> response = c.downloadFile(FILE_ID);
assertEquals(HttpStatus.GONE, response.getStatusCode());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals(PEER_NODE, body.get("ownedBy"));
// Crucially: never reached fileStorage, never returned 403.
verify(fileStorage, never()).retrieveBytes(FILE_ID);
}
@Test
@DisplayName(
"getJobStatus: peer-owned + ownership-denied → 410 (NOT 403) so we don't leak"
+ " job existence")
void getJobStatus_peerOwned_ownershipDenied_returns410NotForbidden() {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
lenient().when(jobOwnershipService.validateJobAccess(JOB_ID)).thenReturn(false);
JobController c = makeController();
ReflectionTestUtils.setField(c, "jobOwnershipService", jobOwnershipService);
ResponseEntity<?> response = c.getJobStatus(JOB_ID);
assertEquals(HttpStatus.GONE, response.getStatusCode());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals(PEER_NODE, body.get("ownedBy"));
}
@Test
@DisplayName(
"cancelJob: peer-owned + ownership-denied → 410 (NOT 403) so we don't leak job"
+ " existence")
void cancelJob_peerOwned_ownershipDenied_returns410NotForbidden() {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(jobQueue.isJobQueued(JOB_ID)).thenReturn(false);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(PEER_NODE)));
lenient().when(jobOwnershipService.validateJobAccess(JOB_ID)).thenReturn(false);
JobController c = makeController();
ReflectionTestUtils.setField(c, "jobOwnershipService", jobOwnershipService);
ResponseEntity<?> response = c.cancelJob(JOB_ID);
assertEquals(HttpStatus.GONE, response.getStatusCode());
Map<?, ?> body = (Map<?, ?>) response.getBody();
assertEquals(PEER_NODE, body.get("ownedBy"));
verify(taskManager, never()).setError(JOB_ID, "Job was cancelled by user");
}
// --------------------------------------------------------------------------------------
// S2: JobStore lookup hardening (local cache + Valkey-fault graceful-degrade)
// --------------------------------------------------------------------------------------
@Test
@DisplayName(
"guardNonOwner caches JobStore.get within TTL window: second call same jobId hits"
+ " cache, not Valkey")
void guardNonOwner_cachesJobStoreLookupWithinTtl() throws Exception {
when(clusterBackplane.localNodeId()).thenReturn(LOCAL_NODE);
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
when(jobStore.get(JOB_ID)).thenReturn(Optional.of(entryOwnedBy(LOCAL_NODE)));
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
JobController c = makeController();
c.downloadFile(FILE_ID);
c.downloadFile(FILE_ID);
c.downloadFile(FILE_ID);
// Three downloads of the same fileId == one HGETALL. Without the cache this would have
// been three Valkey round-trips on the hot download path.
verify(jobStore, times(1)).get(JOB_ID);
}
@Test
@DisplayName(
"guardNonOwner: JobStore.get throws (Valkey timeout) → falls through to local-disk"
+ " path, no 500 leaks to caller")
void guardNonOwner_jobStoreException_fallsThroughToLocalPath() throws Exception {
when(taskManager.findJobKeyByFileId(FILE_ID)).thenReturn(JOB_ID);
// Simulate a Valkey timeout. spring-data-redis surfaces these as runtime exceptions
// wrapping Lettuce errors; the controller must not let any RuntimeException leak out.
when(jobStore.get(JOB_ID)).thenThrow(new RuntimeException("Valkey command timeout"));
when(fileStorage.retrieveBytes(FILE_ID)).thenReturn("payload".getBytes());
ResponseEntity<?> response = makeController().downloadFile(FILE_ID);
// The download must succeed via the local-disk path. A brief Valkey blip cannot break
// every download attempt with 500 - we cleanly degrade to single-node behavior.
assertEquals(HttpStatus.OK, response.getStatusCode());
verify(fileStorage).retrieveBytes(FILE_ID);
// The exception did NOT count as a sticky-miss (it wasn't one; we just couldn't see).
verify(stickyMissRecorder, never()).recordStickyMiss();
}
}
@@ -19,6 +19,8 @@ import org.springframework.test.util.ReflectionTestUtils;
import jakarta.servlet.http.HttpServletRequest;
import stirling.software.common.cluster.ClusterBackplane;
import stirling.software.common.cluster.JobStore;
import stirling.software.common.model.job.JobResult;
import stirling.software.common.service.FileStorage;
import stirling.software.common.service.JobOwnershipService;
@@ -37,6 +39,10 @@ class JobControllerTest {
@Mock private JobOwnershipService jobOwnershipService;
@Mock private ClusterBackplane clusterBackplane;
@Mock private JobStore jobStore;
private MockHttpSession session;
@InjectMocks private JobController controller;
+11 -1
View File
@@ -53,8 +53,13 @@ dependencies {
api 'org.springframework.boot:spring-boot-starter-mail'
api 'org.springframework.boot:spring-boot-starter-cache'
api 'com.github.ben-manes.caffeine:caffeine'
implementation 'org.springframework.boot:spring-boot-starter-data-redis'
api 'io.swagger.core.v3:swagger-core-jakarta:2.2.46'
implementation 'com.bucket4j:bucket4j_jdk17-core:8.18.0'
implementation 'com.bucket4j:bucket4j_jdk17-core:8.19.0'
// Lettuce-backed Bucket4j ProxyManager used by ValkeyRateLimitStore for cluster-wide
// token-bucket rate limiting (parity with in-process Bucket4j semantics; no fixed-window
// boundary doubling).
implementation 'com.bucket4j:bucket4j_jdk17-lettuce:8.19.0'
// https://mvnrepository.com/artifact/com.bucket4j/bucket4j_jdk17
implementation "org.bouncycastle:bcprov-jdk18on:$bouncycastleVersion"
@@ -71,6 +76,11 @@ dependencies {
implementation('com.coveo:saml-client:5.0.0') {
exclude group: 'org.opensaml', module: 'opensaml-core'
}
// Testcontainers: spins up a real Valkey for LiveValkeyIntegrationTest in CI without
// needing a manually-started instance. Tests skip cleanly when Docker is unavailable.
testImplementation 'org.testcontainers:testcontainers:1.21.4'
testImplementation 'org.testcontainers:junit-jupiter:1.21.4'
}
tasks.register('prepareKotlinBuildScriptModel') {}
@@ -0,0 +1,47 @@
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 org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
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). Fires before any Valkey
* bean construction via {@link Ordered#HIGHEST_PRECEDENCE}.
*
* <p>There is no testing/development bypass. Live e2e tests that need cluster mode must inject a
* valid {@code stirling.premium.key} for a test-tier SERVER/ENTERPRISE license. Unit tests stub the
* {@code runningProOrHigher} bean directly.
*/
@Configuration
@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
@Order(Ordered.HIGHEST_PRECEDENCE)
@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 sales@stirlingpdf.com to obtain one), or set"
+ " cluster.enabled=false.");
}
log.info("Cluster license gate: SERVER/ENTERPRISE license verified, cluster mode allowed.");
}
}
@@ -0,0 +1,123 @@
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<>();
// In-flight job count for THIS node.
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);
}
}
/**
* Increment when {@code JobController} returns 410 Gone because the requested job's owner is a
* peer node. Surfaces to {@code stirling_cluster_sticky_miss_total}.
*/
@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();
}
/**
* Publish (or update) the queue depth gauge for {@code lane}. Idempotent - safe to call hot.
* Known lanes (FAST, SLOW, AI) are pre-registered at construction; unknown lanes register on
* first call.
*/
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 {
/** TTL of the node entry in the registry. Set to 3x the heartbeat interval. */
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();
long heartbeatMs =
cluster.getNode() == null ? 10_000L : cluster.getNode().getHeartbeatIntervalMs();
// TTL = 3x heartbeat: tolerate one missed tick 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:10000}")
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);
}
}
// ---------- SmartLifecycle (see class javadoc for ordering rationale) ----------
@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; // stopped first; LettuceConnectionFactory's phase is 0
}
@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,25 @@
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: cluster mode is on AND the configured backplane is Valkey.
*
* <p>Either condition alone is insufficient to load a Valkey bean. With {@code enabled=true} but
* {@code backplane=inprocess}, loading the Valkey beans would crash at boot because there's no
* {@code StringRedisTemplate}; with {@code enabled=false} the whole cluster mode is off. Combining
* the two stops both footguns.
*
* <p>Spring's {@code @ConditionalOnProperty} cannot be applied twice on the same class, so we use
* {@code @ConditionalOnExpression} via this meta-annotation.
*/
@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@ConditionalOnExpression(
"${cluster.enabled:false} and '${cluster.backplane:inprocess}'.equals('valkey')")
public @interface ConditionalOnValkeyBackplane {}
@@ -0,0 +1,62 @@
package stirling.software.proprietary.cluster.valkey;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
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
@ConditionalOnProperty(name = "cluster.enabled", havingValue = "true")
@org.springframework.boot.autoconfigure.condition.ConditionalOnProperty(
name = "cluster.backplane",
havingValue = "valkey")
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();
}
/**
* Disable the local {@code TaskManager#cleanupOldJobs} loop on Valkey-backed clusters: {@link
* ValkeyJobStore} stores every entry with a TTL pExpire and the reverse-index entries share
* that TTL, so Valkey itself evicts expired job state. Running the local cleanup loop on top of
* that would only delete per-node in-memory {@code TaskManager} caches that the cluster-visible
* {@code JobStore} has already authoritative state for.
*/
@Override
public boolean shouldRunLocalCleanup() {
return false;
}
}
@@ -0,0 +1,222 @@
package stirling.software.proprietary.cluster.valkey;
import java.net.URI;
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.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;
/** Wires the LettuceConnectionFactory and StringRedisTemplate for cluster mode. */
@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();
String url = cluster.getValkey().getUrl();
if (url == null || url.isBlank()) {
throw new IllegalStateException("cluster.valkey.url must be set when backplane=valkey");
}
URI uri = URI.create(url);
boolean tls = "rediss".equalsIgnoreCase(uri.getScheme());
int port = uri.getPort() <= 0 ? 6379 : uri.getPort();
RedisStandaloneConfiguration cfg = new RedisStandaloneConfiguration(uri.getHost(), port);
if (uri.getUserInfo() != null) {
String[] parts = uri.getUserInfo().split(":", 2);
if (parts.length == 2) {
cfg.setUsername(parts[0]);
cfg.setPassword(RedisPassword.of(parts[1]));
} else if (parts.length == 1 && !parts[0].isBlank()) {
cfg.setPassword(RedisPassword.of(parts[0]));
}
}
boolean skipCertVerification =
cluster.getValkey().getTls() != null
&& cluster.getValkey().getTls().isSkipCertVerification();
LettuceClientConfiguration clientConfig =
buildClientConfiguration(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, uri.getHost(), port, tls);
log.info(
"Valkey connection configured: {}:{} tls={} verifyPeer={}",
uri.getHost(),
port,
tls,
tls ? clientConfig.getVerifyMode() : "n/a");
return factory;
}
/**
* Builds the Lettuce client configuration with TLS verification pinned. Package-private so unit
* tests can verify the {@code verifyPeer} mode without standing up a real Valkey.
*
* <p>{@code verifyPeer(FULL)} is pinned explicitly so a future Spring Data Redis default change
* cannot silently weaken our TLS handshake. {@code FULL} = X.509 chain + hostname check (per
* Lettuce's {@link SslVerifyMode}). The {@code skipCertVerification} opt-out is for local dev
* with self-signed certs only; production deployments MUST leave it false.
*/
static LettuceClientConfiguration buildClientConfiguration(
boolean tls, boolean skipCertVerification) {
LettuceClientConfiguration.LettuceClientConfigurationBuilder clientBuilder =
LettuceClientConfiguration.builder();
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 of retry. Boot-time only.
*
* <p>Auth-class failures (WRONGPASS / NOAUTH / NOPERM) are unrecoverable and surfaced
* immediately on the first attempt; only transport-level errors (connection refused, timeout,
* host unreachable) get the retry loop.
*
* <p>Package-private so unit tests can drive it with a mocked connection factory.
*/
static void eagerHandshake(
LettuceConnectionFactory factory, String host, int port, boolean tls) {
RuntimeException last = null;
for (int attempt = 1; attempt <= 10; attempt++) {
try {
String pong = factory.getConnection().ping();
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 a Lettuce {@link RedisCommandExecutionException} whose message
* starts with an auth-class server reply (WRONGPASS, NOAUTH, NOPERM). Spring Data Redis wraps
* Lettuce errors in a {@code RedisSystemException}, so the auth signal usually lives one level
* down from the thrown exception.
*
* <p>Checked for a typed alternative: neither Spring Data Redis 4.0.5 nor Lettuce 6.8.2 ships a
* {@code RedisAuthenticationException} on the classpath, so we keep the message-prefix match.
* Revisit when upgrading Spring Data Redis if a typed exception lands upstream.
*/
static boolean isAuthFailure(Throwable t) {
for (Throwable cur = t; cur != null; cur = cur.getCause()) {
if (cur instanceof RedisCommandExecutionException && hasAuthPrefix(cur.getMessage())) {
return true;
}
// Defensive: some translations preserve the original message on the wrapper itself.
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,92 @@
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;
template.execute(RELEASE_SCRIPT, Collections.singletonList(key), value);
}
@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,116 @@
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() {
// SCAN, not KEYS - KEYS blocks the Valkey server for the duration of the walk.
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);
Map<byte[], byte[]> hash =
connection.hashCommands().hGetAll(jobKeyBytes);
List<byte[]> keysToDelete = new ArrayList<>();
keysToDelete.add(jobKeyBytes);
if (hash != null) {
byte[] fileIdsBytes =
hash.get(
"fileIds"
.getBytes(
StandardCharsets
.UTF_8));
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) {
// The shapes we serialize are simple List<String> / Map<String,String>; Jackson
// can encode these without escapes that fail. Surface anything unexpected loud and
// early rather than persisting a half-serialized field that would re-throw on read.
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,62 @@
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) {
// SCAN, not KEYS: KEYS blocks the server until it has walked the entire keyspace.
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,82 @@
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.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.
*
* <p>Replaces the earlier hand-rolled INCR+EXPIRE Lua fixed-window script. The fixed-window impl
* could allow a caller to spend the full bucket at second 59 of one window and the full bucket
* again at second 1 of the next window (effective burst of 2x capacity at boundaries). The Bucket4j
* token bucket refills continuously and removes that boundary doubling, giving cross-node parity
* with the in-process {@code InProcessRateLimitStore} which already 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 yet supported by this rate limit impl)");
}
this.proxyManager = Bucket4jLettuce.casBasedBuilder(redisClient).build();
}
@PreDestroy
void shutdown() {
// Lettuce client lifecycle is owned by Spring (LettuceConnectionFactory#destroy), so we
// only drop the proxy reference. No explicit close needed.
proxyManager = null;
}
@Override
public RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod) {
byte[] key = (PREFIX + bucketKey).getBytes(StandardCharsets.UTF_8);
// Greedy refill of capacity tokens per refillPeriod, matching InProcessRateLimitStore
// semantics (continuously refilling, no fixed-window boundary doubling).
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());
}
}
@@ -0,0 +1,71 @@
package stirling.software.proprietary.cluster;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
/**
* Runtime license gate contract. Verifies cluster mode is gated by the existing {@code
* runningProOrHigher} bean and reports a clear error when the license is missing.
*
* <p>The gate uses reflection-friendly field injection (one optional bean) so the test wires it
* directly without bringing up a full Spring context.
*/
class ClusterLicenseGateTest {
private void injectRunningProOrHigher(ClusterLicenseGate gate, Boolean value) throws Exception {
Field f = ClusterLicenseGate.class.getDeclaredField("runningProOrHigher");
f.setAccessible(true);
f.set(gate, value);
}
private void invokeVerify(ClusterLicenseGate gate) throws Throwable {
Method m = ClusterLicenseGate.class.getDeclaredMethod("verifyLicense");
m.setAccessible(true);
try {
m.invoke(gate);
} catch (InvocationTargetException e) {
throw e.getCause();
}
}
@Test
void serverOrEnterpriseLicense_allowsClusterMode() throws Throwable {
ClusterLicenseGate gate = new ClusterLicenseGate();
injectRunningProOrHigher(gate, Boolean.TRUE);
assertDoesNotThrow(() -> invokeVerify(gate));
}
@Test
void normalLicense_refusesClusterMode_withActionableMessage() throws Exception {
ClusterLicenseGate gate = new ClusterLicenseGate();
injectRunningProOrHigher(gate, Boolean.FALSE);
IllegalStateException ex =
assertThrows(IllegalStateException.class, () -> invokeVerify(gate));
String msg = ex.getMessage();
// The error message must tell the operator exactly what to do.
assertTrue(msg.contains("SERVER"), "message must mention SERVER license tier: " + msg);
assertTrue(msg.contains("ENTERPRISE"), "message must mention ENTERPRISE tier: " + msg);
assertTrue(
msg.contains("stirling.premium.key") || msg.contains("license key"),
"message must explain how to set the license: " + msg);
assertTrue(
msg.contains("cluster.enabled=false"),
"message must offer the opt-out (disable cluster): " + msg);
}
@Test
void saasFlavor_bypassesGate_whenRunningProOrHigherBeanAbsent() throws Throwable {
// In saas builds the runningProOrHigher bean is @Profile("security & !saas") so absent.
// The gate's @Autowired(required=false) leaves the field null. Must not throw.
ClusterLicenseGate gate = new ClusterLicenseGate();
// field stays null (default)
assertDoesNotThrow(() -> invokeVerify(gate));
}
}
@@ -0,0 +1,132 @@
package stirling.software.proprietary.cluster;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import stirling.software.common.model.ApplicationProperties;
/**
* Contract test for cluster metrics. Asserts every named metric is registered and the recorder
* methods write to them, so dashboards do not silently lose a metric to a rename.
*/
class ClusterMetricsTest {
private SimpleMeterRegistry registry;
private ClusterMetrics metrics;
private static final String NODE = "test-node";
@BeforeEach
void setUp() {
registry = new SimpleMeterRegistry();
ApplicationProperties props = new ApplicationProperties();
props.getCluster().getNode().setId(NODE);
metrics = new ClusterMetrics(registry, props);
}
@Test
void registersAllRequiredMeters() {
assertNotNull(registry.find("stirling_cluster_sticky_miss_total").counter());
assertNotNull(registry.find("stirling_cluster_ratelimit_rejected_total").counter());
assertNotNull(registry.find("stirling_cluster_backplane_latency_seconds").timer());
assertNotNull(registry.find("stirling_cluster_job_wait_seconds").timer());
Gauge inflight = registry.find("stirling_cluster_jobs_inflight").tag("node", NODE).gauge();
assertNotNull(inflight, "jobs_inflight gauge with node tag must be registered eagerly");
}
@Test
void registersKnownLaneGaugesEagerly() {
// Lanes (FAST, SLOW, AI) are a fixed enum, so all three gauges must exist at construction
// - dashboards must never have a missing series for a known lane.
for (String lane : new String[] {"FAST", "SLOW", "AI"}) {
Gauge g = registry.find("stirling_cluster_queue_depth").tag("lane", lane).gauge();
assertNotNull(g, "lane gauge must be eagerly registered for " + lane);
assertEquals(0.0, g.value(), "lane gauge default value must be 0 for " + lane);
}
assertEquals(
3,
registry.find("stirling_cluster_queue_depth").gauges().size(),
"exactly the three known lane gauges should be registered at boot");
}
@Test
void recordStickyMissIncrementsCounter() {
metrics.recordStickyMiss();
metrics.recordStickyMiss();
assertEquals(2.0, registry.find("stirling_cluster_sticky_miss_total").counter().count());
}
@Test
void recordRateLimitRejectIncrementsCounter() {
metrics.recordRateLimitReject();
assertEquals(
1.0, registry.find("stirling_cluster_ratelimit_rejected_total").counter().count());
}
@Test
void incrementAndDecrementInflightUpdatesGauge() {
metrics.incrementInflight();
metrics.incrementInflight();
metrics.incrementInflight();
metrics.decrementInflight();
Gauge gauge = registry.find("stirling_cluster_jobs_inflight").tag("node", NODE).gauge();
assertEquals(2.0, gauge.value(), "expected 2 inflight after 3 inc / 1 dec");
}
@Test
void setQueueDepthUpdatesEagerlyRegisteredLaneGauge() {
// The three known-lane gauges (FAST, SLOW, AI) are registered eagerly at construction
// (see registersKnownLaneGaugesEagerly); setQueueDepth only updates the holder value.
metrics.setQueueDepth("FAST", 4);
metrics.setQueueDepth("SLOW", 7);
Gauge fast = registry.find("stirling_cluster_queue_depth").tag("lane", "FAST").gauge();
Gauge slow = registry.find("stirling_cluster_queue_depth").tag("lane", "SLOW").gauge();
assertEquals(4.0, fast.value());
assertEquals(7.0, slow.value());
}
@Test
void setQueueDepthForUnknownLane_lazyRegistersFallbackGauge() {
// Defensive: if a caller passes an unrecognised lane, we still register so we don't lose
// the signal. This is a fallback, not the supported path.
metrics.setQueueDepth("custom-lane", 5);
Gauge g = registry.find("stirling_cluster_queue_depth").tag("lane", "custom-lane").gauge();
assertNotNull(g);
assertEquals(5.0, g.value());
}
@Test
void setQueueDepthIsIdempotentAcrossCalls() {
metrics.setQueueDepth("FAST", 1);
metrics.setQueueDepth("FAST", 2);
metrics.setQueueDepth("FAST", 9);
// Only one gauge per lane, not three.
assertEquals(
1,
registry.find("stirling_cluster_queue_depth").tag("lane", "FAST").gauges().size());
assertEquals(
9.0,
registry.find("stirling_cluster_queue_depth").tag("lane", "FAST").gauge().value());
}
@Test
void backplaneLatencyTimerAcceptsRecordings() {
metrics.backplaneLatency().record(java.time.Duration.ofMillis(7));
metrics.backplaneLatency().record(java.time.Duration.ofMillis(11));
assertEquals(
2L, registry.find("stirling_cluster_backplane_latency_seconds").timer().count());
}
@Test
void jobWaitTimerAcceptsRecordings() {
metrics.jobWaitSeconds().record(java.time.Duration.ofMillis(50));
assertEquals(1L, registry.find("stirling_cluster_job_wait_seconds").timer().count());
}
}
@@ -0,0 +1,126 @@
package stirling.software.proprietary.cluster;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.time.Duration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.test.util.ReflectionTestUtils;
import stirling.software.common.cluster.ClusterNode;
import stirling.software.common.cluster.InstanceRegistry;
import stirling.software.common.model.ApplicationProperties;
/** Verifies the bootstrap registers / heartbeats / deregisters as expected. */
class ClusterNodeBootstrapTest {
private InstanceRegistry registry;
private ApplicationProperties props;
private ClusterNodeBootstrap bootstrap;
@BeforeEach
void setUp() {
registry = mock(InstanceRegistry.class);
props = new ApplicationProperties();
props.getCluster().setEnabled(true);
props.getCluster().getNode().setId("node-test-1");
props.getCluster().getNode().setRole("worker");
// Pin heartbeat to 10s so TTL math is stable across PR2 default changes (TTL = 3x = 30s).
props.getCluster().getNode().setHeartbeatIntervalMs(10_000L);
bootstrap = new ClusterNodeBootstrap(props, registry);
ReflectionTestUtils.setField(bootstrap, "serverPort", 8080);
}
@Test
void registerOnStartupCallsRegistryWithResolvedNodeId() {
bootstrap.registerOnStartup();
ArgumentCaptor<ClusterNode> nodeCaptor = ArgumentCaptor.forClass(ClusterNode.class);
ArgumentCaptor<Duration> ttlCaptor = ArgumentCaptor.forClass(Duration.class);
verify(registry, times(1)).register(nodeCaptor.capture(), ttlCaptor.capture());
ClusterNode captured = nodeCaptor.getValue();
assertEquals("node-test-1", captured.nodeId());
assertTrue(captured.internalAddress().startsWith("http://"));
assertTrue(captured.internalAddress().endsWith(":8080"));
assertEquals("WORKER", captured.role());
assertEquals(30L, ttlCaptor.getValue().toSeconds());
}
@Test
void registerHonoursExplicitInternalAddress() {
props.getCluster().getNode().setInternalAddress("app-1:8080");
bootstrap.registerOnStartup();
ArgumentCaptor<ClusterNode> nodeCaptor = ArgumentCaptor.forClass(ClusterNode.class);
verify(registry).register(nodeCaptor.capture(), any());
assertEquals("http://app-1:8080", nodeCaptor.getValue().internalAddress());
}
@Test
void registerUsesHttpsSchemeWhenConfigured() {
// SE3: nodes that terminate TLS themselves need https:// in the registry so peers can reach
// them. The default (http) is correct for the common LB-terminates-TLS topology.
props.getCluster().getNode().setInternalAddress("app-1:8443");
props.getCluster().getNode().setScheme("https");
ClusterNodeBootstrap httpsBootstrap = new ClusterNodeBootstrap(props, registry);
ReflectionTestUtils.setField(httpsBootstrap, "serverPort", 8443);
httpsBootstrap.registerOnStartup();
ArgumentCaptor<ClusterNode> nodeCaptor = ArgumentCaptor.forClass(ClusterNode.class);
verify(registry).register(nodeCaptor.capture(), any());
assertEquals("https://app-1:8443", nodeCaptor.getValue().internalAddress());
}
@Test
void heartbeatAfterStartup_callsRegister_forSelfHealing() {
// Heartbeat re-invokes register() (idempotent) so a wiped backplane re-populates
// every field, not just lastHeartbeat. Expect 2 register() calls: startup + heartbeat.
bootstrap.start();
bootstrap.registerOnStartup();
bootstrap.heartbeat();
verify(registry, times(2))
.register(
any(ClusterNode.class),
org.mockito.ArgumentMatchers.eq(Duration.ofSeconds(30)));
}
@Test
void smartLifecycleStop_deregisters() {
bootstrap.start();
bootstrap.registerOnStartup();
bootstrap.stop();
verify(registry, times(1)).deregister("node-test-1");
}
@Test
void smartLifecycleStop_beforeStartup_isNoop() {
bootstrap.stop();
verify(registry, never()).deregister(any());
}
@Test
void heartbeatAfterStop_doesNotReRegister() {
// Heartbeat-after-stop race: SmartLifecycle.stop() deregisters, but the @Scheduled
// tick keeps firing during a slow drain. Without a guard, the next tick would
// re-register the dead node and the entry would resurface in the registry until TTL
// expiry. Rolling deploys with slow shutdown = draining nodes keep re-announcing
// themselves indefinitely.
bootstrap.start();
bootstrap.registerOnStartup();
// 1 register from startup.
verify(registry, times(1)).register(any(ClusterNode.class), any(Duration.class));
bootstrap.stop();
verify(registry, times(1)).deregister("node-test-1");
// Critical: next scheduled tick after stop must NOT re-register.
bootstrap.heartbeat();
// Still exactly 1 register call (the startup one); no second register from heartbeat.
verify(registry, times(1)).register(any(ClusterNode.class), any(Duration.class));
}
}
@@ -0,0 +1,147 @@
package stirling.software.proprietary.cluster;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import stirling.software.common.cluster.ClusterBackplane;
import stirling.software.common.cluster.JobStore;
import stirling.software.common.cluster.JobStoreEntry;
import stirling.software.common.cluster.KeyValueCache;
import stirling.software.common.cluster.RateLimitStore;
import stirling.software.common.cluster.RateLimitStore.RateLimitDecision;
import stirling.software.common.cluster.inprocess.InProcessJobStore;
import stirling.software.common.cluster.inprocess.InProcessKeyValueCache;
import stirling.software.common.cluster.inprocess.InProcessRateLimitStore;
/**
* Multi-node CONTRACT validation in a single JVM. Shares the cluster-visible state (JobStore,
* RateLimitStore, KeyValueCache) across two "nodes" - exactly the partition Valkey creates in
* production - and asserts cross-node visibility / global counters / cache propagation.
*
* <p><b>Scope note:</b> this test uses the in-process backplane implementations ({@link
* InProcessJobStore}, {@link InProcessKeyValueCache}, {@link InProcessRateLimitStore}), not the
* Valkey impls. It verifies the CONTRACT every {@code ClusterBackplane} flavor must honor (shared
* map semantics, monotonic counters, evict propagation) and is fast / Docker-free so it runs on
* every PR. The Valkey impls share the same contract by construction (single shared Valkey keyspace
* = single shared {@code ConcurrentHashMap} from the consumer's POV), so a regression here would
* also break the Valkey path.
*
* <p>For Valkey-specific verification (real Lettuce client, MULTI/EXEC atomicity, TTL expiry, WATCH
* race semantics on {@code delete}) see {@code LiveValkeyIntegrationTest}, which spins up a real
* Valkey via Testcontainers.
*
* <p>Result downloads are handled by sticky-session affinity at the load balancer + a {@code 410
* Gone} response on the rare miss (verified in {@code JobControllerOwnershipTest}).
*/
class MultiNodeClusterScenarioTest {
private JobStore sharedJobStore;
private RateLimitStore sharedRateLimit;
private KeyValueCache sharedCache;
private ClusterBackplane backplaneA;
private ClusterBackplane backplaneB;
@BeforeEach
void setUp() {
sharedJobStore = new InProcessJobStore();
sharedRateLimit = new InProcessRateLimitStore();
sharedCache = new InProcessKeyValueCache();
backplaneA = constBackplane("node-A", "valkey");
backplaneB = constBackplane("node-B", "valkey");
}
@Test
@DisplayName("async job created on node-A is readable from node-B via shared JobStore")
void jobStatusVisibleCrossNode() {
JobStoreEntry entry =
new JobStoreEntry(
"job-1",
JobStoreEntry.JobState.RUNNING,
"node-A",
Instant.now(),
null,
null,
List.of("file-1"),
Map.of());
sharedJobStore.put(entry, Duration.ofMinutes(30));
Optional<JobStoreEntry> seenOnB = sharedJobStore.get("job-1");
assertTrue(seenOnB.isPresent(), "node-B must see node-A's job in shared JobStore");
assertEquals("node-A", seenOnB.get().owningNodeId());
assertEquals(JobStoreEntry.JobState.RUNNING, seenOnB.get().state());
}
@Test
@DisplayName("global rate limit - capacity counted once across both nodes")
void rateLimitGlobalAcrossNodes() {
long capacity = 4L;
RateLimitDecision a1 =
sharedRateLimit.tryConsume("user:bob", capacity, Duration.ofMinutes(1));
RateLimitDecision b1 =
sharedRateLimit.tryConsume("user:bob", capacity, Duration.ofMinutes(1));
RateLimitDecision a2 =
sharedRateLimit.tryConsume("user:bob", capacity, Duration.ofMinutes(1));
RateLimitDecision b2 =
sharedRateLimit.tryConsume("user:bob", capacity, Duration.ofMinutes(1));
RateLimitDecision a3 =
sharedRateLimit.tryConsume("user:bob", capacity, Duration.ofMinutes(1));
assertTrue(a1.allowed());
assertTrue(b1.allowed());
assertTrue(a2.allowed());
assertTrue(b2.allowed());
assertFalse(a3.allowed(), "5th request across both nodes must be rejected (limit=4)");
}
@Test
@DisplayName("KeyValueCache populated on A is observed on B; evict on A propagates")
void apiKeyCacheVisibleCrossNode() {
sharedCache.put("apikey", "hash-bob", "bob", Duration.ofSeconds(60));
assertEquals("bob", sharedCache.get("apikey", "hash-bob").orElse(null));
sharedCache.evict("apikey", "hash-bob");
assertFalse(sharedCache.get("apikey", "hash-bob").isPresent());
}
@Test
@DisplayName("backplaneType reports 'valkey' on every node; localNodeId is distinct")
void backplaneType() {
assertEquals("valkey", backplaneA.backplaneType());
assertEquals("valkey", backplaneB.backplaneType());
assertEquals("node-A", backplaneA.localNodeId());
assertEquals("node-B", backplaneB.localNodeId());
assertNotEquals(backplaneA.localNodeId(), backplaneB.localNodeId());
assertNotNull(backplaneA.localNodeId());
}
private ClusterBackplane constBackplane(String nodeId, String type) {
return new ClusterBackplane() {
@Override
public boolean isHealthy() {
return true;
}
@Override
public String backplaneType() {
return type;
}
@Override
public String localNodeId() {
return nodeId;
}
};
}
}
@@ -0,0 +1,502 @@
package stirling.software.proprietary.cluster.valkey;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIf;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import stirling.software.common.cluster.ClusterNode;
import stirling.software.common.cluster.DistributedLock;
import stirling.software.common.cluster.JobStoreEntry;
import stirling.software.common.cluster.RateLimitStore.RateLimitDecision;
import stirling.software.common.model.ApplicationProperties;
/**
* Live integration tests against a real Valkey instance, started by Testcontainers. The
* {@code @EnabledIf} guard probes the Docker daemon via {@link
* DockerClientFactory#isDockerAvailable()} (non-throwing) so the suite skips cleanly when Docker is
* unavailable - without that guard, {@code @Testcontainers} would throw {@code initializationError}
* (test FAILURE, not skip) on CI runners without Docker.
*/
@Testcontainers
@EnabledIf("isDockerAvailable")
class LiveValkeyIntegrationTest {
@Container
static final GenericContainer<?> VALKEY =
new GenericContainer<>(DockerImageName.parse("valkey/valkey:8.0-alpine"))
.withExposedPorts(6379);
static boolean isDockerAvailable() {
return DockerClientFactory.instance().isDockerAvailable();
}
private static LettuceConnectionFactory factoryA;
private static LettuceConnectionFactory factoryB;
private static StringRedisTemplate templateA;
private static StringRedisTemplate templateB;
@BeforeAll
static void connect() {
String host = VALKEY.getHost();
int port = VALKEY.getMappedPort(6379);
factoryA = new LettuceConnectionFactory(new RedisStandaloneConfiguration(host, port));
factoryA.afterPropertiesSet();
factoryB = new LettuceConnectionFactory(new RedisStandaloneConfiguration(host, port));
factoryB.afterPropertiesSet();
templateA = new StringRedisTemplate(factoryA);
templateB = new StringRedisTemplate(factoryB);
// Flush so each run starts clean (test-only)
templateA.getConnectionFactory().getConnection().serverCommands().flushAll();
}
@AfterAll
static void disconnect() {
if (factoryA != null) factoryA.destroy();
if (factoryB != null) factoryB.destroy();
}
@Test
@DisplayName("Valkey reachable and isHealthy() = true after PING round-trip")
void backplaneHealthy() {
ApplicationProperties propsA = newProps("node-A");
ValkeyClusterBackplane bp = new ValkeyClusterBackplane(propsA, templateA);
assertEquals("valkey", bp.backplaneType());
assertEquals("node-A", bp.localNodeId());
assertTrue(bp.isHealthy(), "Valkey must be reachable in the Testcontainers instance");
}
@Test
@DisplayName("JobStore put on connection A, get on connection B reads the same entry")
void jobStoreCrossConnectionVisibility() {
ValkeyJobStore storeA = new ValkeyJobStore(templateA);
ValkeyJobStore storeB = new ValkeyJobStore(templateB);
JobStoreEntry entry =
new JobStoreEntry(
"live-job-1",
JobStoreEntry.JobState.RUNNING,
"node-A",
Instant.now(),
null,
null,
List.of("live-file-1"),
Map.of("k", "v"));
storeA.put(entry, Duration.ofSeconds(30));
Optional<JobStoreEntry> seen = storeB.get("live-job-1");
assertTrue(seen.isPresent(), "storeB on different connection must see storeA's write");
assertEquals("node-A", seen.get().owningNodeId());
assertEquals(JobStoreEntry.JobState.RUNNING, seen.get().state());
// Reverse file→job index
assertEquals("live-job-1", storeB.findJobIdByFileId("live-file-1").orElse(null));
}
@Test
@DisplayName("JobStore entry expires after the configured duration")
void jobStoreTtlExpires() throws InterruptedException {
ValkeyJobStore store = new ValkeyJobStore(templateA);
store.put(
new JobStoreEntry(
"ttl-job",
JobStoreEntry.JobState.PENDING,
"node-A",
Instant.now(),
null,
null,
List.of(),
Map.of()),
Duration.ofSeconds(2));
assertTrue(store.exists("ttl-job"));
// Valkey expiry is lazy / sample-based so a 500 ms margin can race; use ~1 s.
// Poll for up to 3 s so we don't double the suite's wall-clock when Valkey is timely.
long deadline = System.currentTimeMillis() + 3000;
boolean expired = false;
while (System.currentTimeMillis() < deadline) {
if (!store.exists("ttl-job")) {
expired = true;
break;
}
Thread.sleep(100);
}
assertTrue(expired, "entry should TTL-expire within 3 s of a 2 s TTL");
}
@Test
@DisplayName("KeyValueCache propagates across connections; evict observed cross-connection")
void keyValueCacheCrossConnection() {
ValkeyKeyValueCache cacheA = new ValkeyKeyValueCache(templateA);
ValkeyKeyValueCache cacheB = new ValkeyKeyValueCache(templateB);
cacheA.put("apikey", "hash-bob", "bob", Duration.ofSeconds(30));
assertEquals("bob", cacheB.get("apikey", "hash-bob").orElse(null));
cacheA.evict("apikey", "hash-bob");
assertFalse(cacheB.get("apikey", "hash-bob").isPresent());
}
@Test
@DisplayName("RateLimitStore enforces ONE global budget across two instances")
void rateLimitGlobalAcrossInstances() {
ValkeyRateLimitStore storeA = newRateLimitStore(factoryA);
ValkeyRateLimitStore storeB = newRateLimitStore(factoryB);
String key = "live-user:alice";
long capacity = 4;
AtomicInteger allowed = new AtomicInteger();
for (int i = 0; i < 8; i++) {
// alternate consumers
var store = (i % 2 == 0) ? storeA : storeB;
RateLimitDecision d = store.tryConsume(key, capacity, Duration.ofSeconds(30));
if (d.allowed()) allowed.incrementAndGet();
}
assertEquals(
4,
allowed.get(),
"exactly 4 (the global capacity) must be allowed across both instances");
}
@Test
@DisplayName("DistributedLock excludes a second acquirer on a different connection")
void distributedLockMutualExclusion() {
ValkeyDistributedLock lockA = new ValkeyDistributedLock(templateA);
ValkeyDistributedLock lockB = new ValkeyDistributedLock(templateB);
Optional<DistributedLock.LockHandle> heldByA =
lockA.tryAcquire("election-X", Duration.ofSeconds(30));
assertTrue(heldByA.isPresent());
Optional<DistributedLock.LockHandle> heldByB =
lockB.tryAcquire("election-X", Duration.ofSeconds(30));
assertFalse(heldByB.isPresent(), "second acquirer must fail while A holds the lock");
heldByA.get().release();
// After release, B can acquire
Optional<DistributedLock.LockHandle> retry =
lockB.tryAcquire("election-X", Duration.ofSeconds(30));
assertTrue(retry.isPresent());
retry.get().release();
}
@Test
@DisplayName("register is atomic (hash + TTL committed together, no orphan keys on crash)")
void registryRegisterIsAtomic() {
ValkeyInstanceRegistry reg = new ValkeyInstanceRegistry(templateA);
ClusterNode node =
new ClusterNode(
"atomic-node-" + java.util.UUID.randomUUID(),
"10.0.0.99:8080",
Instant.now(),
"BOTH");
reg.register(node, Duration.ofSeconds(30));
// After register returns, the key must have a positive TTL. A TTL of -1 (no expiry)
// would mean the EXPIRE didn't ride along inside the MULTI/EXEC and the entry would
// persist forever past node death.
Long ttlMs =
templateA.getExpire(
"stirling:nodes:" + node.nodeId(),
java.util.concurrent.TimeUnit.MILLISECONDS);
assertNotNull(ttlMs);
assertTrue(
ttlMs > 0 && ttlMs <= 30_000,
"register() must atomically arm TTL; expected (0, 30000] ms, got " + ttlMs);
// Sanity: the hash fields are present too (atomic commit, both sides observable).
Optional<ClusterNode> seen = reg.lookup(node.nodeId());
assertTrue(seen.isPresent(), "hash fields must be visible after atomic register()");
assertEquals("10.0.0.99:8080", seen.get().internalAddress());
reg.deregister(node.nodeId());
}
@Test
@DisplayName("register on connection A is visible from connection B")
void registryCrossConnection() {
ValkeyInstanceRegistry regA = new ValkeyInstanceRegistry(templateA);
ValkeyInstanceRegistry regB = new ValkeyInstanceRegistry(templateB);
ClusterNode node = new ClusterNode("live-node-7", "10.0.0.7:8080", Instant.now(), "BOTH");
regA.register(node, Duration.ofSeconds(30));
Optional<ClusterNode> seen = regB.lookup("live-node-7");
assertTrue(seen.isPresent());
assertEquals("10.0.0.7:8080", seen.get().internalAddress());
boolean inActive =
regB.activeNodes().stream().anyMatch(n -> "live-node-7".equals(n.nodeId()));
assertTrue(inActive);
regA.deregister("live-node-7");
assertFalse(regB.lookup("live-node-7").isPresent());
}
@Test
@DisplayName("Bucket4j: no fixed-window boundary doubling (parity with in-process semantics)")
void rateLimitNoBoundaryDoubling() throws InterruptedException {
// The old Lua INCR+EXPIRE allowed 2x capacity at window boundaries: empty bucket at end
// of window N, full bucket at start of window N+1, observable as 2*capacity within the
// boundary. Token-bucket greedy refill smooths this so total over a short boundary window
// never exceeds capacity + at most one full refill share.
ValkeyRateLimitStore store = newRateLimitStore(factoryA);
String key = "boundary-" + java.util.UUID.randomUUID();
long capacity = 5;
Duration window = Duration.ofMillis(500);
// Drain the bucket in window N.
int firstAllowed = 0;
for (int i = 0; i < 10; i++) {
if (store.tryConsume(key, capacity, window).allowed()) firstAllowed++;
}
assertEquals(capacity, firstAllowed, "must allow exactly capacity tokens initially");
// Wait just past the window. Under fixed-window we'd see another full capacity allowed
// immediately (boundary doubling). Under token-bucket greedy refill we get roughly the
// capacity-per-window rate, not a full burst again.
Thread.sleep(window.toMillis() + 50);
int secondAllowed = 0;
long start = System.nanoTime();
for (int i = 0; i < 20 && (System.nanoTime() - start) < 20_000_000L; i++) {
if (store.tryConsume(key, capacity, window).allowed()) secondAllowed++;
}
// Allow slack but assert we cannot drain a *second* full capacity instantly.
assertTrue(
secondAllowed <= capacity,
"token-bucket must not let a fresh full capacity be consumed instantly across"
+ " the boundary; got "
+ secondAllowed);
}
private ValkeyRateLimitStore newRateLimitStore(LettuceConnectionFactory factory) {
ValkeyRateLimitStore store = new ValkeyRateLimitStore(factory);
store.initProxyManager();
return store;
}
@Test
@DisplayName("JobStore put is atomic (hash + TTL + reverse index visible together)")
void jobStorePutIsAtomic() {
ValkeyJobStore store = new ValkeyJobStore(templateA);
String jobId = "atomic-job-" + java.util.UUID.randomUUID();
String fileId = "atomic-file-" + java.util.UUID.randomUUID();
store.put(
new JobStoreEntry(
jobId,
JobStoreEntry.JobState.PENDING,
"node-A",
Instant.now(),
null,
null,
List.of(fileId),
Map.of("k", "v")),
Duration.ofSeconds(30));
// After put returns, every artifact has to be observable - if any are missing, the
// MULTI/EXEC was not really atomic.
assertTrue(store.exists(jobId), "hash must be visible after put");
Long jobTtl =
templateA.getExpire(
"stirling:job:" + jobId, java.util.concurrent.TimeUnit.MILLISECONDS);
assertNotNull(jobTtl);
assertTrue(jobTtl > 0, "hash must have TTL armed inside the same transaction");
assertEquals(jobId, store.findJobIdByFileId(fileId).orElse(null));
Long indexTtl =
templateA.getExpire(
"stirling:file2job:" + fileId, java.util.concurrent.TimeUnit.MILLISECONDS);
assertNotNull(indexTtl);
assertTrue(indexTtl > 0, "reverse index must also have TTL armed");
}
@Test
@DisplayName(
"JobStore.delete(): WATCH aborts when put() races between read and EXEC, no orphaned"
+ " reverse-index entries")
void jobStoreDeleteWatchRaceRetriesAndCleansUp() {
ValkeyJobStore store = new ValkeyJobStore(templateA);
String jobId = "watch-race-job-" + java.util.UUID.randomUUID();
String originalFile = "orig-file-" + java.util.UUID.randomUUID();
String newFile = "new-file-" + java.util.UUID.randomUUID();
store.put(
new JobStoreEntry(
jobId,
JobStoreEntry.JobState.RUNNING,
"node-A",
Instant.now(),
null,
null,
List.of(originalFile),
Map.of()),
Duration.ofSeconds(30));
// Simulate the race: between delete()'s read and EXEC, another node adds newFile to
// the same job. With WATCH/MULTI/EXEC the first EXEC aborts; the retry sees the
// updated fileIds and deletes both reverse-index entries.
Thread mutator =
new Thread(
() -> {
try {
Thread.sleep(20);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
store.put(
new JobStoreEntry(
jobId,
JobStoreEntry.JobState.RUNNING,
"node-A",
Instant.now(),
null,
null,
List.of(originalFile, newFile),
Map.of()),
Duration.ofSeconds(30));
});
mutator.start();
store.delete(jobId);
try {
mutator.join(2000);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
// Whichever order won, the final state must be self-consistent: either the hash is
// deleted and both reverse-index entries are gone, OR the second put() committed
// after delete and the hash + reverse-index entries for BOTH fileIds are intact.
boolean hashGone = !store.exists(jobId);
boolean origIndexGone = !store.findJobIdByFileId(originalFile).isPresent();
boolean newIndexGone = !store.findJobIdByFileId(newFile).isPresent();
if (hashGone) {
assertTrue(
origIndexGone,
"if hash is deleted, original reverse-index entry must also be gone");
assertTrue(
newIndexGone,
"if hash is deleted after the racing put(), the WATCH retry must catch the"
+ " new fileId and delete its reverse-index entry too");
} else {
// The racing put() committed after delete completed; both indices should point at
// jobId. This is a legitimate outcome - delete and re-put is not an atomic API.
assertEquals(jobId, store.findJobIdByFileId(originalFile).orElse(null));
assertEquals(jobId, store.findJobIdByFileId(newFile).orElse(null));
}
}
@Test
@DisplayName("JobStore.delete() removes hash AND every reverse-index entry atomically")
void jobStoreDeleteRemovesReverseIndexEntries() {
ValkeyJobStore store = new ValkeyJobStore(templateA);
String jobId = "del-atomic-job-" + java.util.UUID.randomUUID();
String fileA = "del-atomic-fileA-" + java.util.UUID.randomUUID();
String fileB = "del-atomic-fileB-" + java.util.UUID.randomUUID();
store.put(
new JobStoreEntry(
jobId,
JobStoreEntry.JobState.COMPLETE,
"node-A",
Instant.now(),
Instant.now(),
null,
List.of(fileA, fileB),
Map.of()),
Duration.ofSeconds(30));
// Sanity: every artifact is in place before delete.
assertTrue(store.exists(jobId));
assertEquals(jobId, store.findJobIdByFileId(fileA).orElse(null));
assertEquals(jobId, store.findJobIdByFileId(fileB).orElse(null));
store.delete(jobId);
// Both the main hash AND every reverse-index entry must be gone; dangling reverse-index
// entries would cause findJobIdByFileId() to return a deleted jobId.
assertFalse(store.exists(jobId), "main hash must be deleted");
assertFalse(
store.findJobIdByFileId(fileA).isPresent(),
"reverse-index entry for fileA must not survive delete()");
assertFalse(
store.findJobIdByFileId(fileB).isPresent(),
"reverse-index entry for fileB must not survive delete()");
assertFalse(
Boolean.TRUE.equals(templateA.hasKey("stirling:file2job:" + fileA)),
"raw reverse-index key for fileA must not survive delete()");
assertFalse(
Boolean.TRUE.equals(templateA.hasKey("stirling:file2job:" + fileB)),
"raw reverse-index key for fileB must not survive delete()");
}
@Test
@DisplayName("JobStore.all() walks the keyspace via SCAN, not KEYS")
void jobStoreAllUsesScanNonBlocking() {
ValkeyJobStore store = new ValkeyJobStore(templateA);
// Seed a handful of keys; the goal is "we get them all back" - the non-blocking property
// of SCAN is a property of the production server, what we verify here is functional parity.
for (int i = 0; i < 15; i++) {
store.put(
new JobStoreEntry(
"scan-job-" + i,
JobStoreEntry.JobState.PENDING,
"node-A",
Instant.now(),
null,
null,
List.of(),
Map.of()),
Duration.ofSeconds(30));
}
long observed = store.all().stream().filter(e -> e.jobId().startsWith("scan-job-")).count();
assertTrue(
observed >= 15,
"SCAN-based all() must surface every inserted job, saw " + observed);
}
@Test
@DisplayName("Valkey unreachable yields isHealthy() = false")
void unreachableBackplaneReportsUnhealthy() {
// Point at a closed port; afterPropertiesSet may succeed but ping will fail.
RedisStandaloneConfiguration cfg = new RedisStandaloneConfiguration("localhost", 16400);
LettuceConnectionFactory dead = new LettuceConnectionFactory(cfg);
dead.afterPropertiesSet();
try {
StringRedisTemplate t = new StringRedisTemplate(dead);
ValkeyClusterBackplane bp = new ValkeyClusterBackplane(newProps("orphan"), t);
assertFalse(bp.isHealthy(), "isHealthy must be false when Valkey is unreachable");
} finally {
dead.destroy();
}
}
private ApplicationProperties newProps(String nodeId) {
ApplicationProperties p = new ApplicationProperties();
p.getCluster().setEnabled(true);
p.getCluster().setBackplane("valkey");
p.getCluster()
.getValkey()
.setUrl("redis://" + VALKEY.getHost() + ":" + VALKEY.getMappedPort(6379));
p.getCluster().getNode().setId(nodeId);
return p;
}
}
@@ -0,0 +1,65 @@
package stirling.software.proprietary.cluster.valkey;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import stirling.software.common.model.ApplicationProperties;
/**
* S3 regression: {@link ValkeyClusterBackplane#isHealthy()} must route through {@code
* template.execute(...)} so the borrowed connection is always returned to the pool. Calling {@code
* getConnectionFactory().getConnection()} directly would leak the connection on every k8s liveness
* probe tick and exhaust the pool under monitoring load.
*/
class ValkeyClusterBackplaneTest {
@Test
void isHealthy_routesThroughTemplateExecute_andDoesNotTouchConnectionFactoryDirectly() {
StringRedisTemplate template = mock(StringRedisTemplate.class);
when(template.execute(any(RedisCallback.class))).thenReturn("PONG");
ApplicationProperties props = new ApplicationProperties();
props.getCluster().getNode().setId("n-1");
ValkeyClusterBackplane bp = new ValkeyClusterBackplane(props, template);
assertTrue(bp.isHealthy());
verify(template, times(1)).execute(any(RedisCallback.class));
// Critical: never bypass the template's connection management.
verify(template, never()).getConnectionFactory();
}
@Test
void isHealthy_returnsFalseWhenExecuteThrows() {
StringRedisTemplate template = mock(StringRedisTemplate.class);
when(template.execute(any(RedisCallback.class))).thenThrow(new RuntimeException("boom"));
ApplicationProperties props = new ApplicationProperties();
props.getCluster().getNode().setId("n-1");
ValkeyClusterBackplane bp = new ValkeyClusterBackplane(props, template);
assertFalse(bp.isHealthy());
}
@Test
void shouldRunLocalCleanup_returnsFalse_valkeyOwnsTtlEviction() {
// Valkey expires job entries via the TTL set in ValkeyJobStore.put(); running the local
// TaskManager.cleanupOldJobs loop on top of that is redundant and would create races
// with cluster-visible state. Default in ClusterBackplane is true; this override flips
// it for the Valkey impl.
StringRedisTemplate template = mock(StringRedisTemplate.class);
ApplicationProperties props = new ApplicationProperties();
props.getCluster().getNode().setId("n-1");
ValkeyClusterBackplane bp = new ValkeyClusterBackplane(props, template);
assertFalse(bp.shouldRunLocalCleanup());
}
}
@@ -0,0 +1,194 @@
package stirling.software.proprietary.cluster.valkey;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.atMost;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import io.lettuce.core.RedisCommandExecutionException;
import io.lettuce.core.SslVerifyMode;
/**
* Unit tests for the auth-fast-fail behaviour of {@link
* ValkeyConnectionConfiguration#eagerHandshake(LettuceConnectionFactory, String, int, boolean)} and
* the auth-detection helper {@link ValkeyConnectionConfiguration#isAuthFailure(Throwable)}.
*
* <p>An auth-class failure (WRONGPASS / NOAUTH / NOPERM) is unrecoverable; retrying for 30 s only
* delays the inevitable boot failure and floods logs. The handshake must surface auth errors after
* exactly one attempt.
*/
class ValkeyConnectionConfigurationTest {
@Test
@DisplayName("WRONGPASS surfaces in one attempt (no 30s retry loop)")
void wrongpass_failsImmediately_withoutRetries() throws Exception {
LettuceConnectionFactory factory = mock(LettuceConnectionFactory.class);
RedisConnection conn = mock(RedisConnection.class);
when(factory.getConnection()).thenReturn(conn);
// Spring Data Redis wraps RedisCommandExecutionException in RedisSystemException; we
// simulate the exact wrapper Lettuce → spring-data-redis produces in production.
RedisCommandExecutionException auth =
new RedisCommandExecutionException("WRONGPASS invalid username-password pair");
when(conn.ping()).thenThrow(new RedisSystemException("Error in execution", auth));
long start = System.nanoTime();
IllegalStateException ex =
assertThrows(
IllegalStateException.class,
() ->
ValkeyConnectionConfiguration.eagerHandshake(
factory, "valkey", 6379, false));
long elapsedMs = (System.nanoTime() - start) / 1_000_000;
// Exactly one ping call. A retry loop would call it 10 times with 3 s sleeps.
verify(factory, times(1)).getConnection();
verify(conn, times(1)).ping();
// Generous 1500 ms bound; the single attempt with a mocked connection is sub-ms in
// practice. The contract is "no 3 s+ sleeps".
assertTrue(
elapsedMs < 1500,
"Auth failure must short-circuit retries; elapsed=" + elapsedMs + " ms");
assertTrue(
ex.getMessage().contains("authentication failed"),
"Error message must explain the auth failure; got: " + ex.getMessage());
verify(factory, atMost(1)).destroy();
}
@Test
@DisplayName("NOAUTH surfaces in one attempt")
void noauth_failsImmediately() {
LettuceConnectionFactory factory = mock(LettuceConnectionFactory.class);
RedisConnection conn = mock(RedisConnection.class);
when(factory.getConnection()).thenReturn(conn);
when(conn.ping())
.thenThrow(
new RedisSystemException(
"Error in execution",
new RedisCommandExecutionException(
"NOAUTH Authentication required.")));
assertThrows(
IllegalStateException.class,
() -> ValkeyConnectionConfiguration.eagerHandshake(factory, "v", 6379, false));
verify(conn, times(1)).ping();
}
@Test
@DisplayName("NOPERM surfaces in one attempt")
void noperm_failsImmediately() {
LettuceConnectionFactory factory = mock(LettuceConnectionFactory.class);
RedisConnection conn = mock(RedisConnection.class);
when(factory.getConnection()).thenReturn(conn);
when(conn.ping())
.thenThrow(
new RedisSystemException(
"Error in execution",
new RedisCommandExecutionException(
"NOPERM this user has no permissions to run the 'ping'"
+ " command")));
assertThrows(
IllegalStateException.class,
() -> ValkeyConnectionConfiguration.eagerHandshake(factory, "v", 6379, false));
verify(conn, times(1)).ping();
}
@Test
@DisplayName("isAuthFailure - direct RedisCommandExecutionException with auth prefix")
void isAuthFailure_directRedisCommandExecutionException() {
assertTrue(
ValkeyConnectionConfiguration.isAuthFailure(
new RedisCommandExecutionException("WRONGPASS bad password")));
assertTrue(
ValkeyConnectionConfiguration.isAuthFailure(
new RedisCommandExecutionException("NOAUTH required")));
assertTrue(
ValkeyConnectionConfiguration.isAuthFailure(
new RedisCommandExecutionException("NOPERM denied")));
}
@Test
@DisplayName("isAuthFailure - wrapped inside RedisSystemException (production path)")
void isAuthFailure_wrappedBySpring() {
assertTrue(
ValkeyConnectionConfiguration.isAuthFailure(
new RedisSystemException(
"Error in execution",
new RedisCommandExecutionException("WRONGPASS bad password"))));
}
@Test
@DisplayName("isAuthFailure - connection errors do NOT count as auth failures")
void isAuthFailure_connectionErrorReturnsFalse() {
// A transport-level failure must continue to retry.
assertFalse(
ValkeyConnectionConfiguration.isAuthFailure(
new RedisSystemException(
"Redis connection failed",
new io.lettuce.core.RedisConnectionException(
"Connection refused"))));
assertFalse(
ValkeyConnectionConfiguration.isAuthFailure(
new IllegalStateException("Valkey PING returned 'foo' (expected PONG)")));
}
@Test
@DisplayName("bad PONG (protocol error) is not an auth failure")
void unexpectedPong_isNotAuthFailure() {
// Sanity: a returned non-PONG string maps to IllegalStateException inside the try block
// and must not be treated as auth, otherwise misclassified protocol errors would skip
// the retry loop too.
assertFalse(
ValkeyConnectionConfiguration.isAuthFailure(
new IllegalStateException("Valkey PING returned 'bar' (expected PONG)")));
}
// --------------------------------------------------------------------------------------
// D5: TLS hostname/chain verification (default ON, opt-out for dev only)
// --------------------------------------------------------------------------------------
@Test
@DisplayName("TLS on, skipCertVerification=false → useSsl + verifyPeer=FULL (default)")
void tls_defaultEnforcesFullPeerVerification() {
LettuceClientConfiguration cfg =
ValkeyConnectionConfiguration.buildClientConfiguration(true, false);
assertTrue(cfg.isUseSsl(), "TLS must be enabled");
// FULL = chain + hostname. CA-only or NONE would be a silent downgrade and is why we
// pin this explicitly rather than relying on the upstream Spring default.
assertSame(SslVerifyMode.FULL, cfg.getVerifyMode());
assertTrue(cfg.isVerifyPeer());
}
@Test
@DisplayName("TLS on, skipCertVerification=true → verifyPeer=NONE (dev override)")
void tls_skipCertVerificationOptOut() {
LettuceClientConfiguration cfg =
ValkeyConnectionConfiguration.buildClientConfiguration(true, true);
assertTrue(cfg.isUseSsl());
// The opt-out path is intentionally available for self-signed local dev certs, but
// requires explicit operator action via cluster.valkey.tls.skip-cert-verification.
assertSame(SslVerifyMode.NONE, cfg.getVerifyMode());
}
@Test
@DisplayName("TLS off → no SSL, verify flag default (skipCertVerification ignored)")
void noTls_ignoresSkipFlag() {
// Without rediss:// we never call useSsl(), so the skip flag is a no-op. Confirming
// here so we cannot accidentally trip TLS off on plain redis:// connections.
LettuceClientConfiguration cfg =
ValkeyConnectionConfiguration.buildClientConfiguration(false, true);
assertFalse(cfg.isUseSsl());
}
}