Add cluster backplane abstraction and interfaces (#6449)

2026-07-15 11:00:47 +02:00 · 2026-05-27 11:54:59 +01:00
parent 7743720f0a
commit 4564ed5bec
40 changed files with 2167 additions and 247 deletions
@@ -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;
+    }
+}
@@ -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.
+ *
+ * <p>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());
+    }
+}
@@ -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) {}
@@ -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<LockHandle> tryAcquire(String lockKey, Duration leaseTime);
+
+    interface LockHandle extends AutoCloseable {
+        void release();
+
+        boolean renew(Duration leaseTime);
+
+        @Override
+        default void close() {
+            release();
+        }
+    }
+}
@@ -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.
+     *
+     * <p>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);
+}
@@ -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<ClusterNode> lookup(String nodeId);
+
+    Collection<ClusterNode> activeNodes();
+
+    void deregister(String nodeId);
+}
@@ -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<JobStoreEntry> get(String jobId);
+
+    void delete(String jobId);
+
+    boolean exists(String jobId);
+
+    /** Reverse lookup: which job owns this result file id? */
+    Optional<String> findJobIdByFileId(String fileId);
+
+    /** Snapshot of every active entry. Used by admin/stats endpoints; may be O(n). */
+    Collection<JobStoreEntry> all();
+}
@@ -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<String> fileIds,
+        Map<String, String> resultMeta) {
+
+    /** Lifecycle states for a job as observed by the cluster. */
+    public enum JobState {
+        PENDING,
+        RUNNING,
+        COMPLETE,
+        FAILED
+    }
+}
@@ -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<String> get(String namespace, String key);
+
+    void evict(String namespace, String key);
+
+    void evictNamespace(String namespace);
+}
@@ -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) {}
+}
@@ -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();
+}
@@ -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();
+    }
+}
@@ -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();
+    }
+}
@@ -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.
+ *
+ * <p>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.
+ *
+ * <p>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<String, LockState> 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<LockHandle> 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);
+        }
+    }
+}
@@ -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<ClusterNode> self = new AtomicReference<>();
+
+    @Override
+    public void register(ClusterNode node, Duration heartbeatTtl) {
+        self.set(node);
+    }
+
+    @Override
+    public Optional<ClusterNode> lookup(String nodeId) {
+        ClusterNode current = self.get();
+        return current != null && current.nodeId().equals(nodeId)
+                ? Optional.of(current)
+                : Optional.empty();
+    }
+
+    @Override
+    public Collection<ClusterNode> 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);
+        }
+    }
+}
@@ -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<String, Holder> 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<JobStoreEntry> 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<String> findJobIdByFileId(String fileId) {
+        for (Map.Entry<String, Holder> e : entries.entrySet()) {
+            Holder h = e.getValue();
+            if (h.isExpired()) {
+                continue;
+            }
+            List<String> fileIds = h.entry.fileIds();
+            if (fileIds != null && fileIds.contains(fileId)) {
+                return Optional.of(e.getKey());
+            }
+        }
+        return Optional.empty();
+    }
+
+    @Override
+    public Collection<JobStoreEntry> all() {
+        List<JobStoreEntry> 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<String, Holder> 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());
+        }
+    }
+}
@@ -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<String, ConcurrentHashMap<String, Expiring>> 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<String> get(String namespace, String key) {
+        Map<String, Expiring> 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<String, Expiring> 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) {}
+}
@@ -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<String, Bucket> buckets =
+            Collections.synchronizedMap(
+                    new LinkedHashMap<String, Bucket>(256, 0.75f, true) {
+                        @Override
+                        protected boolean removeEldestEntry(Map.Entry<String, Bucket> 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();
+    }
+}
@@ -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;
+    }
+}
@@ -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);
+    }
+}
@@ -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}.
+         *
+         * <p>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
@@ -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<Throwable> 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<Resource>} 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);
    }
 }
@@ -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<ResultFile> 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<String> fileIds = new ArrayList<>();
+        if (result.hasFiles()) {
+            for (ResultFile rf : result.getAllResultFiles()) {
+                fileIds.add(rf.getFileId());
+            }
+        }
+        Map<String, String> 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;
    }
 }