Skip to content

Commit 5fd805d

Browse files
BenCodezBenCodez
committed
Fence queued cache snapshots by session generation
1 parent ca9af6c commit 5fd805d

2 files changed

Lines changed: 75 additions & 19 deletions

File tree

‎AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/usercache/UserDataManager.java‎

Lines changed: 28 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -75,11 +75,13 @@ public class UserDataManager {
7575
private final Set<UUID> completedSharedCachePopulations = ConcurrentHashMap.newKeySet();
7676
private final ConcurrentHashMap<UUID, SharedCachePopulationState> sharedCachePopulationStates = new ConcurrentHashMap<>();
7777
/** Live storage identities, maintained from platform join/quit events. */
78-
// TRUE/FALSE records the latest player-event state. Offline tombstones keep a
79-
// concurrent Folia snapshot from re-adding a player whose quit event won the
80-
// race; a later snapshot removes tombstones once Bukkit no longer lists them.
81-
private final ConcurrentHashMap<UUID, Boolean> onlineUserSessions = new ConcurrentHashMap<>();
78+
// Generation-stamped offline tombstones prevent an older queued platform
79+
// snapshot from erasing a newer quit and letting a later stale snapshot
80+
// resurrect that session.
81+
private final ConcurrentHashMap<UUID, OnlineSessionState> onlineUserSessions = new ConcurrentHashMap<>();
82+
private final AtomicLong onlineSessionGeneration = new AtomicLong();
8283
private final Object[] onlineSessionLocks = createOnlineSessionLocks();
84+
private record OnlineSessionState(boolean online, long generation) {}
8385

8486
private static Object[] createOnlineSessionLocks() {
8587
Object[] locks = new Object[64];
@@ -1191,9 +1193,10 @@ public void clearNonNeededCachedUsers() {
11911193
if (storageUuid != null) platformOnline.add(storageUuid);
11921194
}
11931195
}
1196+
long snapshotGeneration = onlineSessionGeneration.get();
11941197
try {
11951198
timer.execute(() -> {
1196-
try { clearNonNeededCachedUsers(reconcileOnlineSnapshot(platformOnline)); }
1199+
try { clearNonNeededCachedUsers(reconcileOnlineSnapshot(platformOnline, snapshotGeneration)); }
11971200
catch (RuntimeException | Error failure) {
11981201
reportDeferredStorageFailure(failure);
11991202
throw failure;
@@ -1212,19 +1215,22 @@ public void clearNonNeededCachedUsers() {
12121215
}
12131216
}
12141217

1215-
private Set<UUID> reconcileOnlineSnapshot(Set<UUID> platformOnline) {
1218+
private Set<UUID> reconcileOnlineSnapshot(Set<UUID> platformOnline, long snapshotGeneration) {
12161219
java.util.HashSet<UUID> online = new java.util.HashSet<>();
12171220
for (UUID uuid : platformOnline) {
12181221
synchronized (onlineSessionLock(uuid)) {
1219-
Boolean state = onlineUserSessions.compute(uuid,
1220-
(ignored, current) -> Boolean.FALSE.equals(current) ? Boolean.FALSE : Boolean.TRUE);
1221-
if (Boolean.TRUE.equals(state)) online.add(uuid);
1222+
OnlineSessionState state = onlineUserSessions.compute(uuid,
1223+
(ignored, current) -> current == null
1224+
? new OnlineSessionState(true, snapshotGeneration) : current);
1225+
if (state.online()) online.add(uuid);
12221226
}
12231227
}
12241228
for (UUID uuid : Set.copyOf(onlineUserSessions.keySet())) {
12251229
synchronized (onlineSessionLock(uuid)) {
1226-
if (Boolean.FALSE.equals(onlineUserSessions.get(uuid)) && !platformOnline.contains(uuid)) {
1227-
onlineUserSessions.remove(uuid, Boolean.FALSE);
1230+
OnlineSessionState state = onlineUserSessions.get(uuid);
1231+
if (state != null && !state.online() && state.generation() <= snapshotGeneration
1232+
&& !platformOnline.contains(uuid)) {
1233+
onlineUserSessions.remove(uuid, state);
12281234
}
12291235
}
12301236
}
@@ -1245,7 +1251,7 @@ private UUID onlineStorageUuid(Player player) {
12451251

12461252
public void markUserOnline(UUID uuid) {
12471253
if (uuid != null) synchronized (onlineSessionLock(uuid)) {
1248-
onlineUserSessions.put(uuid, Boolean.TRUE);
1254+
onlineUserSessions.put(uuid, new OnlineSessionState(true, onlineSessionGeneration.incrementAndGet()));
12491255
}
12501256
}
12511257

@@ -1256,7 +1262,7 @@ public void markUserOnline(Player player) {
12561262

12571263
public void markUserOffline(UUID uuid) {
12581264
if (uuid != null) synchronized (onlineSessionLock(uuid)) {
1259-
onlineUserSessions.put(uuid, Boolean.FALSE);
1265+
onlineUserSessions.put(uuid, new OnlineSessionState(false, onlineSessionGeneration.incrementAndGet()));
12601266
}
12611267
}
12621268

@@ -1269,18 +1275,18 @@ private void clearNonNeededCachedUsers(Set<UUID> onlineSnapshot) {
12691275
plugin.devDebug("Clearing cache for non online players (if any)");
12701276
int removed = 0;
12711277
for (UUID uuid : Set.copyOf(userDataCache.keySet())) {
1272-
if (onlineSnapshot.contains(uuid) || Boolean.TRUE.equals(onlineUserSessions.get(uuid))) continue;
1278+
if (onlineSnapshot.contains(uuid) || isUserOnline(uuid)) continue;
12731279
UserDataCache expected = userDataCache.get(uuid);
12741280
if (expected == null) continue;
12751281
long expectedVersion = expected.getSharedSnapshotVersion();
12761282
java.util.concurrent.atomic.AtomicBoolean retired = new java.util.concurrent.atomic.AtomicBoolean();
12771283
java.util.concurrent.atomic.AtomicBoolean invalidatedForJoin = new java.util.concurrent.atomic.AtomicBoolean();
12781284
withSharedSqlBackendExclusive(uuid, () -> {
1279-
if (Boolean.TRUE.equals(onlineUserSessions.get(uuid))) return;
1285+
if (isUserOnline(uuid)) return;
12801286
UserDataCache current = userDataCache.get(uuid);
12811287
if (current != expected || current.getSharedSnapshotVersion() != expectedVersion) return;
12821288
current.beginRemoval();
1283-
if (Boolean.TRUE.equals(onlineUserSessions.get(uuid))) {
1289+
if (isUserOnline(uuid)) {
12841290
current.cancelRemoval();
12851291
return;
12861292
}
@@ -1291,7 +1297,7 @@ private void clearNonNeededCachedUsers(Set<UUID> onlineSnapshot) {
12911297
throw failure;
12921298
}
12931299
synchronized (onlineSessionLock(uuid)) {
1294-
if (Boolean.TRUE.equals(onlineUserSessions.get(uuid))) {
1300+
if (isUserOnline(uuid)) {
12951301
if (sharedSqlRoute != null) {
12961302
try { current.retireAfterSharedFlush(); }
12971303
catch (RuntimeException | Error failure) {
@@ -1322,6 +1328,11 @@ private void clearNonNeededCachedUsers(Set<UUID> onlineSnapshot) {
13221328
if (removed > 0) plugin.devDebug("Removed " + removed + " cached users who are no longer online");
13231329
}
13241330

1331+
private boolean isUserOnline(UUID uuid) {
1332+
OnlineSessionState state = onlineUserSessions.get(uuid);
1333+
return state != null && state.online();
1334+
}
1335+
13251336
public UserDataCache getPublishedCache(UUID uuid) {
13261337
UserDataCache cache = userDataCache.get(uuid);
13271338
return cache != null && cache.hasPublishedStorageSnapshot() ? cache : null;

‎AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/UserDataManagerCacheCleanupThreadingTest.java‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -128,8 +128,7 @@ void reconcilesSessionMarkersOnlyOnStorageWorker() throws Exception {
128128
java.lang.reflect.Field sessionsField = UserDataManager.class.getDeclaredField("onlineUserSessions");
129129
sessionsField.setAccessible(true);
130130
@SuppressWarnings("unchecked")
131-
ConcurrentHashMap<UUID, Boolean> sessions =
132-
(ConcurrentHashMap<UUID, Boolean>) sessionsField.get(manager);
131+
ConcurrentHashMap<UUID, ?> sessions = (ConcurrentHashMap<UUID, ?>) sessionsField.get(manager);
133132
Server server = mock(Server.class);
134133
try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class)) {
135134
bukkit.when(Bukkit::getServer).thenReturn(server);
@@ -146,6 +145,52 @@ void reconcilesSessionMarkersOnlyOnStorageWorker() throws Exception {
146145
}
147146
}
148147

148+
@Test
149+
void olderQueuedSnapshotCannotEraseNewerQuitTombstone() throws Exception {
150+
AdvancedCorePlugin plugin = mock(AdvancedCorePlugin.class, RETURNS_DEEP_STUBS);
151+
BukkitScheduler scheduler = mock(BukkitScheduler.class);
152+
when(plugin.getBukkitScheduler()).thenReturn(scheduler);
153+
when(plugin.isEnabled()).thenReturn(true);
154+
when(plugin.getOptions().isOnlineMode()).thenReturn(true);
155+
UserDataManager manager = new UserDataManager(plugin);
156+
manager.getTimer().shutdownNow();
157+
java.util.concurrent.ScheduledExecutorService worker = mock(java.util.concurrent.ScheduledExecutorService.class);
158+
java.lang.reflect.Field timerField = UserDataManager.class.getDeclaredField("timer");
159+
timerField.setAccessible(true);
160+
timerField.set(manager, worker);
161+
UUID uuid = UUID.randomUUID();
162+
org.bukkit.entity.Player player = mock(org.bukkit.entity.Player.class);
163+
when(player.getUniqueId()).thenReturn(uuid);
164+
Server server = mock(Server.class);
165+
try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class)) {
166+
bukkit.when(Bukkit::getServer).thenReturn(server);
167+
bukkit.when(Bukkit::getOnlinePlayers)
168+
.thenReturn(java.util.List.of(), java.util.List.of(player), java.util.List.of());
169+
ArgumentCaptor<Runnable> platform = ArgumentCaptor.forClass(Runnable.class);
170+
ArgumentCaptor<Runnable> storage = ArgumentCaptor.forClass(Runnable.class);
171+
172+
manager.clearNonNeededCachedUsers();
173+
manager.clearNonNeededCachedUsers();
174+
verify(scheduler, times(2)).runTask(eq(plugin), platform.capture());
175+
platform.getAllValues().forEach(Runnable::run);
176+
verify(worker, times(2)).execute(storage.capture());
177+
manager.markUserOffline(player);
178+
storage.getAllValues().forEach(Runnable::run);
179+
180+
UserDataCache cache = new UserDataCache(manager, uuid);
181+
cache.updateCache(new java.util.HashMap<>(java.util.Map.of("Points", new DataValueInt(1))));
182+
manager.getUserDataCache().put(uuid, cache);
183+
manager.clearNonNeededCachedUsers();
184+
ArgumentCaptor<Runnable> allPlatformTasks = ArgumentCaptor.forClass(Runnable.class);
185+
verify(scheduler, times(3)).runTask(eq(plugin), allPlatformTasks.capture());
186+
allPlatformTasks.getAllValues().get(2).run();
187+
ArgumentCaptor<Runnable> allStorageTasks = ArgumentCaptor.forClass(Runnable.class);
188+
verify(worker, times(3)).execute(allStorageTasks.capture());
189+
allStorageTasks.getAllValues().get(2).run();
190+
assertFalse(manager.containsKey(uuid));
191+
}
192+
}
193+
149194
@Test
150195
void joinAfterPlatformSnapshotPreventsWorkerEviction() throws Exception {
151196
AdvancedCorePlugin plugin = mock(AdvancedCorePlugin.class, RETURNS_DEEP_STUBS);

0 commit comments

Comments
 (0)