From dd005146f3632923cadfa986334f96605d377055 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 1 Nov 2026 19:36:07 -0600 Subject: [PATCH 1/2] Preserve point placeholder fences across arithmetic paths --- .../user/SharedMysqlPointMutator.java | 230 +++++++++++------- .../votingplugin/user/VotingPluginUser.java | 30 ++- .../ColdOfflinePlaceholderCacheTest.java | 152 ++++++++++++ .../user/SharedMysqlPointMutatorTest.java | 1 + 4 files changed, 323 insertions(+), 90 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java index ca55ae814..6d936456c 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java @@ -124,7 +124,7 @@ int add(VotingPluginUser user, int amount, boolean async, boolean failOnRejected int previousTotal = cachedPoints(user, pointsColumn); int predictedTotal = previousTotal + amount; cachePredictedPoints(user, predictedTotal, pointsColumn); - if (!run(() -> update(user, amount, false, pointsColumn), true)) { + if (!runPointMutation(user, () -> update(user, amount, false, pointsColumn), true)) { discardOptimisticPoints(user, pointsColumn); if (failOnRejectedAdmission) throw new IllegalStateException("Point mutation storage worker is unavailable"); return previousTotal; @@ -383,7 +383,7 @@ boolean setCommittedInColumn(VotingPluginUser user, int value, String pointsColu void cap(VotingPluginUser user, int maximum, boolean async) { String pointsColumn = user.getPointsPath(); - run(() -> capAt(user, maximum, pointsColumn), async); + runPointMutation(user, () -> capAt(user, maximum, pointsColumn), async); } /** @@ -401,7 +401,7 @@ void addAndCap(VotingPluginUser user, int amount, int maximum, boolean async) { int predictedTotal = (int) Math.max(Integer.MIN_VALUE, Math.min((long) previousTotal + amount, maximum)); cachePredictedPoints(user, predictedTotal, pointsColumn); - if (!run(() -> addAndCapAt(user, amount, maximum, pointsColumn), true)) { + if (!runPointMutation(user, () -> addAndCapAt(user, amount, maximum, pointsColumn), true)) { discardOptimisticPoints(user, pointsColumn); } } @@ -414,7 +414,7 @@ boolean remove(VotingPluginUser user, int amount, boolean async) { if (!async) return remove(user, amount); String pointsColumn = user.getPointsPath(); boolean predictedSuccess = cachedPoints(user, pointsColumn) >= amount; - boolean submitted = run(() -> update(user, -amount, true, pointsColumn), true); + boolean submitted = runPointMutation(user, () -> update(user, -amount, true, pointsColumn), true); // Preserve the historical asynchronous API contract: the caller receives // the cached prediction while the conditional database debit runs later. return submitted && predictedSuccess; @@ -964,6 +964,23 @@ private boolean transferAtomically(VotingPluginUser source, VotingPluginUser tar } } + private boolean runPointMutation(VotingPluginUser user, Runnable operation, boolean async) { + var placeholders = plugin.getPlaceholders(); + Consumer admission = placeholders == null ? null + : placeholders.sharedPointAdmissionCompletion(UUID.fromString(user.getUUID())); + boolean accepted = run(() -> { + boolean settled = false; + try { + operation.run(); + settled = true; + } finally { + if (admission != null) admission.accept(settled); + } + }, async); + if (!accepted && admission != null) admission.accept(false); + return accepted; + } + private boolean run(Runnable operation, boolean async) { if (async) { try { @@ -984,25 +1001,34 @@ private boolean update(VotingPluginUser user, int delta, boolean requireNonnegat } private boolean update(VotingPluginUser user, int delta, boolean requireNonnegative, String points) { - drainCache(user, points); - MySQL table = plugin.getMysql(); - StringBuilder sql = new StringBuilder("UPDATE ").append(table.qi(table.getTableName())).append(" SET ") - .append(table.qi(points)).append(" = COALESCE(").append(table.qi(points)).append(", 0) + ? WHERE ") - .append(table.qi("uuid")).append(table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); - if (requireNonnegative) { - sql.append(" AND COALESCE(").append(table.qi(points)).append(", 0) >= ?"); - } - try (Connection connection = requireConnection(table); - PreparedStatement statement = connection.prepareStatement(sql.toString())) { - statement.setInt(1, delta); - statement.setString(2, user.getUUID()); - if (requireNonnegative) statement.setInt(3, -delta); - return statement.executeUpdate() == 1; - } catch (SQLException failure) { - logFailure(failure); - return false; + var publication = beginPointPublication(user); + // Classify this arithmetic statement; an earlier uncertain operation stays fenced. + boolean known = true; + try { + drainCache(user, points); + MySQL table = plugin.getMysql(); + StringBuilder sql = new StringBuilder("UPDATE ").append(table.qi(table.getTableName())).append(" SET ") + .append(table.qi(points)).append(" = COALESCE(").append(table.qi(points)).append(", 0) + ? WHERE ") + .append(table.qi("uuid")).append(table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); + if (requireNonnegative) { + sql.append(" AND COALESCE(").append(table.qi(points)).append(", 0) >= ?"); + } + try (Connection connection = requireConnection(table); + PreparedStatement statement = connection.prepareStatement(sql.toString())) { + statement.setInt(1, delta); + statement.setString(2, user.getUUID()); + if (requireNonnegative) statement.setInt(3, -delta); + known = false; + int affected = statement.executeUpdate(); + known = true; + return affected == 1; + } catch (SQLException failure) { + logFailure(failure); + return false; + } } finally { - discardPointsCache(user, points); + boolean refreshed = SharedMysqlCacheReconciler.tryInvalidateAndRefreshOnWorker(plugin, user.getUUID(), points); + finishPointPublication(publication, user, points, known, refreshed); } } @@ -1020,46 +1046,64 @@ private AddResult addAndReadCommittedResult(VotingPluginUser user, int amount) { } private AddResult addAndReadCommittedResult(VotingPluginUser user, int amount, String points) { - drainCache(user, points); - MySQL table = plugin.getMysql(); - String uuidMatch = table.qi("uuid") + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); - String update = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = COALESCE(" - + table.qi(points) + ", 0) + ? WHERE " + uuidMatch; - String read = "SELECT " + table.qi(points) + " FROM " + table.qi(table.getTableName()) + " WHERE " + uuidMatch; - boolean updateCommitted = false; - Integer committedTotal = null; - try (Connection connection = requireConnection(table); - PreparedStatement updateStatement = connection.prepareStatement(update); - PreparedStatement readStatement = connection.prepareStatement(read)) { - updateStatement.setInt(1, amount); - updateStatement.setString(2, user.getUUID()); - if (updateStatement.executeUpdate() == 1) { - // With JDBC auto-commit, executeUpdate returning one means the mutation - // completed. A later read may still fail after the points have been - // committed, so never turn that outcome into a retryable failure. - updateCommitted = true; - readStatement.setString(1, user.getUUID()); - try (java.sql.ResultSet result = readStatement.executeQuery()) { - if (result.next()) committedTotal = result.getInt(1); + var publication = beginPointPublication(user); + // Classify this arithmetic statement; an earlier uncertain operation stays fenced. + boolean known = true; + boolean refreshed = false; + try { + try { + drainCache(user, points); + } catch (RuntimeException preparationFailure) { + // No arithmetic statement was submitted. Cache retirement may itself + // perform I/O; its failure is not proof this addition ran. + plugin.debug(preparationFailure); + return new AddResult(MutationOutcome.REJECTED, 0); + } + MySQL table = plugin.getMysql(); + String uuidMatch = table.qi("uuid") + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); + String update = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = COALESCE(" + + table.qi(points) + ", 0) + ? WHERE " + uuidMatch; + String read = "SELECT " + table.qi(points) + " FROM " + table.qi(table.getTableName()) + " WHERE " + uuidMatch; + boolean updateCommitted = false; + Integer committedTotal = null; + try (Connection connection = requireConnection(table); + PreparedStatement updateStatement = connection.prepareStatement(update); + PreparedStatement readStatement = connection.prepareStatement(read)) { + updateStatement.setInt(1, amount); + updateStatement.setString(2, user.getUUID()); + known = false; + int affected = updateStatement.executeUpdate(); + known = true; + if (affected == 1) { + // With JDBC auto-commit, executeUpdate returning one means the mutation + // completed. A later read may still fail after the points have been + // committed, so never turn that outcome into a retryable failure. + updateCommitted = true; + readStatement.setString(1, user.getUUID()); + try (java.sql.ResultSet result = readStatement.executeQuery()) { + if (result.next()) committedTotal = result.getInt(1); + } } + } catch (SQLException failure) { + logFailure(failure); + } finally { + refreshed = SharedMysqlCacheReconciler.tryInvalidateAndRefreshOnWorker(plugin, user.getUUID(), points); } - } catch (SQLException failure) { - logFailure(failure); + // Do not evaluate the fallback while the JDBC handle is still held. With a + // one-connection pool, getPoints() may need that same handle after a missing + // row or a failed follow-up read. + if (committedTotal != null) return new AddResult(true, committedTotal); + // A failed acquisition cannot support the fallback read either. Report the + // mutation failure without checking out a second connection and let callers + // complete their callback deterministically. + if (!updateCommitted) return new AddResult(known ? MutationOutcome.REJECTED : MutationOutcome.INDETERMINATE, 0); + // A live mode reload can change getPointsPath() while this queued write is + // running. Keep the fallback on the column that received the credit. + if (!points.equals(user.getPointsPath())) return new AddResult(true, user.getUserData().getInt(points)); + return new AddResult(true, user.getPoints()); } finally { - discardPointsCache(user, points); + finishPointPublication(publication, user, points, known, refreshed); } - // Do not evaluate the fallback while the JDBC handle is still held. With a - // one-connection pool, getPoints() may need that same handle after a missing - // row or a failed follow-up read. - if (committedTotal != null) return new AddResult(true, committedTotal); - // A failed acquisition cannot support the fallback read either. Report the - // mutation failure without checking out a second connection and let callers - // complete their callback deterministically. - if (!updateCommitted) return new AddResult(false, 0); - // A live mode reload can change getPointsPath() while this queued write is - // running. Keep the fallback on the column that received the credit. - if (!points.equals(user.getPointsPath())) return new AddResult(true, user.getUserData().getInt(points)); - return new AddResult(true, user.getPoints()); } enum MutationOutcome { CONFIRMED, REJECTED, INDETERMINATE } @@ -1110,20 +1154,28 @@ private void capAt(VotingPluginUser user, int maximum) { } private void capAt(VotingPluginUser user, int maximum, String points) { - drainCache(user, points); - MySQL table = plugin.getMysql(); - String sql = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = LEAST(COALESCE(" - + table.qi(points) + ", 0), ?) WHERE " + table.qi("uuid") - + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); - try (Connection connection = requireConnection(table); - PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setInt(1, maximum); - statement.setString(2, user.getUUID()); - statement.executeUpdate(); - } catch (SQLException failure) { - logFailure(failure); + var publication = beginPointPublication(user); + // Classify this arithmetic statement; an earlier uncertain operation stays fenced. + boolean known = true; + try { + drainCache(user, points); + MySQL table = plugin.getMysql(); + String sql = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = LEAST(COALESCE(" + + table.qi(points) + ", 0), ?) WHERE " + table.qi("uuid") + + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); + try (Connection connection = requireConnection(table); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setInt(1, maximum); + statement.setString(2, user.getUUID()); + known = false; + statement.executeUpdate(); + known = true; + } catch (SQLException failure) { + logFailure(failure); + } } finally { - discardPointsCache(user, points); + boolean refreshed = SharedMysqlCacheReconciler.tryInvalidateAndRefreshOnWorker(plugin, user.getUUID(), points); + finishPointPublication(publication, user, points, known, refreshed); } } @@ -1132,21 +1184,29 @@ private void addAndCapAt(VotingPluginUser user, int amount, int maximum) { } private void addAndCapAt(VotingPluginUser user, int amount, int maximum, String points) { - drainCache(user, points); - MySQL table = plugin.getMysql(); - String sql = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = LEAST(COALESCE(" - + table.qi(points) + ", 0) + ?, ?) WHERE " + table.qi("uuid") - + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); - try (Connection connection = requireConnection(table); - PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setInt(1, amount); - statement.setInt(2, maximum); - statement.setString(3, user.getUUID()); - statement.executeUpdate(); - } catch (SQLException failure) { - logFailure(failure); + var publication = beginPointPublication(user); + // Classify this arithmetic statement; an earlier uncertain operation stays fenced. + boolean known = true; + try { + drainCache(user, points); + MySQL table = plugin.getMysql(); + String sql = "UPDATE " + table.qi(table.getTableName()) + " SET " + table.qi(points) + " = LEAST(COALESCE(" + + table.qi(points) + ", 0) + ?, ?) WHERE " + table.qi("uuid") + + (table.getDbType() == DbType.POSTGRESQL ? " = ?::uuid" : " = ?"); + try (Connection connection = requireConnection(table); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setInt(1, amount); + statement.setInt(2, maximum); + statement.setString(3, user.getUUID()); + known = false; + statement.executeUpdate(); + known = true; + } catch (SQLException failure) { + logFailure(failure); + } } finally { - discardPointsCache(user, points); + boolean refreshed = SharedMysqlCacheReconciler.tryInvalidateAndRefreshOnWorker(plugin, user.getUUID(), points); + finishPointPublication(publication, user, points, known, refreshed); } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java index d33b1a2f9..97a9dc081 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java @@ -380,7 +380,9 @@ private CompletionStage addPointsStorageAwareAsyncUnfenced(int value, S event.getPoints(), operationId, pointsColumn); if (result.success()) completion.complete(result.total()); else completion.completeExceptionally( - new IllegalStateException("Unable to persist shared MySQL points")); + result.outcome() == SharedMysqlPointMutator.MutationOutcome.REJECTED + ? new PointAdditionNotStartedException() + : new IllegalStateException("Unable to persist shared MySQL points")); } catch (Throwable failure) { completion.completeExceptionally(failure); } @@ -994,16 +996,34 @@ public synchronized void addPointsStorageAware(int value, BiConsumer admission = placeholders == null ? null + : placeholders.sharedPointAdmissionCompletion(UUID.fromString(getUUID())); + var completed = new java.util.concurrent.atomic.AtomicBoolean(); + BiConsumer completeOnce = (success, total) -> { + if (completed.compareAndSet(false, true)) completion.accept(success, total); + }; try { plugin.getTimer().execute(() -> { - SharedMysqlPointMutator.AddResult result = sharedPoints.addCommittedToColumn(this, - event.getPoints(), pointsColumn); + SharedMysqlPointMutator.AddResult result; + boolean settled = false; + try { + result = sharedPoints.addCommittedToColumn(this, event.getPoints(), pointsColumn); + settled = true; + } catch (RuntimeException failure) { + plugin.debug(failure); + result = new SharedMysqlPointMutator.AddResult(false, 0); + } finally { + if (admission != null) admission.accept(settled); + } + var outcome = result; BukkitCompletionScheduler.run(plugin, player, - () -> completion.accept(result.success(), result.total())); + () -> completeOnce.accept(outcome.success(), outcome.total())); }); } catch (RuntimeException rejected) { + if (admission != null) admission.accept(false); plugin.debug(rejected); - BukkitCompletionScheduler.run(plugin, player, () -> completion.accept(false, 0)); + BukkitCompletionScheduler.run(plugin, player, () -> completeOnce.accept(false, 0)); } } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/placeholders/ColdOfflinePlaceholderCacheTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/placeholders/ColdOfflinePlaceholderCacheTest.java index 4baafa69e..4652cf81f 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/placeholders/ColdOfflinePlaceholderCacheTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/placeholders/ColdOfflinePlaceholderCacheTest.java @@ -499,6 +499,158 @@ private void assertCompletedInvalidSnapshot(com.bencodez.simpleapi.sql.data.Data } } + @Test void publicCallbackAdditionAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("callback"); } + @Test void publicStageAdditionAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("stage"); } + @Test void publicLegacyAdditionAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("add"); } + @Test void publicLegacyDebitAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("remove"); } + @Test void publicVoteCapAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("cap"); } + @Test void publicVoteAddAndCapAmbiguityCannotPublishAnUnrelatedRefill() throws Exception { assertPublicArithmeticAmbiguity("addcap"); } + + @Test void confirmedPublicArithmeticPublishesTheReconciledSnapshot() throws Exception { assertPublicArithmeticPhase("confirmed"); } + @Test void rejectedPublicArithmeticPublishesTheReconciledSnapshot() throws Exception { assertPublicArithmeticPhase("rejected"); } + @Test void preparationFailureRemainsRecoverableAcrossPublicArithmetic() throws Exception { assertPublicArithmeticPhase("prepare"); } + @Test void runtimeExecutionFailureRemainsUnavailableAcrossPublicArithmetic() throws Exception { assertPublicArithmeticPhase("runtime"); } + + @Test void rejectedPublicArithmeticAdmissionRecoversWithoutSqlOrRetry() throws Exception { assertPublicArithmeticPhase("admission"); } + @Test void stoppedPublicArithmeticCannotReopenNumericPublication() throws Exception { assertPublicArithmeticPhase("shutdown"); } + + @Test void callbackCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("callback", "drain"); } + @Test void stageCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("stage", "drain"); } + @Test void addCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("add", "drain"); } + @Test void removeCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("remove", "drain"); } + @Test void capCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("cap", "drain"); } + @Test void addcapCacheAdmissionFailureCanRecoverWithoutPointExecution() throws Exception { assertPublicArithmeticOutcome("addcap", "drain"); } + @Test void priorUncertainPointWriteSurvivesCachePreparationFailure() throws Exception { assertPublicArithmeticPhase("drain-uncertain"); } + + private void assertPublicArithmeticPhase(String phase) throws Exception { + for (String operation : List.of("callback", "stage", "add", "remove", "cap", "addcap")) { + assertPublicArithmeticOutcome(operation, phase); + } + } + + private void assertPublicArithmeticAmbiguity(String operation) throws Exception { + assertPublicArithmeticOutcome(operation, "ambiguous"); + } + + private void assertPublicArithmeticOutcome(String operation, String phase) throws Exception { + Fixture f = new Fixture(PlaceholderCacheLevel.AUTOALL, "points"); + when(f.plugin.getStorageType()).thenReturn(com.bencodez.advancedcore.api.user.UserStorage.MYSQL); + org.mockito.Mockito.doReturn(f.player).when(f.user).getPlayer(); + var field = VotingPluginUser.class.getDeclaredField("plugin"); + field.setAccessible(true); + field.set(f.user, f.plugin); + org.mockito.Mockito.doCallRealMethod().when(f.user).addPointsStorageAware( + org.mockito.ArgumentMatchers.anyInt(), org.mockito.ArgumentMatchers.>any()); + var cache = new com.bencodez.advancedcore.api.user.usercache.UserDataCache(null, f.uuid); + cache.updateCache(new java.util.HashMap<>(java.util.Map.of("Points", new com.bencodez.simpleapi.sql.data.DataValueInt(20)))); + when(f.user.getCache()).thenReturn(cache); + var manager = f.plugin.getUserManager().getDataManager(); + when(manager.getUserDataCache()).thenReturn(new java.util.concurrent.ConcurrentHashMap<>(java.util.Map.of(f.uuid, cache))); + doAnswer(call -> { + cache.updateCache(new java.util.HashMap<>(java.util.Map.of("Points", new com.bencodez.simpleapi.sql.data.DataValueInt(36)))); + return null; + }).when(manager).cacheUser(f.uuid, null); + var connection = mock(java.sql.Connection.class); + var statement = mock(java.sql.PreparedStatement.class); + when(f.plugin.getMysql().getTableName()).thenReturn("SyntheticPoints"); + when(f.plugin.getMysql().getMysql().getConnectionManager().getConnection()).thenReturn(connection); + when(connection.prepareStatement(anyString())).thenReturn(statement); + if (phase.equals("prepare")) { + when(connection.prepareStatement(anyString())).thenThrow(new java.sql.SQLException("synthetic prepare rejected")); + } else if (phase.equals("ambiguous")) { + when(statement.executeUpdate()).thenThrow(new java.sql.SQLException("synthetic lost update acknowledgement")); + } else if (phase.equals("runtime")) { + when(statement.executeUpdate()).thenThrow(new IllegalStateException("synthetic driver failure")); + } else { + when(statement.executeUpdate()).thenReturn(phase.equals("confirmed") || phase.equals("shutdown") ? 1 : 0); + var row = mock(java.sql.ResultSet.class); + when(row.next()).thenReturn(true); + when(row.getInt(1)).thenReturn(36); + when(statement.executeQuery()).thenReturn(row); + } + when(f.scheduler.getFoliaLib()).thenReturn(null); + doAnswer(call -> { ((Runnable) call.getArgument(1)).run(); return null; }) + .when(f.scheduler).runTask(eq(f.plugin), any(Runnable.class), eq(f.player)); + if (phase.equals("admission")) { + doThrow(new java.util.concurrent.RejectedExecutionException("synthetic full worker")) + .when(f.timer).execute(any(Runnable.class)); + } + var cacheWrites = new java.util.concurrent.atomic.AtomicInteger(); + boolean drainFailure = phase.startsWith("drain"); + if (drainFailure) { + // AC138's real exclusive cache gate rejects before its writer runs. + cache.configureSharedStorage(values -> cacheWrites.incrementAndGet(), + task -> { throw new IllegalStateException("synthetic cache admission retired"); }); + when(f.user.isCached()).thenReturn(true); + doAnswer(call -> { cache.clearCache(); return null; }).when(manager).removeCache(f.uuid, null); + if (phase.equals("drain-uncertain")) { + f.placeholders.beginSharedPointDebit(f.uuid); + f.placeholders.finishSharedPointDebit(f.uuid, false, false); + } + } + var callbacks = new ArrayList(); + try (var bukkit = mockStatic(Bukkit.class)) { + bukkit.when(Bukkit::getPluginManager).thenReturn(mock(org.bukkit.plugin.PluginManager.class)); + switch (operation) { + case "callback" -> f.user.addPointsStorageAware(16, (java.util.function.BiConsumer) (success, total) -> callbacks.add(success)); + case "stage" -> { + org.mockito.Mockito.doCallRealMethod().when(f.user).addPointsStorageAwareAsync(org.mockito.ArgumentMatchers.anyInt(), org.mockito.ArgumentMatchers.isNull()); + f.user.addPointsStorageAwareAsync(16, null).whenComplete((total, failure) -> callbacks.add(failure == null)); + } + case "add" -> { + org.mockito.Mockito.doCallRealMethod().when(f.user).addPoints(org.mockito.ArgumentMatchers.anyInt(), eq(true)); + f.user.addPoints(16, true); + } + case "remove" -> { + org.mockito.Mockito.doCallRealMethod().when(f.user).removePoints(org.mockito.ArgumentMatchers.anyInt(), eq(true)); + f.user.removePoints(16, true); + } + case "cap", "addcap" -> { + when(f.plugin.getConfigFile().getPointsOnVote()).thenReturn(operation.equals("cap") ? 0 : 16); + when(f.plugin.getConfigFile().getLimitVotePoints()).thenReturn(40); + org.mockito.Mockito.doCallRealMethod().when(f.user).addPoints(); + f.user.addPoints(); + } + default -> throw new AssertionError(operation); + } + } + verify(f.plugin.getMysql().getMysql().getConnectionManager(), never()).getConnection(); + verify(statement, never()).executeUpdate(); + if (phase.equals("admission")) { + assertTrue(f.pending.isEmpty()); + doAnswer(call -> { f.pending.addLast(call.getArgument(0)); return null; }) + .when(f.timer).execute(any(Runnable.class)); + assertEquals("...", f.request("points_nocache")); + assertEquals(1, f.pending.size(), "rejected admission permits one ordinary source-recovery worker"); + } + if (phase.equals("shutdown")) when(f.plugin.isEnabled()).thenReturn(false); + boolean uncertain = phase.equals("ambiguous") || phase.equals("runtime") || phase.equals("drain-uncertain"); + Runnable worker = f.pending.removeFirst(); + if ((phase.equals("runtime") || drainFailure) && !operation.equals("callback") && !operation.equals("stage")) { + org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException.class, worker::run); + } else worker.run(); + assertEquals(operation.equals("callback") || operation.equals("stage") ? List.of(phase.equals("confirmed") || phase.equals("shutdown")) : List.of(), callbacks); + verify(statement, times(phase.equals("prepare") || phase.equals("admission") || drainFailure ? 0 : 1)).executeUpdate(); + if (drainFailure) { + assertEquals(0, cacheWrites.get(), "actual AC138 cache writer was never admitted"); + verify(f.plugin.getMysql().getMysql().getConnectionManager(), never()).getConnection(); + assertEquals("...", f.request("points_nocache")); + if (!uncertain) { + assertEquals(1, f.pending.size(), "known pre-point failure must allow ordinary recovery"); + f.pending.removeFirst().run(); + } + } + if (operation.equals("stage") && !uncertain && !phase.equals("confirmed") && !phase.equals("admission") && !phase.equals("shutdown") && !drainFailure) { + assertEquals("...", f.request("points_nocache"), operation + "/" + phase); + assertEquals(1, f.pending.size(), "known failed stage recovers through the ordinary placeholder worker"); + f.pending.removeFirst().run(); + } + assertEquals(uncertain || phase.equals("shutdown") ? "..." : "36", f.request("points_nocache"), operation + "/" + phase); + f.placeholders.sharedPointRefillCompletion(f.uuid).accept(true); + assertEquals(uncertain || phase.equals("shutdown") ? "..." : "36", f.request("points_nocache")); + assertTrue(f.pending.isEmpty(), "no retry of an ambiguous addition"); + } + @Test void publicReplayAdditionClosesRenderingBeforeWorkerAdmission() throws Exception { Fixture f = new Fixture(PlaceholderCacheLevel.AUTOALL, "points"); when(f.plugin.getStorageType()).thenReturn(com.bencodez.advancedcore.api.user.UserStorage.MYSQL); diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/user/SharedMysqlPointMutatorTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/user/SharedMysqlPointMutatorTest.java index 4a288b404..4dd397136 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/user/SharedMysqlPointMutatorTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/user/SharedMysqlPointMutatorTest.java @@ -503,6 +503,7 @@ void asynchronousAddUsesOnlyCachedPointsOnTheCallerThread() { ScheduledExecutorService persistence = mock(ScheduledExecutorService.class); when(plugin.getTimer()).thenReturn(persistence); VotingPluginUser user = mock(VotingPluginUser.class); + when(user.getUUID()).thenReturn("00000000-0000-0000-0000-000000000001"); UserDataCache cache = mock(UserDataCache.class); DataValue points = mock(DataValue.class); java.util.HashMap values = new java.util.HashMap<>(); From 6e26c9c6c2bc237fc5609a97ad94c61e0feb85a8 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 1 Nov 2026 20:28:38 -0600 Subject: [PATCH 2/2] Use explicit types in point mutation callbacks --- .../votingplugin/user/SharedMysqlPointMutator.java | 11 ++++++----- .../bencodez/votingplugin/user/VotingPluginUser.java | 8 +++++--- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java index 6d936456c..ad7c3172e 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java @@ -21,6 +21,7 @@ import com.bencodez.simpleapi.sql.data.DataValue; import com.bencodez.simpleapi.sql.data.DataValueInt; import com.bencodez.votingplugin.VotingPluginMain; +import com.bencodez.votingplugin.placeholders.PlaceHolders; import com.bencodez.votingplugin.util.BukkitCompletionScheduler; /** Performs point writes that must remain atomic across shared MySQL servers. */ @@ -965,7 +966,7 @@ private boolean transferAtomically(VotingPluginUser source, VotingPluginUser tar } private boolean runPointMutation(VotingPluginUser user, Runnable operation, boolean async) { - var placeholders = plugin.getPlaceholders(); + PlaceHolders placeholders = plugin.getPlaceholders(); Consumer admission = placeholders == null ? null : placeholders.sharedPointAdmissionCompletion(UUID.fromString(user.getUUID())); boolean accepted = run(() -> { @@ -1001,7 +1002,7 @@ private boolean update(VotingPluginUser user, int delta, boolean requireNonnegat } private boolean update(VotingPluginUser user, int delta, boolean requireNonnegative, String points) { - var publication = beginPointPublication(user); + PlaceHolders publication = beginPointPublication(user); // Classify this arithmetic statement; an earlier uncertain operation stays fenced. boolean known = true; try { @@ -1046,7 +1047,7 @@ private AddResult addAndReadCommittedResult(VotingPluginUser user, int amount) { } private AddResult addAndReadCommittedResult(VotingPluginUser user, int amount, String points) { - var publication = beginPointPublication(user); + PlaceHolders publication = beginPointPublication(user); // Classify this arithmetic statement; an earlier uncertain operation stays fenced. boolean known = true; boolean refreshed = false; @@ -1154,7 +1155,7 @@ private void capAt(VotingPluginUser user, int maximum) { } private void capAt(VotingPluginUser user, int maximum, String points) { - var publication = beginPointPublication(user); + PlaceHolders publication = beginPointPublication(user); // Classify this arithmetic statement; an earlier uncertain operation stays fenced. boolean known = true; try { @@ -1184,7 +1185,7 @@ private void addAndCapAt(VotingPluginUser user, int amount, int maximum) { } private void addAndCapAt(VotingPluginUser user, int amount, int maximum, String points) { - var publication = beginPointPublication(user); + PlaceHolders publication = beginPointPublication(user); // Classify this arithmetic statement; an earlier uncertain operation stays fenced. boolean known = true; try { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java index 97a9dc081..b8090224f 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java @@ -21,6 +21,7 @@ import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -42,6 +43,7 @@ import com.bencodez.simpleapi.sql.data.DataValueString; import com.bencodez.simpleapi.time.ParsedDuration; import com.bencodez.votingplugin.VotingPluginMain; +import com.bencodez.votingplugin.placeholders.PlaceHolders; import com.bencodez.votingplugin.events.PlayerReceivePointsEvent; import com.bencodez.votingplugin.events.PlayerSpecialRewardEvent; import com.bencodez.votingplugin.events.PlayerVoteEvent; @@ -996,10 +998,10 @@ public synchronized void addPointsStorageAware(int value, BiConsumer admission = placeholders == null ? null : placeholders.sharedPointAdmissionCompletion(UUID.fromString(getUUID())); - var completed = new java.util.concurrent.atomic.AtomicBoolean(); + AtomicBoolean completed = new AtomicBoolean(); BiConsumer completeOnce = (success, total) -> { if (completed.compareAndSet(false, true)) completion.accept(success, total); }; @@ -1016,7 +1018,7 @@ public synchronized void addPointsStorageAware(int value, BiConsumer completeOnce.accept(outcome.success(), outcome.total())); });