diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataCache.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataCache.java index 3ccaa0c49..d7b54061f 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataCache.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataCache.java @@ -138,6 +138,102 @@ private boolean tryAddChangesBeforeDeferredSharedFlush(Iterable return true; } + /** Supported arithmetic for a bounded, conditional cached integer mutation. */ + public enum IntegerMutation { SET, ADD, ADD_FLOOR_ZERO, DEBIT_POSITIVE } + + /** Previous values are returned so a caller can compensate precisely. */ + public record IntegerMutationResult(boolean applied, java.util.Map previous) { } + + /** + * Atomically compute and enqueue a bounded set of integer changes on a published + * shared cache. Returns null when the cache is cold, retiring, or not bound; a + * DEBIT_POSITIVE result with applied=false means no requested balance was positive. + * No user callback or storage operation is run while the cache monitor is held. + * The notification is staged behind an admitted exclusive flush, just like a + * normal shared change. This is local cache atomicity, not a cross-process transaction. + * If notification submission is rejected after acceptance, the manager records + * the failure and the result still reports the already accepted mutation. + */ + public IntegerMutationResult tryMutateIntegers(java.util.Map operands, + IntegerMutation operation, java.util.function.Consumer> notification) { + Objects.requireNonNull(operands, "operands"); Objects.requireNonNull(operation, "operation"); + if (operands.isEmpty() || operands.size() > 64) throw new IllegalArgumentException("Invalid integer mutation size"); + var ordered = new java.util.LinkedHashMap(); + int copied = 0; + for (var entry : operands.entrySet()) { + if (++copied > 64) throw new IllegalArgumentException("Invalid integer mutation size"); + String key = entry.getKey(); Integer value = entry.getValue(); + if (key == null || key.isEmpty() || value == null) throw new IllegalArgumentException("Invalid integer mutation"); + if (operation == IntegerMutation.DEBIT_POSITIVE && value != -1) + throw new IllegalArgumentException("Positive debit must decrement by one"); + ordered.put(key, value); + } + if (ordered.isEmpty()) throw new IllegalArgumentException("Invalid integer mutation size"); + Runnable notify = null; + IntegerMutationResult result; + synchronized (this) { + if (sharedFlushGate == null || removing || uuid == null || cache == null || cachedChanges == null + || !storageSnapshotPublished) return null; + var previous = new java.util.LinkedHashMap(); + for (String key : ordered.keySet()) { + if (!cache.containsKey(key) || cache.get(key) == null) return null; + DataValue value = cache.get(key); + // Direct snapshot replacement intentionally remains visible to legacy + // readers. It must not let arithmetic forget an accepted write whose + // batch is still in flight. A newer local mutation retains changedAt. + if (!changedAt.containsKey(key) && inFlightValues.containsKey(key)) + value = inFlightValues.get(key); + int number; + if (value.isInt()) number = value.getInt(); + else if (value.isString()) { + try { number = Integer.parseInt(value.getString()); } + catch (NumberFormatException invalid) { number = 0; } + } else number = 0; + previous.put(key, number); + } + var changes = new java.util.ArrayList(); + var changed = new java.util.LinkedHashSet(); + ordered.forEach((key, operand) -> { + int old = previous.get(key); + if (operation == IntegerMutation.DEBIT_POSITIVE && old <= 0) return; + int updated = switch (operation) { + case SET -> operand; + case ADD, DEBIT_POSITIVE -> Math.addExact(old, operand); + case ADD_FLOOR_ZERO -> Math.max(0, Math.addExact(old, operand)); + }; + changes.add(new com.bencodez.advancedcore.api.user.usercache.change.UserDataChangeInt(key, updated)); + changed.add(key); + }); + result = new IntegerMutationResult(!changes.isEmpty(), java.util.Map.copyOf(previous)); + if (changes.isEmpty()) return result; + if (exclusiveFlushesPending != 0) { + for (UserDataChange change : changes) { publishChangeInternal(change); changesAfterExclusiveFlush.add(change); } + if (notification != null) { + var snapshot = java.util.Set.copyOf(changed); + Runnable raw = () -> notification.accept(snapshot); + Runnable captured = manager.captureSharedUserDataNotification(raw); + notificationsAfterExclusiveFlush.add(captured == null ? raw : captured); + } + return result; + } + for (UserDataChange change : changes) addChangeInternal(change, true); + if (notification != null) { + var snapshot = java.util.Set.copyOf(changed); + Runnable raw = () -> notification.accept(snapshot); + Runnable captured = manager.captureSharedUserDataNotification(raw); + notify = captured == null ? raw : captured; + } + } + if (notify != null) { + try { manager.dispatchSharedUserDataNotification(notify); } + catch (RejectedExecutionException rejected) { + // The manager records submission failures. Returning the accepted + // result prevents callers from retrying an already-applied debit. + } + } + return result; + } + private synchronized void addChangeInternal(UserDataChange change, boolean queue) { if (change != null && sharedStorageWriter != null && (cache == null || cachedChanges == null)) { throw new IllegalStateException("Shared user cache is retired"); @@ -319,7 +415,9 @@ private void drainChangesStagedBeforeExclusiveAdmission(ArrayList noti // A checkpoint action may replace the visible cache snapshot. Re-publish // mutations ordered after that checkpoint as they cross back into the // normal queue so read-after-write visibility survives the replacement. + boolean unavailable = !storageSnapshotPublished && cache != null && !cache.containsKey(deferred.getKey()); publishChangeInternal(deferred); + if (unavailable && cache != null) cache.remove(deferred.getKey()); cachedChanges.add(deferred); } Runnable deferredNotification; @@ -600,14 +698,20 @@ private Runnable processChangesInternal(boolean admitted) { // rather than another mutation. It clears changedAt(), so reconcile // that stale snapshot with this completed write instead of leaving a // value visible which will never be persisted. - if (queuedVersion == null || Objects.equals(persistedVersion, queuedVersion)) { + if ((storageSnapshotPublished || cache.containsKey(persistedChange.getKey())) + && (queuedVersion == null || Objects.equals(persistedVersion, queuedVersion))) { cache.put(persistedChange.getKey(), persistedChange.toUserDataValue()); } } } // The in-memory values now represent the successful write. Mark that // replacement so an older in-flight read cannot overwrite them. - persistedMutationVersions.forEach((key, version) -> persistedAt.put(key, version)); + // Fence reads that began before this write completed, even if the + // key's original mutation predates a later mutation of another key. + if (!values.isEmpty()) { + long completedVersion = ++snapshotVersion; + values.keySet().forEach(key -> persistedAt.put(key, completedVersion)); + } persistedMutationVersions.forEach((key, version) -> { if (version.equals(changedAt.get(key))) changedAt.remove(key); }); @@ -615,7 +719,12 @@ private Runnable processChangesInternal(boolean admitted) { changedUser = user; changedKeys = ArrayUtils.convert(keys); } catch (RuntimeException | Error e) { - if (!persisted) requeueChanges(changes); + if (!persisted) { + try { requeueChanges(changes); } + catch (RuntimeException | Error recoveryFailure) { + if (recoveryFailure != e) e.addSuppressed(recoveryFailure); + } + } throw e; } finally { finishInFlightBatch(); @@ -641,6 +750,19 @@ private synchronized void requeueChanges(ArrayList changes) { if (changes == null || changes.isEmpty() || cachedChanges == null) return; Queue restored = new ConcurrentLinkedQueue<>(); restored.addAll(changes); restored.addAll(cachedChanges); cachedChanges = restored; + // A direct replacement may have temporarily exposed an older value while + // this batch was in flight. Once requeued, restore the ordered pending view + // before another arithmetic operation can observe that stale snapshot. + if (!inFlightValues.isEmpty()) { + try { recordSnapshotReplacement(); } + catch (RuntimeException | Error recoveryFailure) { + // A malformed change must not leave a partly rebuilt visible snapshot. + if (cache != null) cache.clear(); + storageSnapshotPublished = false; + replacementVersion = ++snapshotVersion; + throw recoveryFailure; + } + } } private synchronized void finishInFlightBatch() { @@ -706,8 +828,14 @@ private synchronized HashMap updateSharedSnapshot(HashMap expectedVersion) return new HashMap<>(cache); storedDataPresent = snapshotStoredDataPresent; HashMap merged = values == null ? new HashMap<>() : new HashMap<>(values); + // A flush may have claimed pending writes after the read started, removing + // them from the queue before the storage snapshot can observe them. + merged.putAll(inFlightValues); changedAt.forEach((key, version) -> { if (version >= expectedVersion && cache.containsKey(key)) merged.put(key, cache.get(key)); }); - persistedAt.forEach((key, version) -> { if (version >= expectedVersion && cache.containsKey(key)) merged.put(key, cache.get(key)); }); + persistedAt.forEach((key, version) -> { + if ((version > expectedVersion || (!storageSnapshotPublished && version == expectedVersion)) + && cache.containsKey(key)) merged.put(key, cache.get(key)); + }); cache = merged; storageSnapshotPublished = true; recordSnapshotReplacement(); @@ -721,12 +849,17 @@ private void recordSnapshotReplacement() { // not erase the version fence for a mutation that is still waiting in the // write queue. Otherwise an older in-flight batch can overwrite that newer // visible value when it completes. - changedAt.keySet().removeIf(key -> cachedChanges == null - || cachedChanges.stream().noneMatch(change -> key.equals(change.getKey()))); - if (cache != null && cachedChanges != null) { - for (UserDataChange change : cachedChanges) { - cache.put(change.getKey(), change.toUserDataValue()); - } + changedAt.keySet().removeIf(key -> (cachedChanges == null || cachedChanges.stream().noneMatch(change -> key.equals(change.getKey()))) + && changesAfterExclusiveFlush.stream().noneMatch(change -> key.equals(change.getKey()))); + if (cache != null) { + // Queued writes precede changes accepted during an exclusive checkpoint. + // Keep that ordering when a storage snapshot replaces the visible cache. + if (cachedChanges != null) for (UserDataChange change : cachedChanges) + if (storageSnapshotPublished || cache.containsKey(change.getKey())) + cache.put(change.getKey(), change.toUserDataValue()); + for (UserDataChange change : changesAfterExclusiveFlush) + if (storageSnapshotPublished || cache.containsKey(change.getKey())) + cache.put(change.getKey(), change.toUserDataValue()); } } } diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AtomicIntegerMutationTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AtomicIntegerMutationTest.java new file mode 100644 index 000000000..0e31e7546 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AtomicIntegerMutationTest.java @@ -0,0 +1,504 @@ +package com.bencodez.advancedcore.tests.user; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import org.junit.jupiter.api.Test; +import com.bencodez.advancedcore.AdvancedCorePlugin; +import com.bencodez.advancedcore.api.user.usercache.UserDataCache; +import com.bencodez.advancedcore.api.user.usercache.UserDataCache.IntegerMutation; +import com.bencodez.advancedcore.api.user.usercache.UserDataManager; +import com.bencodez.simpleapi.sql.data.DataValue; +import com.bencodez.simpleapi.sql.data.DataValueInt; + +class AtomicIntegerMutationTest { + @Test void directSnapshotReplacementCannotAuthorizeASecondInFlightDebit() throws Exception { + for (boolean failFirst : new boolean[] { false, true }) { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var release = new CountDownLatch(1); var attempts = new AtomicInteger(); + var stored = new ConcurrentHashMap(); stored.put("a", new DataValueInt(1)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { + if (attempts.getAndIncrement() == 0) { entered.countDown(); await(release); if (failFirst) throw new IllegalStateException("write failure"); } + stored.putAll(values); + }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); var executor = Executors.newSingleThreadExecutor(); + try { + assertTrue(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + var user = mock(com.bencodez.advancedcore.api.user.AdvancedCoreUser.class); + when(user.isCached()).thenReturn(true); when(user.getCache()).thenReturn(cache); + var data = new com.bencodez.advancedcore.api.user.UserData(user); + data.setTempCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); data.updateCacheWithTemp(); + assertFalse(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + release.countDown(); + if (failFirst) { assertThrows(ExecutionException.class, () -> flush.get(5, TimeUnit.SECONDS)); cache.processChangesForSharedRuntime(); } + else flush.get(5, TimeUnit.SECONDS); + assertEquals(0, stored.get("a").getInt()); assertEquals(0, cache.getIfPresent("a").getInt()); + } finally { release.countDown(); executor.shutdownNow(); } + } + } + + @Test void newerQueuedAwardTakesPrecedenceOverInFlightDebitAfterReplacement() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var release = new CountDownLatch(1); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { entered.countDown(); await(release); }, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); var executor = Executors.newSingleThreadExecutor(); + try { + cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + var award = cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + assertEquals(0, award.previous().get("a")); assertEquals(1, cache.getIfPresent("a").getInt()); + assertTrue(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + assertEquals(0, cache.getIfPresent("a").getInt()); release.countDown(); flush.get(5, TimeUnit.SECONDS); + assertEquals(0, cache.getIfPresent("a").getInt()); + } finally { release.countDown(); executor.shutdownNow(); } + } + + @Test void failedWriteCannotExposeOldBalanceBeforeRetry() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var release = new CountDownLatch(1); var attempts = new AtomicInteger(); + var stored = new ConcurrentHashMap(); stored.put("a", new DataValueInt(1)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { + if (attempts.getAndIncrement() == 0) { entered.countDown(); await(release); throw new IllegalStateException("write failed"); } + stored.putAll(values); + }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); var executor = Executors.newSingleThreadExecutor(); + try { + cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); cache.updateCache(new HashMap<>(stored)); + release.countDown(); assertThrows(ExecutionException.class, () -> flush.get(5, TimeUnit.SECONDS)); + assertFalse(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + cache.processChangesForSharedRuntime(); assertEquals(0, stored.get("a").getInt()); + } finally { release.countDown(); executor.shutdownNow(); } + } + + @Test void failedRequeueRebuildPreservesOriginalErrorAndFencesCache() { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { throw new IllegalStateException("storage failed"); }, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + var change = mock(com.bencodez.advancedcore.api.user.usercache.change.UserDataChange.class); + when(change.getKey()).thenReturn("a"); var calls = new AtomicInteger(); + when(change.toUserDataValue()).thenAnswer(call -> { + if (calls.incrementAndGet() >= 4) throw new IllegalStateException("rebuild failed"); + return new DataValueInt(2); + }); + cache.addChange(change, true); + var failure = assertThrows(IllegalStateException.class, cache::processChangesForSharedRuntime); + assertEquals("storage failed", failure.getMessage()); + assertEquals(1, failure.getSuppressed().length); assertEquals("rebuild failed", failure.getSuppressed()[0].getMessage()); + assertFalse(cache.hasPublishedStorageSnapshot()); assertNull(cache.getIfPresent("a")); + assertTrue(cache.hasChangesToProcess()); + } + @Test void invalidationKeepsQueuedValueUnavailableUntilRefresh() { + var cache = cache(); + assertTrue(cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null).applied()); + cache.invalidateStorageSnapshot("a"); + assertNull(cache.getIfPresent("a")); assertTrue(cache.hasChangesToProcess()); + cache.updateSharedSnapshot(new HashMap<>(Map.of("a", new DataValueInt(4), "b", new DataValueInt(2))), + cache.getSharedSnapshotVersion()); + assertEquals(2, cache.getIfPresent("a").getInt()); + } + + @Test void platformReadCannotObserveInvalidatedQueuedValue() { + var plugin = mock(AdvancedCorePlugin.class, RETURNS_DEEP_STUBS); var manager = mock(UserDataManager.class); + when(plugin.getUserManager().getDataManager()).thenReturn(manager); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + when(manager.mustDeferSharedStorageAccess()).thenReturn(true); + var cache = new UserDataCache(manager, UUID.randomUUID()); cache.configureSharedStorage(v -> {}, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + var user = mock(com.bencodez.advancedcore.api.user.AdvancedCoreUser.class); + when(user.getPlugin()).thenReturn(plugin); when(user.getCache()).thenReturn(cache); + var data = new com.bencodez.advancedcore.api.user.UserData(user); + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); cache.invalidateStorageSnapshot("a"); + assertThrows(IllegalStateException.class, () -> data.getInt( + com.bencodez.advancedcore.api.user.UserStorage.MYSQL, "a", 0, + com.bencodez.advancedcore.api.user.UserDataFetchMode.DEFAULT)); + } + + @Test void invalidationKeepsStagedValueUnavailableThroughCheckpointDrain() throws Exception { + var cache = cache(); var entered = new CountDownLatch(1); var release = new CountDownLatch(1); + var executor = Executors.newSingleThreadExecutor(); + try { + var checkpoint = executor.submit(() -> cache.flushChangesAndRun(() -> { entered.countDown(); await(release); })); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertTrue(cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null).applied()); + cache.invalidateStorageSnapshot("a"); assertNull(cache.getIfPresent("a")); + release.countDown(); checkpoint.get(5, TimeUnit.SECONDS); + assertNull(cache.getIfPresent("a")); assertTrue(cache.hasChangesToProcess()); + } finally { release.countDown(); executor.shutdownNow(); } + } + + @Test void invalidationKeepsInFlightValueUnavailableAfterWriteCompletes() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var release = new CountDownLatch(1); + var written = new AtomicReference>(); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { entered.countDown(); await(release); written.set(values); }, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + var executor = Executors.newSingleThreadExecutor(); + try { + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); cache.invalidateStorageSnapshot("a"); + assertNull(cache.getIfPresent("a")); release.countDown(); flush.get(5, TimeUnit.SECONDS); + assertNull(cache.getIfPresent("a")); assertEquals(2, written.get().get("a").getInt()); + } finally { release.countDown(); executor.shutdownNow(); } + } + @Test void concurrentOperandGrowthCannotBypassBatchLimit() throws Exception { + var checked = new CountDownLatch(1); var grown = new CountDownLatch(1); + var operands = new ConcurrentHashMap() { + @Override public int size() { + int before = super.size(); checked.countDown(); await(grown); return before; + } + }; + var values = new HashMap(); + for (int i = 0; i < 65; i++) { + values.put("key" + i, new DataValueInt(1)); + if (i < 64) operands.put("key" + i, 1); + } + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var cache = new UserDataCache(manager, UUID.randomUUID()); cache.configureSharedStorage(v -> {}, Runnable::run); + cache.updateCache(values); var executor = Executors.newSingleThreadExecutor(); + try { + var mutation = executor.submit(() -> cache.tryMutateIntegers(operands, IntegerMutation.ADD, null)); + assertTrue(checked.await(5, TimeUnit.SECONDS)); operands.put("key64", 1); grown.countDown(); + var failure = assertThrows(ExecutionException.class, () -> mutation.get(5, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, failure.getCause()); + for (int i = 0; i < 65; i++) assertEquals(1, cache.getIfPresent("key" + i).getInt()); + assertFalse(cache.hasChangesToProcess()); + } finally { grown.countDown(); executor.shutdownNow(); } + } + @Test void notificationRejectionDoesNotHideAnAcceptedDebit() { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + doThrow(new RejectedExecutionException("notification executor unavailable")) + .when(manager).dispatchSharedUserDataNotification(any()); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> {}, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + var result = cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, keys -> {}); + assertTrue(result.applied()); assertEquals(1, result.previous().get("a")); + assertEquals(0, cache.getIfPresent("a").getInt()); assertTrue(cache.hasChangesToProcess()); + } + @Test void checkpointReplacementDoesNotPermitDoubleDebit() throws Exception { + var cache = cache(); var entered = new CountDownLatch(1); var replace = new CountDownLatch(1); + var replaced = new CountDownLatch(1); var finish = new CountDownLatch(1); + var executor = Executors.newSingleThreadExecutor(); + try { + var checkpoint = executor.submit(() -> cache.flushChangesAndRun(() -> { + entered.countDown(); await(replace); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1), "b", new DataValueInt(2)))); + replaced.countDown(); await(finish); + })); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertTrue(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + replace.countDown(); assertTrue(replaced.await(5, TimeUnit.SECONDS)); + assertFalse(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + finish.countDown(); checkpoint.get(5, TimeUnit.SECONDS); + assertEquals(0, cache.getIfPresent("a").getInt()); + } finally { replace.countDown(); finish.countDown(); executor.shutdownNow(); } + } + + @Test void refreshDuringInFlightWriteRetainsPriorIncrement() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + var users = mock(com.bencodez.advancedcore.api.user.UserManager.class); + when(plugin.getUserManager()).thenReturn(users); when(manager.getPlugin()).thenReturn(plugin); + when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var finish = new CountDownLatch(1); + var stored = new ConcurrentHashMap(); + stored.put("a", new DataValueInt(1)); stored.put("b", new DataValueInt(2)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { entered.countDown(); await(finish); stored.putAll(values); }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); + var executor = Executors.newSingleThreadExecutor(); + try { + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + cache.tryMutateIntegers(Map.of("b", 1), IntegerMutation.ADD, null); + long version = cache.getSharedSnapshotVersion(); var stale = new HashMap<>(stored); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + cache.updateSharedSnapshot(stale, version); + assertEquals(2, cache.getIfPresent("a").getInt()); + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + finish.countDown(); flush.get(5, TimeUnit.SECONDS); + cache.processChangesForSharedRuntime(); + assertEquals(3, cache.getIfPresent("a").getInt()); + assertEquals(3, stored.get("a").getInt()); + } finally { finish.countDown(); executor.shutdownNow(); } + } + + @Test void refreshCannotRestoreSpentCreditAfterInFlightWriteCompletes() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var finish = new CountDownLatch(1); + var stored = new ConcurrentHashMap(); + stored.put("a", new DataValueInt(1)); stored.put("b", new DataValueInt(2)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { entered.countDown(); await(finish); stored.putAll(values); }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); var executor = Executors.newSingleThreadExecutor(); + try { + assertTrue(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + cache.tryMutateIntegers(Map.of("b", 1), IntegerMutation.ADD, null); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + long version = cache.getSharedSnapshotVersion(); var stale = new HashMap<>(stored); + finish.countDown(); flush.get(5, TimeUnit.SECONDS); + cache.updateSharedSnapshot(stale, version); + assertEquals(0, cache.getIfPresent("a").getInt()); + assertFalse(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + assertEquals(0, stored.get("a").getInt()); + } finally { finish.countDown(); executor.shutdownNow(); } + } + + @Test void refreshStartedAfterCompletedWriteCanObserveLaterStorageValue() { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> {}, Runnable::run); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(1)))); + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + cache.processChangesForSharedRuntime(); + long version = cache.getSharedSnapshotVersion(); + cache.updateSharedSnapshot(new HashMap<>(Map.of("a", new DataValueInt(9))), version); + assertEquals(9, cache.getIfPresent("a").getInt()); + } + + @Test void successfulRetryFencesRefreshAfterMutationMarkersAreCleared() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var fail = new CountDownLatch(1); + var retryEntered = new CountDownLatch(1); var retryFinish = new CountDownLatch(1); + var attempts = new AtomicInteger(); var stored = new ConcurrentHashMap(); + stored.put("a", new DataValueInt(1)); stored.put("b", new DataValueInt(2)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { + if (attempts.getAndIncrement() == 0) { entered.countDown(); await(fail); throw new IllegalStateException("write failed"); } + retryEntered.countDown(); await(retryFinish); stored.putAll(values); + }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); var executor = Executors.newSingleThreadExecutor(); + try { + assertTrue(cache.tryMutateIntegers(Map.of("a", -1, "b", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + var first = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + cache.updateSharedSnapshot(new HashMap<>(stored), cache.getSharedSnapshotVersion()); + fail.countDown(); assertThrows(ExecutionException.class, () -> first.get(5, TimeUnit.SECONDS)); + var retry = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(retryEntered.await(5, TimeUnit.SECONDS)); + long version = cache.getSharedSnapshotVersion(); var stale = new HashMap<>(stored); + retryFinish.countDown(); retry.get(5, TimeUnit.SECONDS); + cache.updateSharedSnapshot(stale, version); + assertEquals(0, cache.getIfPresent("a").getInt()); + assertEquals(1, cache.getIfPresent("b").getInt()); + assertFalse(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + assertEquals(0, stored.get("a").getInt()); + } finally { fail.countDown(); retryFinish.countDown(); executor.shutdownNow(); } + } + + @Test void checkpointFailureStillPreservesStagedValuesAndNotification() throws Exception { + var cache = cache(); var entered = new CountDownLatch(1); var replace = new CountDownLatch(1); + var notices = new AtomicInteger(); var executor = Executors.newSingleThreadExecutor(); + try { + var checkpoint = executor.submit(() -> cache.flushChangesAndRun(() -> { + entered.countDown(); await(replace); + cache.updateCachePreservingPending(new HashMap<>(Map.of("a", new DataValueInt(1), "b", new DataValueInt(2)))); + throw new IllegalStateException("checkpoint failure"); + })); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertTrue(cache.tryMutateIntegers(Map.of("a", -1, "b", -1), IntegerMutation.DEBIT_POSITIVE, + keys -> notices.incrementAndGet()).applied()); + replace.countDown(); + assertThrows(ExecutionException.class, () -> checkpoint.get(5, TimeUnit.SECONDS)); + assertEquals(0, cache.getIfPresent("a").getInt()); + assertEquals(1, cache.getIfPresent("b").getInt()); + assertEquals(1, notices.get()); assertTrue(cache.hasChangesToProcess()); + } finally { replace.countDown(); executor.shutdownNow(); } + } + + @Test void failedInFlightWriteRequeuesNewerMutationsAfterRefresh() throws Exception { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + when(plugin.getUserManager()).thenReturn(mock(com.bencodez.advancedcore.api.user.UserManager.class)); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(mock(ScheduledExecutorService.class)); + var entered = new CountDownLatch(1); var finish = new CountDownLatch(1); var attempts = new AtomicInteger(); + var stored = new ConcurrentHashMap(); + stored.put("a", new DataValueInt(1)); stored.put("b", new DataValueInt(2)); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> { + if (attempts.getAndIncrement() == 0) { entered.countDown(); await(finish); throw new IllegalStateException("write failed"); } + stored.putAll(values); + }, Runnable::run); + cache.updateCache(new HashMap<>(stored)); var executor = Executors.newSingleThreadExecutor(); + try { + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + cache.tryMutateIntegers(Map.of("b", 1), IntegerMutation.ADD, null); + long version = cache.getSharedSnapshotVersion(); var stale = new HashMap<>(stored); + var flush = executor.submit(cache::processChangesForSharedRuntime); + assertTrue(entered.await(5, TimeUnit.SECONDS)); cache.updateSharedSnapshot(stale, version); + cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null); + finish.countDown(); assertThrows(ExecutionException.class, () -> flush.get(5, TimeUnit.SECONDS)); + assertEquals(3, cache.getIfPresent("a").getInt()); + cache.processChangesForSharedRuntime(); assertEquals(3, stored.get("a").getInt()); + } finally { finish.countDown(); executor.shutdownNow(); } + } + + private static void await(CountDownLatch latch) { + try { if (!latch.await(5, TimeUnit.SECONDS)) throw new IllegalStateException("fixture timeout"); } + catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IllegalStateException(e); } + } + private UserDataCache cache() { + var plugin = mock(AdvancedCorePlugin.class); var manager = mock(UserDataManager.class); + var timer = mock(ScheduledExecutorService.class); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(timer); + doAnswer(call -> { call.getArgument(0).run(); return null; }).when(manager).dispatchSharedUserDataNotification(any()); + var cache = new UserDataCache(manager, UUID.randomUUID()); + cache.configureSharedStorage(values -> {}, Runnable::run); + var values = new HashMap(); values.put("a", new DataValueInt(1)); values.put("b", new DataValueInt(2)); + cache.updateCache(values); return cache; + } + @Test void concurrentUpdatesCannotLoseCredits() throws Exception { + var cache = cache(); var executor = Executors.newFixedThreadPool(2); var start = new CountDownLatch(1); + try { + var first = executor.submit(() -> { start.await(); for (int i=0;i<500;i++) assertNotNull(cache.tryMutateIntegers(Map.of("a",1), IntegerMutation.ADD,null)); return null; }); + var second = executor.submit(() -> { start.await(); for (int i=0;i<500;i++) assertNotNull(cache.tryMutateIntegers(Map.of("a",1), IntegerMutation.ADD,null)); return null; }); + start.countDown(); first.get(5,TimeUnit.SECONDS); second.get(5,TimeUnit.SECONDS); + assertEquals(1001,cache.getIfPresent("a").getInt()); + } finally { executor.shutdownNow(); } + } + @Test void dualDebitReturnsOriginalBalancesAndSkipsEmptyPool() { + var cache=cache(); cache.tryMutateIntegers(Map.of("a",0),IntegerMutation.SET,null); + var result=cache.tryMutateIntegers(Map.of("a",-1,"b",-1),IntegerMutation.DEBIT_POSITIVE,null); + assertTrue(result.applied()); assertEquals(Map.of("a",0,"b",2),result.previous()); + assertEquals(0,cache.getIfPresent("a").getInt()); assertEquals(1,cache.getIfPresent("b").getInt()); + } + @Test void coldRetiredAndMissingKeysFailWithoutMutation() { + var cache=cache(); cache.invalidateStorageSnapshot("a"); + assertNull(cache.tryMutateIntegers(Map.of("a",1),IntegerMutation.ADD,null)); + assertEquals(2,cache.getIfPresent("b").getInt()); + cache.beginRemoval(); assertNull(cache.tryMutateIntegers(Map.of("b",1),IntegerMutation.ADD,null)); + } + @Test void callbackRunsOutsideMonitorAndCanReenter() { + var cache=cache(); var called=new AtomicInteger(); + var result=cache.tryMutateIntegers(Map.of("a",1),IntegerMutation.ADD,keys -> { + assertFalse(Thread.holdsLock(cache)); assertEquals(Set.of("a"),keys); called.incrementAndGet(); + cache.tryMutateIntegers(Map.of("b",1),IntegerMutation.ADD,null); + }); + assertTrue(result.applied()); assertEquals(1,called.get()); assertEquals(3,cache.getIfPresent("b").getInt()); + } + @Test void exclusiveFlushStagesOneBatchAndDelaysNotification() throws Exception { + var cache = cache(); var entered = new CountDownLatch(1); var release = new CountDownLatch(1); + var notices = new AtomicInteger(); var executor = Executors.newSingleThreadExecutor(); + try { + var flush = executor.submit(() -> cache.flushChangesAndRun(() -> { + entered.countDown(); + try { if (!release.await(5, TimeUnit.SECONDS)) throw new IllegalStateException("fixture timeout"); } + catch (InterruptedException failure) { Thread.currentThread().interrupt(); throw new IllegalStateException(failure); } + })); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + var result = cache.tryMutateIntegers(Map.of("a",-1,"b",-1),IntegerMutation.DEBIT_POSITIVE, + keys -> { assertFalse(Thread.holdsLock(cache)); notices.incrementAndGet(); }); + assertTrue(result.applied()); assertEquals(0,cache.getIfPresent("a").getInt()); + assertEquals(1,cache.getIfPresent("b").getInt()); assertEquals(0,notices.get()); + release.countDown(); flush.get(5,TimeUnit.SECONDS); assertEquals(1,notices.get()); + } finally { release.countDown(); executor.shutdownNow(); } + } + @Test void missingOneFieldDoesNotMutateAnyField() { + var cache=cache(); assertNull(cache.tryMutateIntegers(Map.of("a",1,"missing",1),IntegerMutation.ADD,null)); + assertEquals(1,cache.getIfPresent("a").getInt()); + } + @Test void emptyDebitChangesNothingAndAbsoluteSetAndClampRemainAtomic() { + var cache=cache(); cache.tryMutateIntegers(Map.of("a",0,"b",0),IntegerMutation.SET,null); + var result=cache.tryMutateIntegers(Map.of("a",-1,"b",-1),IntegerMutation.DEBIT_POSITIVE,null); + assertFalse(result.applied()); assertEquals(Map.of("a",0,"b",0),result.previous()); + cache.tryMutateIntegers(Map.of("a",-2),IntegerMutation.ADD_FLOOR_ZERO,null); + assertEquals(0,cache.getIfPresent("a").getInt()); + } + + @Test void overflowRejectsWholeBatchBeforeCacheQueueOrNotification() { + for (var operation : List.of(IntegerMutation.ADD, IntegerMutation.ADD_FLOOR_ZERO)) { + for (var boundary : List.of(Integer.MAX_VALUE, Integer.MIN_VALUE)) { + var cache = cache(); + cache.updateCache(new HashMap<>(Map.of("a", new DataValueInt(2), "b", new DataValueInt(boundary)))); + var operands = new LinkedHashMap(); + operands.put("a", 1); + operands.put("b", boundary == Integer.MAX_VALUE ? 1 : -1); + var notifications = new AtomicInteger(); + assertFalse(cache.hasChangesToProcess()); + assertThrows(ArithmeticException.class, () -> cache.tryMutateIntegers(operands, operation, + keys -> notifications.incrementAndGet())); + assertEquals(2, cache.getIfPresent("a").getInt()); + assertEquals(boundary, cache.getIfPresent("b").getInt()); + assertFalse(cache.hasChangesToProcess()); + assertEquals(0, notifications.get()); + } + } + } + + @Test void validBoundarySetDebitAndFloorOperationsRemainSupported() { + var cache = cache(); + assertTrue(cache.tryMutateIntegers(Map.of("a", Integer.MAX_VALUE), IntegerMutation.SET, null).applied()); + assertTrue(cache.tryMutateIntegers(Map.of("a", -1), IntegerMutation.DEBIT_POSITIVE, null).applied()); + assertEquals(Integer.MAX_VALUE - 1, cache.getIfPresent("a").getInt()); + assertTrue(cache.tryMutateIntegers(Map.of("a", 1), IntegerMutation.ADD, null).applied()); + assertEquals(Integer.MAX_VALUE, cache.getIfPresent("a").getInt()); + assertTrue(cache.tryMutateIntegers(Map.of("b", -3), IntegerMutation.ADD_FLOOR_ZERO, null).applied()); + assertEquals(0, cache.getIfPresent("b").getInt()); + } + + @Test void conditionalDebitsAndAwardsAcrossBothFieldsCannotLoseUpdates() throws Exception { + var cache=cache(); var executor=Executors.newFixedThreadPool(2); var start=new CountDownLatch(1); + try { + var debit=executor.submit(() -> { start.await(); return cache.tryMutateIntegers(Map.of("a",-1,"b",-1),IntegerMutation.DEBIT_POSITIVE,null); }); + var award=executor.submit(() -> { start.await(); return cache.tryMutateIntegers(Map.of("a",1,"b",1),IntegerMutation.ADD,null); }); + start.countDown(); assertTrue(debit.get(5,TimeUnit.SECONDS).applied()); assertTrue(award.get(5,TimeUnit.SECONDS).applied()); + assertEquals(1,cache.getIfPresent("a").getInt()); assertEquals(2,cache.getIfPresent("b").getInt()); + } finally { executor.shutdownNow(); } + } + @Test void absoluteSetAndIncrementAreSerializedWithoutLosingEitherOperation() throws Exception { + var cache=cache(); var executor=Executors.newFixedThreadPool(2); var start=new CountDownLatch(1); + try { + var set=executor.submit(() -> { start.await(); return cache.tryMutateIntegers(Map.of("a",10),IntegerMutation.SET,null); }); + var add=executor.submit(() -> { start.await(); return cache.tryMutateIntegers(Map.of("a",1),IntegerMutation.ADD,null); }); + start.countDown(); assertTrue(set.get(5,TimeUnit.SECONDS).applied()); assertTrue(add.get(5,TimeUnit.SECONDS).applied()); + int actual=cache.getIfPresent("a").getInt(); assertTrue(actual==10 || actual==11, "result must match a serial ordering"); + } finally { executor.shutdownNow(); } + } + @Test void rejectedSchedulingStillLeavesTheWholeAcceptedBatchQueued() { + var plugin=mock(AdvancedCorePlugin.class); var manager=mock(UserDataManager.class); var timer=mock(ScheduledExecutorService.class); + when(manager.getPlugin()).thenReturn(plugin); when(manager.getTimer()).thenReturn(timer); + when(timer.schedule(any(Runnable.class),anyLong(),any(TimeUnit.class))).thenThrow(new RejectedExecutionException("synthetic")); + var cache=new UserDataCache(manager,UUID.randomUUID()); cache.configureSharedStorage(values -> {},Runnable::run); + var values=new HashMap(); values.put("a",new DataValueInt(2)); values.put("b",new DataValueInt(2)); cache.updateCache(values); + var result=cache.tryMutateIntegers(Map.of("a",-1,"b",-1),IntegerMutation.DEBIT_POSITIVE,null); + assertTrue(result.applied()); assertEquals(1,cache.getIfPresent("a").getInt()); assertEquals(1,cache.getIfPresent("b").getInt()); + assertTrue(cache.hasChangesToProcess(),"accepted values must remain queued after rejection"); + } + + @Test void negativeDebitMustNotCreateCredits() { + var cache=cache(); assertThrows(IllegalArgumentException.class, + () -> cache.tryMutateIntegers(Map.of("a",1),IntegerMutation.DEBIT_POSITIVE,null)); + assertEquals(1,cache.getIfPresent("a").getInt()); + } +}