From 923e741a90ad87d3ae1717301ded125f6749fc43 Mon Sep 17 00:00:00 2001
From: BenCodez <17074231+BenCodez@users.noreply.github.com>
Date: Thu, 10 Sep 2026 16:29:52 -0600
Subject: [PATCH 01/78] Add ordered async reward injections
---
AdvancedCore/pom.xml | 5 +
.../advancedcore/api/rewards/Reward.java | 209 +++++++++++-
.../api/rewards/injected/RewardInject.java | 72 +++-
.../api/rewards/injected/RewardInjectInt.java | 57 +++-
.../rewards/RewardAsyncInjectionTest.java | 320 ++++++++++++++++++
5 files changed, 643 insertions(+), 20 deletions(-)
create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
diff --git a/AdvancedCore/pom.xml b/AdvancedCore/pom.xml
index 383ffe5bbe..5e33f01bc8 100644
--- a/AdvancedCore/pom.xml
+++ b/AdvancedCore/pom.xml
@@ -85,6 +85,11 @@
${project.name}
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+ 3.5.4
+
org.apache.maven.plugins
maven-shade-plugin
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
index d86d8d2110..bbcdd54e88 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
@@ -8,6 +8,12 @@
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.logging.Level;
import org.bukkit.Bukkit;
import org.bukkit.Material;
@@ -245,6 +251,152 @@ public void giveInjectedRewards(AdvancedCoreUser user, HashMap p
}
}
+ /**
+ * Gives injected rewards in registration order, waiting for each asynchronous
+ * injection before evaluating the next one. Synchronous injections retain
+ * their existing behavior, and post-reward injections run after all normal
+ * injections have completed.
+ *
+ * @param user receiving user
+ * @param placeholders current placeholders
+ * @return completion stage that completes after all injections have run
+ */
+ public CompletionStage giveInjectedRewardsAsync(AdvancedCoreUser user,
+ HashMap placeholders) {
+ List postRewards = new ArrayList<>();
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+
+ for (RewardInject inject : plugin.getRewardHandler().getInjectedRewards()) {
+ if (inject.isPostReward()) {
+ postRewards.add(inject);
+ continue;
+ }
+ sequence = sequence.thenCompose(ignored -> invokeInjectionAsync(inject, user, placeholders))
+ .thenCompose(result -> resumeOnServerThread(user).thenRun(() -> {
+ if (inject.isAddAsPlaceholder() && result != null) addPlaceholder(inject, result, placeholders);
+ }));
+ }
+
+ for (RewardInject inject : postRewards) {
+ sequence = sequence.thenCompose(ignored -> invokeInjectionAsync(inject, user, placeholders))
+ .thenCompose(result -> resumeOnServerThread(user));
+ }
+ return sequence;
+ }
+
+ private CompletionStage
+ *
+ * @param user receiving user
+ * @param rewardOptions reward options
+ * @return completion stage for the complete reward injection chain
+ */
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, RewardOptions rewardOptions) {
+ if (!AdvancedCorePlugin.getInstance().getOptions().isProcessRewards()) {
+ AdvancedCorePlugin.getInstance().debug("Processing rewards is disabled");
+ return CompletableFuture.completedFuture(null);
+ }
+
+ if (rewardOptions == null) rewardOptions = new RewardOptions();
+ if (!rewardOptions.getPlaceholders().containsKey("ExecDate")) {
+ rewardOptions.addPlaceholder("ExecDate", "" + System.currentTimeMillis());
+ }
+ if (!rewardOptions.getPlaceholders().containsKey("date")) {
+ try {
+ LocalDateTime ldt = LocalDateTime.now();
+ Date date = Date.from(ldt.atZone(ZoneId.systemDefault()).toInstant());
+ rewardOptions.addPlaceholder("Date", "" + new SimpleDateFormat(
+ plugin.getOptions().getFormatRewardTimeFormat()).format(date));
+ } catch (Exception e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
+ PlayerRewardEvent event = new PlayerRewardEvent(this, user, rewardOptions);
+ Bukkit.getPluginManager().callEvent(event);
+ if (event.isCancelled()) {
+ plugin.debug("Reward " + name + " was cancelled for " + user.getPlayerName());
+ return CompletableFuture.completedFuture(null);
+ }
+ if (rewardOptions.isCheckTimed() && (checkDelayed(user, rewardOptions.getPlaceholders())
+ || checkTimed(user, rewardOptions.getPlaceholders()))) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (!rewardOptions.isOnlineSet()) rewardOptions.setOnline(user.isOnline());
+ for (RewardPlaceholderHandle handle : plugin.getRewardHandler().getPlaceholders()) {
+ if (handle.isPreProcess()) rewardOptions.addPlaceholder(handle.getKey(), handle.getValue(this, user));
+ }
+
+ boolean allowOffline = false;
+ boolean canGive = true;
+ if (!rewardOptions.isIgnoreRequirements()) {
+ for (RequirementInject inject : plugin.getRewardHandler().getInjectedRequirements()) {
+ try {
+ if (!inject.onRequirementRequest(this, user, getConfig().getConfigData(), rewardOptions)) {
+ canGive = false;
+ if (!inject.isAllowReattempt()) return CompletableFuture.completedFuture(null);
+ allowOffline = true;
+ }
+ } catch (Exception e) {
+ plugin.debug("Failed to check requirement " + inject.getPath());
+ e.printStackTrace();
+ canGive = false;
+ }
+ }
+ }
+ if (plugin.getOptions().isPauseRewards() || (plugin.getOptions().isTreatVanishAsOffline() && user.isVanished())) {
+ checkRewardFile();
+ user.addOfflineRewards(this, rewardOptions.getPlaceholders());
+ return CompletableFuture.completedFuture(null);
+ }
+ if (((((!rewardOptions.isOnline() || rewardOptions.getServer() != null) && !user.isOnline()) || allowOffline)
+ && (!isForceOffline() && !rewardOptions.isForceOffline()))) {
+ if (rewardOptions.isGiveOffline()) {
+ checkRewardFile();
+ user.addOfflineRewards(this, rewardOptions.getPlaceholders());
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+ if (canGive || isForceOffline() || rewardOptions.isForceOffline()) {
+ plugin.debug(name + ": Passed requirements, attempting to give to " + user.getPlayerName() + "/"
+ + user.getUUID());
+ if (hasAsyncRewardInjection()) return giveRewardUserAsync(user, rewardOptions.getPlaceholders(), rewardOptions);
+ try {
+ giveRewardUser(user, rewardOptions.getPlaceholders(), rewardOptions);
+ return CompletableFuture.completedFuture(null);
+ } catch (Throwable failure) {
+ return CompletableFuture.failedFuture(failure);
+ }
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
/**
* Give reward user.
*
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardBuilder.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardBuilder.java
index 0c16f10e0a..21a134e719 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardBuilder.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardBuilder.java
@@ -6,6 +6,8 @@
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.OfflinePlayer;
import org.bukkit.configuration.ConfigurationSection;
@@ -89,6 +91,16 @@ public void send(AdvancedCoreUser user) {
}
}
+ /** Sends this builder's reward and completes when its async injection chain finishes. */
+ public CompletionStage sendAsync(AdvancedCoreUser user) {
+ RewardHandler handler = user.getPlugin().getRewardHandler();
+ if (reward == null) {
+ return data == null ? CompletableFuture.completedFuture(null)
+ : handler.giveRewardAsync(user, data, path, rewardOptions);
+ }
+ return handler.giveRewardAsync(user, reward, rewardOptions);
+ }
+
public void send(AdvancedCoreUser... users) {
for (AdvancedCoreUser user : users) {
send(user);
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
index 9ec68eb337..a79659701f 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
@@ -3,6 +3,8 @@
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Base64;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Bukkit;
import org.bukkit.configuration.Configuration;
@@ -101,6 +103,31 @@ public void giveReward(AdvancedCoreUser user, ConfigurationSection data, String
}
}
+ /**
+ * Asynchronous counterpart for nested rewards that must preserve injection
+ * ordering. The legacy void dispatch methods intentionally remain unchanged.
+ */
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, ConfigurationSection data, String path,
+ RewardOptions rewardOptions) {
+ RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
+ RewardOptions options = context.getOptions();
+ if (path == null || data == null || !plugin.isEnabled()) return CompletableFuture.completedFuture(null);
+
+ if (data.isList(path)) {
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+ for (String nestedReward : new ArrayList<>(data.getStringList(path))) {
+ sequence = sequence.thenCompose(ignored -> giveRewardAsync(user, nestedReward, options));
+ }
+ return sequence;
+ }
+ if (data.isConfigurationSection(path)) {
+ return giveSectionRewardAsync(user, data, path, context);
+ }
+ String nestedReward = data.getString(path, "");
+ return nestedReward.isEmpty() ? CompletableFuture.completedFuture(null)
+ : giveRewardAsync(user, nestedReward, options);
+ }
+
public void giveReward(AdvancedCoreUser user, Reward reward, RewardOptions rewardOptions) {
RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
if (reward == null) {
@@ -116,6 +143,12 @@ public void giveReward(AdvancedCoreUser user, Reward reward, RewardOptions rewar
}
}
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, Reward reward, RewardOptions rewardOptions) {
+ if (reward == null || !plugin.isEnabled()) return CompletableFuture.completedFuture(null);
+ RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
+ return reward.giveRewardAsync(user, context.getOptions());
+ }
+
public void giveReward(AdvancedCoreUser user, String reward, RewardOptions rewardOptions) {
RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
if (reward == null || reward.isEmpty()) {
@@ -130,6 +163,16 @@ public void giveReward(AdvancedCoreUser user, String reward, RewardOptions rewar
giveReward(user, handler.getReward(reward), context.getOptions());
}
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, String reward, RewardOptions rewardOptions) {
+ RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
+ if (reward == null || reward.isEmpty()) return CompletableFuture.completedFuture(null);
+ if (reward.startsWith("/")) {
+ MiscUtils.getInstance().executeConsoleCommands(user.getPlayerName(), reward, context.getPlaceholders());
+ return CompletableFuture.completedFuture(null);
+ }
+ return giveRewardAsync(user, handler.getReward(reward), context.getOptions());
+ }
+
public void givePersistedQueueReward(AdvancedCoreUser user, String reward, RewardOptions rewardOptions) {
RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
if (reward == null || reward.isEmpty()) {
@@ -177,6 +220,43 @@ public void givePersistedQueueReward(AdvancedCoreUser user, String reward, Rewar
giveReward(user, resolved, context.getOptions());
}
+ /** Resolves and awaits a persisted queue item so a failed async injection can be requeued. */
+ public CompletionStage givePersistedQueueRewardAsync(AdvancedCoreUser user, String reward,
+ RewardOptions rewardOptions) {
+ RewardExecutionContext context = new RewardExecutionContext(rewardOptions).initializeOnlineState(user);
+ if (reward == null || reward.isEmpty()) return CompletableFuture.completedFuture(null);
+
+ String rewardName = reward;
+ Boolean generatedSnapshot = null;
+ if (reward.startsWith(QUEUED_REFERENCE_PREFIX)) {
+ String encoded = reward.substring(QUEUED_REFERENCE_PREFIX.length());
+ int modeEnd = encoded.indexOf('/');
+ if (modeEnd > 0) {
+ String mode = encoded.substring(0, modeEnd);
+ String encodedName = encoded.substring(modeEnd + 1);
+ if ((mode.equals("snapshot") || mode.equals("normal")) && !encodedName.isEmpty()) {
+ try {
+ rewardName = new String(Base64.getUrlDecoder().decode(encodedName), StandardCharsets.UTF_8);
+ generatedSnapshot = Boolean.valueOf(mode.equals("snapshot"));
+ } catch (IllegalArgumentException ignored) {
+ rewardName = reward;
+ }
+ }
+ }
+ }
+ Reward resolved;
+ if (generatedSnapshot != null) {
+ resolved = generatedSnapshot.booleanValue() ? handler.getQueuedGeneratedReward(rewardName, user.getUUID())
+ : handler.getReward(rewardName);
+ } else if (handler.rewardExist(rewardName) || handler.hasDirectRewardHandle(rewardName)) {
+ resolved = handler.getReward(rewardName);
+ } else {
+ resolved = handler.getQueuedGeneratedReward(rewardName, user.getUUID());
+ if (resolved == null) resolved = handler.getReward(rewardName);
+ }
+ return giveRewardAsync(user, resolved, context.getOptions());
+ }
+
public void updateReward(Configuration data, String path, RewardOptions rewardOptions) {
if (data == null || path == null || !data.isConfigurationSection(path)) {
return;
@@ -228,4 +308,21 @@ private void giveSectionReward(AdvancedCoreUser user, ConfigurationSection data,
+ user.getUUID());
giveReward(user, reward, options);
}
+
+ private CompletionStage giveSectionRewardAsync(AdvancedCoreUser user, ConfigurationSection data, String path,
+ RewardExecutionContext context) {
+ RewardOptions options = context.getOptions();
+ String rewardName = context.buildRewardName(path);
+ DirectlyDefinedReward direct = handler.getDirectlyDefined(path);
+ SubDirectlyDefinedReward sub = handler.getSubDirectlyDefined(rewardName);
+ SubRewardResolver resolver = handler.getSubRewardResolver();
+ SubDirectlyDefinedReward fileSub = resolver == null ? null : resolver.getFileBackedSubReward(rewardName);
+ if (context.supportsDirectDispatch() && (direct != null || sub != null || fileSub != null)) {
+ Reward selected = direct != null ? direct.getReward() : (sub != null ? sub : fileSub).getReward();
+ return giveRewardAsync(user, selected, options);
+ }
+ Reward selected = new Reward(rewardName, data.getConfigurationSection(path));
+ selected.checkRewardFile();
+ return giveRewardAsync(user, selected, options);
+ }
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardHandler.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardHandler.java
index 885e062615..4312db52ea 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardHandler.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardHandler.java
@@ -7,6 +7,7 @@
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -184,14 +185,27 @@ public void giveReward(AdvancedCoreUser user, ConfigurationSection data, String
rewardExecutor.giveReward(user, data, path, rewardOptions);
}
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, ConfigurationSection data, String path,
+ RewardOptions rewardOptions) {
+ return rewardExecutor.giveRewardAsync(user, data, path, rewardOptions);
+ }
+
public void giveReward(AdvancedCoreUser user, Reward reward, RewardOptions rewardOptions) {
rewardExecutor.giveReward(user, reward, rewardOptions);
}
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, Reward reward, RewardOptions rewardOptions) {
+ return rewardExecutor.giveRewardAsync(user, reward, rewardOptions);
+ }
+
public void giveReward(AdvancedCoreUser user, String reward, RewardOptions rewardOptions) {
rewardExecutor.giveReward(user, reward, rewardOptions);
}
+ public CompletionStage giveRewardAsync(AdvancedCoreUser user, String reward, RewardOptions rewardOptions) {
+ return rewardExecutor.giveRewardAsync(user, reward, rewardOptions);
+ }
+
public void givePersistedQueueReward(AdvancedCoreUser user, PersistedQueueReference rewardReference,
RewardOptions rewardOptions) {
if (rewardReference == null) {
@@ -200,6 +214,14 @@ public void givePersistedQueueReward(AdvancedCoreUser user, PersistedQueueRefere
rewardExecutor.givePersistedQueueReward(user, rewardReference.getReference(), rewardOptions);
}
+ public CompletionStage givePersistedQueueRewardAsync(AdvancedCoreUser user,
+ PersistedQueueReference rewardReference, RewardOptions rewardOptions) {
+ if (rewardReference == null) {
+ return java.util.concurrent.CompletableFuture.completedFuture(null);
+ }
+ return rewardExecutor.givePersistedQueueRewardAsync(user, rewardReference.getReference(), rewardOptions);
+ }
+
public boolean hasDirectRewardHandle(String reward) {
return rewardRegistry.hasDirectRewardHandle(reward);
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
index f3701c64ff..f8b2b51d19 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
@@ -3,6 +3,8 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -30,6 +32,23 @@ private RewardAdvancedRewards() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("AdvancedRewards") {
+ @Override
+ public boolean supportsAsyncRequest() {
+ return true;
+ }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() {
+ return true;
+ }
+
+ @Override
+ public boolean supportsAsyncSynchronization() {
+ // This injector awaits nested rewards, which can invoke this same
+ // shared instance. Serializing that recursive chain would deadlock.
+ return false;
+ }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -42,6 +61,23 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection data, HashMap placeholders) {
+ if (!data.isConfigurationSection(getPath()) && !(isAlwaysForce() && data.contains(getPath(), true))
+ && !isAlwaysForceNoData()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+ for (String rewardName : ArrayUtils.convert(data.getConfigurationSection(getPath()).getKeys(false))) {
+ sequence = sequence.thenCompose(ignored -> handler.giveRewardAsync(user,
+ data.getConfigurationSection(getPath()), rewardName,
+ new RewardOptions().setPlaceholders(placeholders)
+ .setPrefix(reward.getRewardName() + "_AdvancedRewards")));
+ }
+ return sequence.thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSubRewards.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSubRewards.java
index bc681f1fc7..4cb4beb60d 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSubRewards.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSubRewards.java
@@ -2,6 +2,7 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -27,6 +28,16 @@ private RewardSubRewards() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("Rewards") {
+ @Override
+ public boolean supportsAsyncRequest() {
+ return true;
+ }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() {
+ return true;
+ }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -36,6 +47,17 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection data, HashMap placeholders) {
+ if (!data.isConfigurationSection(getPath()) && !(isAlwaysForce() && data.contains(getPath(), true))
+ && !isAlwaysForceNoData()) {
+ return java.util.concurrent.CompletableFuture.completedFuture(null);
+ }
+ return new RewardBuilder(reward.getConfig().getConfigData(), "Rewards").withPrefix(reward.getName())
+ .withPlaceHolder(placeholders).sendAsync(user).thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInject.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInject.java
index 72ba553e53..b96c055c1e 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInject.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInject.java
@@ -103,6 +103,24 @@ public boolean hasValidator() {
public boolean supportsAsyncRequest() {
return false;
}
+
+ /**
+ * Whether this asynchronous implementation is relevant only when its
+ * configured path exists. Existing third-party injectors keep the historic
+ * always-async opt-in behavior unless they explicitly opt into this guard.
+ */
+ public boolean requiresConfiguredDataForAsync() {
+ return false;
+ }
+
+ /**
+ * Whether asynchronous dispatch may serialize this injection across reward
+ * chains. Nested reward injectors opt out because awaiting a child that uses
+ * the same shared injection would otherwise wait on its own unfinished tail.
+ */
+ public boolean supportsAsyncSynchronization() {
+ return true;
+ }
public boolean isEditable() {
return !getEditButtons().isEmpty();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
index 666e041500..179f53b695 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
@@ -310,6 +310,8 @@ public void checkDelayedTimedRewards() {
plugin.debug("Checking timed/delayed for " + getPlayerName());
HashMap timed = getTimedRewards();
HashMap newTimed = new HashMap<>();
+ HashMap replayingTimed = new HashMap<>();
+ HashMap> replayCompletions = new HashMap<>();
for (Entry entry : timed.entrySet()) {
long time = entry.getValue();
@@ -326,7 +328,9 @@ public void checkDelayedTimedRewards() {
.withPlaceHolder(ArrayUtils.fromString(placeholders));
replayOptions.addPlaceholder("date",
"" + new SimpleDateFormat("EEE, d MMM yyyy HH:mm").format(new Date(time)));
- plugin.getRewardHandler().givePersistedQueueReward(this, new PersistedQueueReference(rewardReference), replayOptions);
+ replayingTimed.put(entry.getKey(), time);
+ replayCompletions.put(entry.getKey(), plugin.getRewardHandler().givePersistedQueueRewardAsync(this,
+ new PersistedQueueReference(rewardReference), replayOptions));
String rewardName = rewardReference;
plugin.debug("Giving timed/delayed reward " + rewardName + " for " + getPlayerName()
+ " with placeholders " + ArrayUtils.fromString(placeholders));
@@ -337,6 +341,14 @@ public void checkDelayedTimedRewards() {
}
setTimedRewards(newTimed);
+ for (Entry> replay : replayCompletions.entrySet()) {
+ String rewardEntry = replay.getKey();
+ long time = replayingTimed.get(rewardEntry);
+ replay.getValue().exceptionally(failure -> {
+ restoreTimedReward(rewardEntry, time, failure);
+ return null;
+ });
+ }
}
/**
@@ -371,11 +383,39 @@ public void checkOfflineRewards() {
RewardOptions options = new RewardOptions().setOnline(false).setCheckTimed(false)
.withPlaceHolder(ArrayUtils.fromString(placeholderStr));
- rewardHandler.givePersistedQueueReward(user, new PersistedQueueReference(rewardReference), options);
+ rewardHandler.givePersistedQueueRewardAsync(user, new PersistedQueueReference(rewardReference), options)
+ .exceptionally(failure -> {
+ restoreOfflineReward(rewardEntry, failure);
+ return null;
+ });
}
}
+ /** Restores a queue entry only after an asynchronous replay fails. */
+ private void restoreOfflineReward(String rewardEntry, Throwable failure) {
+ // Match addOfflineRewards' lock so a newly queued reward cannot be lost
+ // while a failed replay is being restored.
+ synchronized (plugin) {
+ ArrayList pending = getOfflineRewards();
+ // Queue entries are allowed to repeat; each failed replay must restore
+ // its own occurrence rather than collapsing identical rewards.
+ pending.add(rewardEntry);
+ setOfflineRewards(pending);
+ }
+ plugin.getLogger().warning("Could not deliver queued offline reward for " + getPlayerName()
+ + "; it will be retried: " + failure.getMessage());
+ }
+
+ /** Restores a due timed entry after an asynchronous replay fails. */
+ private synchronized void restoreTimedReward(String rewardEntry, long time, Throwable failure) {
+ HashMap pending = getTimedRewards();
+ pending.putIfAbsent(rewardEntry, time);
+ setTimedRewards(pending);
+ plugin.getLogger().warning("Could not deliver queued timed reward for " + getPlayerName()
+ + "; it will be retried: " + failure.getMessage());
+ }
+
/**
* Clears the user cache.
*/
@@ -452,7 +492,11 @@ public void forceRunOfflineRewards() {
RewardOptions options = new RewardOptions().setOnline(false).setGiveOffline(false).forceOffline()
.setCheckTimed(false).withPlaceHolder(ArrayUtils.fromString(placeholderStr));
- rewardHandler.givePersistedQueueReward(user, new PersistedQueueReference(rewardReference), options);
+ rewardHandler.givePersistedQueueRewardAsync(user, new PersistedQueueReference(rewardReference), options)
+ .exceptionally(failure -> {
+ restoreOfflineReward(rewardEntry, failure);
+ return null;
+ });
}
}
@@ -1669,4 +1713,4 @@ public void updateTempCacheWithColumns(ArrayList cols) {
getUserData().updateTempCacheWithColumns(cols);
}
-}
\ No newline at end of file
+}
diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
index 870e5b0082..3d814623f4 100644
--- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
+++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
@@ -7,6 +7,7 @@
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.when;
@@ -17,6 +18,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;
@@ -32,6 +34,7 @@
import com.bencodez.advancedcore.api.rewards.RewardHandler;
import com.bencodez.advancedcore.api.rewards.injected.RewardInject;
import com.bencodez.advancedcore.api.rewards.injected.RewardInjectInt;
+import com.bencodez.advancedcore.api.rewards.builtin.RewardSubRewards;
import com.bencodez.advancedcore.api.user.AdvancedCoreUser;
class RewardAsyncInjectionTest {
@@ -201,6 +204,22 @@ void failedAsyncInjectionStopsDependentRewardsAndPropagates() {
assertFalse(dependentRan.get());
}
+ @Test
+ void failingLegacyInjectionDoesNotStopLaterAsyncChainSteps() {
+ AtomicBoolean laterRan = new AtomicBoolean();
+ handler.getInjectedRewards().add(new RewardInject("BrokenLegacy") {
+ @Override
+ public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ throw new IllegalStateException("bad legacy configuration");
+ }
+ });
+ handler.getInjectedRewards().add(asyncInjection("Later", CompletableFuture.completedFuture(null), laterRan));
+
+ reward.giveInjectedRewardsAsync(user, new HashMap<>()).toCompletableFuture().join();
+ assertTrue(laterRan.get());
+ }
+
@Test
void synchronizedAsyncInjectionSerializesThroughCompletion() {
List> completions = new ArrayList<>();
@@ -305,6 +324,60 @@ protected long getServerThreadDispatchTimeoutMillis() {
result.toCompletableFuture().join();
}
+ @Test
+ void nestedPostRewardWaitsForChildRewardBeforeAdvancing() {
+ handler = org.mockito.Mockito.spy(new RewardHandler(plugin));
+ when(plugin.getRewardHandler()).thenReturn(handler);
+ when(user.getPlugin()).thenReturn(plugin);
+ data.createSection("Rewards");
+ CompletableFuture child = new CompletableFuture<>();
+ doReturn(child).when(handler).giveRewardAsync(eq(user), any(ConfigurationSection.class), eq("Rewards"),
+ any());
+ List events = new ArrayList<>();
+ RewardSubRewards.register(handler, plugin);
+ RewardInject after = new RewardInject("After") {
+ @Override
+ public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ events.add("after-child");
+ return null;
+ }
+ };
+ after.postReward();
+ handler.getInjectedRewards().add(after);
+
+ CompletionStage result = reward.giveInjectedRewardsAsync(user, new HashMap<>());
+ assertFalse(result.toCompletableFuture().isDone());
+ assertTrue(events.isEmpty());
+
+ child.complete(null);
+ result.toCompletableFuture().join();
+ assertEquals(List.of("after-child"), events);
+ }
+
+ @Test
+ void nestedAsyncInjectorCanAwaitTheSameSharedInjectorWithoutDeadlocking() {
+ AtomicInteger invocations = new AtomicInteger();
+ RewardInject nested = new RewardInject("Nested") {
+ @Override public boolean supportsAsyncRequest() { return true; }
+ @Override public boolean supportsAsyncSynchronization() { return false; }
+ @Override public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) { return null; }
+ @Override public CompletionStage onRewardRequestAsync(Reward current, AdvancedCoreUser currentUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ if (invocations.incrementAndGet() == 1) {
+ return current.giveInjectedRewardsAsync(currentUser, new HashMap<>()).thenApply(ignored -> null);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+ };
+ nested.synchronize();
+ handler.getInjectedRewards().add(nested);
+
+ reward.giveInjectedRewardsAsync(user, new HashMap<>()).toCompletableFuture().join();
+ assertEquals(2, invocations.get());
+ }
+
private RewardInject asyncInjection(String path, CompletionStage completion, AtomicBoolean invoked) {
return new RewardInject(path) {
@Override public boolean supportsAsyncRequest() { return true; }
diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
index 780107790d..030c207c4b 100644
--- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
+++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
@@ -4,12 +4,15 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.logging.Logger;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -53,6 +56,7 @@ public void setUp() {
when(plugin.getUserManager()).thenReturn(userManager);
when(plugin.getRewardHandler()).thenReturn(rewardHandler);
when(plugin.getStorageType()).thenReturn(UserStorage.MYSQL);
+ when(plugin.getLogger()).thenReturn(mock(Logger.class));
when(userManager.getOfflineRewardsPath()).thenReturn("offlineRewardsPath");
when(configOptions.isProcessRewards()).thenReturn(true);
@@ -66,11 +70,12 @@ void checkOfflineRewards_preservesServerRequirementForNormalReplay() {
rewards.add("VoteReward%placeholders%Server%pair%server-a");
when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT)).thenReturn(rewards);
- user.checkOfflineRewards();
-
ArgumentCaptor referenceCaptor = ArgumentCaptor.forClass(PersistedQueueReference.class);
ArgumentCaptor optionsCaptor = ArgumentCaptor.forClass(RewardOptions.class);
- verify(rewardHandler).givePersistedQueueReward(eq(user), referenceCaptor.capture(), optionsCaptor.capture());
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(CompletableFuture.completedFuture(null));
+ user.checkOfflineRewards();
+ verify(rewardHandler).givePersistedQueueRewardAsync(eq(user), referenceCaptor.capture(), optionsCaptor.capture());
assertEquals("VoteReward", referenceCaptor.getValue().getReference());
RewardOptions options = optionsCaptor.getValue();
assertFalse(options.isForceOffline());
@@ -85,11 +90,12 @@ void forceRunOfflineRewards_stillBypassesNormalReplayChecks() {
rewards.add("VoteReward%placeholders%Server%pair%server-a");
when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT)).thenReturn(rewards);
- user.forceRunOfflineRewards();
-
ArgumentCaptor referenceCaptor = ArgumentCaptor.forClass(PersistedQueueReference.class);
ArgumentCaptor optionsCaptor = ArgumentCaptor.forClass(RewardOptions.class);
- verify(rewardHandler).givePersistedQueueReward(eq(user), referenceCaptor.capture(), optionsCaptor.capture());
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(CompletableFuture.completedFuture(null));
+ user.forceRunOfflineRewards();
+ verify(rewardHandler).givePersistedQueueRewardAsync(eq(user), referenceCaptor.capture(), optionsCaptor.capture());
assertEquals("VoteReward", referenceCaptor.getValue().getReference());
RewardOptions options = optionsCaptor.getValue();
assertTrue(options.isForceOffline());
@@ -143,4 +149,39 @@ void setOfflineRewards_withinLimit() {
user.setOfflineRewards(rewards);
verify(data).setStringList("offlineRewardsPath", rewards);
}
+
+ @Test
+ void failedAsyncOfflineReplayIsRestoredForRetry() {
+ ArrayList initial = new ArrayList<>();
+ initial.add("VoteReward%placeholders%Server%pair%server-a");
+ initial.add("VoteReward%placeholders%Server%pair%server-a");
+ when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT))
+ .thenReturn(new ArrayList<>(initial)).thenReturn(new ArrayList<>());
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(CompletableFuture.failedFuture(new IllegalStateException("temporary")));
+
+ user.checkOfflineRewards();
+
+ ArgumentCaptor> saved = ArgumentCaptor.forClass(ArrayList.class);
+ verify(data, org.mockito.Mockito.times(3)).setStringList(eq("offlineRewardsPath"), saved.capture());
+ assertEquals(initial, saved.getAllValues().get(2));
+ }
+
+ @Test
+ void failedAsyncTimedReplayIsRestoredAfterDueEntriesAreSaved() {
+ long due = System.currentTimeMillis() - 1_000;
+ String entry = "VoteReward%extime%1%placeholders%Server%pair%server-a";
+ ArrayList timed = new ArrayList<>();
+ timed.add(entry + "%ExecutionTime/%" + due);
+ when(data.getStringList("TimedRewards", UserDataFetchMode.DEFAULT))
+ .thenReturn(new ArrayList<>(timed)).thenReturn(new ArrayList<>());
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(CompletableFuture.failedFuture(new IllegalStateException("temporary")));
+
+ user.checkDelayedTimedRewards();
+
+ ArgumentCaptor> saved = ArgumentCaptor.forClass(ArrayList.class);
+ verify(data, org.mockito.Mockito.times(2)).setStringList(eq("TimedRewards"), saved.capture());
+ assertEquals(timed, saved.getAllValues().get(1));
+ }
}
From 31ef89f8e838a88a8f74f322fe476b16f876cd67 Mon Sep 17 00:00:00 2001
From: BenCodez <17074231+BenCodez@users.noreply.github.com>
Date: Thu, 10 Sep 2026 19:18:03 -0600
Subject: [PATCH 03/78] Fix durable async reward replay
---
.../advancedcore/api/rewards/Reward.java | 217 +++++++++-
.../api/rewards/RewardExecutor.java | 10 +-
.../api/rewards/RewardOptions.java | 68 +++-
.../builtin/RewardAdvancedPriority.java | 35 ++
.../builtin/RewardAdvancedRandomReward.java | 22 ++
.../builtin/RewardAdvancedRewards.java | 8 +-
.../rewards/builtin/RewardAdvancedWorld.java | 32 +-
.../api/rewards/builtin/RewardJavascript.java | 21 +
.../api/rewards/builtin/RewardLucky.java | 51 +++
.../api/rewards/builtin/RewardPriority.java | 31 ++
.../api/rewards/builtin/RewardRandom.java | 34 ++
.../rewards/builtin/RewardRandomReward.java | 20 +
.../rewards/builtin/RewardSpecialChance.java | 53 +++
.../RewardInjectConfigurationSection.java | 29 +-
.../injected/RewardInjectStringList.java | 42 +-
.../api/user/AdvancedCoreUser.java | 369 ++++++++++++++----
.../api/user/usercache/UserDataCache.java | 25 ++
.../rewards/RewardAsyncInjectionTest.java | 111 ++++++
.../tests/rewards/RewardExecutorTest.java | 20 +
.../tests/user/AdvancedCoreUserTest.java | 206 +++++++++-
.../user/UserDataCacheSchedulingTest.java | 31 ++
21 files changed, 1321 insertions(+), 114 deletions(-)
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
index 0142b28b62..cf0b541861 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/Reward.java
@@ -6,13 +6,18 @@
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
+import java.util.function.Consumer;
import java.util.logging.Level;
import org.bukkit.Bukkit;
@@ -36,6 +41,9 @@
* The Class Reward.
*/
public class Reward {
+ private static final ThreadLocal ACTIVE_REPLAY_STATE = new ThreadLocal<>();
+ private static final ThreadLocal ACTIVE_REPLAY_KEY = new ThreadLocal<>();
+ private static final String REPLAY_SELECTION_PREFIX = "__advancedcore_replay_selection_";
@Getter
@Setter
@@ -263,34 +271,193 @@ public void giveInjectedRewards(AdvancedCoreUser user, HashMap p
*/
public CompletionStage giveInjectedRewardsAsync(AdvancedCoreUser user,
HashMap placeholders) {
- List postRewards = new ArrayList<>();
- CompletionStage sequence = CompletableFuture.completedFuture(null);
+ return giveInjectedRewardsAsync(user, placeholders, 0, new ReplayState(null), getRewardName());
+ }
+ /**
+ * Resumes a persisted asynchronous reward after the supplied number of
+ * injection stages have completed. Queue recovery uses this checkpoint to
+ * avoid replaying already-applied non-idempotent injections.
+ */
+ private CompletionStage giveInjectedRewardsAsync(AdvancedCoreUser user,
+ HashMap placeholders, int completedStages, ReplayState replayState, String replayKey) {
+ List postRewards = new ArrayList<>();
+ List orderedRewards = new ArrayList<>();
for (RewardInject inject : plugin.getRewardHandler().getInjectedRewards()) {
- if (inject.isPostReward()) {
- postRewards.add(inject);
- continue;
- }
- sequence = sequence.thenCompose(ignored -> invokeInjectionAsync(inject, user, placeholders))
- .thenCompose(result -> resumeOnServerThread(user).thenRun(() -> {
+ if (inject.isPostReward()) postRewards.add(inject);
+ else orderedRewards.add(inject);
+ }
+ orderedRewards.addAll(postRewards);
+ int resumeAfter = Math.max(0, replayState.getCompleted(replayKey, completedStages));
+ AtomicInteger completed = new AtomicInteger(resumeAfter);
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+ for (int index = 0; index < orderedRewards.size(); index++) {
+ if (index < resumeAfter) continue;
+ final RewardInject inject = orderedRewards.get(index);
+ final String injectionKey = replayKey + "/" + index;
+ sequence = sequence.thenCompose(ignored -> invokeInjectionAsync(inject, user, placeholders, replayState,
+ injectionKey))
+ .thenApply(result -> {
if (inject.isAddAsPlaceholder() && result != null) addPlaceholder(inject, result, placeholders);
- }));
+ int checkpoint = completed.incrementAndGet();
+ replayState.setCompleted(replayKey, checkpoint);
+ replayState.persistCheckpoint(placeholders);
+ return result;
+ }).thenCompose(ignored -> resumeOnServerThread(user));
}
+ return sequence.handle((ignored, failure) -> {
+ if (failure == null) return null;
+ RewardReplayFailure nestedFailure = findReplayFailure(failure);
+ if (nestedFailure != null) throw nestedFailure;
+ replayState.setCompleted(replayKey, completed.get());
+ throw new RewardReplayFailure(replayState, placeholders, failure);
+ });
+ }
- for (RewardInject inject : postRewards) {
- sequence = sequence.thenCompose(ignored -> invokeInjectionAsync(inject, user, placeholders))
- .thenCompose(result -> resumeOnServerThread(user));
+ // Kept as a narrow compatibility bridge for internal callers/tests that only
+ // have a single reward checkpoint.
+ private CompletionStage giveInjectedRewardsAsync(AdvancedCoreUser user,
+ HashMap placeholders, int completedStages) {
+ return giveInjectedRewardsAsync(user, placeholders, completedStages, new ReplayState(null), getRewardName());
+ }
+
+ /** Details the last durably safe replay checkpoint when an async chain fails. */
+ public static final class RewardReplayFailure extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+ private final ReplayState replayState;
+ @Getter private final int completedInjectionCount;
+ @Getter private final HashMap replayPlaceholders;
+
+ private RewardReplayFailure(ReplayState replayState, HashMap replayPlaceholders, Throwable cause) {
+ super("Asynchronous reward replay failed after a completed asynchronous reward stage",
+ cause);
+ this.replayState = replayState;
+ this.completedInjectionCount = replayState.highestCompletedCount();
+ this.replayPlaceholders = new HashMap<>(replayPlaceholders);
+ }
+
+ public Map getReplayProgress() { return replayState.copyProgress(); }
+ }
+
+ private static RewardReplayFailure findReplayFailure(Throwable failure) {
+ for (Throwable current = failure; current != null; current = current.getCause()) {
+ if (current instanceof RewardReplayFailure) return (RewardReplayFailure) current;
+ }
+ return null;
+ }
+
+ /** Captures the active nested replay state for deferred child dispatch. */
+ public static ReplayState currentReplayState() { return ACTIVE_REPLAY_STATE.get(); }
+
+ /** Captures the active parent execution path before deferred work runs. */
+ public static String currentReplayKey() { return ACTIVE_REPLAY_KEY.get(); }
+
+ /** Obtains one shared replay state for a group of nested dispatches. */
+ public static ReplayState replayStateFor(RewardOptions options) {
+ ReplayState replayState = options.getAsyncReplayState();
+ if (replayState == null) {
+ replayState = ACTIVE_REPLAY_STATE.get();
+ if (replayState == null) replayState = new ReplayState(options.getAsyncReplayProgress());
+ options.setAsyncReplayState(replayState);
+ }
+ if (options.getAsyncReplayCheckpointConsumer() != null) {
+ replayState.setCheckpointConsumer(options.getAsyncReplayCheckpointConsumer());
+ }
+ return replayState;
+ }
+
+ /**
+ * Records a nondeterministic nested-reward decision in the placeholder
+ * snapshot carried by {@link RewardReplayFailure}. A retry consequently
+ * executes the same selected branch instead of rolling a new outcome after a
+ * child has already performed a side effect.
+ */
+ public static String replaySelection(HashMap placeholders, Supplier selector) {
+ String activeKey = ACTIVE_REPLAY_KEY.get();
+ String storageKey = REPLAY_SELECTION_PREFIX + Base64.getUrlEncoder().withoutPadding()
+ .encodeToString((activeKey == null ? "root" : activeKey).getBytes(StandardCharsets.UTF_8));
+ if (placeholders.containsKey(storageKey)) {
+ String stored = placeholders.get(storageKey);
+ return stored.isEmpty() ? null : new String(Base64.getUrlDecoder().decode(stored), StandardCharsets.UTF_8);
+ }
+ String selected = selector.get();
+ placeholders.put(storageKey, selected == null ? "" : Base64.getUrlEncoder().withoutPadding()
+ .encodeToString(selected.getBytes(StandardCharsets.UTF_8)));
+ return selected;
+ }
+
+ /** Attaches a captured replay state to an option object for a deferred child. */
+ public static RewardOptions withReplayState(RewardOptions options, ReplayState replayState) {
+ if (replayState != null) options.setAsyncReplayState(replayState);
+ return options;
+ }
+
+ /**
+ * Associates a deferred nested reward with a deterministic child occurrence.
+ * A list may intentionally contain the same reward more than once, so its
+ * replay checkpoint must never be shared merely because its name is equal.
+ */
+ public static RewardOptions withReplayState(RewardOptions options, ReplayState replayState,
+ String childOccurrence) {
+ return withReplayState(options, replayState, ACTIVE_REPLAY_KEY.get(), childOccurrence);
+ }
+
+ /**
+ * Associates deferred nested work with the parent path captured while the
+ * injector was invoked. Later completion callbacks do not retain ThreadLocal
+ * context, so they must use this explicit form.
+ */
+ public static RewardOptions withReplayState(RewardOptions options, ReplayState replayState,
+ String parentKey, String childOccurrence) {
+ withReplayState(options, replayState);
+ if (parentKey != null && childOccurrence != null) {
+ options.setAsyncReplayKey(parentKey + "/" + childOccurrence);
+ }
+ return options;
+ }
+
+ /** Internal shared state passed through nested async reward dispatch. */
+ public static final class ReplayState {
+ private final HashMap completed = new HashMap<>();
+ private Consumer checkpointConsumer;
+ private ReplayState(Map initial) { if (initial != null) completed.putAll(initial); }
+ private synchronized int getCompleted(String rewardName, int fallback) {
+ return completed.getOrDefault(rewardName, fallback);
+ }
+ private synchronized void setCompleted(String rewardName, int count) { completed.put(rewardName, count); }
+ private synchronized Map copyProgress() { return new HashMap<>(completed); }
+ private synchronized int highestCompletedCount() {
+ int highest = 0;
+ for (int count : completed.values()) highest = Math.max(highest, count);
+ return highest;
+ }
+ private synchronized void setCheckpointConsumer(Consumer consumer) {
+ checkpointConsumer = consumer;
+ }
+ private void persistCheckpoint(HashMap placeholders) {
+ Consumer consumer;
+ synchronized (this) { consumer = checkpointConsumer; }
+ if (consumer != null) consumer.accept(new ReplayCheckpoint(copyProgress(), placeholders));
+ }
+ }
+
+ /** Immutable durable replay data emitted after every completed injection. */
+ public static final class ReplayCheckpoint {
+ @Getter private final Map replayProgress;
+ @Getter private final HashMap placeholders;
+ private ReplayCheckpoint(Map replayProgress, HashMap placeholders) {
+ this.replayProgress = new HashMap<>(replayProgress);
+ this.placeholders = new HashMap<>(placeholders);
}
- return sequence;
}
private CompletionStage invokeInjectionAsync(RewardInject inject, AdvancedCoreUser user,
- HashMap placeholders) {
+ HashMap placeholders, ReplayState replayState, String injectionKey) {
try {
if (!plugin.isEnabled()) return CompletableFuture.failedFuture(
new IllegalStateException("Plugin disabled before reward injection completed"));
Supplier> request = () -> requestOnServerThread(user,
- () -> requestInjectionAsync(inject, user, placeholders));
+ () -> requestInjectionAsync(inject, user, placeholders, replayState, injectionKey));
CompletionStage result = inject.isSynchronize() && inject.supportsAsyncSynchronization()
? inject.runSynchronizedAsync(request)
: request.get();
@@ -365,7 +532,12 @@ protected long getServerThreadDispatchTimeoutMillis() {
}
private CompletionStage requestInjectionAsync(RewardInject inject, AdvancedCoreUser user,
- HashMap placeholders) {
+ HashMap placeholders, ReplayState replayState, String injectionKey) {
+ ReplayState previous = ACTIVE_REPLAY_STATE.get();
+ String previousKey = ACTIVE_REPLAY_KEY.get();
+ ACTIVE_REPLAY_STATE.set(replayState);
+ ACTIVE_REPLAY_KEY.set(injectionKey);
+ try {
if (inject.supportsAsyncRequest()) {
return inject.onRewardRequestAsync(this, user, getConfig().getConfigData(), placeholders);
}
@@ -378,6 +550,10 @@ private CompletionStage requestInjectionAsync(RewardInject inject, Advan
failure.printStackTrace();
return CompletableFuture.completedFuture(null);
}
+ } finally {
+ if (previous == null) ACTIVE_REPLAY_STATE.remove(); else ACTIVE_REPLAY_STATE.set(previous);
+ if (previousKey == null) ACTIVE_REPLAY_KEY.remove(); else ACTIVE_REPLAY_KEY.set(previousKey);
+ }
}
private void addPlaceholder(RewardInject inject, Object obj, HashMap placeholders) {
@@ -658,7 +834,14 @@ public CompletionStage giveRewardUserAsync(AdvancedCoreUser user, HashMap<
if (placeholders == null) {
return CompletableFuture.completedFuture(null);
}
- return giveInjectedRewardsAsync(user, placeholders)
+ ReplayState replayState = replayStateFor(rewardOptions);
+ String replayKey = rewardOptions.getAsyncReplayKey();
+ if (replayKey == null) {
+ String parentKey = ACTIVE_REPLAY_KEY.get();
+ replayKey = parentKey == null ? getRewardName() : parentKey + "/" + getRewardName();
+ rewardOptions.setAsyncReplayKey(replayKey);
+ }
+ return giveInjectedRewardsAsync(user, placeholders, rewardOptions.getCompletedAsyncInjections(), replayState, replayKey)
.thenRun(() -> plugin.debug("Gave " + user.getPlayerName() + " reward " + name));
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
index a79659701f..faeebd80d1 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardExecutor.java
@@ -115,8 +115,16 @@ public CompletionStage giveRewardAsync(AdvancedCoreUser user, Configuratio
if (data.isList(path)) {
CompletionStage sequence = CompletableFuture.completedFuture(null);
+ Reward.ReplayState replayState = Reward.replayStateFor(options);
+ String parentReplayKey = options.getAsyncReplayKey();
+ if (parentReplayKey == null) parentReplayKey = Reward.currentReplayKey();
+ if (parentReplayKey == null) parentReplayKey = "list:" + path;
+ int rewardIndex = 0;
for (String nestedReward : new ArrayList<>(data.getStringList(path))) {
- sequence = sequence.thenCompose(ignored -> giveRewardAsync(user, nestedReward, options));
+ RewardOptions nestedOptions = options.copyForNestedDispatch(
+ parentReplayKey + "/" + nestedReward + ":" + rewardIndex++);
+ nestedOptions.setAsyncReplayState(replayState);
+ sequence = sequence.thenCompose(ignored -> giveRewardAsync(user, nestedReward, nestedOptions));
}
return sequence;
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardOptions.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardOptions.java
index 014c611637..b26f3b77d7 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardOptions.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/RewardOptions.java
@@ -1,7 +1,9 @@
package com.bencodez.advancedcore.api.rewards;
-import java.util.HashMap;
-import java.util.Map.Entry;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.function.Consumer;
import com.bencodez.advancedcore.AdvancedCorePlugin;
import com.bencodez.simpleapi.array.ArrayUtils;
@@ -41,8 +43,37 @@ public class RewardOptions {
@Getter
private boolean useDefaultWorlds = true;
- @Getter
- private long orginalTrigger = -1;
+ @Getter
+ private long orginalTrigger = -1;
+
+ /**
+ * Number of completed injection stages recorded for a persisted replay. This
+ * is intentionally internal queue metadata: normal reward dispatch always
+ * starts at zero.
+ */
+ @Getter
+ @Setter
+ private int completedAsyncInjections;
+
+ @Getter
+ @Setter
+ private Map asyncReplayProgress = new HashMap<>();
+
+ @Getter
+ @Setter
+ private Reward.ReplayState asyncReplayState;
+
+ /**
+ * Stable execution path for a nested asynchronous reward. The path is queue
+ * metadata only; callers that do not participate in replay leave it unset.
+ */
+ @Getter
+ @Setter
+ private String asyncReplayKey;
+
+ @Getter
+ @Setter
+ private Consumer asyncReplayCheckpointConsumer;
public RewardOptions() {
}
@@ -172,11 +203,30 @@ public String toString() {
}
- public RewardOptions withPlaceHolder(HashMap placeholders2) {
+ public RewardOptions withPlaceHolder(HashMap placeholders2) {
for (Entry entry : placeholders2.entrySet()) {
placeholders.put(entry.getKey(), entry.getValue());
}
- return this;
- }
-
-}
+ return this;
+ }
+
+ /**
+ * Makes an isolated option object for one member of an asynchronously
+ * dispatched list. Mutable placeholders and replay metadata must not leak
+ * from one list member into the next.
+ */
+ RewardOptions copyForNestedDispatch(String replayKey) {
+ RewardOptions copy = new RewardOptions().setCheckTimed(checkTimed).setGiveOffline(giveOffline)
+ .setIgnoreChance(ignoreChance).setIgnoreRequirements(ignoreRequirements).setPrefix(prefix).setSuffix(suffix)
+ .orginalTrigger(orginalTrigger).setPlaceholders(new HashMap<>(placeholders));
+ if (forceOffline) copy.forceOffline();
+ if (!useDefaultWorlds) copy.disableDefaultWorlds();
+ if (onlineSet) copy.setOnline(online);
+ if (!server.isEmpty()) copy.setServer(server);
+ copy.setAsyncReplayState(asyncReplayState);
+ copy.setAsyncReplayKey(replayKey);
+ copy.setAsyncReplayCheckpointConsumer(asyncReplayCheckpointConsumer);
+ return copy;
+ }
+
+}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedPriority.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedPriority.java
index fc1bb37eb7..b1569b4179 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedPriority.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedPriority.java
@@ -2,6 +2,8 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -28,6 +30,15 @@ private RewardAdvancedPriority() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("AdvancedPriority") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward sourceReward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -46,6 +57,30 @@ public String onRewardRequested(Reward sourceReward, AdvancedCoreUser user, Conf
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward sourceReward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ String selectedKey = Reward.replaySelection(placeholders, () -> {
+ for (String key : section.getKeys(false)) {
+ RewardOptions namingOptions = new RewardOptions()
+ .setPrefix(sourceReward.getName() + "_AdvancedPriority");
+ Reward reward = handler.getReward(section, key, namingOptions);
+ if (reward != null && reward.canGiveReward(user,
+ new RewardOptions().withPlaceHolder(placeholders))) {
+ return key;
+ }
+ }
+ return null;
+ });
+ if (selectedKey == null) return CompletableFuture.completedFuture(null);
+ Reward selected = handler.getReward(section, selectedKey, new RewardOptions()
+ .setPrefix(sourceReward.getName() + "_AdvancedPriority"));
+ return selected == null ? CompletableFuture.completedFuture(null)
+ : handler.giveRewardAsync(user, selected, new RewardOptions().setIgnoreChance(true)
+ .setIgnoreRequirements(true).setPrefix(sourceReward.getName() + "_AdvancedPriority")
+ .withPlaceHolder(placeholders)).thenApply(ignored -> selected.getName());
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRandomReward.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRandomReward.java
index 1003eab2bc..fc3ce1af7d 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRandomReward.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRandomReward.java
@@ -4,6 +4,8 @@
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -31,6 +33,15 @@ private RewardAdvancedRandomReward() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("AdvancedRandomReward") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -45,6 +56,17 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ ArrayList rewards = ArrayUtils.convert(section.getKeys(false));
+ if (rewards.isEmpty()) return CompletableFuture.completedFuture(null);
+ String selected = Reward.replaySelection(placeholders,
+ () -> rewards.get(ThreadLocalRandom.current().nextInt(rewards.size())));
+ return handler.giveRewardAsync(user, section, selected, new RewardOptions().setPlaceholders(placeholders)
+ .setPrefix(reward.getRewardName() + "_AdvancedRandomReward")).thenApply(ignored -> selected);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
index f8b2b51d19..c61277d1b1 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedRewards.java
@@ -69,10 +69,16 @@ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreU
return CompletableFuture.completedFuture(null);
}
CompletionStage sequence = CompletableFuture.completedFuture(null);
+ com.bencodez.advancedcore.api.rewards.Reward.ReplayState replayState = Reward.currentReplayState();
+ String parentReplayKey = Reward.currentReplayKey();
+ int rewardIndex = 0;
for (String rewardName : ArrayUtils.convert(data.getConfigurationSection(getPath()).getKeys(false))) {
+ final int childIndex = rewardIndex++;
sequence = sequence.thenCompose(ignored -> handler.giveRewardAsync(user,
data.getConfigurationSection(getPath()), rewardName,
- new RewardOptions().setPlaceholders(placeholders)
+ Reward.withReplayState(new RewardOptions().setPlaceholders(placeholders), replayState,
+ parentReplayKey,
+ rewardName + ":" + childIndex)
.setPrefix(reward.getRewardName() + "_AdvancedRewards")));
}
return sequence.thenApply(ignored -> null);
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedWorld.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedWorld.java
index f0e7cdaa78..1ccd426457 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedWorld.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardAdvancedWorld.java
@@ -2,6 +2,8 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -29,10 +31,19 @@ private RewardAdvancedWorld() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("AdvancedWorld") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward sourceReward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
- for (String key : section.getKeys(false)) {
+ for (String key : section.getKeys(false)) {
plugin.extraDebug("AdvancedWorld: Giving reward " + sourceReward.getName() + "_AdvancedWorld");
section.set(key + ".Worlds", ArrayUtils.convert(new String[] { key }));
handler.giveReward(user, section, key, new RewardOptions().withPlaceHolder(placeholders)
@@ -41,6 +52,25 @@ public String onRewardRequested(Reward sourceReward, AdvancedCoreUser user, Conf
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward sourceReward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+ com.bencodez.advancedcore.api.rewards.Reward.ReplayState replayState = Reward.currentReplayState();
+ String parentReplayKey = Reward.currentReplayKey();
+ int worldIndex = 0;
+ for (String key : section.getKeys(false)) {
+ final int childIndex = worldIndex++;
+ section.set(key + ".Worlds", ArrayUtils.convert(new String[] { key }));
+ sequence = sequence.thenCompose(ignored -> handler.giveRewardAsync(user, section, key,
+ Reward.withReplayState(new RewardOptions().withPlaceHolder(placeholders), replayState,
+ parentReplayKey,
+ key + ":" + childIndex)
+ .setPrefix(sourceReward.getRewardName() + "_AdvancedWorld")));
+ }
+ return sequence.thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardJavascript.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardJavascript.java
index d80fcd9f66..188935d6b7 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardJavascript.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardJavascript.java
@@ -2,6 +2,7 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -55,6 +56,15 @@ public void setValue(Player player, ArrayList value) {
}.addLore("Javascript expressions to run"))));
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("Javascript") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -70,6 +80,17 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ if (!section.getBoolean("Enabled")) return java.util.concurrent.CompletableFuture.completedFuture(null);
+ String path = Reward.replaySelection(placeholders,
+ () -> new JavascriptEngine().addPlayer(user.getOfflinePlayer()).addPlaceholders(placeholders)
+ .getBooleanValue(section.getString("Expression")) ? "TrueRewards" : "FalseRewards");
+ return new RewardBuilder(section, path).withPrefix(reward.getName() + ".Javascript").sendAsync(user)
+ .thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardLucky.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardLucky.java
index fd6fcdf3c0..b3d0373d03 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardLucky.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardLucky.java
@@ -4,6 +4,8 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -35,6 +37,15 @@ private RewardLucky() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("Lucky") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -71,6 +82,46 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ HashMap luckyRewards = new HashMap<>();
+ for (String key : section.getKeys(false)) {
+ if (MessageAPI.isInt(key)) {
+ int num = Integer.parseInt(key);
+ if (num > 0) luckyRewards.put(num, "Lucky." + num);
+ }
+ }
+ String choices = Reward.replaySelection(placeholders, () -> {
+ HashMap selected = new LinkedHashMap<>();
+ for (Entry entry : luckyRewards.entrySet()) {
+ if (MiscUtils.getInstance().checkChance(1, entry.getKey())) selected.put(entry.getValue(), entry.getKey());
+ }
+ selected = ArrayUtils.sortByValuesStr(selected, false);
+ ArrayList paths = new ArrayList<>(selected.keySet());
+ if (reward.getConfig().getConfigData().getBoolean("OnlyOneLucky", false) && paths.size() > 1) {
+ paths.subList(1, paths.size()).clear();
+ }
+ return paths.isEmpty() ? null : String.join("\n", paths);
+ });
+ if (choices == null) return CompletableFuture.completedFuture(null);
+ CompletionStage sequence = CompletableFuture.completedFuture(null);
+ com.bencodez.advancedcore.api.rewards.Reward.ReplayState replayState = Reward.currentReplayState();
+ String parentReplayKey = Reward.currentReplayKey();
+ int luckyIndex = 0;
+ for (String path : choices.split("\\n")) {
+ final int childIndex = luckyIndex++;
+ sequence = sequence.thenCompose(ignored -> {
+ RewardBuilder builder = new RewardBuilder(reward.getConfig().getConfigData(), path)
+ .withPrefix(reward.getName()).withPlaceHolder(placeholders);
+ Reward.withReplayState(builder.getRewardOptions(), replayState, parentReplayKey,
+ path + ":" + childIndex);
+ return builder.sendAsync(user);
+ });
+ }
+ return sequence.thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardPriority.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardPriority.java
index 21a6b98ab6..a9543b1353 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardPriority.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardPriority.java
@@ -2,6 +2,8 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.entity.Player;
@@ -25,6 +27,15 @@ private RewardPriority() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectStringList("Priority") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequest(Reward source, AdvancedCoreUser user, ArrayList list,
HashMap placeholders) {
@@ -38,6 +49,26 @@ public String onRewardRequest(Reward source, AdvancedCoreUser user, ArrayList onRewardRequestAsync(Reward source, AdvancedCoreUser user,
+ ArrayList list, HashMap placeholders) {
+ String selectedName = Reward.replaySelection(placeholders, () -> {
+ for (String rewardName : list) {
+ Reward reward = handler.getReward(rewardName);
+ if (reward != null && reward.canGiveReward(user,
+ new RewardOptions().withPlaceHolder(placeholders))) {
+ return rewardName;
+ }
+ }
+ return null;
+ });
+ if (selectedName == null) return CompletableFuture.completedFuture(null);
+ Reward selected = handler.getReward(selectedName);
+ return selected == null ? CompletableFuture.completedFuture(null)
+ : handler.giveRewardAsync(user, selected, new RewardOptions().withPlaceHolder(placeholders)
+ .setIgnoreChance(true).setIgnoreRequirements(true)).thenApply(ignored -> selected.getName());
+ }
}.asPlaceholder("Priority").addEditButton(
new EditGUIButton(new ItemBuilder(Material.PAPER), new EditGUIValueList("Priority", null) {
@Override
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandom.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandom.java
index e63f8cbb37..ed98e100db 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandom.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandom.java
@@ -3,6 +3,8 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.configuration.ConfigurationSection;
@@ -24,6 +26,15 @@ private RewardRandom() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("Random") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@SuppressWarnings("unchecked")
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
@@ -48,6 +59,29 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @SuppressWarnings("unchecked")
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ String selection = Reward.replaySelection(placeholders, () -> {
+ if (!MiscUtils.getInstance().checkChance(section.getDouble("Chance", 100), 100)) return "fallback";
+ if (!section.getBoolean("PickRandom", true)) return "rewards";
+ ArrayList rewards = (ArrayList) section.getList("Rewards", new ArrayList<>());
+ return rewards == null || rewards.isEmpty() ? "none"
+ : "pick:" + rewards.get(ThreadLocalRandom.current().nextInt(rewards.size()));
+ });
+ if (selection == null || selection.equals("none")) return CompletableFuture.completedFuture(null);
+ if (selection.startsWith("pick:")) {
+ String selected = selection.substring("pick:".length());
+ return selected.isEmpty() ? CompletableFuture.completedFuture(null)
+ : handler.giveRewardAsync(user, selected,
+ new RewardOptions().setPlaceholders(placeholders)).thenApply(ignored -> null);
+ }
+ String path = selection.equals("rewards") ? "Random.Rewards" : "Random.FallBack";
+ return new RewardBuilder(reward.getConfig().getConfigData(), path).withPrefix(reward.getName())
+ .withPlaceHolder(placeholders).sendAsync(user).thenApply(ignored -> null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandomReward.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandomReward.java
index 03dab009e5..f2f39bf124 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandomReward.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardRandomReward.java
@@ -2,6 +2,7 @@
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.concurrent.CompletionStage;
import java.util.concurrent.ThreadLocalRandom;
import org.bukkit.Material;
@@ -25,6 +26,15 @@ private RewardRandomReward() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectStringList("RandomReward") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequest(Reward reward, AdvancedCoreUser user, ArrayList list,
HashMap placeholders) {
@@ -35,6 +45,16 @@ public String onRewardRequest(Reward reward, AdvancedCoreUser user, ArrayList onRewardRequestAsync(Reward reward, AdvancedCoreUser user,
+ ArrayList list, HashMap placeholders) {
+ if (list.isEmpty()) return java.util.concurrent.CompletableFuture.completedFuture(null);
+ String selected = Reward.replaySelection(placeholders,
+ () -> list.get(ThreadLocalRandom.current().nextInt(list.size())));
+ return handler.giveRewardAsync(user, selected, new RewardOptions().setPlaceholders(placeholders))
+ .thenApply(ignored -> selected);
+ }
}.asPlaceholder("RandomReward").priority(20).addEditButton(
new EditGUIButton(new ItemBuilder(Material.PAPER), new EditGUIValueList("RandomReward", null) {
@Override
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSpecialChance.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSpecialChance.java
index ae5be20a20..2ce15237c4 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSpecialChance.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/builtin/RewardSpecialChance.java
@@ -7,6 +7,8 @@
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.Material;
import org.bukkit.configuration.ConfigurationSection;
@@ -34,6 +36,15 @@ private RewardSpecialChance() {
public static void register(RewardHandler handler, AdvancedCorePlugin plugin) {
handler.getInjectedRewards().add(new RewardInjectConfigurationSection("SpecialChance") {
+ @Override
+ public boolean supportsAsyncRequest() { return true; }
+
+ @Override
+ public boolean requiresConfiguredDataForAsync() { return true; }
+
+ @Override
+ public boolean supportsAsyncSynchronization() { return false; }
+
@Override
public String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
HashMap placeholders) {
@@ -71,6 +82,48 @@ public String onRewardRequested(Reward reward, AdvancedCoreUser user, Configurat
return null;
}
+ @Override
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ double totalChance = 0;
+ LinkedHashMap map = new LinkedHashMap<>();
+ for (String key : section.getKeys(false)) {
+ String path = key;
+ key = key.replaceAll("_", ".");
+ if (MessageAPI.isDouble(key)) {
+ double chance = Double.valueOf(key);
+ totalChance += chance;
+ map.put(chance, path);
+ }
+ }
+ Set> copy = new HashSet<>(map.entrySet());
+ double currentNum = 0;
+ map.clear();
+ for (Entry entry : copy) {
+ currentNum += entry.getKey();
+ map.put(currentNum, entry.getValue());
+ }
+ final double chanceTotal = totalChance;
+ String selection = Reward.replaySelection(placeholders, () -> {
+ double randomNum = ThreadLocalRandom.current().nextDouble(chanceTotal);
+ for (Entry entry : map.entrySet()) {
+ if (randomNum <= entry.getKey()) return entry.getValue() + "\n" + entry.getKey();
+ }
+ return null;
+ });
+ if (selection == null) return CompletableFuture.completedFuture(null);
+ String[] selected = selection.split("\n", 2);
+ if (selected.length != 2) return CompletableFuture.completedFuture(null);
+ for (Entry entry : map.entrySet()) {
+ if (entry.getValue().equals(selected[0])) {
+ return new RewardBuilder(section, entry.getValue())
+ .withPrefix(reward.getName() + "_SpecialChance").withPlaceHolder(placeholders)
+ .withPlaceHolder("chance", selected[1]).sendAsync(user).thenApply(ignored -> null);
+ }
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
@Override
public ArrayList subRewards(DefinedReward direct) {
ArrayList subs = new ArrayList<>();
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectConfigurationSection.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectConfigurationSection.java
index 8a2d6bb889..b6e11843c7 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectConfigurationSection.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectConfigurationSection.java
@@ -1,6 +1,8 @@
package com.bencodez.advancedcore.api.rewards.injected;
-import java.util.HashMap;
+import java.util.HashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.configuration.ConfigurationSection;
@@ -25,7 +27,28 @@ public String onRewardRequest(Reward reward, AdvancedCoreUser user, Configuratio
return null;
}
- public abstract String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
- HashMap placeholders);
+ public abstract String onRewardRequested(Reward reward, AdvancedCoreUser user, ConfigurationSection section,
+ HashMap placeholders);
+
+ /** Completion-aware counterpart for section injections. */
+ public CompletionStage onRewardRequestedAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection section, HashMap placeholders) {
+ try {
+ return CompletableFuture.completedFuture(onRewardRequested(reward, user, section, placeholders));
+ } catch (Throwable failure) {
+ return CompletableFuture.failedFuture(failure);
+ }
+ }
+
+ @Override
+ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection data, HashMap placeholders) {
+ if (!data.isConfigurationSection(getPath()) && !(isAlwaysForce() && data.contains(getPath(), true))
+ && !isAlwaysForceNoData()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return onRewardRequestedAsync(reward, user, data.getConfigurationSection(getPath()), placeholders)
+ .thenApply(result -> result);
+ }
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectStringList.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectStringList.java
index c92f2638b4..7bc405ad0c 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectStringList.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/rewards/injected/RewardInjectStringList.java
@@ -1,7 +1,10 @@
package com.bencodez.advancedcore.api.rewards.injected;
-import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import org.bukkit.configuration.ConfigurationSection;
@@ -28,12 +31,26 @@ public RewardInjectStringList(String path, ArrayList defaultValue) {
this.defaultValue = defaultValue;
}
- public abstract String onRewardRequest(Reward reward, AdvancedCoreUser user, ArrayList num,
- HashMap placeholders);
+ public abstract String onRewardRequest(Reward reward, AdvancedCoreUser user, ArrayList num,
+ HashMap placeholders);
+
+ /**
+ * Completion-aware counterpart for list injections. Existing list injections
+ * retain their synchronous behavior unless they opt in by overriding this
+ * method.
+ */
+ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreUser user, ArrayList value,
+ HashMap placeholders) {
+ try {
+ return CompletableFuture.completedFuture(onRewardRequest(reward, user, value, placeholders));
+ } catch (Throwable failure) {
+ return CompletableFuture.failedFuture(failure);
+ }
+ }
@SuppressWarnings("unchecked")
@Override
- public String onRewardRequest(Reward reward, AdvancedCoreUser user, ConfigurationSection data,
+ public String onRewardRequest(Reward reward, AdvancedCoreUser user, ConfigurationSection data,
HashMap placeholders) {
if (data.isList(getPath()) || (isAlwaysForce() && data.contains(getPath(), true)) || isAlwaysForceNoData()) {
ArrayList value = (ArrayList) data.getList(getPath(), getDefaultValue());
@@ -43,6 +60,19 @@ public String onRewardRequest(Reward reward, AdvancedCoreUser user, Configuratio
}
return null;
- }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public CompletionStage onRewardRequestAsync(Reward reward, AdvancedCoreUser user,
+ ConfigurationSection data, HashMap placeholders) {
+ if (!data.isList(getPath()) && !(isAlwaysForce() && data.contains(getPath(), true)) && !isAlwaysForceNoData()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ List> stored = data.getList(getPath(), getDefaultValue());
+ ArrayList value = new ArrayList<>();
+ for (Object entry : stored) value.add(String.valueOf(entry));
+ return onRewardRequestAsync(reward, user, value, placeholders).thenApply(result -> result);
+ }
}
diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
index 179f53b695..de3d3103e2 100644
--- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
+++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/api/user/AdvancedCoreUser.java
@@ -10,9 +10,12 @@
import java.util.Base64;
import java.util.Date;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.bukkit.Bukkit;
import org.bukkit.Effect;
@@ -56,6 +59,10 @@
public class AdvancedCoreUser {
private static final String QUEUED_REFERENCE_PREFIX = "\\AdvancedCoreQueue/1/";
+ private static final String ASYNC_PROGRESS_DELIMITER = "%asyncprogress%";
+ private static final String ASYNC_RETRY_DELIMITER = "%asyncretry%";
+ private final HashMap offlineReplayInFlight = new HashMap<>();
+ private final HashSet timedReplayInFlight = new HashSet<>();
/**
* User data fetch mode for this user.
@@ -249,7 +256,7 @@ public void addPermission(String permission, long delay) {
* given
*/
public synchronized void addTimedReward(Reward reward, HashMap placeholders, long epochMilli) {
- HashMap timed = getTimedRewards();
+ HashMap timed = new HashMap<>(getTimedRewards());
String rewardName = queuedRewardReference(reward);
rewardName += "%extime%" + System.currentTimeMillis();
@@ -265,6 +272,136 @@ private String queuedRewardReference(Reward reward) {
+ encodedName;
}
+ private static QueuedReplay parseQueuedReplay(String storedReference) {
+ int marker = storedReference.lastIndexOf(ASYNC_PROGRESS_DELIMITER);
+ if (marker < 0) return new QueuedReplay(storedReference, 0, new HashMap<>());
+ String value = storedReference.substring(marker + ASYNC_PROGRESS_DELIMITER.length());
+ if (value.startsWith("v2-")) {
+ try {
+ HashMap progress = new HashMap<>();
+ String decoded = new String(Base64.getUrlDecoder().decode(value.substring(3)), StandardCharsets.UTF_8);
+ for (String line : decoded.split("\\n")) {
+ String[] pair = line.split("\\t", 2);
+ if (pair.length == 2) progress.put(pair[0], Integer.parseInt(pair[1]));
+ }
+ return new QueuedReplay(storedReference.substring(0, marker), 0, progress);
+ } catch (IllegalArgumentException ignored) {
+ return new QueuedReplay(storedReference, 0, new HashMap<>());
+ }
+ }
+ try {
+ int progress = Integer.parseInt(value);
+ return progress > 0 ? new QueuedReplay(storedReference.substring(0, marker), progress, new HashMap<>())
+ : new QueuedReplay(storedReference.substring(0, marker), 0, new HashMap<>());
+ } catch (NumberFormatException ignored) {
+ // Preserve malformed legacy values as a normal reward reference instead
+ // of accidentally skipping an arbitrary portion of a reward chain.
+ return new QueuedReplay(storedReference, 0, new HashMap<>());
+ }
+ }
+
+ private static String encodeAsyncReplayProgress(Map progress) {
+ if (progress.isEmpty()) return "";
+ StringBuilder encoded = new StringBuilder();
+ for (Entry entry : progress.entrySet()) {
+ if (entry.getValue() != null && entry.getValue() > 0) {
+ encoded.append(entry.getKey()).append('\t').append(entry.getValue()).append('\n');
+ }
+ }
+ return encoded.length() == 0 ? "" : "v2-" + Base64.getUrlEncoder().withoutPadding()
+ .encodeToString(encoded.toString().getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static String stripTimedExecutionMarker(String storedReference) {
+ int marker = storedReference.indexOf("%extime%");
+ if (marker < 0) return stripAsyncRetryMarker(storedReference);
+ int valueStart = marker + "%extime%".length();
+ int nextMarker = storedReference.indexOf('%', valueStart);
+ return stripAsyncRetryMarker(storedReference.substring(0, marker)
+ + (nextMarker < 0 ? "" : storedReference.substring(nextMarker)));
+ }
+
+ private static String stripAsyncRetryMarker(String storedReference) {
+ int marker = storedReference.indexOf(ASYNC_RETRY_DELIMITER);
+ if (marker < 0) return storedReference;
+ int valueStart = marker + ASYNC_RETRY_DELIMITER.length();
+ int nextMarker = storedReference.indexOf('%', valueStart);
+ return storedReference.substring(0, marker) + (nextMarker < 0 ? "" : storedReference.substring(nextMarker));
+ }
+
+ private static int asyncRetryCount(String rewardEntry) {
+ int marker = rewardEntry.indexOf(ASYNC_RETRY_DELIMITER);
+ if (marker < 0) return 0;
+ int valueStart = marker + ASYNC_RETRY_DELIMITER.length();
+ int nextMarker = rewardEntry.indexOf('%', valueStart);
+ try { return Integer.parseInt(rewardEntry.substring(valueStart, nextMarker < 0 ? rewardEntry.length() : nextMarker)); }
+ catch (NumberFormatException ignored) { return 0; }
+ }
+
+ private static String withAsyncRetryCount(String rewardEntry, int count) {
+ int placeholders = rewardEntry.indexOf("%placeholders%");
+ String reference = placeholders < 0 ? rewardEntry : rewardEntry.substring(0, placeholders);
+ String suffix = placeholders < 0 ? "" : rewardEntry.substring(placeholders);
+ return stripAsyncRetryMarker(reference) + ASYNC_RETRY_DELIMITER + count + suffix;
+ }
+
+ private static String withAsyncReplayProgress(String rewardEntry, Throwable failure) {
+ Reward.RewardReplayFailure replayFailure = replayFailure(failure);
+ int completed = completedAsyncInjections(failure);
+ String serializedProgress = replayFailure == null ? "" : encodeAsyncReplayProgress(replayFailure.getReplayProgress());
+ if (completed <= 0 && serializedProgress.isEmpty()) return rewardEntry;
+ int placeholders = rewardEntry.indexOf("%placeholders%");
+ String storedReference = placeholders < 0 ? rewardEntry : rewardEntry.substring(0, placeholders);
+ String suffix = placeholders < 0 ? "" : rewardEntry.substring(placeholders);
+ if (replayFailure != null) suffix = "%placeholders%" + ArrayUtils.makeString(replayFailure.getReplayPlaceholders());
+ QueuedReplay queuedReplay = parseQueuedReplay(stripAsyncRetryMarker(storedReference));
+ if (!serializedProgress.isEmpty()) return queuedReplay.rewardReference + ASYNC_PROGRESS_DELIMITER
+ + serializedProgress + suffix;
+ return queuedReplay.rewardReference + ASYNC_PROGRESS_DELIMITER
+ + Math.max(queuedReplay.completedAsyncInjections, completed) + suffix;
+ }
+
+ private static String withAsyncReplayProgress(String rewardEntry, Map progress,
+ HashMap placeholders) {
+ int marker = rewardEntry.indexOf("%placeholders%");
+ String reference = marker < 0 ? rewardEntry : rewardEntry.substring(0, marker);
+ QueuedReplay queuedReplay = parseQueuedReplay(stripAsyncRetryMarker(reference));
+ String serialized = encodeAsyncReplayProgress(progress);
+ return queuedReplay.rewardReference + (serialized.isEmpty() ? "" : ASYNC_PROGRESS_DELIMITER + serialized)
+ + "%placeholders%" + ArrayUtils.makeString(placeholders);
+ }
+
+ private static int completedAsyncInjections(Throwable failure) {
+ Throwable current = failure;
+ while (current != null) {
+ if (current instanceof Reward.RewardReplayFailure) {
+ return ((Reward.RewardReplayFailure) current).getCompletedInjectionCount();
+ }
+ current = current.getCause();
+ }
+ return 0;
+ }
+
+ private static Reward.RewardReplayFailure replayFailure(Throwable failure) {
+ for (Throwable current = failure; current != null; current = current.getCause()) {
+ if (current instanceof Reward.RewardReplayFailure) return (Reward.RewardReplayFailure) current;
+ }
+ return null;
+ }
+
+ private static final class QueuedReplay {
+ private final String rewardReference;
+ private final int completedAsyncInjections;
+ private final Map asyncReplayProgress;
+
+ private QueuedReplay(String rewardReference, int completedAsyncInjections,
+ Map asyncReplayProgress) {
+ this.rewardReference = rewardReference;
+ this.completedAsyncInjections = completedAsyncInjections;
+ this.asyncReplayProgress = asyncReplayProgress;
+ }
+ }
+
/**
* Adds an unclaimed choice reward.
*
@@ -309,46 +446,44 @@ public void cacheIfNeeded() {
public void checkDelayedTimedRewards() {
plugin.debug("Checking timed/delayed for " + getPlayerName());
HashMap timed = getTimedRewards();
- HashMap newTimed = new HashMap<>();
- HashMap replayingTimed = new HashMap<>();
- HashMap> replayCompletions = new HashMap<>();
for (Entry entry : timed.entrySet()) {
long time = entry.getValue();
if (time != 0) {
Date timeDate = new Date(time);
- if (new Date().after(timeDate)) {
+ if (new Date().after(timeDate) && claimTimedReward(entry.getKey(), time)) {
String[] data = entry.getKey().split("%placeholders%", 2);
- String rewardReference = data[0].split("%extime%", 2)[0];
+ QueuedReplay queuedReplay = parseQueuedReplay(stripTimedExecutionMarker(data[0]));
+ String rewardReference = queuedReplay.rewardReference;
String placeholders = "";
if (data.length > 1) {
placeholders = data[1];
}
RewardOptions replayOptions = new RewardOptions().setCheckTimed(false)
.withPlaceHolder(ArrayUtils.fromString(placeholders));
+ replayOptions.setCompletedAsyncInjections(queuedReplay.completedAsyncInjections);
+ replayOptions.setAsyncReplayProgress(queuedReplay.asyncReplayProgress);
replayOptions.addPlaceholder("date",
"" + new SimpleDateFormat("EEE, d MMM yyyy HH:mm").format(new Date(time)));
- replayingTimed.put(entry.getKey(), time);
- replayCompletions.put(entry.getKey(), plugin.getRewardHandler().givePersistedQueueRewardAsync(this,
- new PersistedQueueReference(rewardReference), replayOptions));
+ // Keep the due entry durable while asynchronous stages are running. A
+ // checkpoint can change its serialized key (for example by adding the
+ // v2 progress marker), so completion and failure must follow this
+ // reference rather than the original entry key.
+ AtomicReference currentEntry = new AtomicReference<>(entry.getKey());
+ replayOptions.setAsyncReplayCheckpointConsumer(
+ checkpoint -> checkpointTimedReward(currentEntry, time, checkpoint));
+ plugin.getRewardHandler().givePersistedQueueRewardAsync(this,
+ new PersistedQueueReference(rewardReference), replayOptions).whenComplete((ignored, failure) -> {
+ if (failure == null) completeTimedReward(currentEntry.get(), time);
+ else restoreTimedReward(currentEntry.get(), time, failure);
+ });
String rewardName = rewardReference;
plugin.debug("Giving timed/delayed reward " + rewardName + " for " + getPlayerName()
+ " with placeholders " + ArrayUtils.fromString(placeholders));
- } else {
- newTimed.put(entry.getKey(), time);
}
}
}
- setTimedRewards(newTimed);
- for (Entry> replay : replayCompletions.entrySet()) {
- String rewardEntry = replay.getKey();
- long time = replayingTimed.get(rewardEntry);
- replay.getValue().exceptionally(failure -> {
- restoreTimedReward(rewardEntry, time, failure);
- return null;
- });
- }
}
/**
@@ -362,56 +497,150 @@ public void checkOfflineRewards() {
if (isCheckWorld()) {
setCheckWorld(false);
}
- ArrayList rewards = getOfflineRewards();
- if (rewards.isEmpty()) {
- return;
- }
-
- setOfflineRewards(new ArrayList<>());
- RewardHandler rewardHandler = plugin.getRewardHandler();
- AdvancedCoreUser user = this;
+ dispatchOfflineRewards(false);
+ }
+ private void dispatchOfflineRewards(boolean force) {
+ ArrayList rewards = new ArrayList<>(getOfflineRewards());
for (String rewardEntry : rewards) {
if (rewardEntry == null || rewardEntry.equals("null")) {
continue;
}
+ if (!claimOfflineReward(rewardEntry)) continue;
String[] parts = rewardEntry.split("%placeholders%", 2);
- String rewardReference = parts[0];
+ QueuedReplay queuedReplay = parseQueuedReplay(parts[0]);
+ String rewardReference = queuedReplay.rewardReference;
String placeholderStr = parts.length > 1 ? parts[1] : "";
RewardOptions options = new RewardOptions().setOnline(false).setCheckTimed(false)
.withPlaceHolder(ArrayUtils.fromString(placeholderStr));
-
- rewardHandler.givePersistedQueueRewardAsync(user, new PersistedQueueReference(rewardReference), options)
- .exceptionally(failure -> {
- restoreOfflineReward(rewardEntry, failure);
- return null;
+ if (force) options.setGiveOffline(false).forceOffline();
+ options.setCompletedAsyncInjections(queuedReplay.completedAsyncInjections);
+ options.setAsyncReplayProgress(queuedReplay.asyncReplayProgress);
+ AtomicReference currentEntry = new AtomicReference<>(rewardEntry);
+ options.setAsyncReplayCheckpointConsumer(checkpoint -> checkpointOfflineReward(currentEntry, checkpoint));
+
+ plugin.getRewardHandler().givePersistedQueueRewardAsync(this,
+ new PersistedQueueReference(rewardReference), options).whenComplete((ignored, failure) -> {
+ if (failure == null) completeOfflineReward(currentEntry.get());
+ else restoreOfflineReward(currentEntry.get(), failure);
});
}
+ }
+
+ private void checkpointOfflineReward(AtomicReference currentEntry, Reward.ReplayCheckpoint checkpoint) {
+ synchronized (plugin) {
+ String current = currentEntry.get();
+ String updated = withAsyncReplayProgress(current, checkpoint.getReplayProgress(), checkpoint.getPlaceholders());
+ ArrayList pending = getOfflineRewards();
+ int index = pending.indexOf(current);
+ if (index >= 0) {
+ pending.set(index, updated);
+ int claimed = offlineReplayInFlight.getOrDefault(current, 0);
+ // Migrate this occurrence only. Identical legacy queue entries share a
+ // serialized key, but may each already be executing; moving the whole
+ // count would make the remaining old entry appear unclaimed.
+ if (claimed <= 1) offlineReplayInFlight.remove(current);
+ else offlineReplayInFlight.put(current, claimed - 1);
+ offlineReplayInFlight.put(updated, offlineReplayInFlight.getOrDefault(updated, 0) + 1);
+ currentEntry.set(updated);
+ setOfflineRewardsDurably(pending);
+ }
+ }
+ }
+ private boolean claimOfflineReward(String rewardEntry) {
+ synchronized (plugin) {
+ int occurrences = 0;
+ for (String pending : getOfflineRewards()) if (rewardEntry.equals(pending)) occurrences++;
+ int claimed = offlineReplayInFlight.getOrDefault(rewardEntry, 0);
+ if (claimed >= occurrences) return false;
+ offlineReplayInFlight.put(rewardEntry, claimed + 1);
+ return true;
+ }
}
- /** Restores a queue entry only after an asynchronous replay fails. */
+ private void completeOfflineReward(String rewardEntry) {
+ synchronized (plugin) {
+ ArrayList pending = getOfflineRewards();
+ pending.remove(rewardEntry);
+ setOfflineRewards(pending);
+ releaseOfflineReward(rewardEntry);
+ }
+ }
+
+ /** Retains and updates a queue entry only after an asynchronous replay fails. */
private void restoreOfflineReward(String rewardEntry, Throwable failure) {
// Match addOfflineRewards' lock so a newly queued reward cannot be lost
// while a failed replay is being restored.
synchronized (plugin) {
ArrayList pending = getOfflineRewards();
- // Queue entries are allowed to repeat; each failed replay must restore
- // its own occurrence rather than collapsing identical rewards.
- pending.add(rewardEntry);
+ int index = pending.indexOf(rewardEntry);
+ if (index >= 0) pending.set(index, withAsyncReplayProgress(rewardEntry, failure));
+ releaseOfflineReward(rewardEntry);
setOfflineRewards(pending);
}
plugin.getLogger().warning("Could not deliver queued offline reward for " + getPlayerName()
+ "; it will be retried: " + failure.getMessage());
}
+ private void releaseOfflineReward(String rewardEntry) {
+ int claimed = offlineReplayInFlight.getOrDefault(rewardEntry, 0);
+ if (claimed <= 1) offlineReplayInFlight.remove(rewardEntry);
+ else offlineReplayInFlight.put(rewardEntry, claimed - 1);
+ }
+
+ private synchronized boolean claimTimedReward(String rewardEntry, long time) {
+ if (timedReplayInFlight.contains(rewardEntry) || !Long.valueOf(time).equals(getTimedRewards().get(rewardEntry))) {
+ return false;
+ }
+ timedReplayInFlight.add(rewardEntry);
+ return true;
+ }
+
+ private synchronized void completeTimedReward(String rewardEntry, long time) {
+ HashMap pending = getTimedRewards();
+ if (Long.valueOf(time).equals(pending.get(rewardEntry))) {
+ pending.remove(rewardEntry);
+ setTimedRewards(pending);
+ }
+ timedReplayInFlight.remove(rewardEntry);
+ }
+
+ /**
+ * Persists every completed asynchronous stage before the next one can run.
+ * Timed reward keys are map keys, so migrate the exact current key while
+ * retaining its execution time and its in-flight claim.
+ */
+ private synchronized void checkpointTimedReward(AtomicReference currentEntry, long time,
+ Reward.ReplayCheckpoint checkpoint) {
+ String current = currentEntry.get();
+ String updated = withAsyncReplayProgress(current, checkpoint.getReplayProgress(), checkpoint.getPlaceholders());
+ if (current.equals(updated)) return;
+ HashMap pending = getTimedRewards();
+ if (!Long.valueOf(time).equals(pending.get(current))) return;
+ pending.remove(current);
+ pending.put(updated, time);
+ setTimedRewardsDurably(pending);
+ timedReplayInFlight.remove(current);
+ timedReplayInFlight.add(updated);
+ currentEntry.set(updated);
+ }
+
/** Restores a due timed entry after an asynchronous replay fails. */
private synchronized void restoreTimedReward(String rewardEntry, long time, Throwable failure) {
HashMap pending = getTimedRewards();
- pending.putIfAbsent(rewardEntry, time);
+ int retry = Math.min(8, asyncRetryCount(rewardEntry) + 1);
+ long retryDelay = Math.min(TimeUnit.MINUTES.toMillis(5), TimeUnit.SECONDS.toMillis(1L << retry));
+ long retryTime = System.currentTimeMillis() + retryDelay;
+ pending.remove(rewardEntry);
+ pending.put(withAsyncRetryCount(withAsyncReplayProgress(rewardEntry, failure), retry), retryTime);
setTimedRewards(pending);
+ timedReplayInFlight.remove(rewardEntry);
+ // A due entry restored after its original timer fired needs its own retry
+ // timer; waiting for reconnect would strand it indefinitely.
+ loadTimedDelayedTimer(retryTime);
plugin.getLogger().warning("Could not deliver queued timed reward for " + getPlayerName()
+ "; it will be retried: " + failure.getMessage());
}
@@ -471,33 +700,7 @@ public void forceRunOfflineRewards() {
}
setCheckWorld(false);
- ArrayList rewards = getOfflineRewards();
- if (rewards.isEmpty()) {
- return;
- }
-
- setOfflineRewards(new ArrayList<>());
- RewardHandler rewardHandler = plugin.getRewardHandler();
- AdvancedCoreUser user = this;
-
- for (String rewardEntry : rewards) {
- if (rewardEntry == null || rewardEntry.equals("null")) {
- continue;
- }
-
- String[] parts = rewardEntry.split("%placeholders%", 2);
- String rewardReference = parts[0];
- String placeholderStr = parts.length > 1 ? parts[1] : "";
-
- RewardOptions options = new RewardOptions().setOnline(false).setGiveOffline(false).forceOffline()
- .setCheckTimed(false).withPlaceHolder(ArrayUtils.fromString(placeholderStr));
-
- rewardHandler.givePersistedQueueRewardAsync(user, new PersistedQueueReference(rewardReference), options)
- .exceptionally(failure -> {
- restoreOfflineReward(rewardEntry, failure);
- return null;
- });
- }
+ dispatchOfflineRewards(true);
}
/**
@@ -1604,6 +1807,15 @@ public void setLastOnline(long online) {
* @param offlineRewards the offline rewards
*/
public void setOfflineRewards(ArrayList offlineRewards) {
+ setOfflineRewards(offlineRewards, true);
+ }
+
+ /** Writes an async replay checkpoint before the next stage can execute. */
+ private void setOfflineRewardsDurably(ArrayList offlineRewards) {
+ persistReplayCheckpoint(() -> setOfflineRewards(offlineRewards, false));
+ }
+
+ private void setOfflineRewards(ArrayList offlineRewards, boolean queue) {
// MySQL TEXT max length is 65535 bytes
int maxLength = 65535;
String str = String.join("%line%", offlineRewards);
@@ -1614,7 +1826,8 @@ public void setOfflineRewards(ArrayList offlineRewards) {
str = String.join("%line%", offlineRewards);
}
- data.setStringList(plugin.getUserManager().getOfflineRewardsPath(), offlineRewards);
+ if (queue) data.setStringList(plugin.getUserManager().getOfflineRewardsPath(), offlineRewards);
+ else data.setStringList(plugin.getUserManager().getOfflineRewardsPath(), offlineRewards, false);
}
/**
@@ -1642,6 +1855,25 @@ public void setRepeatAmount(Reward reward, int amount) {
* @param timed the timed rewards
*/
public void setTimedRewards(HashMap timed) {
+ setTimedRewards(timed, true);
+ }
+
+ /** Writes an async replay checkpoint before the next stage can execute. */
+ private void setTimedRewardsDurably(HashMap timed) {
+ persistReplayCheckpoint(() -> setTimedRewards(timed, false));
+ }
+
+ /**
+ * A cached user may already have an older queued value for this key. Drain it
+ * before the direct write so the delayed cache flush cannot replace a replay
+ * checkpoint that an asynchronous reward stage has already relied on.
+ */
+ private void persistReplayCheckpoint(Runnable write) {
+ if (isCached()) getCache().flushChangesAndRun(write);
+ else write.run();
+ }
+
+ private void setTimedRewards(HashMap timed, boolean queue) {
ArrayList timedRewards = new ArrayList<>();
for (Entry entry : timed.entrySet()) {
@@ -1651,7 +1883,8 @@ public void setTimedRewards(HashMap timed) {
timedRewards.add(str);
}
- data.setStringList("TimedRewards", timedRewards);
+ if (queue) data.setStringList("TimedRewards", timedRewards);
+ else data.setStringList("TimedRewards", timedRewards, false);
}
/**
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 b2b60078bd..c58b17fc9d 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
@@ -100,6 +100,31 @@ public synchronized void clearChanges() {
}
}
+ /**
+ * Drains every queued or in-flight cache batch before running a synchronous
+ * write. The action runs while holding the cache monitor, preventing an old
+ * queued value from being persisted after a durable replacement.
+ */
+ public void flushChangesAndRun(Runnable action) {
+ if (action == null) return;
+ while (true) {
+ processChanges();
+ synchronized (this) {
+ while (inFlightBatches > 0) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while flushing cached user changes", e);
+ }
+ }
+ if (cachedChanges != null && !cachedChanges.isEmpty()) continue;
+ action.run();
+ return;
+ }
+ }
+ }
+
public void displayCache() {
manager.getPlugin().devDebug(displayCacheStringList().toString());
}
diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
index 3d814623f4..e0389a42ab 100644
--- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
+++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardAsyncInjectionTest.java
@@ -35,6 +35,7 @@
import com.bencodez.advancedcore.api.rewards.injected.RewardInject;
import com.bencodez.advancedcore.api.rewards.injected.RewardInjectInt;
import com.bencodez.advancedcore.api.rewards.builtin.RewardSubRewards;
+import com.bencodez.advancedcore.api.rewards.builtin.RewardRandomReward;
import com.bencodez.advancedcore.api.user.AdvancedCoreUser;
class RewardAsyncInjectionTest {
@@ -355,6 +356,109 @@ public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
assertEquals(List.of("after-child"), events);
}
+ @Test
+ void everyNestedRewardInjectorWaitsForItsSelectedChild() {
+ handler = org.mockito.Mockito.spy(new RewardHandler(plugin));
+ when(plugin.getRewardHandler()).thenReturn(handler);
+ when(user.getPlugin()).thenReturn(plugin);
+ data.set("RandomReward", new ArrayList<>(List.of("child")));
+ CompletableFuture child = new CompletableFuture<>();
+ doReturn(child).when(handler).giveRewardAsync(eq(user), eq("child"), any());
+ List events = new ArrayList<>();
+ RewardRandomReward.register(handler, plugin);
+ handler.getInjectedRewards().add(new RewardInject("After") {
+ @Override public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ events.add("after-child");
+ return null;
+ }
+ }.postReward());
+
+ CompletionStage result = reward.giveInjectedRewardsAsync(user, new HashMap<>());
+ assertFalse(result.toCompletableFuture().isDone());
+ assertTrue(events.isEmpty());
+ child.complete(null);
+ result.toCompletableFuture().join();
+ assertEquals(List.of("after-child"), events);
+ }
+
+ @Test
+ void replayCheckpointSkipsAlreadyAppliedInjectionAfterLaterFailure() throws Exception {
+ AtomicInteger alreadyApplied = new AtomicInteger();
+ AtomicBoolean fail = new AtomicBoolean(true);
+ RewardInject applied = new RewardInject("Money") {
+ @Override public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ alreadyApplied.incrementAndGet();
+ return "receipt-1";
+ }
+ };
+ applied.asPlaceholder("receipt");
+ handler.getInjectedRewards().add(applied);
+ handler.getInjectedRewards().add(new RewardInject("Durable") {
+ @Override public boolean supportsAsyncRequest() { return true; }
+ @Override public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) { return null; }
+ @Override public CompletionStage onRewardRequestAsync(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ return fail.get() ? CompletableFuture.failedFuture(new IllegalStateException("temporary"))
+ : CompletableFuture.completedFuture(null);
+ }
+ });
+ java.lang.reflect.Method replay = Reward.class.getDeclaredMethod("giveInjectedRewardsAsync",
+ AdvancedCoreUser.class, HashMap.class, int.class);
+ replay.setAccessible(true);
+ CompletionStage first = (CompletionStage) replay.invoke(reward, user, new HashMap<>(), 0);
+ Throwable failure = assertThrows(java.util.concurrent.CompletionException.class,
+ () -> first.toCompletableFuture().join());
+ Reward.RewardReplayFailure checkpoint = findCheckpoint(failure);
+ assertEquals(1, checkpoint.getCompletedInjectionCount());
+ assertEquals("receipt-1", checkpoint.getReplayPlaceholders().get("receipt"));
+ fail.set(false);
+ CompletionStage resumed = (CompletionStage) replay.invoke(reward, user, new HashMap<>(),
+ checkpoint.getCompletedInjectionCount());
+ resumed.toCompletableFuture().join();
+ assertEquals(1, alreadyApplied.get());
+ }
+
+ @Test
+ void replayProgressDoesNotSuppressASecondSameNamedChildOccurrence() throws Exception {
+ AtomicInteger deliveries = new AtomicInteger();
+ handler.getInjectedRewards().add(new RewardInject("Deliver") {
+ @Override public boolean supportsAsyncRequest() { return true; }
+ @Override public Object onRewardRequest(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) { return null; }
+ @Override public CompletionStage onRewardRequestAsync(Reward ignored, AdvancedCoreUser ignoredUser,
+ ConfigurationSection ignoredData, HashMap ignoredPlaceholders) {
+ deliveries.incrementAndGet();
+ return CompletableFuture.completedFuture(null);
+ }
+ });
+ Class> stateType = Class.forName("com.bencodez.advancedcore.api.rewards.Reward$ReplayState");
+ java.lang.reflect.Constructor> constructor = stateType.getDeclaredConstructor(java.util.Map.class);
+ constructor.setAccessible(true);
+ Object state = constructor.newInstance((Object) null);
+ java.lang.reflect.Method replay = Reward.class.getDeclaredMethod("giveInjectedRewardsAsync",
+ AdvancedCoreUser.class, HashMap.class, int.class, stateType, String.class);
+ replay.setAccessible(true);
+ ((CompletionStage) replay.invoke(reward, user, new HashMap<>(), 0, state,
+ "root/Rewards/child:0")).toCompletableFuture().join();
+ ((CompletionStage) replay.invoke(reward, user, new HashMap<>(), 0, state,
+ "root/Rewards/child:1")).toCompletableFuture().join();
+ assertEquals(2, deliveries.get());
+ }
+
+ @Test
+ void replaySelectionKeepsTheFirstNondeterministicChoice() {
+ HashMap placeholders = new HashMap<>();
+ AtomicInteger selections = new AtomicInteger();
+ assertEquals("first", Reward.replaySelection(placeholders,
+ () -> selections.incrementAndGet() == 1 ? "first" : "second"));
+ assertEquals("first", Reward.replaySelection(placeholders,
+ () -> selections.incrementAndGet() == 1 ? "first" : "second"));
+ assertEquals(1, selections.get());
+ }
+
@Test
void nestedAsyncInjectorCanAwaitTheSameSharedInjectorWithoutDeadlocking() {
AtomicInteger invocations = new AtomicInteger();
@@ -390,4 +494,11 @@ private RewardInject asyncInjection(String path, CompletionStage complet
}
};
}
+
+ private Reward.RewardReplayFailure findCheckpoint(Throwable failure) {
+ for (Throwable current = failure; current != null; current = current.getCause()) {
+ if (current instanceof Reward.RewardReplayFailure) return (Reward.RewardReplayFailure) current;
+ }
+ throw new AssertionError("missing replay checkpoint", failure);
+ }
}
diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardExecutorTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardExecutorTest.java
index 52926d2b3c..8dafc8ece8 100644
--- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardExecutorTest.java
+++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/RewardExecutorTest.java
@@ -17,6 +17,7 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;
@@ -124,6 +125,25 @@ public void listRewardDispatchesEveryNamedReward() {
assertTrue(options.isOnlineSet());
}
+ @Test
+ public void asyncListDispatchUsesASeparateReplayContextForEveryEntry() {
+ YamlConfiguration data = new YamlConfiguration();
+ data.set("Rewards", new ArrayList<>(List.of("Child", "Child")));
+ Reward child = mock(Reward.class);
+ when(handler.getReward("Child")).thenReturn(child);
+ when(child.giveRewardAsync(eq(user), any(RewardOptions.class)))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ executor.giveRewardAsync(user, data, "Rewards", new RewardOptions()).toCompletableFuture().join();
+
+ ArgumentCaptor captured = ArgumentCaptor.forClass(RewardOptions.class);
+ verify(child, org.mockito.Mockito.times(2)).giveRewardAsync(eq(user), captured.capture());
+ List children = captured.getAllValues();
+ assertFalse(children.get(0) == children.get(1));
+ assertFalse(children.get(0).getAsyncReplayKey().equals(children.get(1).getAsyncReplayKey()));
+ assertSame(children.get(0).getAsyncReplayState(), children.get(1).getAsyncReplayState());
+ }
+
@Test
public void stringRewardDispatchesNamedReward() {
YamlConfiguration data = new YamlConfiguration();
diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
index 030c207c4b..5218386c3f 100644
--- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
+++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/user/AdvancedCoreUserTest.java
@@ -10,8 +10,12 @@
import static org.mockito.Mockito.when;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import org.junit.jupiter.api.BeforeEach;
@@ -20,6 +24,7 @@
import com.bencodez.advancedcore.AdvancedCoreConfigOptions;
import com.bencodez.advancedcore.AdvancedCorePlugin;
+import com.bencodez.advancedcore.api.rewards.Reward;
import com.bencodez.advancedcore.api.rewards.RewardHandler;
import com.bencodez.advancedcore.api.rewards.RewardOptions;
import com.bencodez.advancedcore.api.user.AdvancedCoreUser;
@@ -37,6 +42,7 @@ public class AdvancedCoreUserTest {
private UserDataManager dataManager;
private UserData data;
private RewardHandler rewardHandler;
+ private ScheduledExecutorService delayedTimer;
private AdvancedCoreUser user;
@BeforeEach
@@ -46,6 +52,7 @@ public void setUp() {
dataManager = mock(UserDataManager.class);
data = mock(UserData.class);
rewardHandler = mock(RewardHandler.class);
+ delayedTimer = mock(ScheduledExecutorService.class);
MySQL mysql = mock(MySQL.class);
AdvancedCoreConfigOptions configOptions = mock(AdvancedCoreConfigOptions.class);
@@ -55,6 +62,7 @@ public void setUp() {
when(userManager.getDataManager()).thenReturn(dataManager);
when(plugin.getUserManager()).thenReturn(userManager);
when(plugin.getRewardHandler()).thenReturn(rewardHandler);
+ when(rewardHandler.getDelayedTimer()).thenReturn(delayedTimer);
when(plugin.getStorageType()).thenReturn(UserStorage.MYSQL);
when(plugin.getLogger()).thenReturn(mock(Logger.class));
when(userManager.getOfflineRewardsPath()).thenReturn("offlineRewardsPath");
@@ -104,6 +112,23 @@ void forceRunOfflineRewards_stillBypassesNormalReplayChecks() {
assertEquals("server-a", options.getPlaceholders().get("Server"));
}
+ @Test
+ void queuedAsyncReplayResumesAtItsStoredCheckpoint() {
+ ArrayList rewards = new ArrayList<>();
+ rewards.add("VoteReward%asyncprogress%2%placeholders%Server%pair%server-a");
+ when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT)).thenReturn(rewards);
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(CompletableFuture.completedFuture(null));
+
+ ArgumentCaptor reference = ArgumentCaptor.forClass(PersistedQueueReference.class);
+ ArgumentCaptor options = ArgumentCaptor.forClass(RewardOptions.class);
+ user.checkOfflineRewards();
+
+ verify(rewardHandler).givePersistedQueueRewardAsync(eq(user), reference.capture(), options.capture());
+ assertEquals("VoteReward", reference.getValue().getReference());
+ assertEquals(2, options.getValue().getCompletedAsyncInjections());
+ }
+
@Test
void setOfflineRewards_emptyList() {
ArrayList rewards = new ArrayList<>();
@@ -155,16 +180,15 @@ void failedAsyncOfflineReplayIsRestoredForRetry() {
ArrayList initial = new ArrayList<>();
initial.add("VoteReward%placeholders%Server%pair%server-a");
initial.add("VoteReward%placeholders%Server%pair%server-a");
- when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT))
- .thenReturn(new ArrayList<>(initial)).thenReturn(new ArrayList<>());
+ when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT)).thenReturn(new ArrayList<>(initial));
when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
any(RewardOptions.class))).thenReturn(CompletableFuture.failedFuture(new IllegalStateException("temporary")));
user.checkOfflineRewards();
ArgumentCaptor> saved = ArgumentCaptor.forClass(ArrayList.class);
- verify(data, org.mockito.Mockito.times(3)).setStringList(eq("offlineRewardsPath"), saved.capture());
- assertEquals(initial, saved.getAllValues().get(2));
+ verify(data, org.mockito.Mockito.atLeastOnce()).setStringList(eq("offlineRewardsPath"), saved.capture());
+ assertTrue(saved.getAllValues().stream().allMatch(value -> value.size() == initial.size()));
}
@Test
@@ -173,15 +197,181 @@ void failedAsyncTimedReplayIsRestoredAfterDueEntriesAreSaved() {
String entry = "VoteReward%extime%1%placeholders%Server%pair%server-a";
ArrayList timed = new ArrayList<>();
timed.add(entry + "%ExecutionTime/%" + due);
- when(data.getStringList("TimedRewards", UserDataFetchMode.DEFAULT))
- .thenReturn(new ArrayList<>(timed)).thenReturn(new ArrayList<>());
+ when(data.getStringList("TimedRewards", UserDataFetchMode.DEFAULT)).thenReturn(new ArrayList<>(timed));
when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
any(RewardOptions.class))).thenReturn(CompletableFuture.failedFuture(new IllegalStateException("temporary")));
user.checkDelayedTimedRewards();
ArgumentCaptor> saved = ArgumentCaptor.forClass(ArrayList.class);
- verify(data, org.mockito.Mockito.times(2)).setStringList(eq("TimedRewards"), saved.capture());
- assertEquals(timed, saved.getAllValues().get(1));
+ verify(data).setStringList(eq("TimedRewards"), saved.capture());
+ String restored = saved.getValue().get(0);
+ assertTrue(restored.contains("%asyncretry%1%"));
+ assertTrue(restored.contains("VoteReward%extime%1"));
+ verify(delayedTimer).schedule(any(Runnable.class), org.mockito.ArgumentMatchers.anyLong(),
+ eq(TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ void pendingAsyncOfflineReplayRemainsDurableAndIsNotDispatchedTwice() {
+ ArrayList rewards = new ArrayList<>();
+ rewards.add("VoteReward%placeholders%Server%pair%server-a");
+ when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT)).thenReturn(rewards);
+ CompletableFuture pending = new CompletableFuture<>();
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(pending);
+
+ user.checkOfflineRewards();
+ user.checkOfflineRewards();
+
+ verify(rewardHandler, org.mockito.Mockito.times(1)).givePersistedQueueRewardAsync(eq(user),
+ any(PersistedQueueReference.class), any(RewardOptions.class));
+ verify(data, org.mockito.Mockito.never()).setStringList(eq("offlineRewardsPath"), any());
+ pending.complete(null);
+ assertTrue(rewards.isEmpty());
+ verify(data).setStringList(eq("offlineRewardsPath"), any());
+ }
+
+ @Test
+ void offlineCheckpointMigratesOnlyItsOwnDuplicateInFlightClaim() throws Exception {
+ String entry = "VoteReward%placeholders%Server%pair%server-a";
+ ArrayList persisted = new ArrayList<>();
+ persisted.add(entry);
+ persisted.add(entry);
+ when(data.getStringList("offlineRewardsPath", UserDataFetchMode.DEFAULT))
+ .thenAnswer(ignored -> new ArrayList<>(persisted));
+ org.mockito.Mockito.doAnswer(invocation -> {
+ persisted.clear();
+ persisted.addAll(invocation.getArgument(1));
+ return null;
+ }).when(data).setStringList(eq("offlineRewardsPath"), any(), eq(false));
+ CompletableFuture pending = new CompletableFuture<>();
+ when(rewardHandler.givePersistedQueueRewardAsync(eq(user), any(PersistedQueueReference.class),
+ any(RewardOptions.class))).thenReturn(pending);
+
+ ArgumentCaptor