Skip to content

Preserve point placeholder fences across callback and legacy arithmetic #1741

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BenCodez wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Preserve point placeholder fences across callback and legacy arithmetic #1741

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dd00514
Preserve point placeholder fences across arithmetic paths
BenCodez Nov 2, 2026
6e26c9c
Use explicit types in point mutation callbacks
BenCodez Nov 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 146 additions & 85 deletions VotingPlugin/src/main/java/com/bencodez/votingplugin/user/SharedMysqlPointMutator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down Expand Up @@ -124,7 +125,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;
Expand Down Expand Up @@ -383,7 +384,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);
}

/**
Expand All @@ -401,7 +402,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);
}
}
Expand All @@ -414,7 +415,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;
Expand Down Expand Up @@ -964,6 +965,23 @@ private boolean transferAtomically(VotingPluginUser source, VotingPluginUser tar
}
}

private boolean runPointMutation(VotingPluginUser user, Runnable operation, boolean async) {
PlaceHolders placeholders = plugin.getPlaceholders();
Consumer<Boolean> 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 {
Expand All @@ -984,25 +1002,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;
PlaceHolders 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);
}
}

Expand All @@ -1020,46 +1047,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);
PlaceHolders 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 }
Expand Down Expand Up @@ -1110,20 +1155,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);
PlaceHolders 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);
}
}

Expand All @@ -1132,21 +1185,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);
PlaceHolders 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);
}
}

Expand Down
32 changes: 27 additions & 5 deletions VotingPlugin/src/main/java/com/bencodez/votingplugin/user/VotingPluginUser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -380,7 +382,9 @@ private CompletionStage<Integer> 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);
}
Expand Down Expand Up @@ -994,16 +998,34 @@ public synchronized void addPointsStorageAware(int value, BiConsumer<Boolean, In
}
Player player = getPlayer();
String pointsColumn = getPointsPath();
PlaceHolders placeholders = plugin.getPlaceholders();
Consumer<Boolean> admission = placeholders == null ? null
: placeholders.sharedPointAdmissionCompletion(UUID.fromString(getUUID()));
AtomicBoolean completed = new AtomicBoolean();
BiConsumer<Boolean, Integer> 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);
}
SharedMysqlPointMutator.AddResult 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));
}
}

Expand Down
Loading
Loading