From 4564ed5bec18e8cd665667321906a381e2c3d5f8 Mon Sep 17 00:00:00 2001 From: Anthony Stirling <77850077+Frooodle@users.noreply.github.com> Date: Wed, 27 May 2026 11:54:59 +0100 Subject: [PATCH] Add cluster backplane abstraction and interfaces (#6449) --- app/common/build.gradle | 3 + .../common/cluster/ClusterBackplane.java | 24 ++ .../common/cluster/ClusterConfig.java | 63 ++++ .../software/common/cluster/ClusterNode.java | 12 + .../common/cluster/DistributedLock.java | 21 ++ .../software/common/cluster/FileStore.java | 45 +++ .../common/cluster/InstanceRegistry.java | 18 + .../software/common/cluster/JobStore.java | 24 ++ .../common/cluster/JobStoreEntry.java | 30 ++ .../common/cluster/KeyValueCache.java | 16 + .../common/cluster/RateLimitStore.java | 18 + .../common/cluster/StickyMissRecorder.java | 7 + .../inprocess/InProcessClusterBackplane.java | 31 ++ .../InProcessClusterConfiguration.java | 65 ++++ .../inprocess/InProcessDistributedLock.java | 128 +++++++ .../inprocess/InProcessInstanceRegistry.java | 42 +++ .../cluster/inprocess/InProcessJobStore.java | 96 +++++ .../inprocess/InProcessKeyValueCache.java | 55 +++ .../inprocess/InProcessRateLimitStore.java | 49 +++ .../cluster/inprocess/LocalDiskFileStore.java | 128 +++++++ .../LocalDiskFileStoreConfiguration.java | 29 ++ .../common/model/ApplicationProperties.java | 102 ++++++ .../software/common/service/FileStorage.java | 331 ++++++------------ .../software/common/service/TaskManager.java | 108 +++++- .../common/architecture/ArchitectureTest.java | 13 + .../BackplaneContractCompilationTest.java | 51 +++ .../cluster/ClusterConfigValidationTest.java | 65 ++++ .../common/cluster/ClusterPropertiesTest.java | 62 ++++ ...InProcessConfigurationConditionalTest.java | 90 +++++ .../InProcessDistributedLockTest.java | 167 +++++++++ .../InProcessInstanceRegistryTest.java | 28 ++ .../inprocess/InProcessJobStoreTest.java | 91 +++++ .../inprocess/InProcessKeyValueCacheTest.java | 41 +++ .../InProcessRateLimitStoreTest.java | 56 +++ .../inprocess/LocalDiskFileStoreTest.java | 42 +++ .../service/FileStorageDelegationTest.java | 27 ++ .../common/service/FileStorageTest.java | 30 +- .../TaskManagerJobStoreDelegationTest.java | 120 +++++++ .../common/service/TaskManagerTest.java | 70 ++++ .../src/main/resources/settings.yml.template | 16 + 40 files changed, 2167 insertions(+), 247 deletions(-) create mode 100644 app/common/src/main/java/stirling/software/common/cluster/ClusterBackplane.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/ClusterNode.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/DistributedLock.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/FileStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/InstanceRegistry.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/JobStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/JobStoreEntry.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/KeyValueCache.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/RateLimitStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/StickyMissRecorder.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterBackplane.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterConfiguration.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessDistributedLock.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistry.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessJobStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCache.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStore.java create mode 100644 app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreConfiguration.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/BackplaneContractCompilationTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/ClusterPropertiesTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/InProcessConfigurationConditionalTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessDistributedLockTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistryTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessJobStoreTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCacheTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStoreTest.java create mode 100644 app/common/src/test/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreTest.java create mode 100644 app/common/src/test/java/stirling/software/common/service/FileStorageDelegationTest.java create mode 100644 app/common/src/test/java/stirling/software/common/service/TaskManagerJobStoreDelegationTest.java diff --git a/app/common/build.gradle b/app/common/build.gradle index 508a97825..245a16234 100644 --- a/app/common/build.gradle +++ b/app/common/build.gradle @@ -78,6 +78,9 @@ dependencies { runtimeOnly "com.stirling:jpdfium-natives-${platform}:1.0.1" } + // Bucket4j (local in-process token bucket for RateLimitStore default impl) + implementation 'com.bucket4j:bucket4j_jdk17-core:8.19.0' + // ArchUnit: enforces module dependency direction (see ArchitectureTest) testImplementation 'com.tngtech.archunit:archunit-junit5:1.4.2' } diff --git a/app/common/src/main/java/stirling/software/common/cluster/ClusterBackplane.java b/app/common/src/main/java/stirling/software/common/cluster/ClusterBackplane.java new file mode 100644 index 000000000..098608270 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/ClusterBackplane.java @@ -0,0 +1,24 @@ +package stirling.software.common.cluster; + +/** Health and identity facade for the active cluster backplane. */ +public interface ClusterBackplane { + + /** Returns {@code true} when the backplane is reachable; used for health endpoints. */ + boolean isHealthy(); + + /** Returns {@code "inprocess"} or {@code "valkey"}. */ + String backplaneType(); + + /** Returns this JVM's stable node id (matches {@code Cluster.resolvedNodeId()}). */ + String localNodeId(); + + /** + * Whether this JVM should run the local {@link + * stirling.software.common.service.TaskManager#cleanupOldJobs()} loop. Distributed backplanes + * own job expiry via their own TTL, so they should override this to return {@code false}. + * Defaults to {@code true} so in-process behavior is preserved without an explicit override. + */ + default boolean shouldRunLocalCleanup() { + return true; + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java b/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java new file mode 100644 index 000000000..127fbfbfb --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java @@ -0,0 +1,63 @@ +package stirling.software.common.cluster; + +import org.springframework.context.annotation.Configuration; + +import jakarta.annotation.PostConstruct; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.ApplicationProperties.Cluster; + +/** + * Validates that cluster mode is internally consistent. + * + *

Cluster settings are bound on the central {@link ApplicationProperties} under {@code + * cluster.*}; this class reads {@link ApplicationProperties#getCluster()} and runs guards in {@link + * PostConstruct}. When {@code cluster.enabled=false} (the default) all checks are skipped so a + * single-instance install needs no new config. + */ +@Slf4j +@Configuration +@RequiredArgsConstructor +public class ClusterConfig { + + private final ApplicationProperties applicationProperties; + + @PostConstruct + void validate() { + Cluster cluster = applicationProperties.getCluster(); + if (!cluster.isEnabled()) { + return; + } + String backplane = cluster.getBackplane(); + if ("valkey".equalsIgnoreCase(backplane)) { + String url = cluster.getValkey() == null ? null : cluster.getValkey().getUrl(); + if (url == null || url.isBlank()) { + throw new IllegalStateException( + "cluster.enabled=true with backplane=valkey requires" + + " cluster.valkey.url to be set (e.g." + + " redis://valkey:6379)."); + } + } else if ("inprocess".equalsIgnoreCase(backplane)) { + // enabled+inprocess only coordinates the local JVM; cross-node lookups will 410. + log.warn( + "cluster.enabled=true with backplane=inprocess - only the local" + + " JVM is coordinated. Cross-node lookups and the file proxy will fail." + + " Use backplane=valkey for real multi-node deployments."); + } else { + // Fail fast on typos like "valky" so Spring doesn't later report a cryptic + // "no ClusterBackplane bean" - the operator-facing error names the bad value. + throw new IllegalStateException( + "cluster.enabled=true with unknown backplane '" + + backplane + + "'. Valid values: inprocess | valkey."); + } + log.info( + "Cluster mode enabled (backplane={}, nodeRole={}, nodeId={}).", + backplane, + cluster.resolvedRole(), + cluster.resolvedNodeId()); + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/ClusterNode.java b/app/common/src/main/java/stirling/software/common/cluster/ClusterNode.java new file mode 100644 index 000000000..c008d0e99 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/ClusterNode.java @@ -0,0 +1,12 @@ +package stirling.software.common.cluster; + +import java.time.Instant; + +/** + * Snapshot of a peer node as recorded in the {@link InstanceRegistry}. + * + * @param internalAddress {@code host:port} the node listens on for {@code /internal/cluster/**} + * @param role one of {@code WEB}, {@code WORKER}, {@code BOTH} + */ +public record ClusterNode( + String nodeId, String internalAddress, Instant lastHeartbeat, String role) {} diff --git a/app/common/src/main/java/stirling/software/common/cluster/DistributedLock.java b/app/common/src/main/java/stirling/software/common/cluster/DistributedLock.java new file mode 100644 index 000000000..4c77e0f93 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/DistributedLock.java @@ -0,0 +1,21 @@ +package stirling.software.common.cluster; + +import java.time.Duration; +import java.util.Optional; + +/** Cluster-wide mutual exclusion primitive; non-reentrant by contract. */ +public interface DistributedLock { + + Optional tryAcquire(String lockKey, Duration leaseTime); + + interface LockHandle extends AutoCloseable { + void release(); + + boolean renew(Duration leaseTime); + + @Override + default void close() { + release(); + } + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/FileStore.java b/app/common/src/main/java/stirling/software/common/cluster/FileStore.java new file mode 100644 index 000000000..f576e1e97 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/FileStore.java @@ -0,0 +1,45 @@ +package stirling.software.common.cluster; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; + +/** Low-level storage seam for result/job files. */ +public interface FileStore { + + /** Stored file record. */ + record Stored(String fileId, long size) {} + + /** Store the given stream and return a generated file id and total bytes written. */ + Stored store(InputStream in, String originalName) throws IOException; + + /** + * Store the file at {@code source} and return a generated file id and total bytes written. + * + *

Default implementation opens {@code source} as a stream and delegates to {@link + * #store(InputStream, String)}. Local-disk implementations should override to use a direct + * file-to-file copy ({@code Files.copy(source, dest)} can use {@code sendfile(2)} on Linux), + * which avoids the two-memory-copy hit of streaming a disk-backed upload through the JVM heap. + */ + default Stored store(Path source, String originalName) throws IOException { + try (InputStream in = Files.newInputStream(source)) { + return store(in, originalName); + } + } + + /** Open the stored file for streaming reads. Caller closes. */ + InputStream retrieve(String fileId) throws IOException; + + /** Load the stored file into a byte array. */ + byte[] retrieveBytes(String fileId) throws IOException; + + /** Size of the stored file in bytes. */ + long size(String fileId) throws IOException; + + /** Delete the stored file. Returns true if a file was removed. */ + boolean delete(String fileId); + + /** Whether the file id exists in the store. */ + boolean exists(String fileId); +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/InstanceRegistry.java b/app/common/src/main/java/stirling/software/common/cluster/InstanceRegistry.java new file mode 100644 index 000000000..289a35269 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/InstanceRegistry.java @@ -0,0 +1,18 @@ +package stirling.software.common.cluster; + +import java.time.Duration; +import java.util.Collection; +import java.util.Optional; + +/** Maps {@code nodeId} to its internal cluster address, with TTL'd heartbeats. */ +public interface InstanceRegistry { + + /** Register or refresh this node. Idempotent so a wiped backplane self-heals on next tick. */ + void register(ClusterNode node, Duration heartbeatTtl); + + Optional lookup(String nodeId); + + Collection activeNodes(); + + void deregister(String nodeId); +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/JobStore.java b/app/common/src/main/java/stirling/software/common/cluster/JobStore.java new file mode 100644 index 000000000..03083c06e --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/JobStore.java @@ -0,0 +1,24 @@ +package stirling.software.common.cluster; + +import java.time.Duration; +import java.util.Collection; +import java.util.Optional; + +/** Cluster-visible storage for job status and result metadata, with TTL'd entries. */ +public interface JobStore { + + /** Persist or overwrite a job entry. {@code ttl} sets the lifetime of the entry. */ + void put(JobStoreEntry entry, Duration ttl); + + Optional get(String jobId); + + void delete(String jobId); + + boolean exists(String jobId); + + /** Reverse lookup: which job owns this result file id? */ + Optional findJobIdByFileId(String fileId); + + /** Snapshot of every active entry. Used by admin/stats endpoints; may be O(n). */ + Collection all(); +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/JobStoreEntry.java b/app/common/src/main/java/stirling/software/common/cluster/JobStoreEntry.java new file mode 100644 index 000000000..e730938b6 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/JobStoreEntry.java @@ -0,0 +1,30 @@ +package stirling.software.common.cluster; + +import java.time.Instant; +import java.util.List; +import java.util.Map; + +/** + * Cluster-visible projection of a job's status and result metadata, as persisted in {@link + * JobStore}. + * + * @param owningNodeId the node id that originally executed the job + */ +public record JobStoreEntry( + String jobId, + JobState state, + String owningNodeId, + Instant createdAt, + Instant completedAt, + String error, + List fileIds, + Map resultMeta) { + + /** Lifecycle states for a job as observed by the cluster. */ + public enum JobState { + PENDING, + RUNNING, + COMPLETE, + FAILED + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/KeyValueCache.java b/app/common/src/main/java/stirling/software/common/cluster/KeyValueCache.java new file mode 100644 index 000000000..fae44ad82 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/KeyValueCache.java @@ -0,0 +1,16 @@ +package stirling.software.common.cluster; + +import java.time.Duration; +import java.util.Optional; + +/** Short-TTL namespaced key/value cache backed by the cluster backplane. */ +public interface KeyValueCache { + + void put(String namespace, String key, String value, Duration ttl); + + Optional get(String namespace, String key); + + void evict(String namespace, String key); + + void evictNamespace(String namespace); +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/RateLimitStore.java b/app/common/src/main/java/stirling/software/common/cluster/RateLimitStore.java new file mode 100644 index 000000000..82351961a --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/RateLimitStore.java @@ -0,0 +1,18 @@ +package stirling.software.common.cluster; + +import java.time.Duration; + +/** Token-bucket rate limiting backed by the cluster backplane. */ +public interface RateLimitStore { + + /** + * Attempt to consume one token from the bucket identified by {@code bucketKey}. + * + * @param bucketKey opaque key identifying the bucket (e.g. {@code api:user:123}) + * @param capacity bucket capacity + * @param refillPeriod time window over which {@code capacity} tokens refill + */ + RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod); + + record RateLimitDecision(boolean allowed, long remainingTokens, long nanosToWaitForRefill) {} +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/StickyMissRecorder.java b/app/common/src/main/java/stirling/software/common/cluster/StickyMissRecorder.java new file mode 100644 index 000000000..cba9b3ceb --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/StickyMissRecorder.java @@ -0,0 +1,7 @@ +package stirling.software.common.cluster; + +/** Records one increment per sticky-session miss (a 410 Gone for a job owned by another node). */ +@FunctionalInterface +public interface StickyMissRecorder { + void recordStickyMiss(); +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterBackplane.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterBackplane.java new file mode 100644 index 000000000..8cc147b89 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterBackplane.java @@ -0,0 +1,31 @@ +package stirling.software.common.cluster.inprocess; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.cluster.ClusterBackplane; +import stirling.software.common.model.ApplicationProperties; + +@Slf4j +public class InProcessClusterBackplane implements ClusterBackplane { + + private final ApplicationProperties applicationProperties; + + public InProcessClusterBackplane(ApplicationProperties applicationProperties) { + this.applicationProperties = applicationProperties; + } + + @Override + public boolean isHealthy() { + return true; + } + + @Override + public String backplaneType() { + return "inprocess"; + } + + @Override + public String localNodeId() { + return applicationProperties.getCluster().resolvedNodeId(); + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterConfiguration.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterConfiguration.java new file mode 100644 index 000000000..005ec319c --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterConfiguration.java @@ -0,0 +1,65 @@ +package stirling.software.common.cluster.inprocess; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.cluster.ClusterBackplane; +import stirling.software.common.cluster.DistributedLock; +import stirling.software.common.cluster.InstanceRegistry; +import stirling.software.common.cluster.JobStore; +import stirling.software.common.cluster.KeyValueCache; +import stirling.software.common.cluster.RateLimitStore; +import stirling.software.common.model.ApplicationProperties; + +/** + * Default cluster backplane wiring: every interface gets an {@code InProcess*} bean. Active when + * cluster mode is off or {@code cluster.backplane=inprocess}. + */ +@Slf4j +@Configuration +@ConditionalOnExpression( + "!${cluster.enabled:false} ||" + + " '${cluster.backplane:inprocess}'.equalsIgnoreCase('inprocess')") +public class InProcessClusterConfiguration { + + @Bean + @ConditionalOnMissingBean + public ClusterBackplane clusterBackplane(ApplicationProperties applicationProperties) { + log.info("Cluster backplane: in-process (single node)"); + return new InProcessClusterBackplane(applicationProperties); + } + + @Bean + @ConditionalOnMissingBean + public JobStore jobStore() { + return new InProcessJobStore(); + } + + @Bean + @ConditionalOnMissingBean + public RateLimitStore rateLimitStore() { + return new InProcessRateLimitStore(); + } + + @Bean + @ConditionalOnMissingBean + public DistributedLock distributedLock() { + return new InProcessDistributedLock(); + } + + @Bean + @ConditionalOnMissingBean + public KeyValueCache keyValueCache() { + return new InProcessKeyValueCache(); + } + + @Bean + @ConditionalOnMissingBean + public InstanceRegistry instanceRegistry() { + return new InProcessInstanceRegistry(); + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessDistributedLock.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessDistributedLock.java new file mode 100644 index 000000000..d937b3f39 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessDistributedLock.java @@ -0,0 +1,128 @@ +package stirling.software.common.cluster.inprocess; + +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; + +import stirling.software.common.cluster.DistributedLock; + +/** + * In-process {@link DistributedLock}, non-reentrant per the interface contract, with lease-expiry + * semantics that mirror a SET-NX-EX style distributed backend. + * + *

Each lock state carries a per-acquire {@code ownerToken} and an {@code expiryNanos}; another + * caller can take over once the lease has elapsed even if the original holder never called {@link + * LockHandle#release()}. This matters mostly for parity with the Valkey-backed implementation + * (Redis {@code SETEX} auto-expires the key); within a single JVM a crashed holder takes its lock + * state with it, but tests and code that rely on the {@code leaseTime} parameter still need it to + * be honored. + * + *

Expiry is lazy: an expired lock state lingers in the map until the next acquire attempt for + * the same key replaces it. Per-key cleanup also happens on explicit {@link LockHandle#release()}, + * so a balanced acquire/release workload keeps the map size bounded. + */ +public class InProcessDistributedLock implements DistributedLock { + + private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + private final AtomicLong tokenSeq = new AtomicLong(); + + /** + * Lease state for a single acquired lock. {@code ownerToken} prevents a former holder from + * releasing or renewing a lock now owned by someone else after lease expiry; {@code + * expiryNanos} is read/written only inside {@link ConcurrentHashMap#compute} so the bin lock + * provides the necessary happens-before guarantee. + */ + private static final class LockState { + final long ownerToken; + long expiryNanos; + + LockState(long ownerToken, long expiryNanos) { + this.ownerToken = ownerToken; + this.expiryNanos = expiryNanos; + } + } + + @Override + public Optional tryAcquire(String lockKey, Duration leaseTime) { + long token = tokenSeq.incrementAndGet(); + long nowNanos = System.nanoTime(); + long expiryNanos = nowNanos + leaseTime.toNanos(); + boolean[] acquired = {false}; + locks.compute( + lockKey, + (k, existing) -> { + if (existing == null || existing.expiryNanos - nowNanos <= 0L) { + // No lock, or the previous lease has expired - we take it. Subtraction + // form avoids the long-overflow trap that would bite a naive + // expiryNanos <= nowNanos comparison around System.nanoTime() rollover. + acquired[0] = true; + return new LockState(token, expiryNanos); + } + return existing; + }); + if (!acquired[0]) { + return Optional.empty(); + } + return Optional.of(new InProcessHandle(lockKey, token)); + } + + private void releaseInternal(String lockKey, long token) { + locks.compute( + lockKey, + (k, existing) -> { + if (existing == null || existing.ownerToken != token) { + // Already removed, expired-and-replaced, or never ours. + return existing; + } + return null; + }); + } + + private boolean renewInternal(String lockKey, long token, Duration leaseTime) { + long nowNanos = System.nanoTime(); + boolean[] renewed = {false}; + locks.compute( + lockKey, + (k, existing) -> { + if (existing == null + || existing.ownerToken != token + || existing.expiryNanos - nowNanos <= 0L) { + // Lock is gone or expired; renewal is a no-op so the caller can detect it. + return existing; + } + existing.expiryNanos = nowNanos + leaseTime.toNanos(); + renewed[0] = true; + return existing; + }); + return renewed[0]; + } + + private final class InProcessHandle implements LockHandle { + private final String lockKey; + private final long token; + private boolean released; + + InProcessHandle(String lockKey, long token) { + this.lockKey = lockKey; + this.token = token; + } + + @Override + public synchronized void release() { + if (released) { + return; + } + released = true; + releaseInternal(lockKey, token); + } + + @Override + public synchronized boolean renew(Duration leaseTime) { + if (released) { + return false; + } + return renewInternal(lockKey, token, leaseTime); + } + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistry.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistry.java new file mode 100644 index 000000000..d46801f8b --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistry.java @@ -0,0 +1,42 @@ +package stirling.software.common.cluster.inprocess; + +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; + +import stirling.software.common.cluster.ClusterNode; +import stirling.software.common.cluster.InstanceRegistry; + +public class InProcessInstanceRegistry implements InstanceRegistry { + + private final AtomicReference self = new AtomicReference<>(); + + @Override + public void register(ClusterNode node, Duration heartbeatTtl) { + self.set(node); + } + + @Override + public Optional lookup(String nodeId) { + ClusterNode current = self.get(); + return current != null && current.nodeId().equals(nodeId) + ? Optional.of(current) + : Optional.empty(); + } + + @Override + public Collection activeNodes() { + ClusterNode current = self.get(); + return current == null ? Collections.emptyList() : Collections.singletonList(current); + } + + @Override + public void deregister(String nodeId) { + ClusterNode current = self.get(); + if (current != null && current.nodeId().equals(nodeId)) { + self.set(null); + } + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessJobStore.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessJobStore.java new file mode 100644 index 000000000..500716e8e --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessJobStore.java @@ -0,0 +1,96 @@ +package stirling.software.common.cluster.inprocess; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.cluster.JobStore; +import stirling.software.common.cluster.JobStoreEntry; + +@Slf4j +public class InProcessJobStore implements JobStore { + + private final ConcurrentHashMap entries = new ConcurrentHashMap<>(); + + @Override + public void put(JobStoreEntry entry, Duration ttl) { + Instant expiry = ttl == null ? Instant.MAX : Instant.now().plus(ttl); + entries.put(entry.jobId(), new Holder(entry, expiry)); + } + + @Override + public Optional get(String jobId) { + Holder h = entries.get(jobId); + if (h == null) { + return Optional.empty(); + } + if (h.isExpired()) { + entries.remove(jobId, h); + return Optional.empty(); + } + return Optional.of(h.entry); + } + + @Override + public void delete(String jobId) { + entries.remove(jobId); + } + + @Override + public boolean exists(String jobId) { + return get(jobId).isPresent(); + } + + @Override + public Optional findJobIdByFileId(String fileId) { + for (Map.Entry e : entries.entrySet()) { + Holder h = e.getValue(); + if (h.isExpired()) { + continue; + } + List fileIds = h.entry.fileIds(); + if (fileIds != null && fileIds.contains(fileId)) { + return Optional.of(e.getKey()); + } + } + return Optional.empty(); + } + + @Override + public Collection all() { + List result = new ArrayList<>(entries.size()); + for (Holder h : entries.values()) { + if (!h.isExpired()) { + result.add(h.entry); + } + } + return result; + } + + /** Drop entries whose TTL has elapsed. Called by the {@code TaskManager} cleanup scheduler. */ + public int purgeExpired() { + int removed = 0; + Instant now = Instant.now(); + for (Map.Entry e : entries.entrySet()) { + if (!e.getValue().expiry.equals(Instant.MAX) && e.getValue().expiry.isBefore(now)) { + if (entries.remove(e.getKey(), e.getValue())) { + removed++; + } + } + } + return removed; + } + + private record Holder(JobStoreEntry entry, Instant expiry) { + boolean isExpired() { + return !expiry.equals(Instant.MAX) && expiry.isBefore(Instant.now()); + } + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCache.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCache.java new file mode 100644 index 000000000..845d27288 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCache.java @@ -0,0 +1,55 @@ +package stirling.software.common.cluster.inprocess; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import stirling.software.common.cluster.KeyValueCache; + +public class InProcessKeyValueCache implements KeyValueCache { + + private final ConcurrentHashMap> namespaces = + new ConcurrentHashMap<>(); + + @Override + public void put(String namespace, String key, String value, Duration ttl) { + Instant expiry = ttl == null ? Instant.MAX : Instant.now().plus(ttl); + namespaces + .computeIfAbsent(namespace, n -> new ConcurrentHashMap<>()) + .put(key, new Expiring(value, expiry)); + } + + @Override + public Optional get(String namespace, String key) { + Map ns = namespaces.get(namespace); + if (ns == null) { + return Optional.empty(); + } + Expiring e = ns.get(key); + if (e == null) { + return Optional.empty(); + } + if (e.expiry.isBefore(Instant.now())) { + ns.remove(key, e); + return Optional.empty(); + } + return Optional.of(e.value); + } + + @Override + public void evict(String namespace, String key) { + Map ns = namespaces.get(namespace); + if (ns != null) { + ns.remove(key); + } + } + + @Override + public void evictNamespace(String namespace) { + namespaces.remove(namespace); + } + + private record Expiring(String value, Instant expiry) {} +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStore.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStore.java new file mode 100644 index 000000000..b2cd4cd74 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStore.java @@ -0,0 +1,49 @@ +package stirling.software.common.cluster.inprocess; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import io.github.bucket4j.Bandwidth; +import io.github.bucket4j.Bucket; +import io.github.bucket4j.ConsumptionProbe; +import io.github.bucket4j.local.LocalBucketBuilder; + +import stirling.software.common.cluster.RateLimitStore; + +/** Bucket4j-backed token bucket implementation of {@link RateLimitStore}. */ +public class InProcessRateLimitStore implements RateLimitStore { + + /** Cap to bound memory; oldest accessed buckets are evicted. */ + private static final int MAX_BUCKETS = 10_000; + + private final Map buckets = + Collections.synchronizedMap( + new LinkedHashMap(256, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() > MAX_BUCKETS; + } + }); + + @Override + public RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod) { + String compositeKey = bucketKey + "|" + capacity + "|" + refillPeriod.toNanos(); + Bucket bucket = + buckets.computeIfAbsent(compositeKey, k -> buildBucket(capacity, refillPeriod)); + ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1); + return new RateLimitDecision( + probe.isConsumed(), + probe.getRemainingTokens(), + probe.isConsumed() ? 0L : probe.getNanosToWaitForRefill()); + } + + private static Bucket buildBucket(long capacity, Duration refillPeriod) { + Bandwidth limit = + Bandwidth.builder().capacity(capacity).refillGreedy(capacity, refillPeriod).build(); + LocalBucketBuilder builder = Bucket.builder(); + builder.addLimit(limit); + return builder.build(); + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStore.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStore.java new file mode 100644 index 000000000..6d8d0d91f --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStore.java @@ -0,0 +1,128 @@ +package stirling.software.common.cluster.inprocess; + +import java.io.BufferedInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.UUID; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.cluster.FileStore; + +/** Local-disk {@link FileStore} storing files under a base directory keyed by a UUID file id. */ +@Slf4j +public class LocalDiskFileStore implements FileStore { + + private final String baseDirPath; + + public LocalDiskFileStore(String baseDirPath) { + this.baseDirPath = baseDirPath; + } + + @Override + public Stored store(InputStream in, String originalName) throws IOException { + String fileId = UUID.randomUUID().toString(); + Path filePath = resolve(fileId); + Files.createDirectories(filePath.getParent()); + boolean success = false; + try { + long size = Files.copy(in, filePath); + success = true; + return new Stored(fileId, size); + } finally { + if (!success) { + try { + Files.deleteIfExists(filePath); + } catch (IOException cleanupEx) { + log.warn( + "Failed to clean up partial file {} after store failure", + filePath, + cleanupEx); + } + } + } + } + + /** + * File-to-file copy. {@link Files#copy(Path, Path, java.nio.file.CopyOption...)} can use {@code + * sendfile(2)} on Linux for a zero-copy kernel transfer when source and destination share a + * filesystem, avoiding the streaming overhead of pulling the bytes through the JVM heap. Reads + * the source size before copying so the post-copy stat is unnecessary. + */ + @Override + public Stored store(Path source, String originalName) throws IOException { + String fileId = UUID.randomUUID().toString(); + Path filePath = resolve(fileId); + Files.createDirectories(filePath.getParent()); + long size = Files.size(source); + boolean success = false; + try { + Files.copy(source, filePath); + success = true; + return new Stored(fileId, size); + } finally { + if (!success) { + try { + Files.deleteIfExists(filePath); + } catch (IOException cleanupEx) { + log.warn( + "Failed to clean up partial file {} after store failure", + filePath, + cleanupEx); + } + } + } + } + + @Override + public InputStream retrieve(String fileId) throws IOException { + return new BufferedInputStream(Files.newInputStream(resolve(fileId))); + } + + @Override + public byte[] retrieveBytes(String fileId) throws IOException { + Path filePath = resolve(fileId); + if (!Files.exists(filePath)) { + throw new IOException("File not found with ID: " + fileId); + } + return Files.readAllBytes(filePath); + } + + @Override + public long size(String fileId) throws IOException { + Path filePath = resolve(fileId); + if (!Files.exists(filePath)) { + throw new IOException("File not found with ID: " + fileId); + } + return Files.size(filePath); + } + + @Override + public boolean delete(String fileId) { + try { + return Files.deleteIfExists(resolve(fileId)); + } catch (IOException e) { + log.error("Error deleting file with ID: {}", fileId, e); + return false; + } + } + + @Override + public boolean exists(String fileId) { + return Files.exists(resolve(fileId)); + } + + public Path resolve(String fileId) { + if (fileId.contains("..") || fileId.contains("/") || fileId.contains("\\")) { + throw new IllegalArgumentException("Invalid file ID"); + } + Path basePath = Path.of(baseDirPath).normalize().toAbsolutePath(); + Path resolvedPath = basePath.resolve(fileId).normalize(); + if (!resolvedPath.startsWith(basePath)) { + throw new IllegalArgumentException("File ID resolves to an invalid path"); + } + return resolvedPath; + } +} diff --git a/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreConfiguration.java b/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreConfiguration.java new file mode 100644 index 000000000..2df824663 --- /dev/null +++ b/app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreConfiguration.java @@ -0,0 +1,29 @@ +package stirling.software.common.cluster.inprocess; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import stirling.software.common.cluster.FileStore; + +/** + * Always-on wiring for the per-node local-disk {@link FileStore}. Active when {@code + * cluster.artifactStore=local} (the default; {@code matchIfMissing=true}). The S3 artifact-store + * supplies its own bean when {@code cluster.artifactStore=s3}. + */ +@Configuration +@ConditionalOnProperty( + prefix = "cluster", + name = "artifactStore", + havingValue = "local", + matchIfMissing = true) +public class LocalDiskFileStoreConfiguration { + + @Bean + @ConditionalOnMissingBean + public FileStore fileStore(@Value("${stirling.tempDir:/tmp/stirling-files}") String tempDir) { + return new LocalDiskFileStore(tempDir); + } +} diff --git a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java index 227ba7709..e298edd07 100644 --- a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java +++ b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java @@ -13,6 +13,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; import java.util.Locale; +import java.util.UUID; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -77,6 +78,7 @@ public class ApplicationProperties { private PdfEditor pdfEditor = new PdfEditor(); private AiEngine aiEngine = new AiEngine(); private InternalApi internalApi = new InternalApi(); + private Cluster cluster = new Cluster(); @Bean public PropertySource dynamicYamlPropertySource(ConfigurableEnvironment environment) @@ -254,6 +256,106 @@ public class ApplicationProperties { private int longRunningTimeoutSeconds = 600; } + /** + * Cluster backplane configuration. All keys live under the top-level {@code cluster.*} prefix + * (e.g. env var {@code CLUSTER_ENABLED}). The master switch is {@link #enabled} and defaults to + * off; when off the in-process backplane is wired and no other cluster keys are required. + */ + @Data + public static class Cluster { + + /** Master switch. When {@code false} (default) the in-process backplane is wired. */ + private boolean enabled = false; + + /** Backplane implementation selector. Valid values: {@code inprocess} | {@code valkey}. */ + private String backplane = "inprocess"; + + /** + * Transient cluster job-artifact store selector. Valid values: {@code local} | {@code s3}. + * + *

This is distinct from {@code storage.provider}, which selects the backend for + * persistent user-uploaded files. The two switches exist because the user-facing storage + * feature is optional ({@code storage.enabled=false} is common) but every multi-node + * cluster still needs a shared artifact store to serve cross-node downloads. Both + * implementations share credentials from {@code storage.s3.*} when set to {@code s3}. + */ + private String artifactStore = "local"; + + private Valkey valkey = new Valkey(); + private Node node = new Node(); + + private transient String cachedNodeId; + + public NodeRole resolvedRole() { + if (node == null || node.getRole() == null) { + return NodeRole.BOTH; + } + String value = node.getRole().trim().toUpperCase(Locale.ROOT); + try { + return NodeRole.valueOf(value); + } catch (IllegalArgumentException ex) { + return NodeRole.BOTH; + } + } + + public synchronized String resolvedNodeId() { + if (node != null && node.getId() != null && !node.getId().isBlank()) { + return node.getId(); + } + if (cachedNodeId == null) { + cachedNodeId = UUID.randomUUID().toString(); + } + return cachedNodeId; + } + + public enum NodeRole { + WEB, + WORKER, + BOTH + } + + @Data + public static class Valkey { + /** + * {@code redis://host:6379} or {@code rediss://...} for TLS. Required when cluster mode + * is on and backplane is valkey. + */ + private String url = ""; + + private Tls tls = new Tls(); + + @Data + public static class Tls { + /** + * When {@code true}, skip Valkey/Redis TLS certificate verification (dev/test + * only). Leave {@code false} in production. + */ + private boolean skipCertVerification = false; + } + } + + @Data + public static class Node { + /** Optional explicit node id. Blank = auto-generated UUID at startup. */ + private String id = ""; + + /** {@code web} | {@code worker} | {@code both}. */ + private String role = "both"; + + /** + * Internal cluster address advertised in the instance registry (host:port). Blank = + * derived at startup. + */ + private String internalAddress = ""; + + /** {@code http} | {@code https} - scheme used when peers call this node. */ + private String scheme = "http"; + + /** Heartbeat publish interval for the instance registry, in milliseconds. */ + private long heartbeatIntervalMs = 5000; + } + } + /** * HTTP timeouts for loopback calls to internal Stirling API endpoints, used by the AI workflow * executor and the pipeline processor. A bounded read timeout prevents a hung tool (e.g. an diff --git a/app/common/src/main/java/stirling/software/common/service/FileStorage.java b/app/common/src/main/java/stirling/software/common/service/FileStorage.java index beb8028f7..e77e6d6fc 100644 --- a/app/common/src/main/java/stirling/software/common/service/FileStorage.java +++ b/app/common/src/main/java/stirling/software/common/service/FileStorage.java @@ -1,15 +1,13 @@ package stirling.software.common.service; -import java.io.BufferedInputStream; -import java.io.BufferedOutputStream; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; -import java.io.OutputStream; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.UUID; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicReference; -import org.springframework.beans.factory.annotation.Value; import org.springframework.core.io.Resource; import org.springframework.stereotype.Service; import org.springframework.web.multipart.MultipartFile; @@ -18,9 +16,11 @@ import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBo import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import stirling.software.common.cluster.FileStore; + /** * Service for storing and retrieving files with unique file IDs. Used by the AutoJobPostMapping - * system to handle file references. + * system to handle file references. Disk I/O is delegated to the injected {@link FileStore} bean. */ @Service @RequiredArgsConstructor @@ -30,251 +30,150 @@ public class FileStorage { /** Holds the result of a stream-to-disk store operation: the file ID and the bytes written. */ public record StoredFile(String fileId, long size) {} - @Value("${stirling.tempDir:/tmp/stirling-files}") - private String tempDirPath; - private final FileOrUploadService fileOrUploadService; + private final FileStore fileStore; - /** - * Store a file and return its unique ID - * - * @param file The file to store - * @return The unique ID assigned to the file - * @throws IOException If there is an error storing the file - */ public String storeFile(MultipartFile file) throws IOException { - String fileId = generateFileId(); - Path filePath = getFilePath(fileId); - - // Ensure the directory exists - Files.createDirectories(filePath.getParent()); - - // Transfer the file to the storage location - file.transferTo(filePath.toFile()); - - log.debug("Stored file with ID: {}", fileId); - return fileId; - } - - /** - * Store a byte array as a file and return its unique ID - * - * @param bytes The byte array to store - * @param originalName The original name of the file (for extension) - * @return The unique ID assigned to the file - * @throws IOException If there is an error storing the file - */ - public String storeBytes(byte[] bytes, String originalName) throws IOException { - String fileId = generateFileId(); - Path filePath = getFilePath(fileId); - - // Ensure the directory exists - Files.createDirectories(filePath.getParent()); - - // Write the bytes to the file - Files.write(filePath, bytes); - - log.debug("Stored byte array with ID: {}", fileId); - return fileId; - } - - /** - * Retrieve a file by its ID as a MultipartFile - * - * @param fileId The ID of the file to retrieve - * @return The file as a MultipartFile - * @throws IOException If the file doesn't exist or can't be read - */ - public MultipartFile retrieveFile(String fileId) throws IOException { - Path filePath = getFilePath(fileId); - - if (!Files.exists(filePath)) { - throw new IOException("File not found with ID: " + fileId); + // Fast path: when Spring buffered the multipart to disk (typical for large uploads), the + // backing Resource exposes a real File. Hand the Path to the FileStore so it can do a + // file-to-file copy (Linux sendfile, no copy through Java heap) rather than streaming + // the bytes through an 8K buffer. Falls back to the InputStream path for in-memory + // multiparts, exotic Resource impls, and anything that does not back onto a File. + Resource res; + try { + res = file.getResource(); + } catch (RuntimeException ignored) { + res = null; } + if (res != null && res.isFile()) { + try { + FileStore.Stored stored = + fileStore.store(res.getFile().toPath(), file.getOriginalFilename()); + log.debug("Stored file with ID: {} (fast path)", stored.fileId()); + return stored.fileId(); + } catch (IOException ex) { + // Some Resource impls advertise isFile()=true but throw on getFile(); fall through. + log.debug("Resource fast path failed, falling back to stream copy", ex); + } + } + try (InputStream in = file.getInputStream()) { + FileStore.Stored stored = fileStore.store(in, file.getOriginalFilename()); + log.debug("Stored file with ID: {}", stored.fileId()); + return stored.fileId(); + } + } - byte[] fileData = Files.readAllBytes(filePath); + public String storeBytes(byte[] bytes, String originalName) throws IOException { + FileStore.Stored stored = fileStore.store(new ByteArrayInputStream(bytes), originalName); + log.debug("Stored byte array with ID: {}", stored.fileId()); + return stored.fileId(); + } + + public MultipartFile retrieveFile(String fileId) throws IOException { + byte[] fileData = fileStore.retrieveBytes(fileId); return fileOrUploadService.toMockMultipartFile(fileId, fileData); } - /** - * Retrieve a file by its ID as a byte array - * - * @param fileId The ID of the file to retrieve - * @return The file as a byte array - * @throws IOException If the file doesn't exist or can't be read - */ public byte[] retrieveBytes(String fileId) throws IOException { - Path filePath = getFilePath(fileId); - - if (!Files.exists(filePath)) { - throw new IOException("File not found with ID: " + fileId); - } - - return Files.readAllBytes(filePath); + return fileStore.retrieveBytes(fileId); } - /** - * Retrieve a file by its ID as a streaming InputStream. The caller is responsible for closing - * the returned stream. - * - * @param fileId The ID of the file to retrieve - * @return A buffered InputStream for the file - * @throws IOException If the file doesn't exist or can't be read - */ public InputStream retrieveInputStream(String fileId) throws IOException { - Path filePath = getFilePath(fileId); - // Let Files.newInputStream throw NoSuchFileException naturally — avoids TOCTOU race - // between exists-check and open when another thread may delete concurrently. - return new BufferedInputStream(Files.newInputStream(filePath)); + return fileStore.retrieve(fileId); } - /** - * Store data from an InputStream as a file and return its unique ID and byte count. Streams - * directly to disk without buffering the entire content in heap. - * - * @param inputStream The input stream to read from - * @param originalName The original name of the file (unused, kept for API symmetry) - * @return A {@link StoredFile} containing the file ID and the number of bytes written - * @throws IOException If there is an error storing the file - */ public StoredFile storeInputStream(InputStream inputStream, String originalName) throws IOException { - String fileId = generateFileId(); - Path filePath = getFilePath(fileId); - Files.createDirectories(filePath.getParent()); - long size = Files.copy(inputStream, filePath); - log.debug("Stored input stream with ID: {}", fileId); - return new StoredFile(fileId, size); + FileStore.Stored stored = fileStore.store(inputStream, originalName); + log.debug("Stored input stream with ID: {}", stored.fileId()); + return new StoredFile(stored.fileId(), stored.size()); } public String storeFromStreamingBody(StreamingResponseBody body, String originalName) throws IOException { - String fileId = generateFileId(); - Path filePath = getFilePath(fileId); - Files.createDirectories(filePath.getParent()); - boolean success = false; - try (OutputStream os = new BufferedOutputStream(Files.newOutputStream(filePath))) { - body.writeTo(os); - success = true; - } finally { - if (!success) { + // Hold Throwable not IOException: an unchecked failure (NPE, IllegalState, OOM, etc.) + // from the body writer would otherwise close the pipe with EOF and the consumer would + // return a truncated file with no error surfaced to the caller. + AtomicReference bodyError = new AtomicReference<>(); + try (PipedOutputStream out = new PipedOutputStream(); + PipedInputStream in = new PipedInputStream(out, 8192)) { + var executor = Executors.newSingleThreadExecutor(Thread.ofVirtual().factory()); + java.util.concurrent.Future task = null; + try { + task = + executor.submit( + () -> { + try { + body.writeTo(out); + } catch (Throwable ex) { + bodyError.set(ex); + } finally { + try { + out.close(); + } catch (IOException ignored) { + // closed on the consumer side too + } + } + }); + FileStore.Stored stored = fileStore.store(in, originalName); + Throwable writerErr = bodyError.get(); + if (writerErr != null) { + // Body failed mid-write: the FileStore persisted a truncated entry. + // Best-effort delete so we don't leak partial files; never let cleanup + // mask the original writer error. + try { + fileStore.delete(stored.fileId()); + } catch (RuntimeException cleanupEx) { + log.warn( + "Failed to delete partial file {} after writer error: {}", + stored.fileId(), + cleanupEx.getMessage()); + } + if (writerErr instanceof IOException ioe) { + throw ioe; + } + throw new IOException( + "StreamingResponseBody writer failed: " + writerErr.getMessage(), + writerErr); + } + log.debug("Stored StreamingResponseBody with ID: {}", stored.fileId()); + return stored.fileId(); + } finally { + // Interrupt and join the writer task: shutdown() alone returns immediately and a + // failed store leaves the writer running, leaking a thread per failed upload. + if (task != null && !task.isDone()) { + task.cancel(true); + } + executor.shutdown(); try { - Files.deleteIfExists(filePath); - } catch (IOException cleanupEx) { - log.warn( - "Failed to clean up partial file {} after store failure", - filePath, - cleanupEx); + if (!executor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) { + executor.shutdownNow(); + } + } catch (InterruptedException ie) { + executor.shutdownNow(); + Thread.currentThread().interrupt(); } } } - log.debug("Stored StreamingResponseBody with ID: {}", fileId); - return fileId; } - /** - * Persist a {@link Resource} body to disk, returning the generated file ID. Used by the async - * job pipeline to capture {@code ResponseEntity} results produced by controllers. - */ public String storeFromResource(Resource resource, String originalName) throws IOException { - String fileId = generateFileId(); - Path filePath = getFilePath(fileId); - Files.createDirectories(filePath.getParent()); - boolean success = false; try (InputStream in = resource.getInputStream()) { - Files.copy(in, filePath); - success = true; - } finally { - if (!success) { - try { - Files.deleteIfExists(filePath); - } catch (IOException cleanupEx) { - log.warn( - "Failed to clean up partial file {} after store failure", - filePath, - cleanupEx); - } - } + FileStore.Stored stored = fileStore.store(in, originalName); + log.debug("Stored Resource with ID: {}", stored.fileId()); + return stored.fileId(); } - log.debug("Stored Resource with ID: {}", fileId); - return fileId; } - /** - * Delete a file by its ID - * - * @param fileId The ID of the file to delete - * @return true if the file was deleted, false otherwise - */ public boolean deleteFile(String fileId) { - try { - Path filePath = getFilePath(fileId); - return Files.deleteIfExists(filePath); - } catch (IOException e) { - log.error("Error deleting file with ID: {}", fileId, e); - return false; - } + return fileStore.delete(fileId); } - /** - * Check if a file exists by its ID - * - * @param fileId The ID of the file to check - * @return true if the file exists, false otherwise - */ public boolean fileExists(String fileId) { - Path filePath = getFilePath(fileId); - return Files.exists(filePath); + return fileStore.exists(fileId); } - /** - * Get the size of a file by its ID without loading the content into memory - * - * @param fileId The ID of the file - * @return The size of the file in bytes - * @throws IOException If the file doesn't exist or can't be read - */ public long getFileSize(String fileId) throws IOException { - Path filePath = getFilePath(fileId); - - if (!Files.exists(filePath)) { - throw new IOException("File not found with ID: " + fileId); - } - - return Files.size(filePath); - } - - /** - * Get the path for a file ID - * - * @param fileId The ID of the file - * @return The path to the file - * @throws IllegalArgumentException if fileId contains path traversal characters or resolves - * outside base directory - */ - private Path getFilePath(String fileId) { - // Validate fileId to prevent path traversal - if (fileId.contains("..") || fileId.contains("/") || fileId.contains("\\")) { - throw new IllegalArgumentException("Invalid file ID"); - } - - Path basePath = Path.of(tempDirPath).normalize().toAbsolutePath(); - Path resolvedPath = basePath.resolve(fileId).normalize(); - - // Ensure resolved path is within the base directory - if (!resolvedPath.startsWith(basePath)) { - throw new IllegalArgumentException("File ID resolves to an invalid path"); - } - - return resolvedPath; - } - - /** - * Generate a unique file ID - * - * @return A unique file ID - */ - private String generateFileId() { - return UUID.randomUUID().toString(); + return fileStore.size(fileId); } } diff --git a/app/common/src/main/java/stirling/software/common/service/TaskManager.java b/app/common/src/main/java/stirling/software/common/service/TaskManager.java index 36a1d1857..790e0626a 100644 --- a/app/common/src/main/java/stirling/software/common/service/TaskManager.java +++ b/app/common/src/main/java/stirling/software/common/service/TaskManager.java @@ -3,9 +3,13 @@ package stirling.software.common.service; import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.ZoneId; import java.time.temporal.ChronoUnit; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -16,6 +20,7 @@ import java.util.concurrent.TimeUnit; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.http.MediaType; import org.springframework.stereotype.Service; @@ -26,6 +31,10 @@ import jakarta.annotation.PreDestroy; 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.JobStoreEntry.JobState; import stirling.software.common.model.job.JobResult; import stirling.software.common.model.job.JobStats; import stirling.software.common.model.job.ResultFile; @@ -40,20 +49,20 @@ public class TaskManager { private int jobResultExpiryMinutes = 30; private final FileStorage fileStorage; + private final JobStore jobStore; + private final ClusterBackplane clusterBackplane; private final ScheduledExecutorService cleanupExecutor = Executors.newSingleThreadScheduledExecutor( Thread.ofVirtual().name("task-cleanup-", 0).factory()); - /** Initialize the task manager and start the cleanup scheduler */ - public TaskManager(FileStorage fileStorage) { + @Autowired + public TaskManager( + FileStorage fileStorage, JobStore jobStore, ClusterBackplane clusterBackplane) { this.fileStorage = fileStorage; + this.jobStore = jobStore; + this.clusterBackplane = clusterBackplane; - // Schedule periodic cleanup of old job results - cleanupExecutor.scheduleAtFixedRate( - this::cleanupOldJobs, - 10, // Initial delay - 10, // Interval - TimeUnit.MINUTES); + cleanupExecutor.scheduleAtFixedRate(this::cleanupOldJobs, 10, 10, TimeUnit.MINUTES); log.debug( "Task manager initialized with job result expiry of {} minutes", @@ -66,7 +75,9 @@ public class TaskManager { * @param jobId The job ID */ public void createTask(String jobId) { - jobResults.put(jobId, JobResult.createNew(jobId)); + JobResult result = JobResult.createNew(jobId); + jobResults.put(jobId, result); + writeThrough(jobId, result); log.debug("Created task with job ID: {}", jobId); } @@ -79,6 +90,7 @@ public class TaskManager { public void setResult(String jobId, Object result) { JobResult jobResult = getOrCreateJobResult(jobId); jobResult.completeWithResult(result); + writeThrough(jobId, jobResult); log.debug("Set result for job ID: {}", jobId); } @@ -101,6 +113,7 @@ public class TaskManager { extractZipToIndividualFiles(fileId, originalFileName); if (!extractedFiles.isEmpty()) { jobResult.completeWithFiles(extractedFiles); + writeThrough(jobId, jobResult); log.debug( "Set multiple file results for job ID: {} with {} files extracted from" + " ZIP", @@ -127,6 +140,7 @@ public class TaskManager { "Failed to get file size for job {}: {}. Using size 0.", jobId, e.getMessage()); jobResult.completeWithSingleFile(fileId, originalFileName, contentType, 0); } + writeThrough(jobId, jobResult); } /** @@ -138,6 +152,7 @@ public class TaskManager { public void setMultipleFileResults(String jobId, List resultFiles) { JobResult jobResult = getOrCreateJobResult(jobId); jobResult.completeWithFiles(resultFiles); + writeThrough(jobId, jobResult); log.debug( "Set multiple file results for job ID: {} with {} files", jobId, @@ -153,6 +168,7 @@ public class TaskManager { public void setError(String jobId, String error) { JobResult jobResult = getOrCreateJobResult(jobId); jobResult.failWithError(error); + writeThrough(jobId, jobResult); log.debug("Set error for job ID: {}: {}", jobId, error); } @@ -169,6 +185,7 @@ public class TaskManager { // If no result or error has been set, mark it as complete with an empty result jobResult.completeWithResult("Task completed successfully"); } + writeThrough(jobId, jobResult); log.debug("Marked job ID: {} as complete", jobId); } @@ -205,6 +222,7 @@ public class TaskManager { JobResult jobResult = jobResults.get(jobId); if (jobResult != null) { jobResult.addNote(note); + writeThrough(jobId, jobResult); log.debug("Added note to job ID: {}: {}", jobId, note); return true; } @@ -295,8 +313,11 @@ public class TaskManager { return jobResults.computeIfAbsent(jobId, JobResult::createNew); } - /** Clean up old completed job results */ + /** Clean up old completed job results. No-op in cluster mode; the backplane TTL owns expiry. */ public void cleanupOldJobs() { + if (clusterBackplane != null && !clusterBackplane.shouldRunLocalCleanup()) { + return; + } LocalDateTime expiryThreshold = LocalDateTime.now().minus(jobResultExpiryMinutes, ChronoUnit.MINUTES); int removedCount = 0; @@ -315,6 +336,9 @@ public class TaskManager { // Remove the job result jobResults.remove(entry.getKey()); + if (jobStore != null) { + jobStore.delete(entry.getKey()); + } removedCount++; } } @@ -327,6 +351,53 @@ public class TaskManager { } } + /** Mirror the in-memory {@code JobResult} into the cluster-visible {@link JobStore}. */ + private void writeThrough(String jobId, JobResult result) { + if (jobStore == null) { + return; + } + try { + jobStore.put(toEntry(jobId, result), Duration.ofMinutes(jobResultExpiryMinutes)); + } catch (RuntimeException ex) { + log.warn("JobStore write-through failed for job {}: {}", jobId, ex.getMessage()); + } + } + + private JobStoreEntry toEntry(String jobId, JobResult result) { + JobState state; + if (result.isComplete()) { + state = result.getError() != null ? JobState.FAILED : JobState.COMPLETE; + } else { + state = JobState.PENDING; + } + Instant createdAt = toInstant(result.getCreatedAt()); + Instant completedAt = toInstant(result.getCompletedAt()); + List fileIds = new ArrayList<>(); + if (result.hasFiles()) { + for (ResultFile rf : result.getAllResultFiles()) { + fileIds.add(rf.getFileId()); + } + } + Map meta = new HashMap<>(); + if (result.getNotes() != null && !result.getNotes().isEmpty()) { + meta.put("notesCount", Integer.toString(result.getNotes().size())); + } + String owningNodeId = clusterBackplane == null ? "local" : clusterBackplane.localNodeId(); + return new JobStoreEntry( + jobId, + state, + owningNodeId, + createdAt, + completedAt, + result.getError(), + fileIds, + meta); + } + + private Instant toInstant(LocalDateTime ldt) { + return ldt == null ? null : ldt.atZone(ZoneId.systemDefault()).toInstant(); + } + /** Shutdown the cleanup executor */ @PreDestroy public void shutdown() { @@ -370,7 +441,7 @@ public class TaskManager { while ((entry = zipIn.getNextEntry()) != null) { if (!entry.isDirectory()) { String contentType = determineContentType(entry.getName()); - // storeInputStream returns the fileId and byte count — no extra stat needed + // storeInputStream returns the fileId and byte count - no extra stat needed FileStorage.StoredFile stored = fileStorage.storeInputStream(zipIn, entry.getName()); @@ -458,7 +529,8 @@ public class TaskManager { } /** - * Find the job key that owns a given file ID. + * Find the job key that owns a given file ID. Checks the local in-memory map first, then falls + * back to the cluster-visible {@link JobStore}. * * @param fileId file identifier to look up * @return scoped job key if found, otherwise null @@ -474,6 +546,18 @@ public class TaskManager { } } } + if (jobStore != null) { + // Propagate JobStore failures: returning null on a backplane outage would conflate + // "no such file" with "lookup unavailable" and the caller would respond 404 to a + // transient blip that should be retried. Let Spring's exception handler surface a + // 5xx so clients know to retry. + try { + return jobStore.findJobIdByFileId(fileId).orElse(null); + } catch (RuntimeException e) { + log.warn("JobStore findJobIdByFileId failed for {}: {}", fileId, e.getMessage()); + throw e; + } + } return null; } } diff --git a/app/common/src/test/java/stirling/software/common/architecture/ArchitectureTest.java b/app/common/src/test/java/stirling/software/common/architecture/ArchitectureTest.java index 68f22dfe8..285407dca 100644 --- a/app/common/src/test/java/stirling/software/common/architecture/ArchitectureTest.java +++ b/app/common/src/test/java/stirling/software/common/architecture/ArchitectureTest.java @@ -56,4 +56,17 @@ class ArchitectureTest { .resideInAPackage("stirling.software.saas.."); rule.check(commonClasses); } + + @Test + void clusterInterfacesHaveNoImplementationDependencies() { + ArchRule rule = + noClasses() + .that() + .resideInAPackage("stirling.software.common.cluster..") + .should() + .dependOnClassesThat() + .resideInAnyPackage( + "stirling.software.proprietary..", "stirling.software.saas.."); + rule.check(commonClasses); + } } diff --git a/app/common/src/test/java/stirling/software/common/cluster/BackplaneContractCompilationTest.java b/app/common/src/test/java/stirling/software/common/cluster/BackplaneContractCompilationTest.java new file mode 100644 index 000000000..9da2c41cf --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/BackplaneContractCompilationTest.java @@ -0,0 +1,51 @@ +package stirling.software.common.cluster; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Instant; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +class BackplaneContractCompilationTest { + + @Test + void jobStoreEntryRecordRoundTrips() { + Instant now = Instant.now(); + JobStoreEntry entry = + new JobStoreEntry( + "job-1", + JobStoreEntry.JobState.PENDING, + "node-a", + now, + null, + null, + List.of("file-1"), + Map.of("k", "v")); + assertEquals("job-1", entry.jobId()); + assertEquals(JobStoreEntry.JobState.PENDING, entry.state()); + assertEquals("node-a", entry.owningNodeId()); + assertEquals(now, entry.createdAt()); + assertEquals(List.of("file-1"), entry.fileIds()); + assertEquals("v", entry.resultMeta().get("k")); + } + + @Test + void clusterNodeRecordRoundTrips() { + Instant heartbeat = Instant.now(); + ClusterNode node = new ClusterNode("node-a", "10.0.0.1:8080", heartbeat, "BOTH"); + assertEquals("node-a", node.nodeId()); + assertEquals("10.0.0.1:8080", node.internalAddress()); + assertEquals(heartbeat, node.lastHeartbeat()); + assertEquals("BOTH", node.role()); + } + + @Test + void rateLimitDecisionRecordRoundTrips() { + RateLimitStore.RateLimitDecision d = new RateLimitStore.RateLimitDecision(true, 7, 0L); + assertEquals(true, d.allowed()); + assertEquals(7, d.remainingTokens()); + assertEquals(0L, d.nanosToWaitForRefill()); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java b/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java new file mode 100644 index 000000000..d87e2e8bb --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java @@ -0,0 +1,65 @@ +package stirling.software.common.cluster; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.lang.reflect.Method; + +import org.junit.jupiter.api.Test; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.ApplicationProperties.Cluster; + +class ClusterConfigValidationTest { + + @Test + void validationPassesWhenDisabled() { + ApplicationProperties props = new ApplicationProperties(); + ClusterConfig config = new ClusterConfig(props); + assertDoesNotThrow(() -> invokeValidate(config)); + } + + @Test + void validationFailsWhenValkeyEnabledWithoutUrl() { + ApplicationProperties props = new ApplicationProperties(); + Cluster cluster = props.getCluster(); + cluster.setEnabled(true); + cluster.setBackplane("valkey"); + ClusterConfig config = new ClusterConfig(props); + assertThrows(IllegalStateException.class, () -> invokeValidate(config)); + } + + @Test + void validationPassesWhenValkeyEnabledWithUrl() { + ApplicationProperties props = new ApplicationProperties(); + Cluster cluster = props.getCluster(); + cluster.setEnabled(true); + cluster.setBackplane("valkey"); + cluster.getValkey().setUrl("redis://localhost:6379"); + ClusterConfig config = new ClusterConfig(props); + assertDoesNotThrow(() -> invokeValidate(config)); + } + + @Test + void validationPassesWhenInProcessEnabled() { + ApplicationProperties props = new ApplicationProperties(); + Cluster cluster = props.getCluster(); + cluster.setEnabled(true); + cluster.setBackplane("inprocess"); + ClusterConfig config = new ClusterConfig(props); + assertDoesNotThrow(() -> invokeValidate(config)); + } + + private void invokeValidate(ClusterConfig config) throws Exception { + Method m = ClusterConfig.class.getDeclaredMethod("validate"); + m.setAccessible(true); + try { + m.invoke(config); + } catch (java.lang.reflect.InvocationTargetException ex) { + if (ex.getCause() instanceof RuntimeException re) { + throw re; + } + throw ex; + } + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/ClusterPropertiesTest.java b/app/common/src/test/java/stirling/software/common/cluster/ClusterPropertiesTest.java new file mode 100644 index 000000000..8237a2252 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/ClusterPropertiesTest.java @@ -0,0 +1,62 @@ +package stirling.software.common.cluster; + +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 org.junit.jupiter.api.Test; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.ApplicationProperties.Cluster; + +class ClusterPropertiesTest { + + @Test + void defaultsAreDisabledAndInprocess() { + Cluster props = new ApplicationProperties().getCluster(); + assertFalse(props.isEnabled()); + assertEquals("inprocess", props.getBackplane()); + assertEquals("local", props.getArtifactStore()); + assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole()); + assertEquals("", props.getValkey().getUrl()); + assertFalse(props.getValkey().getTls().isSkipCertVerification()); + assertEquals("both", props.getNode().getRole()); + assertEquals("http", props.getNode().getScheme()); + assertEquals(5000L, props.getNode().getHeartbeatIntervalMs()); + } + + @Test + void resolvedRoleParsesCaseInsensitively() { + Cluster props = new ApplicationProperties().getCluster(); + props.getNode().setRole("WEB"); + assertEquals(Cluster.NodeRole.WEB, props.resolvedRole()); + + props.getNode().setRole("web"); + assertEquals(Cluster.NodeRole.WEB, props.resolvedRole()); + + props.getNode().setRole("Worker"); + assertEquals(Cluster.NodeRole.WORKER, props.resolvedRole()); + + props.getNode().setRole("garbage"); + assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole()); + + props.getNode().setRole(null); + assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole()); + } + + @Test + void resolvedNodeIdIsStableAcrossCalls() { + Cluster props = new ApplicationProperties().getCluster(); + String first = props.resolvedNodeId(); + String second = props.resolvedNodeId(); + assertNotNull(first); + assertEquals(first, second); + } + + @Test + void resolvedNodeIdHonoursExplicitId() { + Cluster props = new ApplicationProperties().getCluster(); + props.getNode().setId("abc"); + assertEquals("abc", props.resolvedNodeId()); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/InProcessConfigurationConditionalTest.java b/app/common/src/test/java/stirling/software/common/cluster/InProcessConfigurationConditionalTest.java new file mode 100644 index 000000000..4a5a73ce3 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/InProcessConfigurationConditionalTest.java @@ -0,0 +1,90 @@ +package stirling.software.common.cluster; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import stirling.software.common.cluster.inprocess.InProcessClusterConfiguration; +import stirling.software.common.model.ApplicationProperties; + +/** + * Verifies the {@link InProcessClusterConfiguration} conditional wiring: in-process beans wire when + * cluster mode is off or {@code backplane=inprocess}, and are skipped when {@code + * backplane=valkey}. + */ +class InProcessConfigurationConditionalTest { + + private final ApplicationContextRunner runner = + new ApplicationContextRunner() + .withConfiguration( + org.springframework.boot.autoconfigure.AutoConfigurations.of( + PropertyPlaceholderAutoConfiguration.class)) + .withUserConfiguration( + TestAppPropertiesConfig.class, + ClusterConfig.class, + InProcessClusterConfiguration.class); + + @Test + void inProcessBeansWireWhenClusterDisabled() { + runner.run( + context -> + assertThat(context) + .hasNotFailed() + .hasSingleBean(ClusterBackplane.class) + .hasSingleBean(JobStore.class) + .hasSingleBean(RateLimitStore.class) + .hasSingleBean(DistributedLock.class) + .hasSingleBean(KeyValueCache.class) + .hasSingleBean(InstanceRegistry.class)); + } + + @Test + void inProcessBeansWireWhenEnabledWithInProcessBackplane() { + runner.withPropertyValues("cluster.enabled=true", "cluster.backplane=inprocess") + .run( + context -> + assertThat(context) + .hasNotFailed() + .hasSingleBean(ClusterBackplane.class) + .hasSingleBean(JobStore.class) + .hasSingleBean(RateLimitStore.class) + .hasSingleBean(DistributedLock.class) + .hasSingleBean(KeyValueCache.class) + .hasSingleBean(InstanceRegistry.class)); + } + + @Test + void inProcessBeansSkippedWhenEnabledWithDistributedBackplane() { + runner.withPropertyValues( + "cluster.enabled=true", + "cluster.backplane=valkey", + "cluster.valkey.url=redis://localhost:6379") + .run( + context -> + assertThat(context) + .hasNotFailed() + .doesNotHaveBean(ClusterBackplane.class) + .doesNotHaveBean(JobStore.class) + .doesNotHaveBean(RateLimitStore.class) + .doesNotHaveBean(DistributedLock.class) + .doesNotHaveBean(KeyValueCache.class) + .doesNotHaveBean(InstanceRegistry.class)); + } + + /** + * Hand-rolled {@link ApplicationProperties} bean: the production class loads YAML at startup + * via a {@code @PostConstruct} hook that isn't appropriate for the slice runner, so we wire a + * defaults-only instance here. + */ + @Configuration + static class TestAppPropertiesConfig { + @Bean + ApplicationProperties applicationProperties() { + return new ApplicationProperties(); + } + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessDistributedLockTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessDistributedLockTest.java new file mode 100644 index 000000000..30b738a3b --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessDistributedLockTest.java @@ -0,0 +1,167 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.Test; + +import stirling.software.common.cluster.DistributedLock; + +class InProcessDistributedLockTest { + + @Test + void acquireReleaseAcquire() { + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow(); + h1.release(); + assertTrue(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent()); + } + + @Test + void reentryFromSameThreadFails() { + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow(); + Optional reentry = lock.tryAcquire("k", Duration.ofSeconds(30)); + assertFalse(reentry.isPresent(), "in-process lock must be non-reentrant"); + h1.release(); + // After release, anyone can acquire again. + assertTrue(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent()); + } + + @Test + void secondAcquireFromAnotherThreadFails() throws InterruptedException { + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow(); + + CountDownLatch done = new CountDownLatch(1); + AtomicBoolean acquired = new AtomicBoolean(true); + Thread t = + new Thread( + () -> { + Optional attempt = + lock.tryAcquire("k", Duration.ofSeconds(30)); + acquired.set(attempt.isPresent()); + attempt.ifPresent(DistributedLock.LockHandle::release); + done.countDown(); + }); + t.start(); + assertTrue(done.await(2, TimeUnit.SECONDS)); + assertFalse(acquired.get()); + h1.release(); + } + + @Test + void leaseExpiryAllowsTakeoverEvenWithoutRelease() throws InterruptedException { + // Acquire with a short lease, never call release, then try to acquire again after the + // lease has elapsed. Matches Redis SET-NX-EX semantics - the second caller gets the lock + // because the first lease auto-expired. 250ms lease + 350ms wait gives CI generous slack. + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofMillis(250)).orElseThrow(); + Thread.sleep(350); + Optional takeover = + lock.tryAcquire("k", Duration.ofSeconds(30)); + assertTrue( + takeover.isPresent(), + "expired lease must release the lock so a new caller can take over"); + // Calling release() on the original handle after takeover must be a no-op (token check). + h1.release(); + // The takeover holder is still the legitimate owner. + assertFalse(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent()); + takeover.get().release(); + } + + @Test + void renewExtendsLease() throws InterruptedException { + // Acquire with a short lease, renew it before it expires, then verify the lock is still + // held past the original expiry point. 200ms initial + renew to 2s + wait 350ms. + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofMillis(200)).orElseThrow(); + assertTrue(h1.renew(Duration.ofSeconds(2)), "renew on a held lease must succeed"); + Thread.sleep(350); + assertFalse( + lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent(), + "renew should have pushed expiry well past the original 200ms"); + h1.release(); + } + + @Test + void renewAfterReleaseFails() { + DistributedLock lock = new InProcessDistributedLock(); + DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow(); + h1.release(); + assertFalse(h1.renew(Duration.ofSeconds(30)), "renew on a released handle must fail"); + } + + /** + * Concurrency stress: many threads contending on the same key with each holder respecting the + * lease (hold << lease). The lock behaves as a strict mutex in this regime so asserting + * mutual exclusion is meaningful. A separate test ({@link + * #leaseExpiryAllowsTakeoverEvenWithoutRelease}) covers the takeover-across-expiry branch, + * which legitimately allows two holders momentarily and is split-brain behaviour inherent to + * any lease-based lock. + */ + @Test + void concurrentContentionPreservesMutualExclusion() throws InterruptedException { + DistributedLock lock = new InProcessDistributedLock(); + int threads = 16; + int attemptsPerThread = 200; + // Lease far exceeds any plausible hold time, so the takeover branch never triggers in + // this test and the lock acts as a strict mutex. + Duration lease = Duration.ofSeconds(5); + java.util.concurrent.atomic.AtomicInteger concurrentHolders = + new java.util.concurrent.atomic.AtomicInteger(); + java.util.concurrent.atomic.AtomicInteger maxConcurrent = + new java.util.concurrent.atomic.AtomicInteger(); + java.util.concurrent.atomic.AtomicInteger acquires = + new java.util.concurrent.atomic.AtomicInteger(); + java.util.concurrent.atomic.AtomicReference firstFailure = + new java.util.concurrent.atomic.AtomicReference<>(); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threads); + + for (int i = 0; i < threads; i++) { + new Thread( + () -> { + try { + start.await(); + for (int j = 0; j < attemptsPerThread; j++) { + Optional h = + lock.tryAcquire("hot", lease); + if (h.isPresent()) { + int now = concurrentHolders.incrementAndGet(); + maxConcurrent.accumulateAndGet(now, Math::max); + acquires.incrementAndGet(); + // Trivial critical section; well within lease. + concurrentHolders.decrementAndGet(); + h.get().release(); + } + } + } catch (Throwable t) { + firstFailure.compareAndSet(null, t); + } finally { + done.countDown(); + } + }, + "lock-stress-" + i) + .start(); + } + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS), "stress workers must finish in time"); + org.junit.jupiter.api.Assertions.assertNull(firstFailure.get(), "no worker may throw"); + org.junit.jupiter.api.Assertions.assertEquals( + 1, + maxConcurrent.get(), + "mutual exclusion violated: more than one holder observed simultaneously"); + assertTrue( + acquires.get() > 0, + "at least some acquires must succeed under contention (saw " + + acquires.get() + + ")"); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistryTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistryTest.java new file mode 100644 index 000000000..c19eb444f --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistryTest.java @@ -0,0 +1,28 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.time.Instant; + +import org.junit.jupiter.api.Test; + +import stirling.software.common.cluster.ClusterNode; + +class InProcessInstanceRegistryTest { + + @Test + void registerThenLookupAndActiveNodes() { + InProcessInstanceRegistry registry = new InProcessInstanceRegistry(); + ClusterNode node = new ClusterNode("node-1", "127.0.0.1:8080", Instant.now(), "BOTH"); + registry.register(node, Duration.ofSeconds(30)); + + assertTrue(registry.lookup("node-1").isPresent()); + assertEquals("node-1", registry.lookup("node-1").get().nodeId()); + assertEquals(1, registry.activeNodes().size()); + + registry.deregister("node-1"); + assertTrue(registry.lookup("node-1").isEmpty()); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessJobStoreTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessJobStoreTest.java new file mode 100644 index 000000000..33eaebe35 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessJobStoreTest.java @@ -0,0 +1,91 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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 org.junit.jupiter.api.Test; + +import stirling.software.common.cluster.JobStoreEntry; + +class InProcessJobStoreTest { + + private final InProcessJobStore store = new InProcessJobStore(); + + @Test + void putGetDeleteExistsRoundTrip() { + JobStoreEntry entry = entry("job-1"); + store.put(entry, Duration.ofMinutes(30)); + + assertTrue(store.exists("job-1")); + assertEquals(entry, store.get("job-1").orElseThrow()); + + store.delete("job-1"); + assertFalse(store.exists("job-1")); + } + + @Test + void ttlExpiry() throws InterruptedException { + store.put(entry("job-2"), Duration.ofMillis(50)); + Thread.sleep(100); + assertFalse(store.get("job-2").isPresent()); + } + + @Test + void purgeExpiredRemovesOnlyStaleEntries() throws InterruptedException { + store.put(entry("job-fresh"), Duration.ofMinutes(30)); + store.put(entry("job-stale"), Duration.ofMillis(20)); + Thread.sleep(80); + + int removed = store.purgeExpired(); + assertEquals(1, removed); + assertTrue(store.exists("job-fresh")); + } + + @Test + void findJobIdByFileIdReturnsTheRightJob() { + store.put( + new JobStoreEntry( + "job-a", + JobStoreEntry.JobState.COMPLETE, + "node-1", + Instant.now(), + Instant.now(), + null, + List.of("file-1", "file-2"), + Map.of()), + Duration.ofMinutes(30)); + store.put( + new JobStoreEntry( + "job-b", + JobStoreEntry.JobState.COMPLETE, + "node-1", + Instant.now(), + Instant.now(), + null, + List.of("file-3"), + Map.of()), + Duration.ofMinutes(30)); + + assertEquals("job-a", store.findJobIdByFileId("file-1").orElseThrow()); + assertEquals("job-b", store.findJobIdByFileId("file-3").orElseThrow()); + assertFalse(store.findJobIdByFileId("missing").isPresent()); + } + + private JobStoreEntry entry(String id) { + return new JobStoreEntry( + id, + JobStoreEntry.JobState.PENDING, + "node-1", + Instant.now(), + null, + null, + List.of(), + Map.of()); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCacheTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCacheTest.java new file mode 100644 index 000000000..f0afc9789 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCacheTest.java @@ -0,0 +1,41 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +import stirling.software.common.cluster.KeyValueCache; + +class InProcessKeyValueCacheTest { + + @Test + void putGetEvict() { + KeyValueCache cache = new InProcessKeyValueCache(); + cache.put("apikey", "a", "userA", Duration.ofMinutes(1)); + assertEquals("userA", cache.get("apikey", "a").orElseThrow()); + + cache.evict("apikey", "a"); + assertFalse(cache.get("apikey", "a").isPresent()); + } + + @Test + void ttlExpiry() throws InterruptedException { + KeyValueCache cache = new InProcessKeyValueCache(); + cache.put("ns", "k", "v", Duration.ofMillis(40)); + Thread.sleep(80); + assertFalse(cache.get("ns", "k").isPresent()); + } + + @Test + void evictNamespace() { + KeyValueCache cache = new InProcessKeyValueCache(); + cache.put("ns", "a", "1", Duration.ofMinutes(1)); + cache.put("ns", "b", "2", Duration.ofMinutes(1)); + cache.evictNamespace("ns"); + assertFalse(cache.get("ns", "a").isPresent()); + assertFalse(cache.get("ns", "b").isPresent()); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStoreTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStoreTest.java new file mode 100644 index 000000000..8aecf259b --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStoreTest.java @@ -0,0 +1,56 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +import stirling.software.common.cluster.RateLimitStore; +import stirling.software.common.cluster.RateLimitStore.RateLimitDecision; + +class InProcessRateLimitStoreTest { + + @Test + void firstNConsumesAllowed() { + RateLimitStore store = new InProcessRateLimitStore(); + for (int i = 0; i < 5; i++) { + assertTrue(store.tryConsume("k", 5, Duration.ofSeconds(60)).allowed(), "i=" + i); + } + assertFalse(store.tryConsume("k", 5, Duration.ofSeconds(60)).allowed()); + } + + @Test + void remainingTokensDecrements() { + RateLimitStore store = new InProcessRateLimitStore(); + RateLimitDecision d1 = store.tryConsume("k", 5, Duration.ofSeconds(60)); + RateLimitDecision d2 = store.tryConsume("k", 5, Duration.ofSeconds(60)); + assertTrue(d1.allowed()); + assertTrue(d2.allowed()); + assertEquals(4, d1.remainingTokens()); + assertEquals(3, d2.remainingTokens()); + } + + @Test + void refillRestoresTokens() throws InterruptedException { + RateLimitStore store = new InProcessRateLimitStore(); + // Capacity 2 with smooth refill over 100 ms -> ~1 token per 50 ms. + for (int i = 0; i < 2; i++) { + assertTrue(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed()); + } + assertFalse(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed()); + Thread.sleep(150); + assertTrue(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed()); + } + + @Test + void deniedConsumeReportsWaitNanos() { + RateLimitStore store = new InProcessRateLimitStore(); + assertTrue(store.tryConsume("wait", 1, Duration.ofSeconds(10)).allowed()); + RateLimitDecision denied = store.tryConsume("wait", 1, Duration.ofSeconds(10)); + assertFalse(denied.allowed()); + assertTrue(denied.nanosToWaitForRefill() > 0L); + } +} diff --git a/app/common/src/test/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreTest.java b/app/common/src/test/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreTest.java new file mode 100644 index 000000000..296d012c8 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreTest.java @@ -0,0 +1,42 @@ +package stirling.software.common.cluster.inprocess; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.file.Path; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import stirling.software.common.cluster.FileStore; + +class LocalDiskFileStoreTest { + + @Test + void storeRetrieveSizeDeleteExistsRoundTrip(@TempDir Path dir) throws IOException { + LocalDiskFileStore store = new LocalDiskFileStore(dir.toString()); + byte[] payload = "hello-bytes".getBytes(); + + FileStore.Stored stored = store.store(new ByteArrayInputStream(payload), "x.txt"); + assertEquals(payload.length, stored.size()); + assertTrue(store.exists(stored.fileId())); + assertEquals(payload.length, store.size(stored.fileId())); + assertArrayEquals(payload, store.retrieveBytes(stored.fileId())); + + assertTrue(store.delete(stored.fileId())); + assertFalse(store.exists(stored.fileId())); + } + + @Test + void traversalIdsAreRejected(@TempDir Path dir) { + LocalDiskFileStore store = new LocalDiskFileStore(dir.toString()); + assertThrows(IllegalArgumentException.class, () -> store.resolve("../foo")); + assertThrows(IllegalArgumentException.class, () -> store.resolve("a/b")); + assertThrows(IllegalArgumentException.class, () -> store.resolve("a\\b")); + } +} diff --git a/app/common/src/test/java/stirling/software/common/service/FileStorageDelegationTest.java b/app/common/src/test/java/stirling/software/common/service/FileStorageDelegationTest.java new file mode 100644 index 000000000..a929a15af --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/service/FileStorageDelegationTest.java @@ -0,0 +1,27 @@ +package stirling.software.common.service; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.mockito.Mockito.mock; + +import java.io.IOException; +import java.nio.file.Path; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import stirling.software.common.cluster.inprocess.LocalDiskFileStore; + +class FileStorageDelegationTest { + + @Test + void storeBytesThenRetrieveBytesRoundTripsThroughFileStore(@TempDir Path tempDir) + throws IOException { + FileStorage fs = + new FileStorage( + mock(FileOrUploadService.class), + new LocalDiskFileStore(tempDir.toString())); + byte[] payload = "round-trip".getBytes(); + String id = fs.storeBytes(payload, "x.bin"); + assertArrayEquals(payload, fs.retrieveBytes(id)); + } +} diff --git a/app/common/src/test/java/stirling/software/common/service/FileStorageTest.java b/app/common/src/test/java/stirling/software/common/service/FileStorageTest.java index 5b06747a5..ace0dfa56 100644 --- a/app/common/src/test/java/stirling/software/common/service/FileStorageTest.java +++ b/app/common/src/test/java/stirling/software/common/service/FileStorageTest.java @@ -3,6 +3,7 @@ package stirling.software.common.service; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.Mockito.*; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; @@ -13,29 +14,30 @@ import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; -import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.MockitoAnnotations; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; import org.springframework.http.MediaType; -import org.springframework.test.util.ReflectionTestUtils; import org.springframework.web.multipart.MultipartFile; +import stirling.software.common.cluster.inprocess.LocalDiskFileStore; + class FileStorageTest { @TempDir Path tempDir; @Mock private FileOrUploadService fileOrUploadService; - @InjectMocks private FileStorage fileStorage; + private FileStorage fileStorage; private MultipartFile mockFile; @BeforeEach - void setUp() { + void setUp() throws IOException { MockitoAnnotations.openMocks(this); - ReflectionTestUtils.setField(fileStorage, "tempDirPath", tempDir.toString()); + fileStorage = + new FileStorage(fileOrUploadService, new LocalDiskFileStore(tempDir.toString())); // Create a mock MultipartFile mockFile = mock(MultipartFile.class); @@ -47,17 +49,7 @@ class FileStorageTest { void testStoreFile() throws IOException { // Arrange byte[] fileContent = "Test PDF content".getBytes(); - when(mockFile.getBytes()).thenReturn(fileContent); - - // Set up mock to handle transferTo by writing the file - doAnswer( - invocation -> { - java.io.File file = invocation.getArgument(0); - Files.write(file.toPath(), fileContent); - return null; - }) - .when(mockFile) - .transferTo(any(java.io.File.class)); + when(mockFile.getInputStream()).thenReturn(new ByteArrayInputStream(fileContent)); // Act String fileId = fileStorage.storeFile(mockFile); @@ -65,7 +57,7 @@ class FileStorageTest { // Assert assertNotNull(fileId); assertTrue(Files.exists(tempDir.resolve(fileId))); - verify(mockFile).transferTo(any(java.io.File.class)); + assertArrayEquals(fileContent, Files.readAllBytes(tempDir.resolve(fileId))); } @Test @@ -247,11 +239,11 @@ class FileStorageTest { filesBefore = s.count(); } - // Act + Assert: IOException must propagate out — not be swallowed. + // Act + Assert: IOException must propagate out - not be swallowed. assertThrows( IOException.class, () -> fileStorage.storeFromResource(flakyResource, "n.pdf")); - // Assert: no partial file lingers under the storage directory — the finally + // Assert: no partial file lingers under the storage directory - the finally // branch's deleteIfExists must have cleaned it up. long filesAfter; try (Stream s = Files.list(tempDir)) { diff --git a/app/common/src/test/java/stirling/software/common/service/TaskManagerJobStoreDelegationTest.java b/app/common/src/test/java/stirling/software/common/service/TaskManagerJobStoreDelegationTest.java new file mode 100644 index 000000000..1316c3c13 --- /dev/null +++ b/app/common/src/test/java/stirling/software/common/service/TaskManagerJobStoreDelegationTest.java @@ -0,0 +1,120 @@ +package stirling.software.common.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +import java.time.LocalDateTime; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.test.util.ReflectionTestUtils; + +import stirling.software.common.cluster.ClusterBackplane; +import stirling.software.common.cluster.JobStoreEntry; +import stirling.software.common.cluster.JobStoreEntry.JobState; +import stirling.software.common.cluster.inprocess.InProcessClusterBackplane; +import stirling.software.common.cluster.inprocess.InProcessJobStore; +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.job.JobResult; + +class TaskManagerJobStoreDelegationTest { + + @Mock private FileStorage fileStorage; + + private InProcessJobStore jobStore; + private ClusterBackplane backplane; + private TaskManager taskManager; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + jobStore = spy(new InProcessJobStore()); + backplane = new InProcessClusterBackplane(new ApplicationProperties()); + taskManager = new TaskManager(fileStorage, jobStore, backplane); + ReflectionTestUtils.setField(taskManager, "jobResultExpiryMinutes", 30); + } + + @Test + void createTaskWritesPendingEntry() { + taskManager.createTask("job-1"); + JobStoreEntry entry = jobStore.get("job-1").orElseThrow(); + assertEquals(JobState.PENDING, entry.state()); + assertEquals(backplane.localNodeId(), entry.owningNodeId()); + } + + @Test + void setCompleteFlipsToComplete() { + taskManager.createTask("job-2"); + taskManager.setResult("job-2", "ok"); + taskManager.setComplete("job-2"); + JobStoreEntry entry = jobStore.get("job-2").orElseThrow(); + assertEquals(JobState.COMPLETE, entry.state()); + } + + @Test + void setErrorFlipsToFailed() { + taskManager.createTask("job-3"); + taskManager.setError("job-3", "boom"); + JobStoreEntry entry = jobStore.get("job-3").orElseThrow(); + assertEquals(JobState.FAILED, entry.state()); + assertEquals("boom", entry.error()); + } + + @Test + void cleanupOldJobsIsNoopWhenBackplaneIsNotInProcess() { + ClusterBackplane mockedValkeyBackplane = + new ClusterBackplane() { + @Override + public boolean isHealthy() { + return true; + } + + @Override + public String backplaneType() { + return "valkey"; + } + + @Override + public String localNodeId() { + return "node-1"; + } + + @Override + public boolean shouldRunLocalCleanup() { + return false; + } + }; + TaskManager tm = new TaskManager(fileStorage, jobStore, mockedValkeyBackplane); + ReflectionTestUtils.setField(tm, "jobResultExpiryMinutes", 30); + tm.createTask("job-4"); + tm.setComplete("job-4"); + ageJobPastExpiry(tm, "job-4"); + tm.cleanupOldJobs(); + // cleanup must short-circuit before touching jobStore in cluster mode; the backplane + // TTL owns expiry there. If the gate fired correctly, delete is never called. + verify(jobStore, never()).delete(any()); + } + + @Test + void cleanupOldJobsDeletesFromJobStoreWhenBackplaneIsInProcess() { + taskManager.createTask("job-5"); + taskManager.setComplete("job-5"); + ageJobPastExpiry(taskManager, "job-5"); + taskManager.cleanupOldJobs(); + verify(jobStore).delete("job-5"); + } + + @SuppressWarnings("unchecked") + private static void ageJobPastExpiry(TaskManager tm, String jobId) { + var jobResults = + (java.util.Map) ReflectionTestUtils.getField(tm, "jobResults"); + JobResult result = jobResults.get(jobId); + ReflectionTestUtils.setField(result, "completedAt", LocalDateTime.now().minusHours(2)); + ReflectionTestUtils.setField(result, "complete", true); + } +} diff --git a/app/common/src/test/java/stirling/software/common/service/TaskManagerTest.java b/app/common/src/test/java/stirling/software/common/service/TaskManagerTest.java index 7a61270d1..9d880d345 100644 --- a/app/common/src/test/java/stirling/software/common/service/TaskManagerTest.java +++ b/app/common/src/test/java/stirling/software/common/service/TaskManagerTest.java @@ -1,20 +1,28 @@ package stirling.software.common.service; import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.*; import java.time.LocalDateTime; import java.util.Map; +import java.util.Optional; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.MockitoAnnotations; import org.springframework.http.MediaType; import org.springframework.test.util.ReflectionTestUtils; +import stirling.software.common.cluster.ClusterBackplane; +import stirling.software.common.cluster.JobStore; +import stirling.software.common.cluster.JobStoreEntry; +import stirling.software.common.cluster.JobStoreEntry.JobState; import stirling.software.common.model.job.JobResult; import stirling.software.common.model.job.JobStats; import stirling.software.common.model.job.ResultFile; @@ -22,6 +30,8 @@ import stirling.software.common.model.job.ResultFile; class TaskManagerTest { @Mock private FileStorage fileStorage; + @Mock private JobStore jobStore; + @Mock private ClusterBackplane clusterBackplane; @InjectMocks private TaskManager taskManager; @@ -30,6 +40,10 @@ class TaskManagerTest { @BeforeEach void setUp() { closeable = MockitoAnnotations.openMocks(this); + // Treat the backplane as in-process so cleanupOldJobs is not short-circuited. + lenient().when(clusterBackplane.backplaneType()).thenReturn("inprocess"); + lenient().when(clusterBackplane.localNodeId()).thenReturn("test-node"); + lenient().when(clusterBackplane.shouldRunLocalCleanup()).thenReturn(true); ReflectionTestUtils.setField(taskManager, "jobResultExpiryMinutes", 30); } @@ -270,6 +284,33 @@ class TaskManagerTest { verify(fileStorage).deleteFile("file-id"); } + @Test + void testCleanupOldJobs_NoOpWhenBackplaneOwnsExpiry() { + // When the backplane reports it should NOT run local cleanup (e.g. a distributed + // backplane with its own TTL), the cleanup loop must leave local state untouched. + when(clusterBackplane.shouldRunLocalCleanup()).thenReturn(false); + + // Seed an old completed job that would normally be removed. + String oldJobId = "old-job-distributed"; + taskManager.createTask(oldJobId); + JobResult oldJob = taskManager.getJobResult(oldJobId); + ReflectionTestUtils.setField(oldJob, "completedAt", LocalDateTime.now().minusHours(1)); + ReflectionTestUtils.setField(oldJob, "complete", true); + + Map jobResultsMap = + (Map) ReflectionTestUtils.getField(taskManager, "jobResults"); + assertNotNull(jobResultsMap); + assertTrue(jobResultsMap.containsKey(oldJobId)); + + // Act + taskManager.cleanupOldJobs(); + + // Assert: nothing was removed locally, and no jobStore.delete was issued. + assertTrue(jobResultsMap.containsKey(oldJobId)); + verify(jobStore, never()).delete(anyString()); + verify(fileStorage, never()).deleteFile(anyString()); + } + @Test void testShutdown() { // This mainly tests that the shutdown method doesn't throw exceptions @@ -310,4 +351,33 @@ class TaskManagerTest { // Assert assertFalse(result); } + + @Test + void testWriteThroughOnUpdate() { + // Mutating calls must write through to the injected JobStore. + String jobId = "write-through-job"; + taskManager.createTask(jobId); + taskManager.setResult(jobId, "done"); + + ArgumentCaptor captor = ArgumentCaptor.forClass(JobStoreEntry.class); + verify(jobStore, atLeast(2)).put(captor.capture(), any()); + + JobStoreEntry last = captor.getValue(); + assertEquals(jobId, last.jobId()); + assertEquals(JobState.COMPLETE, last.state()); + assertEquals("test-node", last.owningNodeId()); + } + + @Test + void testFindJobKeyByFileId_FallsBackToJobStore() { + // When the file id is not in the local map, TaskManager delegates to JobStore. + String fileId = "remote-file-id"; + String expectedJobKey = "remote-job-key"; + when(jobStore.findJobIdByFileId(fileId)).thenReturn(Optional.of(expectedJobKey)); + + String actual = taskManager.findJobKeyByFileId(fileId); + + assertEquals(expectedJobKey, actual); + verify(jobStore).findJobIdByFileId(fileId); + } } diff --git a/app/core/src/main/resources/settings.yml.template b/app/core/src/main/resources/settings.yml.template index 2540c2f51..b6eca4332 100644 --- a/app/core/src/main/resources/settings.yml.template +++ b/app/core/src/main/resources/settings.yml.template @@ -330,6 +330,22 @@ aiEngine: url: http://localhost:5001 # URL of the Python AI engine timeoutSeconds: 120 # Timeout in seconds for AI engine requests +# Cluster configuration. NOT YET ENABLED - scaffolding for later work. Leave at defaults. +cluster: + enabled: false # Master switch. 'false' (default) wires the in-process backplane and skips all cluster checks. Single-instance installs do not need to change anything here. + backplane: inprocess # Backplane implementation: 'inprocess' (single JVM only) or 'valkey' (multi-node via Valkey/Redis) + artifactStore: local # Transient cluster job-artifact backend: 'local' (per-node disk; single-node only) or 's3' (shared object store; required for multi-node). Distinct from 'storage.provider' which controls persistent user uploads - when both are 's3' they share the storage.s3.* credentials block. Multi-node deployments MUST set this to 's3'. + valkey: + url: "" # Valkey/Redis URL, e.g. 'redis://valkey:6379' or 'rediss://...' for TLS. Required when enabled=true and backplane=valkey. + tls: + skipCertVerification: false # set to 'true' to skip TLS certificate verification on Valkey connections (dev/test only) + node: + id: "" # Optional explicit node id. Blank = auto-generated UUID at startup. + role: both # 'web' (serves HTTP), 'worker' (runs jobs), or 'both' (default) + internalAddress: "" # host:port advertised in the instance registry for peer-to-peer cluster traffic. Blank = derived at startup. + scheme: http # 'http' or 'https' - scheme peers use to call this node's /internal/cluster/** endpoints + heartbeatIntervalMs: 5000 # Heartbeat publish interval for the instance registry (ms) + pdfEditor: fallback-font: classpath:/static/fonts/NotoSans-Regular.ttf # Override to point at a custom fallback font cache: