diff --git a/candybox-admin-api/src/main/java/me/predatorray/candybox/admin/MetricsScraper.java b/candybox-admin-api/src/main/java/me/predatorray/candybox/admin/MetricsScraper.java index ce676e5..d711405 100644 --- a/candybox-admin-api/src/main/java/me/predatorray/candybox/admin/MetricsScraper.java +++ b/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> series = new ConcurrentHashMap<>(); + private final KeyedSlidingWindow 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> seriesFor(Collection names) { Map> out = new LinkedHashMap<>(); + Map> snapshot = series.snapshot(); for (String name : names) { List matches = new ArrayList<>(); - for (Map.Entry> e : series.entrySet()) { + for (Map.Entry> e : snapshot.entrySet()) { if (!e.getKey().name.equals(name)) { continue; } - List 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 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())); } } diff --git a/candybox-client/src/main/java/me/predatorray/candybox/client/ClusterRouter.java b/candybox-client/src/main/java/me/predatorray/candybox/client/ClusterRouter.java index a9d0c63..ab401a4 100644 --- a/candybox-client/src/main/java/me/predatorray/candybox/client/ClusterRouter.java +++ b/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 partitionCache = new ConcurrentHashMap<>(); + private final TtlCache partitionCache; private final ConcurrentMap 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 outside 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) { - } } diff --git a/candybox-client/src/test/java/me/predatorray/candybox/client/ClusterRouterConcurrencyTest.java b/candybox-client/src/test/java/me/predatorray/candybox/client/ClusterRouterConcurrencyTest.java new file mode 100644 index 0000000..25d8da9 --- /dev/null +++ b/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 outside 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); + } + } +} diff --git a/candybox-common/src/main/java/me/predatorray/candybox/common/auth/FileCredentialStore.java b/candybox-common/src/main/java/me/predatorray/candybox/common/auth/FileCredentialStore.java index ee0ae7d..4ecd2a4 100644 --- a/candybox-common/src/main/java/me/predatorray/candybox/common/auth/FileCredentialStore.java +++ b/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(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; } diff --git a/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/BoundedLruCache.java b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/BoundedLruCache.java new file mode 100644 index 0000000..3e83d45 --- /dev/null +++ b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/BoundedLruCache.java @@ -0,0 +1,103 @@ +/* + * 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.common.concurrent; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * A thread-safe, fixed-capacity cache that evicts the least-recently-used entry once it is full. + * + *

This is the business-logic-free extraction of the access-ordered {@link LinkedHashMap} idiom + * (the {@code removeEldestEntry} override) that callers previously hand-rolled and wrapped in + * {@link java.util.Collections#synchronizedMap}. Access order makes both {@link #get} and + * {@link #put} structural modifications, so every operation is serialized on the instance lock — + * unlike a plain {@code synchronizedMap}, reads here genuinely need the lock too, which is exactly + * why the locking is encapsulated rather than left to the caller. + * + *

Null keys and null values are rejected so {@link #get} returning {@code null} unambiguously + * means "absent". + * + * @param key type + * @param value type + */ +public final class BoundedLruCache { + + private final int capacity; + private final LinkedHashMap map; + + /** + * @param capacity the maximum number of entries to retain; must be positive + */ + public BoundedLruCache(int capacity) { + if (capacity <= 0) { + throw new IllegalArgumentException("capacity must be positive: " + capacity); + } + this.capacity = capacity; + // access-order=true so iteration/eviction order tracks recency of use. + this.map = new LinkedHashMap<>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() > BoundedLruCache.this.capacity; + } + }; + } + + /** The configured maximum number of entries. */ + public int capacity() { + return capacity; + } + + /** + * Returns the value cached under {@code key}, marking it most-recently-used, or {@code null} if + * absent. + */ + public synchronized V get(K key) { + return map.get(requireNonNull(key, "key")); + } + + /** + * Stores {@code value} under {@code key} as the most-recently-used entry, evicting the + * least-recently-used entry if the cache is at capacity. + * + * @return the previous value mapped to {@code key}, or {@code null} if there was none + */ + public synchronized V put(K key, V value) { + return map.put(requireNonNull(key, "key"), requireNonNull(value, "value")); + } + + /** Whether {@code key} currently has a cached value (and marks it most-recently-used). */ + public synchronized boolean containsKey(K key) { + return map.get(requireNonNull(key, "key")) != null; + } + + /** The current number of cached entries (never exceeds {@link #capacity()}). */ + public synchronized int size() { + return map.size(); + } + + /** Removes all entries. */ + public synchronized void clear() { + map.clear(); + } + + private static T requireNonNull(T value, String what) { + if (value == null) { + throw new NullPointerException(what + " must not be null"); + } + return value; + } +} diff --git a/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindow.java b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindow.java new file mode 100644 index 0000000..bb73345 --- /dev/null +++ b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindow.java @@ -0,0 +1,108 @@ +/* + * 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.common.concurrent; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * A thread-safe store of per-key bounded sliding windows: each key maps to the most recent + * {@code windowSize} appended values, oldest evicted first. This is the business-logic-free + * extraction of the {@code ConcurrentHashMap>} guarded by {@code synchronized(deque)} + * idiom used to buffer rolling-window samples. + * + *

Concurrency contract: the outer map is a {@link ConcurrentHashMap}, and each per-key deque is + * its own monitor — every append, eviction and snapshot of a given key's window happens while holding + * that deque's intrinsic lock, so a snapshot never observes a torn window and a CME is impossible. + * Different keys never contend. + * + * @param key type + * @param value type stored in each window + */ +public final class KeyedSlidingWindow { + + private final int windowSize; + private final ConcurrentMap> windows = new ConcurrentHashMap<>(); + + /** + * @param windowSize the maximum number of values retained per key; must be positive + */ + public KeyedSlidingWindow(int windowSize) { + if (windowSize <= 0) { + throw new IllegalArgumentException("windowSize must be positive: " + windowSize); + } + this.windowSize = windowSize; + } + + /** The configured per-key window capacity. */ + public int windowSize() { + return windowSize; + } + + /** + * Appends {@code value} to {@code key}'s window, evicting the oldest values until the window is + * within capacity. + */ + public void append(K key, V value) { + Deque window = windows.computeIfAbsent(key, k -> new ArrayDeque<>(windowSize)); + synchronized (window) { + window.addLast(value); + while (window.size() > windowSize) { + window.removeFirst(); + } + } + } + + /** + * Returns an oldest-to-newest snapshot of {@code key}'s window, or an empty list if the key has no + * values. + */ + public List snapshot(K key) { + Deque window = windows.get(key); + if (window == null) { + return List.of(); + } + synchronized (window) { + return new ArrayList<>(window); + } + } + + /** + * Returns a consistent snapshot of every key's window (each list oldest-to-newest), in no + * particular key order. Each window is copied under its own lock. + */ + public Map> snapshot() { + Map> out = new LinkedHashMap<>(); + for (Map.Entry> e : windows.entrySet()) { + Deque window = e.getValue(); + synchronized (window) { + out.put(e.getKey(), new ArrayList<>(window)); + } + } + return out; + } + + /** Removes all windows. */ + public void clear() { + windows.clear(); + } +} diff --git a/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/PeriodicGate.java b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/PeriodicGate.java new file mode 100644 index 0000000..cb6d62a --- /dev/null +++ b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/PeriodicGate.java @@ -0,0 +1,78 @@ +/* + * 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.common.concurrent; + +import me.predatorray.candybox.common.Clock; + +/** + * A throttle that runs an action at most once per fixed interval, no matter how many threads call it + * in between. This is the business-logic-free extraction of the volatile-deadline + double-checked + * lock idiom used to rate-limit periodic side effects (such as re-stat'ing a credentials file) off + * the hot read path. + * + *

The hot path is a single volatile read of the next-due deadline; only when the deadline has + * passed does a caller take the lock, re-check the deadline under it (so concurrent callers do not + * all fire), advance the deadline, and run the action while still holding the lock. Holding + * the lock across the action guarantees two runs never overlap even if an action outlasts its own + * interval, so callers must keep the action short and must not acquire locks that could invert + * ordering with this gate's monitor. + * + *

The deadline is advanced before the action runs, so a throwing action still consumes + * its interval (it will not be retried until the next interval elapses) and the exception propagates + * to the caller that happened to run it. + */ +public final class PeriodicGate { + + private final Clock clock; + private final long intervalMillis; + private volatile long nextDueAtMillis; + + /** + * @param clock the time source governing the interval (injected for deterministic tests) + * @param intervalMillis the minimum spacing between two runs of the action; must be non-negative + * (zero runs the action on every call) + */ + public PeriodicGate(Clock clock, long intervalMillis) { + if (intervalMillis < 0) { + throw new IllegalArgumentException("intervalMillis must be non-negative: " + intervalMillis); + } + this.clock = clock; + this.intervalMillis = intervalMillis; + this.nextDueAtMillis = clock.currentTimeMillis() + intervalMillis; + } + + /** + * Runs {@code action} if at least {@code intervalMillis} has elapsed since the last run, otherwise + * returns without running it. At most one of any number of concurrent callers runs the action per + * interval. + * + * @return {@code true} if this call ran the action, {@code false} if it was throttled + */ + public boolean runIfDue(Runnable action) { + if (clock.currentTimeMillis() < nextDueAtMillis) { + return false; + } + synchronized (this) { + long now = clock.currentTimeMillis(); + if (now < nextDueAtMillis) { + return false; + } + nextDueAtMillis = now + intervalMillis; + action.run(); + return true; + } + } +} diff --git a/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/TtlCache.java b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/TtlCache.java new file mode 100644 index 0000000..4e60355 --- /dev/null +++ b/candybox-common/src/main/java/me/predatorray/candybox/common/concurrent/TtlCache.java @@ -0,0 +1,123 @@ +/* + * 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.common.concurrent; + +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; +import me.predatorray.candybox.common.Clock; + +/** + * A thread-safe read-through cache whose entries expire a fixed time after they are written. This is + * the business-logic-free extraction of the {@code ConcurrentHashMap} idiom + * that several stores hand-rolled to avoid hitting their backing service (ZooKeeper, the cluster) on + * every request. + * + *

Staleness model: an entry is served until {@code clock.currentTimeMillis() >= expiresAt}, after + * which it is treated as absent and reloaded. Expiry is checked lazily on access; expired entries are + * not actively reaped, but a reload overwrites them. + * + *

Loaders run outside the cache's internal locks. {@link #get(Object, Function)} reads the + * (lock-free) map, and only if the entry is missing or stale does it invoke the loader, then publishes + * the result with {@link ConcurrentMap#put}. It deliberately does not wrap the loader in + * {@link ConcurrentMap#computeIfAbsent}, which would run the (potentially blocking, potentially + * remote) loader while holding a bin lock — the very anti-pattern this class exists to avoid. The + * trade-off is that concurrent misses for the same key may each load once; the last write wins. Null + * values are rejected; cache "absence" of a present-but-empty result should be modelled with an + * {@link Optional} value type. + * + * @param key type + * @param value type + */ +public final class TtlCache { + + private final Clock clock; + private final long ttlMillis; + private final ConcurrentMap> entries = new ConcurrentHashMap<>(); + + /** + * @param clock the time source governing expiry (injected for deterministic tests) + * @param ttlMillis how long after a write an entry stays fresh; must be non-negative (zero means + * every entry is immediately stale, i.e. the cache never serves a hit) + */ + public TtlCache(Clock clock, long ttlMillis) { + if (ttlMillis < 0) { + throw new IllegalArgumentException("ttlMillis must be non-negative: " + ttlMillis); + } + this.clock = clock; + this.ttlMillis = ttlMillis; + } + + /** + * Returns the fresh cached value for {@code key}, loading and caching it via {@code loader} on a + * miss or expiry. The loader is invoked without holding any cache lock. + * + * @param loader computes the value for a key on a miss; must not return {@code null} + */ + public V get(K key, Function loader) { + Entry existing = entries.get(key); + long now = clock.currentTimeMillis(); + if (existing != null && now < existing.expiresAtMillis) { + return existing.value; + } + V loaded = loader.apply(key); + if (loaded == null) { + throw new NullPointerException("loader returned null for key " + key); + } + entries.put(key, new Entry<>(loaded, now + ttlMillis)); + return loaded; + } + + /** + * Returns the fresh cached value for {@code key}, or empty if absent or expired. Never loads. + */ + public Optional getIfFresh(K key) { + Entry existing = entries.get(key); + if (existing != null && clock.currentTimeMillis() < existing.expiresAtMillis) { + return Optional.of(existing.value); + } + return Optional.empty(); + } + + /** Stores {@code value} under {@code key}, fresh for the configured TTL from now. */ + public void put(K key, V value) { + if (value == null) { + throw new NullPointerException("value must not be null"); + } + entries.put(key, new Entry<>(value, clock.currentTimeMillis() + ttlMillis)); + } + + /** Drops any cached entry for {@code key} (e.g. after a write the cache must not serve stale). */ + public void invalidate(K key) { + entries.remove(key); + } + + /** Drops every cached entry. */ + public void clear() { + entries.clear(); + } + + private static final class Entry { + final V value; + final long expiresAtMillis; + + Entry(V value, long expiresAtMillis) { + this.value = value; + this.expiresAtMillis = expiresAtMillis; + } + } +} diff --git a/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/BoundedLruCacheTest.java b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/BoundedLruCacheTest.java new file mode 100644 index 0000000..574b5fc --- /dev/null +++ b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/BoundedLruCacheTest.java @@ -0,0 +1,116 @@ +/* + * 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.common.concurrent; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.concurrent.ConcurrentLinkedQueue; +import org.junit.jupiter.api.Test; + +class BoundedLruCacheTest { + + @Test + void rejectsNonPositiveCapacity() { + assertThatThrownBy(() -> new BoundedLruCache<>(0)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void rejectsNullKeyAndValue() { + BoundedLruCache cache = new BoundedLruCache<>(4); + assertThatThrownBy(() -> cache.put(null, "v")).isInstanceOf(NullPointerException.class); + assertThatThrownBy(() -> cache.put("k", null)).isInstanceOf(NullPointerException.class); + assertThatThrownBy(() -> cache.get(null)).isInstanceOf(NullPointerException.class); + } + + @Test + void evictsLeastRecentlyUsedOnceFull() { + BoundedLruCache cache = new BoundedLruCache<>(3); + cache.put(1, "a"); + cache.put(2, "b"); + cache.put(3, "c"); + // Touch 1 so 2 becomes the least-recently-used. + assertThat(cache.get(1)).isEqualTo("a"); + cache.put(4, "d"); // evicts 2 + + assertThat(cache.size()).isEqualTo(3); + assertThat(cache.get(2)).isNull(); + assertThat(cache.get(1)).isEqualTo("a"); + assertThat(cache.get(3)).isEqualTo("c"); + assertThat(cache.get(4)).isEqualTo("d"); + } + + @Test + void overwriteUpdatesValueWithoutGrowing() { + BoundedLruCache cache = new BoundedLruCache<>(2); + cache.put(1, "a"); + assertThat(cache.put(1, "a2")).isEqualTo("a"); + assertThat(cache.size()).isEqualTo(1); + assertThat(cache.get(1)).isEqualTo("a2"); + } + + // ---- multi-threaded ------------------------------------------------------------------------- + + @Test + void neverExceedsCapacityUnderConcurrentDistinctWrites() throws Exception { + int capacity = 64; + BoundedLruCache cache = new BoundedLruCache<>(capacity); + // Many threads hammer a key space far larger than capacity; size must stay bounded and the + // map must never corrupt (no exceptions, no over-capacity reads). + int threads = 8; + int perThread = 20_000; + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + int maxObserved = ConcurrencyTestSupport.runConcurrently(threads, failures, t -> { + int localMax = 0; + for (int i = 0; i < perThread; i++) { + int key = (t * 7 + i) % (capacity * 8); + cache.put(key, i); + if (i % 16 == 0) { + cache.get((t * 3 + i) % (capacity * 8)); + } + localMax = Math.max(localMax, cache.size()); + } + return localMax; + }); + + assertThat(failures).isEmpty(); + assertThat(maxObserved).isLessThanOrEqualTo(capacity); + assertThat(cache.size()).isLessThanOrEqualTo(capacity); + } + + @Test + void concurrentReadersAndWritersOnSharedKeysStayConsistent() throws Exception { + // Access-order LRU mutates structure on get(); if get() were not serialized this would throw + // ConcurrentModificationException or corrupt the linked list. Drive heavy get+put on a small + // shared key set to surface any such race. + BoundedLruCache cache = new BoundedLruCache<>(16); + for (int i = 0; i < 16; i++) { + cache.put(i, i); + } + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + ConcurrencyTestSupport.runConcurrently(10, failures, t -> { + for (int i = 0; i < 50_000; i++) { + int key = i % 16; + cache.get(key); + cache.put(key, i); + } + return 0; + }); + assertThat(failures).isEmpty(); + assertThat(cache.size()).isLessThanOrEqualTo(16); + } +} diff --git a/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/ConcurrencyTestSupport.java b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/ConcurrencyTestSupport.java new file mode 100644 index 0000000..49a809e --- /dev/null +++ b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/ConcurrencyTestSupport.java @@ -0,0 +1,68 @@ +/* + * 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.common.concurrent; + +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Helpers for the concurrency unit tests: spin up worker threads, release them in lockstep so they + * collide on the structure under test, and surface any thrown error to the asserting thread. + */ +final class ConcurrencyTestSupport { + + private ConcurrencyTestSupport() { + } + + @FunctionalInterface + interface Worker { + /** Runs the worker body for {@code threadIndex}, returning a per-thread integer to aggregate. */ + int run(int threadIndex) throws Exception; + } + + /** + * Runs {@code threads} copies of {@code worker}, started together at a {@link CyclicBarrier} to + * maximize contention. Any throwable from a worker is added to {@code failures} (the asserting + * test decides how to react). Returns the maximum value any worker returned. + */ + static int runConcurrently(int threads, Queue failures, Worker worker) + throws InterruptedException { + CyclicBarrier barrier = new CyclicBarrier(threads); + AtomicInteger max = new AtomicInteger(Integer.MIN_VALUE); + List workers = new ArrayList<>(threads); + for (int i = 0; i < threads; i++) { + int index = i; + Thread thread = new Thread(() -> { + try { + barrier.await(); + int result = worker.run(index); + max.accumulateAndGet(result, Math::max); + } catch (Throwable t) { + failures.add(t); + } + }, "concurrency-test-" + i); + workers.add(thread); + } + workers.forEach(Thread::start); + for (Thread thread : workers) { + thread.join(); + } + return max.get(); + } +} diff --git a/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindowTest.java b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindowTest.java new file mode 100644 index 0000000..e7876b0 --- /dev/null +++ b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/KeyedSlidingWindowTest.java @@ -0,0 +1,104 @@ +/* + * 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.common.concurrent; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.junit.jupiter.api.Test; + +class KeyedSlidingWindowTest { + + @Test + void rejectsNonPositiveWindow() { + assertThatThrownBy(() -> new KeyedSlidingWindow<>(0)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void keepsOnlyMostRecentValuesPerKey() { + KeyedSlidingWindow w = new KeyedSlidingWindow<>(3); + for (int i = 1; i <= 5; i++) { + w.append("a", i); + } + w.append("b", 99); + assertThat(w.snapshot("a")).containsExactly(3, 4, 5); + assertThat(w.snapshot("b")).containsExactly(99); + assertThat(w.snapshot("missing")).isEmpty(); + } + + @Test + void snapshotAllReturnsEveryKey() { + KeyedSlidingWindow w = new KeyedSlidingWindow<>(2); + w.append("a", 1); + w.append("a", 2); + w.append("b", 3); + assertThat(w.snapshot()).containsOnlyKeys("a", "b"); + assertThat(w.snapshot().get("a")).containsExactly(1, 2); + } + + // ---- multi-threaded ------------------------------------------------------------------------- + + @Test + void concurrentAppendsToOneKeyStayBoundedAndUntorn() throws Exception { + // Many threads append to the same key while others snapshot it; the window must never exceed + // capacity and a snapshot must never throw or observe a torn deque. + int window = 50; + KeyedSlidingWindow w = new KeyedSlidingWindow<>(window); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + int maxSnapshot = ConcurrencyTestSupport.runConcurrently(8, failures, t -> { + int localMax = 0; + for (int i = 0; i < 50_000; i++) { + if ((t & 1) == 0) { + w.append("hot", i); + } else { + List snap = w.snapshot("hot"); + localMax = Math.max(localMax, snap.size()); + } + } + return localMax; + }); + + assertThat(failures).isEmpty(); + assertThat(maxSnapshot).isLessThanOrEqualTo(window); + assertThat(w.snapshot("hot")).hasSize(window); + } + + @Test + void concurrentAppendsAcrossManyKeysDoNotInterfere() throws Exception { + int window = 10; + KeyedSlidingWindow w = new KeyedSlidingWindow<>(window); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + ConcurrencyTestSupport.runConcurrently(8, failures, t -> { + for (int i = 0; i < 20_000; i++) { + w.append(i % 64, i); + if (i % 200 == 0) { + w.snapshot(); + } + } + return 0; + }); + + assertThat(failures).isEmpty(); + for (List values : w.snapshot().values()) { + assertThat(values).hasSizeLessThanOrEqualTo(window); + } + } +} diff --git a/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/PeriodicGateTest.java b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/PeriodicGateTest.java new file mode 100644 index 0000000..bf9399e --- /dev/null +++ b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/PeriodicGateTest.java @@ -0,0 +1,86 @@ +/* + * 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.common.concurrent; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import me.predatorray.candybox.common.ManualClock; +import org.junit.jupiter.api.Test; + +class PeriodicGateTest { + + @Test + void runsOnlyOncePerInterval() { + ManualClock clock = new ManualClock(0); + PeriodicGate gate = new PeriodicGate(clock, 100); + AtomicInteger runs = new AtomicInteger(); + + assertThat(gate.runIfDue(runs::incrementAndGet)).isFalse(); // not yet due + clock.advance(100); + assertThat(gate.runIfDue(runs::incrementAndGet)).isTrue(); + assertThat(gate.runIfDue(runs::incrementAndGet)).isFalse(); // throttled + clock.advance(99); + assertThat(gate.runIfDue(runs::incrementAndGet)).isFalse(); + clock.advance(1); + assertThat(gate.runIfDue(runs::incrementAndGet)).isTrue(); + assertThat(runs).hasValue(2); + } + + @Test + void throwingActionStillConsumesItsInterval() { + ManualClock clock = new ManualClock(0); + PeriodicGate gate = new PeriodicGate(clock, 10); + clock.advance(10); + try { + gate.runIfDue(() -> { + throw new IllegalStateException("boom"); + }); + } catch (IllegalStateException expected) { + // propagated to the caller that ran it + } + AtomicInteger runs = new AtomicInteger(); + // The interval was consumed despite the throw; not due again until the clock advances. + assertThat(gate.runIfDue(runs::incrementAndGet)).isFalse(); + } + + // ---- multi-threaded ------------------------------------------------------------------------- + + @Test + void aSwarmAtTheDeadlineFiresTheActionExactlyOnce() throws Exception { + // The crux of the double-checked-lock idiom: with the clock pinned exactly at the deadline, + // every caller passes the lock-free outer check, but the inner re-check under the lock must + // let exactly ONE of them run the action. A broken DCL (missing inner check) fires it + // once-per-caller. Repeated across many fresh gates to defeat luck. + for (int round = 0; round < 500; round++) { + ManualClock clock = new ManualClock(0); + PeriodicGate gate = new PeriodicGate(clock, 100); + clock.advance(100); // exactly due + AtomicInteger runs = new AtomicInteger(); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + int fired = ConcurrencyTestSupport.runConcurrently(16, failures, t -> { + boolean ran = gate.runIfDue(runs::incrementAndGet); + return ran ? 1 : 0; + }); + + assertThat(failures).isEmpty(); + assertThat(runs.get()).as("round %d action runs", round).isEqualTo(1); + assertThat(fired).isEqualTo(1); // exactly one caller reports it ran + } + } +} diff --git a/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/TtlCacheTest.java b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/TtlCacheTest.java new file mode 100644 index 0000000..2b3917a --- /dev/null +++ b/candybox-common/src/test/java/me/predatorray/candybox/common/concurrent/TtlCacheTest.java @@ -0,0 +1,140 @@ +/* + * 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.common.concurrent; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import me.predatorray.candybox.common.ManualClock; +import org.junit.jupiter.api.Test; + +class TtlCacheTest { + + @Test + void servesHitUntilTtlElapsesThenReloads() { + ManualClock clock = new ManualClock(1000); + TtlCache cache = new TtlCache<>(clock, 100); + AtomicInteger loads = new AtomicInteger(); + + assertThat(cache.get("k", k -> loads.incrementAndGet())).isEqualTo(1); + assertThat(cache.get("k", k -> loads.incrementAndGet())).isEqualTo(1); // hit, no reload + assertThat(loads).hasValue(1); + + clock.advance(99); + assertThat(cache.get("k", k -> loads.incrementAndGet())).isEqualTo(1); // still fresh + assertThat(loads).hasValue(1); + + clock.advance(1); // now == expiry → stale + assertThat(cache.get("k", k -> loads.incrementAndGet())).isEqualTo(2); // reloaded + assertThat(loads).hasValue(2); + } + + @Test + void getIfFreshNeverLoadsAndRespectsExpiry() { + ManualClock clock = new ManualClock(0); + TtlCache cache = new TtlCache<>(clock, 50); + assertThat(cache.getIfFresh("k")).isEmpty(); + cache.put("k", "v"); + assertThat(cache.getIfFresh("k")).contains("v"); + clock.advance(50); + assertThat(cache.getIfFresh("k")).isEmpty(); + } + + @Test + void invalidateForcesReload() { + ManualClock clock = new ManualClock(0); + TtlCache cache = new TtlCache<>(clock, 10_000); + AtomicInteger loads = new AtomicInteger(); + cache.get("k", k -> loads.incrementAndGet()); + cache.invalidate("k"); + cache.get("k", k -> loads.incrementAndGet()); + assertThat(loads).hasValue(2); + } + + @Test + void zeroTtlNeverServesHit() { + ManualClock clock = new ManualClock(0); + TtlCache cache = new TtlCache<>(clock, 0); + AtomicInteger loads = new AtomicInteger(); + cache.get("k", k -> loads.incrementAndGet()); + cache.get("k", k -> loads.incrementAndGet()); + assertThat(loads).hasValue(2); + } + + @Test + void rejectsNullLoadedValue() { + TtlCache cache = new TtlCache<>(new ManualClock(0), 100); + assertThatThrownBy(() -> cache.get("k", k -> null)) + .isInstanceOf(NullPointerException.class); + } + + // ---- multi-threaded ------------------------------------------------------------------------- + + @Test + void concurrentMissesAllReturnAConsistentValueAndCacheConverges() throws Exception { + // With a cold cache and TTL that never expires during the test, many threads racing on the + // same key may each load once (documented), but every returned value must be a valid load and + // subsequent reads must hit. We assert no torn reads / exceptions and that the cache ends warm. + ManualClock clock = new ManualClock(0); + TtlCache cache = new TtlCache<>(clock, 1_000_000); + AtomicInteger loads = new AtomicInteger(); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + ConcurrencyTestSupport.runConcurrently(16, failures, t -> { + for (int i = 0; i < 10_000; i++) { + Integer v = cache.get("hot", k -> loads.incrementAndGet()); + if (v == null || v <= 0) { + throw new AssertionError("unexpected value " + v); + } + } + return 0; + }); + + assertThat(failures).isEmpty(); + // The cache is warm: a read does not trigger a further load. + int loadsAfter = loads.get(); + cache.get("hot", k -> loads.incrementAndGet()); + assertThat(loads.get()).isEqualTo(loadsAfter); + } + + @Test + void concurrentGetAndInvalidateDoesNotCorruptOrLeakStale() throws Exception { + // Race read-through against invalidate under an expiring TTL with a moving clock. Any stale + // value served must still be a legitimate load; the structure must not throw. + ManualClock clock = new ManualClock(0); + TtlCache cache = new TtlCache<>(clock, 5); + AtomicInteger loads = new AtomicInteger(); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + ConcurrencyTestSupport.runConcurrently(12, failures, t -> { + for (int i = 0; i < 20_000; i++) { + int key = i % 8; + if ((t & 1) == 0) { + cache.get(key, k -> loads.incrementAndGet()); + } else { + cache.invalidate(key); + } + if (i % 100 == 0) { + clock.advance(1); + } + } + return 0; + }); + assertThat(failures).isEmpty(); + } +} diff --git a/candybox-coordination/src/main/java/me/predatorray/candybox/coordination/CoordinationCas.java b/candybox-coordination/src/main/java/me/predatorray/candybox/coordination/CoordinationCas.java new file mode 100644 index 0000000..612cad3 --- /dev/null +++ b/candybox-coordination/src/main/java/me/predatorray/candybox/coordination/CoordinationCas.java @@ -0,0 +1,105 @@ +/* + * 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.coordination; + +import java.util.Optional; +import java.util.function.Supplier; + +/** + * Optimistic read-modify-write helpers over a {@link CoordinationService}: the {@code get} the current + * version, then {@code create}-if-absent-or-{@code compareAndSet}, retrying a bounded number of times + * on {@link CasConflictException} idiom that several stores hand-rolled. Business-logic-free; callers + * decide how to treat exhaustion (wrap, swallow, or propagate the {@link CasConflictException}). + * + *

These do not back off between attempts. They are intended for short, low-contention races (an + * owner publishing its own state, an ACL edit); a conflict means a peer wrote concurrently, so a + * fresh re-read and retry converges quickly. Use {@code maxRetries = 0} for a single attempt that + * propagates the first conflict (the caller then typically swallows it and converges on a later pass). + */ +public final class CoordinationCas { + + private CoordinationCas() { + } + + /** + * Sets {@code key} to a freshly computed value, whether or not it already exists, retrying on a + * concurrent-write conflict. The value is supplied lazily so a retry can recompute it against the + * just-observed state if needed (constant values may ignore the re-read). + * + * @param coordination the backing service + * @param key the key to upsert + * @param value supplies the bytes to write; re-invoked on each attempt + * @param maxRetries how many times to retry after the first conflict (0 = single attempt) + * @return the new version written + * @throws CasConflictException if every attempt (1 + {@code maxRetries}) lost a race + */ + public static long upsert(CoordinationService coordination, String key, + Supplier value, int maxRetries) { + requireNonNegative(maxRetries); + for (int attempt = 0; ; attempt++) { + try { + Optional current = coordination.get(key); + if (current.isEmpty()) { + return coordination.create(key, value.get()); + } + return coordination.compareAndSet(key, value.get(), current.get().version()); + } catch (CasConflictException raced) { + if (attempt >= maxRetries) { + throw raced; + } + } + } + } + + /** {@link #upsert(CoordinationService, String, Supplier, int)} with a constant value. */ + public static long upsert(CoordinationService coordination, String key, byte[] value, + int maxRetries) { + return upsert(coordination, key, () -> value, maxRetries); + } + + /** + * Deletes {@code key} if it currently exists, retrying on a concurrent-write conflict. A key that + * is already absent is a no-op. + * + * @param maxRetries how many times to retry after the first conflict (0 = single attempt) + * @return {@code true} if a value was deleted, {@code false} if the key was already absent + * @throws CasConflictException if every attempt (1 + {@code maxRetries}) lost a race + */ + public static boolean deleteIfPresent(CoordinationService coordination, String key, + int maxRetries) { + requireNonNegative(maxRetries); + for (int attempt = 0; ; attempt++) { + try { + Optional current = coordination.get(key); + if (current.isEmpty()) { + return false; + } + coordination.delete(key, current.get().version()); + return true; + } catch (CasConflictException raced) { + if (attempt >= maxRetries) { + throw raced; + } + } + } + } + + private static void requireNonNegative(int maxRetries) { + if (maxRetries < 0) { + throw new IllegalArgumentException("maxRetries must be non-negative: " + maxRetries); + } + } +} diff --git a/candybox-coordination/src/test/java/me/predatorray/candybox/coordination/CoordinationCasTest.java b/candybox-coordination/src/test/java/me/predatorray/candybox/coordination/CoordinationCasTest.java new file mode 100644 index 0000000..c9ac9be --- /dev/null +++ b/candybox-coordination/src/test/java/me/predatorray/candybox/coordination/CoordinationCasTest.java @@ -0,0 +1,185 @@ +/* + * 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.coordination; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; +import me.predatorray.candybox.coordination.fake.InMemoryCoordinationService; +import org.junit.jupiter.api.Test; + +class CoordinationCasTest { + + private static byte[] bytes(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private static String str(byte[] b) { + return new String(b, StandardCharsets.UTF_8); + } + + @Test + void upsertCreatesWhenAbsentThenOverwrites() { + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + assertThat(CoordinationCas.upsert(cs, "k", bytes("v0"), 3)).isEqualTo(0); + assertThat(CoordinationCas.upsert(cs, "k", bytes("v1"), 3)).isEqualTo(1); + assertThat(cs.get("k").map(v -> str(v.value()))).contains("v1"); + } + + @Test + void deleteIfPresentReportsWhetherItRemovedAnything() { + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + assertThat(CoordinationCas.deleteIfPresent(cs, "k", 3)).isFalse(); + cs.create("k", bytes("v")); + assertThat(CoordinationCas.deleteIfPresent(cs, "k", 3)).isTrue(); + assertThat(cs.get("k")).isEmpty(); + } + + @Test + void singleAttemptPropagatesConflictButOneRetryRecovers() { + // Deterministically inject a race: between our get() and our compareAndSet(), an "interloper" + // bumps the version. With maxRetries=0 the conflict propagates; with a retry the re-read sees + // the new version and succeeds. + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + cs.create("k", bytes("seed")); // version 0 + + AtomicInteger valueCalls = new AtomicInteger(); + Supplier racyValue = () -> { + if (valueCalls.getAndIncrement() == 0) { + // A concurrent writer wins the race exactly once, on our first attempt. + cs.compareAndSet("k", bytes("interloper"), 0); + } + return bytes("mine"); + }; + + assertThatThrownBy(() -> CoordinationCas.upsert(cs, "k", racyValue, 0)) + .isInstanceOf(CasConflictException.class); + + // Reset and prove a single retry absorbs the same race. + InMemoryCoordinationService cs2 = new InMemoryCoordinationService(); + cs2.create("k", bytes("seed")); + AtomicInteger calls2 = new AtomicInteger(); + Supplier racyValue2 = () -> { + if (calls2.getAndIncrement() == 0) { + cs2.compareAndSet("k", bytes("interloper"), 0); + } + return bytes("mine"); + }; + long version = CoordinationCas.upsert(cs2, "k", racyValue2, 1); + assertThat(version).isEqualTo(2); // 0 seed, 1 interloper, 2 us + assertThat(str(cs2.get("k").orElseThrow().value())).isEqualTo("mine"); + } + + @Test + void rejectsNegativeMaxRetries() { + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + assertThatThrownBy(() -> CoordinationCas.upsert(cs, "k", bytes("v"), -1)) + .isInstanceOf(IllegalArgumentException.class); + } + + // ---- multi-threaded ------------------------------------------------------------------------- + + @Test + void concurrentUpsertsAllApplyExactlyOnceWithRetries() throws Exception { + // Every thread upserts a distinct value to the same key, with generous retries. The decisive + // invariant: the final coordination version equals the number of successful writes (threads), + // i.e. no upsert was silently dropped and none applied twice. A broken retry loop would lose + // writes and leave the version short. + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + int threads = 16; + int maxRetries = 10_000; // bounded but far above any realistic conflict streak + CyclicBarrier barrier = new CyclicBarrier(threads); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + List workers = new ArrayList<>(); + AtomicInteger successes = new AtomicInteger(); + + for (int i = 0; i < threads; i++) { + int id = i; + Thread thread = new Thread(() -> { + try { + barrier.await(); + CoordinationCas.upsert(cs, "shared", bytes("node-" + id), maxRetries); + successes.incrementAndGet(); + } catch (Throwable t) { + failures.add(t); + } + }); + workers.add(thread); + } + workers.forEach(Thread::start); + for (Thread t : workers) { + t.join(); + } + + assertThat(failures).isEmpty(); + assertThat(successes).hasValue(threads); + Optional finalValue = cs.get("shared"); + assertThat(finalValue).isPresent(); + // create()=v0, then one compareAndSet per remaining successful writer ⇒ version == threads-1. + assertThat(finalValue.get().version()).isEqualTo((long) threads - 1); + assertThat(str(finalValue.get().value())).startsWith("node-"); + } + + @Test + void concurrentUpsertAndDeleteNeverThrowWithRetries() throws Exception { + // Mix upserts and deletes on the same key under contention; with retries, no operation should + // surface a CasConflictException to the caller (each converges on a fresh read). + InMemoryCoordinationService cs = new InMemoryCoordinationService(); + int threads = 12; + int iterations = 2_000; + // Each attempt makes progress (the fake serializes CAS, so some caller always wins a round); + // a thread therefore wins within a bounded number of retries, but a tight per-op cap would + // still flake under 12-way contention. Use an effectively-unbounded cap to assert "retries + // converge without ever surfacing a conflict". + int maxRetries = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(threads); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + List workers = new ArrayList<>(); + + for (int i = 0; i < threads; i++) { + int id = i; + Thread thread = new Thread(() -> { + try { + barrier.await(); + for (int n = 0; n < iterations; n++) { + if ((id + n) % 2 == 0) { + CoordinationCas.upsert(cs, "shared", bytes("w" + id + "-" + n), maxRetries); + } else { + CoordinationCas.deleteIfPresent(cs, "shared", maxRetries); + } + } + } catch (Throwable t) { + failures.add(t); + } + }); + workers.add(thread); + } + workers.forEach(Thread::start); + for (Thread t : workers) { + t.join(); + } + + assertThat(failures).isEmpty(); + } +} diff --git a/candybox-lsm/src/main/java/me/predatorray/candybox/lsm/engine/BoxEngine.java b/candybox-lsm/src/main/java/me/predatorray/candybox/lsm/engine/BoxEngine.java index af50bca..a4d3ed0 100644 --- a/candybox-lsm/src/main/java/me/predatorray/candybox/lsm/engine/BoxEngine.java +++ b/candybox-lsm/src/main/java/me/predatorray/candybox/lsm/engine/BoxEngine.java @@ -19,7 +19,6 @@ import java.io.InputStream; import java.io.OutputStream; import java.util.ArrayList; -import java.util.Collections; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; @@ -47,6 +46,7 @@ import me.predatorray.candybox.common.RangeTombstone; import me.predatorray.candybox.common.SegmentRef; import me.predatorray.candybox.common.Validation; +import me.predatorray.candybox.common.concurrent.BoundedLruCache; import me.predatorray.candybox.common.config.CandyboxConfig; import me.predatorray.candybox.common.config.LedgerRole; import me.predatorray.candybox.common.exception.BusyException; @@ -123,13 +123,8 @@ public final class BoxEngine implements AutoCloseable { private final AtomicLong stallRejectionCount = new AtomicLong(); // Bounded idempotency cache: token -> already-applied result, so a retried put is a no-op. - private final Map idempotencyCache = Collections.synchronizedMap( - new LinkedHashMap<>(16, 0.75f, true) { - @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - return size() > IDEMPOTENCY_CACHE_SIZE; - } - }); + private final BoundedLruCache idempotencyCache = + new BoundedLruCache<>(IDEMPOTENCY_CACHE_SIZE); private BoxEngine(BoxName box, CandyboxConfig config, LedgerStore ledgerStore, HybridLogicalClock hlc, Clock clock, Manifest manifest, WriteAheadLog wal) { @@ -262,16 +257,18 @@ public CandyMetadata putCandy(CandyKey key, InputStream data, String contentType ObjectAcl acl) { Validation.checkCandyKey(key, config.sizeLimits()); Validation.checkUserMetadata(userMetadata, config.sizeLimits()); - if (idempotencyToken != null) { - CandyMetadata cached = idempotencyCache.get(idempotencyToken); - if (cached != null) { - return cached; - } + CandyMetadata cached = idempotentResult(idempotencyToken); + if (cached != null) { + return cached; } Map metadata = userMetadata == null ? Map.of() : Map.copyOf(userMetadata); lock.writeLock().lock(); try { + CandyMetadata replay = idempotentResult(idempotencyToken); + if (replay != null) { + return replay; + } rejectIfStalled(); SyrupWriteResult written = syrupManager.writeCandy(data); Validation.checkCandySize(written.contentLength(), config.sizeLimits()); @@ -464,14 +461,16 @@ public CandyMetadata completeMultipartUpload(String uploadId, java.util.List new CandyNotFoundException(box.value(), src.value())); @@ -737,14 +738,16 @@ public CandyMetadata zeroCopyPut(CandyKey dst, List parts, String contentT ObjectAcl acl, String idempotencyToken) { Validation.checkCandyKey(dst, config.sizeLimits()); Map metadata = userMetadata == null ? Map.of() : Map.copyOf(userMetadata); - if (idempotencyToken != null) { - CandyMetadata cached = idempotencyCache.get(idempotencyToken); - if (cached != null) { - return cached; - } + CandyMetadata cached = idempotentResult(idempotencyToken); + if (cached != null) { + return cached; } lock.writeLock().lock(); try { + CandyMetadata replay = idempotentResult(idempotencyToken); + if (replay != null) { + return replay; + } rejectIfStalled(); Hlc stamp = hlc.tick(); CandyLocator dstLocator = new CandyLocator(stamp, LocatorType.PUT, contentType, metadata, @@ -1267,6 +1270,16 @@ public void close() { // ---- internals ------------------------------------------------------------------------- + /** + * A previously stored idempotent result for {@code token}, or {@code null} if none (or no token). + * Checked both on the unlocked fast path (to skip the write lock for an obvious retry) and again + * under the write lock, so two concurrent retries that both miss the fast path cannot both apply + * the mutation — the second sees the first's cached result and returns it. + */ + private CandyMetadata idempotentResult(String token) { + return token == null ? null : idempotencyCache.get(token); + } + private void rejectIfStalled() { int l0 = manifest.current().level0().size(); if (l0 >= config.l0StallThreshold()) { diff --git a/candybox-lsm/src/test/java/me/predatorray/candybox/lsm/engine/BoxEngineConcurrencyTest.java b/candybox-lsm/src/test/java/me/predatorray/candybox/lsm/engine/BoxEngineConcurrencyTest.java new file mode 100644 index 0000000..126160d --- /dev/null +++ b/candybox-lsm/src/test/java/me/predatorray/candybox/lsm/engine/BoxEngineConcurrencyTest.java @@ -0,0 +1,106 @@ +/* + * 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.lsm.engine; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicReferenceArray; +import me.predatorray.candybox.bookkeeper.fake.InMemoryLedgerStore; +import me.predatorray.candybox.common.BoxName; +import me.predatorray.candybox.common.CandyKey; +import me.predatorray.candybox.common.ManualClock; +import me.predatorray.candybox.common.config.CandyboxConfig; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class BoxEngineConcurrencyTest { + + private final InMemoryLedgerStore store = new InMemoryLedgerStore(); + private final BoxName box = BoxName.of("race-box"); + private BoxEngine engine; + + @AfterEach + void tearDown() { + if (engine != null) { + engine.close(); + } + store.close(); + } + + private static byte[] bytes(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + /** + * Two concurrent retries of the same write (same idempotency token) must apply the mutation + * exactly once. Before the inside-the-write-lock re-check, both callers could pass the unlocked + * fast-path check while the cache was still cold and then each take the lock and write — a double + * apply (two HLCs, {@code puts} incremented twice). With the re-check, the second caller observes + * the first's cached result under the lock and returns it unchanged. + */ + @Test + void concurrentRetriesWithSameIdempotencyTokenApplyExactlyOnce() throws Exception { + // A ManualClock that never advances forces the two writes to differ only by HLC logical + // counter, so a double-apply is unmistakable: the two returned metadatas would not be equal + // and puts would jump by two. + engine = BoxEngine.createNew(box, CandyboxConfig.defaults(), store, 1, new ManualClock(1000), 1L); + + int iterations = 300; + for (int i = 0; i < iterations; i++) { + String key = "k/" + i; + String token = "token-" + i; + long putsBefore = engine.stats().puts(); + + CyclicBarrier barrier = new CyclicBarrier(2); + AtomicReferenceArray results = new AtomicReferenceArray<>(2); + AtomicReferenceArray errors = new AtomicReferenceArray<>(2); + + Runnable body0 = put(key, token, results, errors, 0, barrier); + Runnable body1 = put(key, token, results, errors, 1, barrier); + Thread t0 = new Thread(body0, "put-0"); + Thread t1 = new Thread(body1, "put-1"); + t0.start(); + t1.start(); + t0.join(); + t1.join(); + + assertThat(errors.get(0)).as("thread 0 error at i=%d", i).isNull(); + assertThat(errors.get(1)).as("thread 1 error at i=%d", i).isNull(); + + long applied = engine.stats().puts() - putsBefore; + assertThat(applied).as("puts applied for token %s", token).isEqualTo(1); + // Both callers observe the identical winning result (same HLC, same metadata). + assertThat(results.get(0)).as("results agree at i=%d", i).isEqualTo(results.get(1)); + } + } + + private Runnable put(String key, String token, AtomicReferenceArray results, + AtomicReferenceArray errors, int slot, CyclicBarrier barrier) { + return () -> { + try { + barrier.await(); + CandyMetadata result = engine.putCandy(CandyKey.of(key), bytes("payload-" + key), + "text/plain", Map.of(), token); + results.set(slot, result); + } catch (Throwable t) { + errors.set(slot, t); + } + }; + } +} diff --git a/candybox-s3-gateway/src/main/java/me/predatorray/candybox/s3/S3AccessControl.java b/candybox-s3-gateway/src/main/java/me/predatorray/candybox/s3/S3AccessControl.java index c30efa4..d4a27a5 100644 --- a/candybox-s3-gateway/src/main/java/me/predatorray/candybox/s3/S3AccessControl.java +++ b/candybox-s3-gateway/src/main/java/me/predatorray/candybox/s3/S3AccessControl.java @@ -19,9 +19,8 @@ import java.util.EnumSet; import java.util.List; import java.util.Locale; -import java.util.Map; import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; +import me.predatorray.candybox.common.SystemClock; import me.predatorray.candybox.common.auth.BoxAcl; import me.predatorray.candybox.common.auth.Grant; import me.predatorray.candybox.common.auth.ObjectAcl; @@ -29,6 +28,7 @@ import me.predatorray.candybox.common.auth.Principal; import me.predatorray.candybox.common.auth.Resource; import me.predatorray.candybox.common.auth.StandardAuthorizer; +import me.predatorray.candybox.common.concurrent.TtlCache; import me.predatorray.candybox.s3.S3Router.S3Action; /** @@ -53,10 +53,8 @@ final class S3AccessControl { private final boolean enabled; private final StandardAuthorizer authorizer; private final CandyStore store; - private final Map cache = new ConcurrentHashMap<>(); - - private record CachedAcl(Optional acl, long expiresAtMillis) { - } + private final TtlCache> cache = + new TtlCache<>(SystemClock.INSTANCE, CACHE_TTL_MILLIS); S3AccessControl(boolean enabled, CandyStore store) { this.enabled = enabled; @@ -69,18 +67,11 @@ boolean enabled() { } private Optional boxAcl(String box) { - long now = System.currentTimeMillis(); - CachedAcl cached = cache.get(box); - if (cached != null && now < cached.expiresAtMillis) { - return cached.acl; - } - Optional acl = store.getBoxAcl(box); - cache.put(box, new CachedAcl(acl, now + CACHE_TTL_MILLIS)); - return acl; + return cache.get(box, store::getBoxAcl); } void invalidate(String box) { - cache.remove(box); + cache.invalidate(box); } /** Authorizes one routed request; throws {@code AccessDenied} otherwise. */ diff --git a/candybox-server/src/main/java/me/predatorray/candybox/server/BoxAclStore.java b/candybox-server/src/main/java/me/predatorray/candybox/server/BoxAclStore.java index 5d6de58..727d6ab 100644 --- a/candybox-server/src/main/java/me/predatorray/candybox/server/BoxAclStore.java +++ b/candybox-server/src/main/java/me/predatorray/candybox/server/BoxAclStore.java @@ -16,15 +16,14 @@ package me.predatorray.candybox.server; import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import me.predatorray.candybox.common.Clock; import me.predatorray.candybox.common.auth.BoxAcl; +import me.predatorray.candybox.common.concurrent.TtlCache; import me.predatorray.candybox.common.exception.CandyboxException; import me.predatorray.candybox.coordination.CandyboxKeys; import me.predatorray.candybox.coordination.CasConflictException; +import me.predatorray.candybox.coordination.CoordinationCas; import me.predatorray.candybox.coordination.CoordinationService; -import me.predatorray.candybox.coordination.VersionedValue; /** * Box ACL documents in the coordination service ({@code acls/}), with a small read-through @@ -38,57 +37,35 @@ final class BoxAclStore { private static final int CAS_RETRIES = 3; private final CoordinationService coordination; - private final Clock clock; - private final ConcurrentMap cache = new ConcurrentHashMap<>(); - - private record CachedAcl(Optional acl, long expiresAtMillis) { - } + private final TtlCache> cache; BoxAclStore(CoordinationService coordination, Clock clock) { this.coordination = coordination; - this.clock = clock; + this.cache = new TtlCache<>(clock, CACHE_TTL_MILLIS); } /** The Box's ACL document, or empty when none was ever written (legacy Box). */ Optional get(String box) { - long now = clock.currentTimeMillis(); - CachedAcl cached = cache.get(box); - if (cached != null && now < cached.expiresAtMillis) { - return cached.acl; - } - Optional acl = coordination.get(CandyboxKeys.boxAclKey(box)) - .map(v -> BoxAcl.fromBytes(v.value())); - cache.put(box, new CachedAcl(acl, now + CACHE_TTL_MILLIS)); - return acl; + return cache.get(box, b -> coordination.get(CandyboxKeys.boxAclKey(b)) + .map(v -> BoxAcl.fromBytes(v.value()))); } /** Creates or replaces the Box's ACL document (CAS with a short retry on races). */ void set(String box, BoxAcl acl) { - String key = CandyboxKeys.boxAclKey(box); - byte[] bytes = acl.toBytes(); - for (int attempt = 0; ; attempt++) { - try { - Optional current = coordination.get(key); - if (current.isEmpty()) { - coordination.create(key, bytes); - } else { - coordination.compareAndSet(key, bytes, current.get().version()); - } - cache.remove(box); - return; - } catch (CasConflictException raced) { - if (attempt == CAS_RETRIES) { - throw new CandyboxException("Concurrent ACL update on Box " + box, raced); - } - } + try { + CoordinationCas.upsert(coordination, CandyboxKeys.boxAclKey(box), acl.toBytes(), + CAS_RETRIES); + } catch (CasConflictException raced) { + throw new CandyboxException("Concurrent ACL update on Box " + box, raced); } + cache.invalidate(box); } /** Writes the initial ACL for a fresh Box; loses quietly if one already exists. */ void seed(String box, BoxAcl acl) { try { coordination.create(CandyboxKeys.boxAclKey(box), acl.toBytes()); - cache.remove(box); + cache.invalidate(box); } catch (CasConflictException alreadySeeded) { // a concurrent creator won; their ACL stands } @@ -96,20 +73,11 @@ void seed(String box, BoxAcl acl) { /** Drops the Box's ACL document when the Box is deleted; absent is fine. */ void delete(String box) { - String key = CandyboxKeys.boxAclKey(box); - for (int attempt = 0; ; attempt++) { - try { - Optional current = coordination.get(key); - if (current.isPresent()) { - coordination.delete(key, current.get().version()); - } - cache.remove(box); - return; - } catch (CasConflictException raced) { - if (attempt == CAS_RETRIES) { - return; // someone else is mutating it concurrently with the delete; let it be - } - } + try { + CoordinationCas.deleteIfPresent(coordination, CandyboxKeys.boxAclKey(box), CAS_RETRIES); + } catch (CasConflictException raced) { + return; // someone else is mutating it concurrently with the delete; let it be } + cache.invalidate(box); } } diff --git a/candybox-server/src/main/java/me/predatorray/candybox/server/CandyboxNode.java b/candybox-server/src/main/java/me/predatorray/candybox/server/CandyboxNode.java index 1562bbb..bed94f9 100644 --- a/candybox-server/src/main/java/me/predatorray/candybox/server/CandyboxNode.java +++ b/candybox-server/src/main/java/me/predatorray/candybox/server/CandyboxNode.java @@ -42,6 +42,7 @@ import me.predatorray.candybox.coordination.BoxDescriptor; import me.predatorray.candybox.coordination.CandyboxKeys; import me.predatorray.candybox.coordination.CasConflictException; +import me.predatorray.candybox.coordination.CoordinationCas; import me.predatorray.candybox.coordination.CoordinationService; import me.predatorray.candybox.coordination.VersionedValue; import me.predatorray.candybox.common.serial.BinaryReader; @@ -694,24 +695,16 @@ boolean renameMarkerPresent(String box, String token) { } void deleteRenameMarker(String box, String token) { - String key = CandyboxKeys.renameMarkerKey(box, token); - coordination.get(key).ifPresent(v -> { - try { - coordination.delete(key, v.version()); - } catch (CasConflictException raced) { - // Already deleted (or rewritten); nothing to do. - } - }); + try { + CoordinationCas.deleteIfPresent(coordination, CandyboxKeys.renameMarkerKey(box, token), 0); + } catch (CasConflictException raced) { + // Already deleted (or rewritten); nothing to do. + } } private void casPut(String key, byte[] value) { - Optional current = coordination.get(key); try { - if (current.isEmpty()) { - coordination.create(key, value); - } else { - coordination.compareAndSet(key, value, current.get().version()); - } + CoordinationCas.upsert(coordination, key, value, 0); } catch (CasConflictException raced) { // A concurrent writer won; the next pass republishes. }