From fd1676d24aafbaaccc6fdfe4a04bba3bd5a32462 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 17:00:56 -0600 Subject: [PATCH 1/6] Make user cache change flushing race-safe --- .../api/user/usercache/UserDataCache.java | 137 ++++++++++-------- 1 file changed, 75 insertions(+), 62 deletions(-) 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 c4a00bcc7a..2a41db03ac 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 @@ -6,6 +6,7 @@ import java.util.Queue; import java.util.UUID; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import com.bencodez.advancedcore.api.user.AdvancedCoreUser; @@ -22,7 +23,7 @@ public class UserDataCache { private Queue cachedChanges; - private UserDataManager manager; + private final UserDataManager manager; private boolean scheduled = false; @Getter private UUID uuid; @@ -35,6 +36,9 @@ public UserDataCache(UserDataManager manager, UUID uuid) { } public synchronized void addChange(UserDataChange change, boolean queue) { + if (change == null || cache == null || cachedChanges == null) { + return; + } cache.put(change.getKey(), change.toUserDataValue()); if (queue) { cachedChanges.add(change); @@ -42,11 +46,10 @@ public synchronized void addChange(UserDataChange change, boolean queue) { scheduleChanges(); } } - } - public UserDataCache cache() { - if (uuid != null) { + public synchronized UserDataCache cache() { + if (uuid != null && cache != null) { AdvancedCoreUser user = getUser(); ArrayList keys = user.getUserData().getKeys(); HashMap data = user.getUserData().getValues(); @@ -58,22 +61,18 @@ public UserDataCache cache() { DataValue dataValue = data.get(key); manager.getPlugin().devDebug("Caching " + dataValue.getTypeName() + " " + key + " for " + uuid.toString() + ", value: " + dataValue.toString()); - // temp try/catch to prevent plugin failures try { - if (cache.containsKey(key)) { - if (!cache.get(key).toString().equals(dataValue.toString())) { - changedKeys.add(key); - } + if (cache.containsKey(key) && !cache.get(key).toString().equals(dataValue.toString())) { + changedKeys.add(key); } } catch (Exception e) { - e.printStackTrace(); + manager.getPlugin().debug(e); } cache.put(key, dataValue); } else { manager.getPlugin().devDebug("Loading default cache value for " + key + " for " + uuid.toString()); cache.put(key, dataKey.getDefault()); } - } if (!changedKeys.isEmpty()) { manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(changedKeys)); @@ -85,14 +84,16 @@ public UserDataCache cache() { return this; } - public void clearCache() { + public synchronized void clearCache() { if (hasChangesToProcess()) { processChanges(); } - cache.clear(); + if (cache != null) { + cache.clear(); + } } - public void clearChanges() { + public synchronized void clearChanges() { if (hasChangesToProcess()) { processChanges(); } @@ -102,9 +103,12 @@ public void displayCache() { manager.getPlugin().devDebug(displayCacheStringList().toString()); } - public ArrayList displayCacheStringList() { + public synchronized ArrayList displayCacheStringList() { ArrayList list = new ArrayList<>(); list.add("Current cache for " + uuid + ": "); + if (cache == null) { + return list; + } for (Entry entry : getCache().entrySet()) { if (entry.getValue().isBoolean()) { list.add(entry.getKey() + "=" + entry.getValue().getBoolean()); @@ -117,88 +121,97 @@ public ArrayList displayCacheStringList() { return list; } - public void dump() { + public synchronized void dump() { if (hasChangesToProcess()) { processChanges(); } cache = null; cachedChanges = null; uuid = null; + scheduled = false; } public AdvancedCoreUser getUser() { return manager.getPlugin().getUserManager().getUser(uuid, false); } - public boolean hasCache() { - return !cache.isEmpty(); + public synchronized boolean hasCache() { + return cache != null && !cache.isEmpty(); } - public boolean hasChangesToProcess() { - return !cachedChanges.isEmpty(); + public synchronized boolean hasChangesToProcess() { + return cachedChanges != null && !cachedChanges.isEmpty(); } - public boolean isCached(String key) { - if (cache != null) { - return cache.containsKey(key); - } - return false; + public synchronized boolean isCached(String key) { + return cache != null && cache.containsKey(key); } public void processChanges() { - if (uuid != null) { - if (cachedChanges.size() > 0) { - manager.getPlugin() - .extraDebug("Processing changes for " + uuid.toString() + ", Changes: " + cachedChanges.size()); - AdvancedCoreUser user = getUser(); - HashMap values = new HashMap<>(); - ArrayList keys = new ArrayList<>(); - while (!cachedChanges.isEmpty()) { - UserDataChange change = cachedChanges.poll(); - values.put(change.getKey(), change.toUserDataValue()); - keys.add(change.getKey()); - change.dump(); - // manager.getPlugin().extraDebug("Processing change for " + change.getKey()); - } - if (!values.isEmpty()) { - user.getUserData().setValues(values); - } - manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys)); + UUID currentUuid; + ArrayList changes = new ArrayList<>(); + synchronized (this) { + currentUuid = uuid; + if (currentUuid == null || cachedChanges == null || cachedChanges.isEmpty()) { + return; } + UserDataChange change; + while ((change = cachedChanges.poll()) != null) { + changes.add(change); + } + } + + manager.getPlugin().extraDebug("Processing changes for " + currentUuid + ", Changes: " + changes.size()); + AdvancedCoreUser user = manager.getPlugin().getUserManager().getUser(currentUuid, false); + HashMap values = new HashMap<>(); + ArrayList keys = new ArrayList<>(); + for (UserDataChange change : changes) { + values.put(change.getKey(), change.toUserDataValue()); + keys.add(change.getKey()); + change.dump(); } + if (!values.isEmpty()) { + user.getUserData().setValues(values); + } + manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys)); } public void processChangesAsync() { - if (uuid != null && cachedChanges.size() > 0) { - manager.getPlugin().getTimer().execute(new Runnable() { - - @Override - public void run() { - processChanges(); - } - }); + if (uuid != null && hasChangesToProcess()) { + manager.getPlugin().getTimer().execute(this::processChanges); } } - private void scheduleChanges() { + private synchronized void scheduleChanges() { + if (scheduled || cachedChanges == null || cachedChanges.isEmpty()) { + return; + } manager.getPlugin().debug("Schedule changes"); scheduled = true; - - manager.getTimer().schedule(new Runnable() { - - @Override - public void run() { + try { + manager.getTimer().schedule(() -> { try { processChanges(); - scheduled = false; } catch (Exception e) { manager.getPlugin().debug(e); + } finally { + onScheduledFlushComplete(); } - } - }, 3, TimeUnit.SECONDS); + }, 3, TimeUnit.SECONDS); + } catch (RejectedExecutionException e) { + scheduled = false; + manager.getPlugin().debug(e); + } + } + + private synchronized void onScheduledFlushComplete() { + scheduled = false; + if (cachedChanges != null && !cachedChanges.isEmpty()) { + scheduleChanges(); + } } - public void updateCache(HashMap tempCache) { - cache = tempCache; + public synchronized void updateCache(HashMap tempCache) { + cache = tempCache == null ? new HashMap<>() : new HashMap<>(tempCache); } } From 3ddabdb67a4009386037e5be5dc658f51b81d3e9 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 17:01:18 -0600 Subject: [PATCH 2/6] Test user cache flush scheduling --- .../user/UserDataCacheSchedulingTest.java | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java new file mode 100644 index 0000000000..d33d7a8936 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java @@ -0,0 +1,74 @@ +package com.bencodez.advancedcore.tests.user; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +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.UserDataManager; +import com.bencodez.advancedcore.api.user.usercache.change.UserDataChangeString; +import com.bencodez.simpleapi.sql.data.DataValue; +import com.bencodez.simpleapi.sql.data.DataValueString; + +public class UserDataCacheSchedulingTest { + + @Test + @SuppressWarnings("rawtypes") + public void changesQueuedDuringScheduledFlushAreRescheduled() { + AdvancedCorePlugin plugin = mock(AdvancedCorePlugin.class); + UserDataManager manager = mock(UserDataManager.class); + ScheduledExecutorService timer = mock(ScheduledExecutorService.class); + ScheduledFuture future = mock(ScheduledFuture.class); + List scheduledTasks = new ArrayList<>(); + + when(manager.getPlugin()).thenReturn(plugin); + when(manager.getTimer()).thenReturn(timer); + doAnswer(invocation -> { + scheduledTasks.add(invocation.getArgument(0)); + return future; + }).when(timer).schedule(any(Runnable.class), eq(3L), eq(TimeUnit.SECONDS)); + + UserDataCache cache = spy(new UserDataCache(manager, UUID.randomUUID())); + doAnswer(invocation -> { + cache.addChange(new UserDataChangeString("second", "two"), true); + return null; + }).when(cache).processChanges(); + + cache.addChange(new UserDataChangeString("first", "one"), true); + assertEquals(1, scheduledTasks.size()); + + scheduledTasks.get(0).run(); + + assertEquals(2, scheduledTasks.size(), "a change queued during the flush must schedule another flush"); + } + + @Test + public void updateCacheDefensivelyCopiesInput() { + UserDataManager manager = mock(UserDataManager.class); + UserDataCache cache = new UserDataCache(manager, UUID.randomUUID()); + HashMap values = new HashMap<>(); + values.put("PlayerName", new DataValueString("Ben")); + + cache.updateCache(values); + values.clear(); + + assertTrue(cache.isCached("PlayerName")); + } +} From 439729dc94c002bec939bd6bcafe5bdd01034508 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 18:06:07 -0600 Subject: [PATCH 3/6] Preserve user changes when flush preparation fails --- .../api/user/usercache/UserDataCache.java | 47 ++++++++++++++----- 1 file changed, 34 insertions(+), 13 deletions(-) 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 2a41db03ac..8c809d9b03 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 @@ -161,19 +161,40 @@ public void processChanges() { } } - manager.getPlugin().extraDebug("Processing changes for " + currentUuid + ", Changes: " + changes.size()); - AdvancedCoreUser user = manager.getPlugin().getUserManager().getUser(currentUuid, false); - HashMap values = new HashMap<>(); - ArrayList keys = new ArrayList<>(); - for (UserDataChange change : changes) { - values.put(change.getKey(), change.toUserDataValue()); - keys.add(change.getKey()); - change.dump(); - } - if (!values.isEmpty()) { - user.getUserData().setValues(values); - } - manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys)); + boolean persisted = false; + try { + manager.getPlugin().extraDebug("Processing changes for " + currentUuid + ", Changes: " + changes.size()); + AdvancedCoreUser user = manager.getPlugin().getUserManager().getUser(currentUuid, false); + HashMap values = new HashMap<>(); + ArrayList keys = new ArrayList<>(); + for (UserDataChange change : changes) { + values.put(change.getKey(), change.toUserDataValue()); + keys.add(change.getKey()); + } + if (!values.isEmpty()) { + user.getUserData().setValues(values); + } + persisted = true; + manager.getPlugin().getUserManager().onChange(user, ArrayUtils.convert(keys)); + for (UserDataChange change : changes) { + change.dump(); + } + } catch (RuntimeException | Error e) { + if (!persisted) { + requeueChanges(changes); + } + throw e; + } + } + + 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; } public void processChangesAsync() { From fe9d4ddad6ca0e56f71c37cc4561cc7822aa2127 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 18:06:26 -0600 Subject: [PATCH 4/6] Test failed user cache flush restoration --- .../user/UserDataCacheSchedulingTest.java | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java index d33d7a8936..41f508dd64 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java @@ -1,6 +1,7 @@ package com.bencodez.advancedcore.tests.user; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; @@ -17,12 +18,16 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import com.bencodez.advancedcore.AdvancedCorePlugin; +import com.bencodez.advancedcore.api.user.AdvancedCoreUser; +import com.bencodez.advancedcore.api.user.UserManager; import com.bencodez.advancedcore.api.user.usercache.UserDataCache; import com.bencodez.advancedcore.api.user.usercache.UserDataManager; +import com.bencodez.advancedcore.api.user.usercache.change.UserDataChange; import com.bencodez.advancedcore.api.user.usercache.change.UserDataChangeString; import com.bencodez.simpleapi.sql.data.DataValue; import com.bencodez.simpleapi.sql.data.DataValueString; @@ -59,6 +64,45 @@ public void changesQueuedDuringScheduledFlushAreRescheduled() { assertEquals(2, scheduledTasks.size(), "a change queued during the flush must schedule another flush"); } + @Test + @SuppressWarnings("rawtypes") + public void failedFlushPreparationRestoresDrainedChanges() { + AdvancedCorePlugin plugin = mock(AdvancedCorePlugin.class); + UserManager userManager = mock(UserManager.class); + AdvancedCoreUser user = mock(AdvancedCoreUser.class); + UserDataManager manager = mock(UserDataManager.class); + ScheduledExecutorService timer = mock(ScheduledExecutorService.class); + ScheduledFuture future = mock(ScheduledFuture.class); + UUID uuid = UUID.randomUUID(); + + when(manager.getPlugin()).thenReturn(plugin); + when(manager.getTimer()).thenReturn(timer); + when(plugin.getUserManager()).thenReturn(userManager); + when(userManager.getUser(uuid, false)).thenReturn(user); + doReturn(future).when(timer).schedule(any(Runnable.class), eq(3L), eq(TimeUnit.SECONDS)); + + AtomicInteger conversions = new AtomicInteger(); + UserDataChange failingChange = new UserDataChange("Custom") { + @Override + public void dump() { + } + + @Override + public DataValue toUserDataValue() { + if (conversions.getAndIncrement() == 0) { + return new DataValueString("value"); + } + throw new IllegalStateException("conversion failed"); + } + }; + + UserDataCache cache = new UserDataCache(manager, uuid); + cache.addChange(failingChange, true); + + assertThrows(IllegalStateException.class, cache::processChanges); + assertTrue(cache.hasChangesToProcess(), "failed preparation must leave the drained change queued for retry"); + } + @Test public void updateCacheDefensivelyCopiesInput() { UserDataManager manager = mock(UserDataManager.class); From 618cb024e17c4ac7bc0c84f96fb1b4f249ac7c11 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 18:31:39 -0600 Subject: [PATCH 5/6] Wait for in-flight user cache flushes before dump --- .../api/user/usercache/UserDataCache.java | 37 ++++++++++++++++--- 1 file changed, 31 insertions(+), 6 deletions(-) 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 8c809d9b03..b2b60078bd 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 @@ -25,6 +25,7 @@ public class UserDataCache { private final UserDataManager manager; private boolean scheduled = false; + private int inFlightBatches = 0; @Getter private UUID uuid; @@ -121,14 +122,28 @@ public synchronized ArrayList displayCacheStringList() { return list; } - public synchronized void dump() { - if (hasChangesToProcess()) { + public void dump() { + while (true) { processChanges(); + synchronized (this) { + while (inFlightBatches > 0) { + try { + wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + if (cachedChanges != null && !cachedChanges.isEmpty()) { + continue; + } + cache = null; + cachedChanges = null; + uuid = null; + scheduled = false; + return; + } } - cache = null; - cachedChanges = null; - uuid = null; - scheduled = false; } public AdvancedCoreUser getUser() { @@ -159,6 +174,7 @@ public void processChanges() { while ((change = cachedChanges.poll()) != null) { changes.add(change); } + inFlightBatches++; } boolean persisted = false; @@ -184,6 +200,8 @@ public void processChanges() { requeueChanges(changes); } throw e; + } finally { + finishInFlightBatch(); } } @@ -197,6 +215,13 @@ private synchronized void requeueChanges(ArrayList changes) { cachedChanges = restored; } + private synchronized void finishInFlightBatch() { + if (inFlightBatches > 0) { + inFlightBatches--; + } + notifyAll(); + } + public void processChangesAsync() { if (uuid != null && hasChangesToProcess()) { manager.getPlugin().getTimer().execute(this::processChanges); From fbc7a52ee433159bd3306770604cd918cf52c4ce Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 24 Aug 2026 18:32:13 -0600 Subject: [PATCH 6/6] Test dump waits for in-flight cache batches --- .../user/UserDataCacheSchedulingTest.java | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java index 41f508dd64..eaf79a2554 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataCacheSchedulingTest.java @@ -1,6 +1,7 @@ package com.bencodez.advancedcore.tests.user; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -15,6 +16,7 @@ import java.util.HashMap; import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -24,6 +26,7 @@ import com.bencodez.advancedcore.AdvancedCorePlugin; import com.bencodez.advancedcore.api.user.AdvancedCoreUser; +import com.bencodez.advancedcore.api.user.UserData; import com.bencodez.advancedcore.api.user.UserManager; import com.bencodez.advancedcore.api.user.usercache.UserDataCache; import com.bencodez.advancedcore.api.user.usercache.UserDataManager; @@ -103,6 +106,83 @@ public DataValue toUserDataValue() { assertTrue(cache.hasChangesToProcess(), "failed preparation must leave the drained change queued for retry"); } + @Test + @SuppressWarnings("rawtypes") + public void dumpWaitsForFailedInFlightBatchToRequeueAndRetry() throws Exception { + AdvancedCorePlugin plugin = mock(AdvancedCorePlugin.class); + UserManager userManager = mock(UserManager.class); + AdvancedCoreUser user = mock(AdvancedCoreUser.class); + UserData userData = mock(UserData.class); + UserDataManager manager = mock(UserDataManager.class); + ScheduledExecutorService timer = mock(ScheduledExecutorService.class); + ScheduledFuture future = mock(ScheduledFuture.class); + UUID uuid = UUID.randomUUID(); + + when(manager.getPlugin()).thenReturn(plugin); + when(manager.getTimer()).thenReturn(timer); + when(plugin.getUserManager()).thenReturn(userManager); + when(userManager.getUser(uuid, false)).thenReturn(user); + when(user.getUserData()).thenReturn(userData); + doReturn(future).when(timer).schedule(any(Runnable.class), eq(3L), eq(TimeUnit.SECONDS)); + + CountDownLatch inFlight = new CountDownLatch(1); + CountDownLatch releaseFailure = new CountDownLatch(1); + CountDownLatch dumpFinished = new CountDownLatch(1); + AtomicInteger conversions = new AtomicInteger(); + AtomicInteger dumps = new AtomicInteger(); + UserDataChange change = new UserDataChange("Custom") { + @Override + public void dump() { + dumps.incrementAndGet(); + } + + @Override + public DataValue toUserDataValue() { + int call = conversions.incrementAndGet(); + if (call == 2) { + inFlight.countDown(); + try { + releaseFailure.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + throw new IllegalStateException("conversion failed while batch is in flight"); + } + return new DataValueString("value"); + } + }; + + UserDataCache cache = new UserDataCache(manager, uuid); + cache.addChange(change, true); + + Thread flushThread = new Thread(() -> { + try { + cache.processChanges(); + } catch (IllegalStateException expected) { + // The failed batch must be restored for dump() to retry. + } + }); + flushThread.start(); + assertTrue(inFlight.await(1, TimeUnit.SECONDS)); + + Thread dumpThread = new Thread(() -> { + cache.dump(); + dumpFinished.countDown(); + }); + dumpThread.start(); + assertFalse(dumpFinished.await(100, TimeUnit.MILLISECONDS), + "dump must wait while a drained batch is still in flight"); + + releaseFailure.countDown(); + flushThread.join(1000); + dumpThread.join(1000); + + assertEquals(0L, dumpFinished.getCount(), "dump should finish after the failed batch is restored and retried"); + assertEquals(1, dumps.get(), "the restored change must be persisted and dumped exactly once before disposal"); + assertFalse(cache.hasChangesToProcess()); + } + @Test public void updateCacheDefensivelyCopiesInput() { UserDataManager manager = mock(UserDataManager.class);