Extract concurrency utilities, add multithreaded tests, fix two races

candybox-admin-api/src/main/java/me/predatorray/candybox/admin/MetricsScraper.java

@@ -20,18 +20,16 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.time.Duration;
-import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Deque;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import me.predatorray.candybox.common.concurrent.KeyedSlidingWindow;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,7 +61,7 @@ final class MetricsScraper implements AutoCloseable {
     private final long pollIntervalMillis;
     private final HttpClient http;
     private final ScheduledExecutorService scheduler;
-    private final Map<SeriesKey, Deque<Sample>> series = new ConcurrentHashMap<>();
+    private final KeyedSlidingWindow<SeriesKey, Sample> series;
     private final String scrapeToken;
     private volatile String latestText = "";
 
@@ -78,6 +76,7 @@ final class MetricsScraper implements AutoCloseable {
         this.targets = List.copyOf(targets);
         this.pollIntervalMillis = pollIntervalMillis;
         this.windowSamples = windowSamples;
+        this.series = new KeyedSlidingWindow<>(windowSamples);
         this.http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)).build();
         this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
             Thread t = new Thread(r, "admin-metrics-scraper");
@@ -116,17 +115,14 @@ String latestText() {
      */
     Map<String, List<Series>> seriesFor(Collection<String> names) {
         Map<String, List<Series>> out = new LinkedHashMap<>();
+        Map<SeriesKey, List<Sample>> snapshot = series.snapshot();
         for (String name : names) {
             List<Series> matches = new ArrayList<>();
-            for (Map.Entry<SeriesKey, Deque<Sample>> e : series.entrySet()) {
+            for (Map.Entry<SeriesKey, List<Sample>> e : snapshot.entrySet()) {
                 if (!e.getKey().name.equals(name)) {
                     continue;
                 }
-                List<Sample> snapshot;
-                synchronized (e.getValue()) {
-                    snapshot = new ArrayList<>(e.getValue());
-                }
-                matches.add(new Series(e.getKey().name, e.getKey().labels, snapshot));
+                matches.add(new Series(e.getKey().name, e.getKey().labels, e.getValue()));
             }
             out.put(name, matches);
         }
@@ -173,14 +169,7 @@ private void scrapeOnce() {
 
     private void ingest(String text, long ingestMillis) {
         for (PrometheusText.Sample s : PrometheusText.parse(text)) {
-            SeriesKey key = new SeriesKey(s.name(), s.labels());
-            Deque<Sample> window = series.computeIfAbsent(key, k -> new ArrayDeque<>(windowSamples));
-            synchronized (window) {
-                window.addLast(new Sample(ingestMillis, s.value()));
-                while (window.size() > windowSamples) {
-                    window.removeFirst();
-                }
-            }
+            series.append(new SeriesKey(s.name(), s.labels()), new Sample(ingestMillis, s.value()));
         }
     }
 

candybox-client/src/main/java/me/predatorray/candybox/client/ClusterRouter.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import me.predatorray.candybox.common.Clock;
+import me.predatorray.candybox.common.concurrent.TtlCache;
 import me.predatorray.candybox.common.exception.CandyboxException;
 import me.predatorray.candybox.common.exception.NotOwnerException;
 import me.predatorray.candybox.coordination.CandyboxKeys;
@@ -44,19 +45,16 @@ final class ClusterRouter implements Router {
 
     private final Transport transport;
     private final CoordinationService coordination;
-    private final long cacheTtlMillis;
-    private final Clock clock;
     private final MessageCodec codec = new MessageCodec();
 
-    private final ConcurrentMap<String, CachedAddress> partitionCache = new ConcurrentHashMap<>();
+    private final TtlCache<String, NodeAddress> partitionCache;
     private final ConcurrentMap<String, Connection> connections = new ConcurrentHashMap<>();
 
     ClusterRouter(Transport transport, CoordinationService coordination, long cacheTtlMillis,
                   Clock clock) {
         this.transport = transport;
         this.coordination = coordination;
-        this.cacheTtlMillis = cacheTtlMillis;
-        this.clock = clock;
+        this.partitionCache = new TtlCache<>(clock, cacheTtlMillis);
     }
 
     @Override
@@ -67,8 +65,7 @@ public Message callPartition(String box, int partition, Message request) {
             Message response = send(address, request);
             if (response instanceof Message.MovedResponse moved) {
                 address = addressOfNode(moved.ownerNodeId());
-                partitionCache.put(cacheKey,
-                        new CachedAddress(address, clock.currentTimeMillis() + cacheTtlMillis));
+                partitionCache.put(cacheKey, address);
                 continue;
             }
             return response;
@@ -83,16 +80,15 @@ public Message callAny(Message request) {
     }
 
     private NodeAddress resolveOwner(String box, int partition, String cacheKey) {
-        CachedAddress cached = partitionCache.get(cacheKey);
-        if (cached != null && clock.currentTimeMillis() < cached.expiry()) {
-            return cached.address();
+        NodeAddress cached = partitionCache.getIfFresh(cacheKey).orElse(null);
+        if (cached != null) {
+            return cached;
         }
         LeaseInfo holder = coordination.leaseHolder(CandyboxKeys.ownerResource(box, partition))
                 .orElseThrow(() -> new NotOwnerException("box " + box + " partition " + partition
                         + " has no current owner"));
         NodeAddress address = addressOfNode(holder.ownerNodeId());
-        partitionCache.put(cacheKey,
-                new CachedAddress(address, clock.currentTimeMillis() + cacheTtlMillis));
+        partitionCache.put(cacheKey, address);
         return address;
     }
 
@@ -111,8 +107,7 @@ private NodeAddress anyMember() {
     }
 
     private Message send(NodeAddress address, Message request) {
-        Connection connection = connections.computeIfAbsent(address.key(),
-                k -> transport.connect(address.host(), address.port()));
+        Connection connection = acquireConnection(address);
         try {
             Frame response = connection.call(codec.encode(request));
             return codec.decode(response);
@@ -130,6 +125,33 @@ private Message send(NodeAddress address, Message request) {
         }
     }
 
+    /**
+     * Returns a pooled connection to {@code address}, opening one on a miss. The blocking
+     * {@link Transport#connect} runs <em>outside</em> the connection map's locks: doing it inside
+     * {@link ConcurrentMap#computeIfAbsent} would hold a bin lock across network I/O (stalling
+     * unrelated keys that hash to the same bin) and is forbidden from updating the map reentrantly.
+     * The cost of this lock-free approach is that two callers racing on a cold address may both
+     * connect; the loser's surplus connection is closed and the winner's is shared.
+     */
+    private Connection acquireConnection(NodeAddress address) {
+        String key = address.key();
+        Connection existing = connections.get(key);
+        if (existing != null) {
+            return existing;
+        }
+        Connection fresh = transport.connect(address.host(), address.port());
+        Connection raced = connections.putIfAbsent(key, fresh);
+        if (raced != null) {
+            try {
+                fresh.close();
+            } catch (RuntimeException ignored) {
+                // best effort: discard the connection that lost the race
+            }
+            return raced;
+        }
+        return fresh;
+    }
+
     @Override
     public void close() {
         for (Connection connection : connections.values()) {
@@ -156,7 +178,4 @@ static NodeAddress parse(String hostPort) {
                     Integer.parseInt(hostPort.substring(colon + 1)));
         }
     }
-
-    private record CachedAddress(NodeAddress address, long expiry) {
-    }
 }

candybox-client/src/test/java/me/predatorray/candybox/client/ClusterRouterConcurrencyTest.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright (c) 2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package me.predatorray.candybox.client;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import me.predatorray.candybox.common.SystemClock;
+import me.predatorray.candybox.coordination.CandyboxKeys;
+import me.predatorray.candybox.coordination.fake.InMemoryCoordinationService;
+import me.predatorray.candybox.protocol.Frame;
+import me.predatorray.candybox.protocol.Message;
+import me.predatorray.candybox.protocol.MessageCodec;
+import me.predatorray.candybox.protocol.transport.Connection;
+import me.predatorray.candybox.protocol.transport.Transport;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Guards the fix for the blocking-{@code computeIfAbsent} defect: {@link ClusterRouter} must open
+ * connections <em>outside</em> the connection map's locks. If {@code connect()} ran inside
+ * {@link java.util.concurrent.ConcurrentMap#computeIfAbsent}, two callers racing on the same cold
+ * address could never have their connects in flight at the same time (the second blocks on the bin
+ * lock the first holds), and the barrier below would never trip.
+ */
+class ClusterRouterConcurrencyTest {
+
+    private static final MessageCodec CODEC = new MessageCodec();
+
+    @Test
+    void connectsForTheSameAddressAreNotSerializedByThePoolLock() throws Exception {
+        InMemoryCoordinationService coordination = new InMemoryCoordinationService();
+        coordination.registerMember(2, "127.0.0.1:2002".getBytes(StandardCharsets.UTF_8));
+        coordination.tryAcquireLease(CandyboxKeys.ownerResource("b", 0), 2, 60_000);
+
+        // connect() parks on a 2-party barrier: it can only complete if BOTH callers are inside
+        // connect() simultaneously — i.e. neither is blocked behind a map lock held by the other.
+        CyclicBarrier bothConnecting = new CyclicBarrier(2);
+        AtomicInteger connectStarts = new AtomicInteger();
+        AtomicInteger closes = new AtomicInteger();
+
+        Transport barrierTransport = new Transport() {
+            @Override
+            public Connection connect(String host, int port) {
+                connectStarts.incrementAndGet();
+                try {
+                    bothConnecting.await(5, TimeUnit.SECONDS);
+                } catch (TimeoutException | BrokenBarrierException e) {
+                    throw new AssertionError("connects were serialized — connect() ran under a lock", e);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(e);
+                }
+                return new Connection() {
+                    @Override
+                    public Frame call(Frame request) {
+                        return CODEC.encode(new Message.OkResponse());
+                    }
+
+                    @Override
+                    public void close() {
+                        closes.incrementAndGet();
+                    }
+                };
+            }
+
+            @Override
+            public void close() {
+            }
+        };
+
+        try (ClusterRouter router = new ClusterRouter(barrierTransport, coordination, 60_000,
+                SystemClock.INSTANCE)) {
+            CyclicBarrier start = new CyclicBarrier(2);
+            Runnable call = () -> {
+                try {
+                    start.await();
+                } catch (InterruptedException | BrokenBarrierException e) {
+                    throw new RuntimeException(e);
+                }
+                Message resp = router.callPartition("b", 0, new Message.GetCandyRequest("b", "k"));
+                assertThat(resp).isInstanceOf(Message.OkResponse.class);
+            };
+            Thread a = new Thread(call, "router-a");
+            Thread b = new Thread(call, "router-b");
+            a.start();
+            b.start();
+            a.join(10_000);
+            b.join(10_000);
+            assertThat(a.isAlive()).as("thread a finished").isFalse();
+            assertThat(b.isAlive()).as("thread b finished").isFalse();
+
+            // Both callers connected concurrently (the barrier tripped), and exactly one surplus
+            // connection — the loser of the putIfAbsent race — was closed.
+            assertThat(connectStarts).hasValue(2);
+            assertThat(closes).hasValue(1);
+
+            // The pooled connection is reused on a later call: no new connect.
+            router.callPartition("b", 0, new Message.GetCandyRequest("b", "k"));
+            assertThat(connectStarts).hasValue(2);
+        }
+    }
+}

candybox-common/src/main/java/me/predatorray/candybox/common/auth/FileCredentialStore.java

@@ -25,6 +25,8 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import me.predatorray.candybox.common.SystemClock;
+import me.predatorray.candybox.common.concurrent.PeriodicGate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,12 +59,12 @@ public final class FileCredentialStore implements CredentialStore, S3KeyStore {
 
     private final Path file;
     private volatile Snapshot snapshot;
-    private volatile long nextRecheckAtMillis;
+    private final PeriodicGate recheckGate;
 
     public FileCredentialStore(Path file) {
         this.file = file;
         this.snapshot = Snapshot.load(file);
-        this.nextRecheckAtMillis = System.currentTimeMillis() + RECHECK_INTERVAL_MILLIS;
+        this.recheckGate = new PeriodicGate(SystemClock.INSTANCE, RECHECK_INTERVAL_MILLIS);
     }
 
     @Override
@@ -86,15 +88,8 @@ public Optional<S3Key> s3Key(String accessKeyId) {
     }
 
     private Snapshot current() {
-        long now = System.currentTimeMillis();
-        if (now >= nextRecheckAtMillis) {
-            synchronized (this) {
-                if (now >= nextRecheckAtMillis) {
-                    nextRecheckAtMillis = now + RECHECK_INTERVAL_MILLIS;
-                    reloadIfChanged();
-                }
-            }
-        }
+        // Re-stat the file at most once per interval (off the hot read path) and reload on change.
+        recheckGate.runIfDue(this::reloadIfChanged);
         return snapshot;
     }