From d679f610d6c6c557d23a2e42d7dbc6e8f96a5db1 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 1 Nov 2026 19:18:40 -0600 Subject: [PATCH 1/2] Test shared-cache refill after invalidation --- .../SharedCacheCleanupPrimaryThreadTest.java | 72 ++++++++++++++++++- 1 file changed, 70 insertions(+), 2 deletions(-) diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java index 0b847d67b..3104ce42f 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java @@ -910,9 +910,26 @@ void invalidatedCompletedCacheQueuesOneWorkerRefillOnPrimaryThreadAccess() throw Set completedPopulations = (Set) completed.get(manager); completedPopulations.add(uuid); SqlUserBackend backend = mock(SqlUserBackend.class); + com.bencodez.advancedcore.core.user.storage.SqlUserStorage storage = + mock(com.bencodez.advancedcore.core.user.storage.SqlUserStorage.class); + UserManager users = mock(UserManager.class); + AdvancedCoreUser user = mock(AdvancedCoreUser.class); + when(plugin.getUserManager()).thenReturn(users); + when(plugin.getStorageType()).thenReturn(UserStorage.MYSQL); + when(users.getDataManager()).thenReturn(manager); + when(users.getUser(uuid, false)).thenReturn(user); + when(user.getPlugin()).thenReturn(plugin); + when(user.getUUID()).thenReturn(uuid.toString()); + when(user.getUserData()).thenReturn(new UserData(user)); when(backend.isOpen()).thenReturn(true); - manager.bindSharedSqlBackend(backend, (user, operation) -> { - throw new AssertionError("primary thread must not admit cache storage work"); + when(backend.storageType()).thenReturn(UserStorage.MYSQL); + when(backend.user(uuid)).thenReturn(storage); + when(storage.readRow(UserStorage.MYSQL)).thenReturn(java.util.List.of( + new com.bencodez.simpleapi.sql.Column("Points", new DataValueInt(57)))); + Thread primary = Thread.currentThread(); + manager.bindSharedSqlBackend(backend, (id, operation) -> { + assertTrue(Thread.currentThread() != primary, "storage admission must not run on the primary thread"); + operation.run(); }); Server server = mock(Server.class); try (var bukkit = mockStatic(Bukkit.class)) { @@ -927,6 +944,57 @@ void invalidatedCompletedCacheQueuesOneWorkerRefillOnPrimaryThreadAccess() throw assertSame(cache, manager.getCache(uuid)); verify(worker, times(1)).execute(any(Runnable.class)); assertFalse(cache.hasPublishedStorageSnapshot()); + verify(backend, never()).user(any()); + ArgumentCaptor task = ArgumentCaptor.forClass(Runnable.class); + verify(worker).execute(task.capture()); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + executor.submit(task.getValue()).get(5, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + verify(storage, times(2)).readRow(UserStorage.MYSQL); + assertTrue(cache.hasPublishedStorageSnapshot()); + assertEquals(57, cache.getIfPresent("Points").getInt()); + // The second worker task delivers the refresh notification, not another refill. + verify(worker, times(2)).execute(any(Runnable.class)); + assertSame(cache, manager.getCache(uuid)); + verify(worker, times(2)).execute(any(Runnable.class)); + + // A second invalidation during an older read must fence that read's snapshot. + org.mockito.Mockito.clearInvocations(worker); + cache.invalidateStorageSnapshot("Points"); + AtomicInteger reads = new AtomicInteger(); + when(storage.readRow(UserStorage.MYSQL)).thenAnswer(invocation -> { + if (reads.incrementAndGet() == 2) cache.invalidateStorageSnapshot("Points"); + return java.util.List.of(new com.bencodez.simpleapi.sql.Column("Points", new DataValueInt(31))); + }); + manager.getCache(uuid); + ArgumentCaptor staleTask = ArgumentCaptor.forClass(Runnable.class); + verify(worker).execute(staleTask.capture()); + ExecutorService staleExecutor = Executors.newSingleThreadExecutor(); + try { + staleExecutor.submit(staleTask.getValue()).get(5, TimeUnit.SECONDS); + } finally { + staleExecutor.shutdownNow(); + } + assertFalse(cache.hasPublishedStorageSnapshot()); + assertNull(cache.getIfPresent("Points")); + org.mockito.Mockito.clearInvocations(worker); + org.mockito.Mockito.doReturn(java.util.List.of( + new com.bencodez.simpleapi.sql.Column("Points", new DataValueInt(72)))) + .when(storage).readRow(UserStorage.MYSQL); + manager.getCache(uuid); + ArgumentCaptor freshTask = ArgumentCaptor.forClass(Runnable.class); + verify(worker).execute(freshTask.capture()); + ExecutorService freshExecutor = Executors.newSingleThreadExecutor(); + try { + freshExecutor.submit(freshTask.getValue()).get(5, TimeUnit.SECONDS); + } finally { + freshExecutor.shutdownNow(); + } + assertTrue(cache.hasPublishedStorageSnapshot()); + assertEquals(72, cache.getIfPresent("Points").getInt()); } } From 3a8cf105e60c9b0029387116bea67354cd706d3b Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 1 Nov 2026 20:14:12 -0600 Subject: [PATCH 2/2] Test pending writes and queued cache notifications --- .../SharedCacheCleanupPrimaryThreadTest.java | 45 ++++++++++++++++--- 1 file changed, 40 insertions(+), 5 deletions(-) diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java index 3104ce42f..e52bdfe35 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/SharedCacheCleanupPrimaryThreadTest.java @@ -902,7 +902,8 @@ void invalidatedCompletedCacheQueuesOneWorkerRefillOnPrimaryThreadAccess() throw timer.set(manager, worker); UUID uuid = UUID.randomUUID(); UserDataCache cache = new UserDataCache(manager, uuid); - cache.updateCache(new HashMap<>(Map.of("Points", new DataValueInt(38)))); + cache.updateCache(new HashMap<>(Map.of("Points", new DataValueInt(38), + "Rating", new DataValueInt(38), "Coins", new DataValueInt(4)))); manager.getUserDataCache().put(uuid, cache); Field completed = UserDataManager.class.getDeclaredField("completedSharedCachePopulations"); completed.setAccessible(true); @@ -924,19 +925,33 @@ void invalidatedCompletedCacheQueuesOneWorkerRefillOnPrimaryThreadAccess() throw when(backend.isOpen()).thenReturn(true); when(backend.storageType()).thenReturn(UserStorage.MYSQL); when(backend.user(uuid)).thenReturn(storage); - when(storage.readRow(UserStorage.MYSQL)).thenReturn(java.util.List.of( - new com.bencodez.simpleapi.sql.Column("Points", new DataValueInt(57)))); + AtomicInteger initialReads = new AtomicInteger(); + when(storage.readRow(UserStorage.MYSQL)).thenAnswer(invocation -> { + // The second read occurs after the refresh's initial pending-write flush. + // Queue a new write while the authoritative read is in flight. + if (initialReads.incrementAndGet() == 2) { + assertTrue(cache.tryAddChangeBeforeDeferredSharedFlush(new UserDataChangeInt("Coins", 9))); + } + return java.util.List.of( + new com.bencodez.simpleapi.sql.Column("Points", new DataValueInt(57)), + new com.bencodez.simpleapi.sql.Column("Rating", new DataValueInt(57)), + new com.bencodez.simpleapi.sql.Column("Coins", new DataValueInt(4))); + }); Thread primary = Thread.currentThread(); manager.bindSharedSqlBackend(backend, (id, operation) -> { assertTrue(Thread.currentThread() != primary, "storage admission must not run on the primary thread"); operation.run(); }); + cache.configureSharedStorage(values -> storage.writeValues(UserStorage.MYSQL, values), Runnable::run); Server server = mock(Server.class); try (var bukkit = mockStatic(Bukkit.class)) { bukkit.when(Bukkit::getServer).thenReturn(server); bukkit.when(Bukkit::isPrimaryThread).thenReturn(true); assertSame(cache, manager.getCache(uuid)); verify(worker, never()).execute(any(Runnable.class)); + // A queued Points write is flushed before the refill reads storage. + assertTrue(cache.tryAddChangeBeforeDeferredSharedFlush(new UserDataChangeInt("Points", 57))); + assertTrue(cache.hasChangesToProcess()); cache.invalidateStorageSnapshot("Points"); assertSame(cache, manager.getCache(uuid)); verify(worker, times(1)).execute(any(Runnable.class)); @@ -954,10 +969,30 @@ void invalidatedCompletedCacheQueuesOneWorkerRefillOnPrimaryThreadAccess() throw executor.shutdownNow(); } verify(storage, times(2)).readRow(UserStorage.MYSQL); + assertNull(manager.getLastDeferredStorageFailure()); assertTrue(cache.hasPublishedStorageSnapshot()); assertEquals(57, cache.getIfPresent("Points").getInt()); - // The second worker task delivers the refresh notification, not another refill. - verify(worker, times(2)).execute(any(Runnable.class)); + assertEquals(9, cache.getIfPresent("Coins").getInt()); + assertTrue(cache.hasChangesToProcess()); + @SuppressWarnings("unchecked") + ArgumentCaptor> flushed = + ArgumentCaptor.forClass(HashMap.class); + verify(storage).writeValues(eq(UserStorage.MYSQL), flushed.capture()); + assertEquals(Set.of("Points"), flushed.getValue().keySet()); + assertEquals(57, flushed.getValue().get("Points").getInt()); + // The second task delivers the refresh notification after lifecycle admission. + ArgumentCaptor notifications = ArgumentCaptor.forClass(Runnable.class); + verify(worker, times(2)).execute(notifications.capture()); + ExecutorService notifier = Executors.newSingleThreadExecutor(); + try { + notifier.submit(notifications.getAllValues().get(1)).get(5, TimeUnit.SECONDS); + } finally { + notifier.shutdownNow(); + } + verify(users).onChange(user, "Points"); + ArgumentCaptor changed = ArgumentCaptor.forClass(String[].class); + verify(users, times(2)).onChange(eq(user), changed.capture()); + assertEquals(Set.of("Rating", "Coins"), Set.of(changed.getAllValues().get(1))); assertSame(cache, manager.getCache(uuid)); verify(worker, times(2)).execute(any(Runnable.class));