From bc39dde1260eb4e34196de425a7117997ce8b0bd Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:19 -0600 Subject: [PATCH 01/14] Add shared reward execution result --- .../advancedcore/core/reward/SharedRewardResult.java | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardResult.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardResult.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardResult.java new file mode 100644 index 0000000000..b65a643dff --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardResult.java @@ -0,0 +1,11 @@ +package com.bencodez.advancedcore.core.reward; + +/** Outcome of a platform-neutral reward execution. */ +public enum SharedRewardResult { + /** The requested work completed and may be checkpointed by its parent. */ + COMPLETED, + /** The request was durably deferred and must not be checkpointed as delivered. */ + DEFERRED, + /** Requirements or chance intentionally prevented delivery. */ + NOT_ELIGIBLE +} From ea1189735f6935ce6e560c2552cfe7d3d56799e4 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:25 -0600 Subject: [PATCH 02/14] Add shared reward execution context --- .../core/reward/SharedRewardContext.java | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardContext.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardContext.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardContext.java new file mode 100644 index 0000000000..6f24331aba --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardContext.java @@ -0,0 +1,35 @@ +package com.bencodez.advancedcore.core.reward; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; + +/** Platform-neutral identity and placeholder state for one logical reward execution. */ +public final class SharedRewardContext { + private final UUID userId; + private final String playerName; + private final HashMap placeholders; + + public SharedRewardContext(UUID userId, String playerName, Map placeholders) { + this.userId = Objects.requireNonNull(userId, "userId"); + this.playerName = playerName; + this.placeholders = new HashMap<>(); + if (placeholders != null) { + this.placeholders.putAll(placeholders); + } + } + + public UUID userId() { + return userId; + } + + public String playerName() { + return playerName; + } + + /** Mutable execution-local placeholders; durable adapters decide when to persist them. */ + public HashMap placeholders() { + return placeholders; + } +} From 08c127fdd1f3e9e7f6d6bdac7add3f2e0aa91adc Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:32 -0600 Subject: [PATCH 03/14] Add shared reward platform boundary --- .../core/reward/SharedRewardPlatform.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java new file mode 100644 index 0000000000..f69dff7ba2 --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java @@ -0,0 +1,24 @@ +package com.bencodez.advancedcore.core.reward; + +import java.time.Duration; +import java.util.UUID; +import java.util.concurrent.CompletionStage; +import java.util.function.Supplier; + +/** Native operations needed by the platform-neutral reward orchestrator. */ +public interface SharedRewardPlatform { + boolean isOnline(UUID userId); + + /** Returns a value in the range [0, 1). */ + double nextChanceRoll(); + + /** + * Runs {@code operation} after the delay and completes only when the operation's + * returned stage completes. Implementations must not report task submission as + * completion. + */ + CompletionStage delay(Duration delay, + Supplier> operation); + + boolean isShuttingDown(); +} From 1970935373606ceb57fdbed356f1d68a0b103bda Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:41 -0600 Subject: [PATCH 04/14] Add shared reward durability boundary --- .../core/reward/SharedRewardDurability.java | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java new file mode 100644 index 0000000000..a43d5d4b4c --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java @@ -0,0 +1,30 @@ +package com.bencodez.advancedcore.core.reward; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +/** + * Adapter over the existing durable replay owner. The orchestrator stores no + * replay queue or cursor itself; #317's replay state can implement this boundary. + */ +public interface SharedRewardDurability { + SharedRewardDurability NONE = new SharedRewardDurability() { + @Override public int completedSteps(String executionPath) { return 0; } + @Override public CompletionStage checkpoint(String executionPath, int completedSteps, + SharedRewardContext context) { return CompletableFuture.completedFuture(null); } + @Override public CompletionStage defer(String executionPath, int nextStep, + SharedRewardContext context) { return CompletableFuture.failedFuture( + new IllegalStateException("Offline reward deferral is unavailable")); } + @Override public boolean durable() { return false; } + }; + + int completedSteps(String executionPath); + + /** Completes only after progress and placeholder state are durable. */ + CompletionStage checkpoint(String executionPath, int completedSteps, SharedRewardContext context); + + /** Completes only after the still-pending reward occurrence is durable. */ + CompletionStage defer(String executionPath, int nextStep, SharedRewardContext context); + + boolean durable(); +} From 2e274751c5777bc1c671a46c17b52fdc0ece14c7 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:47 -0600 Subject: [PATCH 05/14] Add shared reward requirement boundary --- .../advancedcore/core/reward/SharedRewardRequirement.java | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java new file mode 100644 index 0000000000..db1698cb3c --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java @@ -0,0 +1,8 @@ +package com.bencodez.advancedcore.core.reward; + +import java.util.concurrent.CompletionStage; + +@FunctionalInterface +public interface SharedRewardRequirement { + CompletionStage test(SharedRewardContext context); +} From b7dae4940beb8f8a878229bf2e24cd70642bba23 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:51 -0600 Subject: [PATCH 06/14] Add shared reward step abstraction --- .../core/reward/SharedRewardStep.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardStep.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardStep.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardStep.java new file mode 100644 index 0000000000..09a2f7e217 --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardStep.java @@ -0,0 +1,18 @@ +package com.bencodez.advancedcore.core.reward; + +import java.util.Objects; +import java.util.concurrent.CompletionStage; + +/** One already-configured reward operation. Native adapters provide the action. */ +public record SharedRewardStep(String id, boolean requiresOnlinePlayer, Action action) { + public SharedRewardStep { + Objects.requireNonNull(id, "id"); + Objects.requireNonNull(action, "action"); + if (id.isBlank()) throw new IllegalArgumentException("Reward step id cannot be blank"); + } + + @FunctionalInterface + public interface Action { + CompletionStage execute(SharedRewardContext context, String executionPath); + } +} From 3b36667924d3feb731fe1f5494d3b1a53d848450 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:42:58 -0600 Subject: [PATCH 07/14] Add shared reward execution plan --- .../core/reward/SharedRewardPlan.java | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java new file mode 100644 index 0000000000..aad234889a --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java @@ -0,0 +1,28 @@ +package com.bencodez.advancedcore.core.reward; + +import java.time.Duration; +import java.util.List; +import java.util.Objects; + +/** + * Platform-neutral plan prepared by the existing reward configuration layer. + * Parsing and native item/sound/effect behavior intentionally remain outside it. + */ +public record SharedRewardPlan(String id, double chance, Duration delay, + List requirements, List steps) { + public SharedRewardPlan { + Objects.requireNonNull(id, "id"); + Objects.requireNonNull(delay, "delay"); + requirements = List.copyOf(Objects.requireNonNull(requirements, "requirements")); + steps = List.copyOf(Objects.requireNonNull(steps, "steps")); + if (id.isBlank()) throw new IllegalArgumentException("Reward plan id cannot be blank"); + if (chance < 0.0 || chance > 1.0 || Double.isNaN(chance)) { + throw new IllegalArgumentException("chance must be between 0 and 1"); + } + if (delay.isNegative()) throw new IllegalArgumentException("delay cannot be negative"); + } + + public static SharedRewardPlan immediate(String id, List steps) { + return new SharedRewardPlan(id, 1.0, Duration.ZERO, List.of(), steps); + } +} From b9d4306ee98eff01cdc12b687aae8de554e8dfc1 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:43:26 -0600 Subject: [PATCH 08/14] Add platform-neutral reward orchestrator --- .../core/reward/SharedRewardOrchestrator.java | 162 ++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java new file mode 100644 index 0000000000..05de1d5b55 --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -0,0 +1,162 @@ +package com.bencodez.advancedcore.core.reward; + +import java.time.Duration; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +/** + * Sequences already-configured reward work without Bukkit dependencies. Native + * actions remain in platform adapters, while durability remains owned by the + * injected replay adapter. + */ +public final class SharedRewardOrchestrator { + private final SharedRewardPlatform platform; + + public SharedRewardOrchestrator(SharedRewardPlatform platform) { + this.platform = Objects.requireNonNull(platform, "platform"); + } + + public CompletionStage execute(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability) { + Objects.requireNonNull(plan, "plan"); + Objects.requireNonNull(context, "context"); + SharedRewardDurability replay = durability == null ? SharedRewardDurability.NONE : durability; + return execute(plan, context, replay, plan.id()); + } + + public CompletionStage executeNested(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability, String parentPath) { + Objects.requireNonNull(parentPath, "parentPath"); + String path = parentPath.isBlank() ? plan.id() : parentPath + "/" + plan.id(); + return execute(plan, context, durability == null ? SharedRewardDurability.NONE : durability, path); + } + + private CompletionStage execute(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability, String executionPath) { + if (platform.isShuttingDown()) { + return failed("Reward platform is shutting down"); + } + + int resume = durability.completedSteps(executionPath); + if (resume < 0 || resume > plan.steps().size()) { + return CompletableFuture.failedFuture(new IllegalStateException( + "Invalid durable reward cursor " + resume + " for " + executionPath)); + } + + CompletionStage eligible = resume > 0 + ? CompletableFuture.completedFuture(Boolean.TRUE) + : evaluateRequirements(plan.requirements(), context, 0); + return eligible.thenCompose(requirementsPassed -> { + if (!requirementsPassed.booleanValue()) { + return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); + } + if (resume == 0 && plan.chance() < 1.0 && platform.nextChanceRoll() >= plan.chance()) { + return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); + } + + java.util.function.Supplier> work = + () -> executeSteps(plan, context, durability, executionPath, resume); + Duration delay = resume == 0 ? plan.delay() : Duration.ZERO; + if (delay.isZero()) { + return work.get(); + } + try { + CompletionStage delayed = platform.delay(delay, work); + if (delayed == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward platform returned null delay stage")); + } + return delayed; + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + }); + } + + private CompletionStage evaluateRequirements(List requirements, + SharedRewardContext context, int index) { + if (index >= requirements.size()) { + return CompletableFuture.completedFuture(Boolean.TRUE); + } + CompletionStage stage; + try { + stage = requirements.get(index).test(context); + if (stage == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward requirement returned null completion stage")); + } + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + return stage.thenCompose(passed -> Boolean.TRUE.equals(passed) + ? evaluateRequirements(requirements, context, index + 1) + : CompletableFuture.completedFuture(Boolean.FALSE)); + } + + private CompletionStage executeSteps(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability, String executionPath, int index) { + if (platform.isShuttingDown()) { + return failed("Reward platform shut down before execution completed"); + } + if (index >= plan.steps().size()) { + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + } + + SharedRewardStep step = plan.steps().get(index); + if (step.requiresOnlinePlayer() && !platform.isOnline(context.userId())) { + if (!durability.durable()) { + return failed("Player became unavailable during non-durable reward step " + step.id()); + } + CompletionStage deferred; + try { + deferred = durability.defer(executionPath, index, context); + if (deferred == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null deferral stage")); + } + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + return deferred.thenApply(ignored -> SharedRewardResult.DEFERRED); + } + + String stepPath = executionPath + "/" + step.id() + ":" + index; + CompletionStage action; + try { + action = step.action().execute(context, stepPath); + if (action == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward step returned null completion stage: " + step.id())); + } + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + + return action.thenCompose(result -> { + if (result == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward step returned null result: " + step.id())); + } + if (result == SharedRewardResult.DEFERRED) { + return CompletableFuture.completedFuture(SharedRewardResult.DEFERRED); + } + CompletionStage checkpoint; + try { + checkpoint = durability.checkpoint(executionPath, index + 1, context); + if (checkpoint == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null checkpoint stage")); + } + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + return checkpoint.thenCompose(ignored -> executeSteps(plan, context, durability, executionPath, index + 1)); + }); + } + + private CompletionStage failed(String message) { + return CompletableFuture.failedFuture(new IllegalStateException(message)); + } +} From da23f3c0dd893fd48e28fd832702f185f0363d46 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:43:52 -0600 Subject: [PATCH 09/14] Test platform-neutral reward orchestration --- .../rewards/SharedRewardOrchestratorTest.java | 234 ++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java new file mode 100644 index 0000000000..32581cc33e --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java @@ -0,0 +1,234 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; + +import com.bencodez.advancedcore.core.reward.SharedRewardContext; +import com.bencodez.advancedcore.core.reward.SharedRewardDurability; +import com.bencodez.advancedcore.core.reward.SharedRewardOrchestrator; +import com.bencodez.advancedcore.core.reward.SharedRewardPlan; +import com.bencodez.advancedcore.core.reward.SharedRewardPlatform; +import com.bencodez.advancedcore.core.reward.SharedRewardResult; +import com.bencodez.advancedcore.core.reward.SharedRewardStep; + +class SharedRewardOrchestratorTest { + @Test + void runsRequirementsDelayActionsAndDurableCheckpointsInOrder() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardContext context = context(); + SharedRewardPlan plan = new SharedRewardPlan("root", 1.0, Duration.ofSeconds(3), + List.of(ctx -> { + events.add("requirement"); + return CompletableFuture.completedFuture(Boolean.TRUE); + }), + List.of(step("command", true, events), step("message", true, events))); + + SharedRewardResult result = orchestrator.execute(plan, context, durability).toCompletableFuture().join(); + + assertEquals(SharedRewardResult.COMPLETED, result); + assertEquals(List.of("requirement", "delay:3", "command", "checkpoint:root:1", "message", + "checkpoint:root:2"), events); + } + + @Test + void defersBeforePlayerBoundWorkWhenOffline() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + platform.online = false; + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardPlan plan = SharedRewardPlan.immediate("offline", List.of(step("player-command", true, events))); + + SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); + + assertEquals(SharedRewardResult.DEFERRED, result); + assertEquals(List.of("defer:offline:0"), events); + } + + @Test + void disconnectAfterCompletedStepDefersOnlyRemainingSuffix() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardStep first = new SharedRewardStep("first", true, (ctx, path) -> { + events.add("first"); + platform.online = false; + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }); + SharedRewardPlan plan = SharedRewardPlan.immediate("disconnect", + List.of(first, step("second", true, events))); + + SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); + + assertEquals(SharedRewardResult.DEFERRED, result); + assertEquals(List.of("first", "checkpoint:disconnect:1", "defer:disconnect:1"), events); + } + + @Test + void partialFailureNeverCheckpointsOrRunsLaterWork() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardStep failure = new SharedRewardStep("failure", false, (ctx, path) -> { + events.add("failure"); + return CompletableFuture.failedFuture(new IllegalStateException("boom")); + }); + SharedRewardPlan plan = SharedRewardPlan.immediate("partial", + List.of(step("first", false, events), failure, step("third", false, events))); + + assertThrows(CompletionException.class, + () -> orchestrator.execute(plan, context(), durability).toCompletableFuture().join()); + assertEquals(List.of("first", "checkpoint:partial:1", "failure"), events); + } + + @Test + void delayedWorkFailsClosedWhenShutdownStartsBeforeCallback() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + platform.shutdownBeforeDelayedCallback = true; + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardPlan plan = new SharedRewardPlan("shutdown", 1.0, Duration.ofSeconds(1), List.of(), + List.of(step("never", false, events))); + + assertThrows(CompletionException.class, + () -> orchestrator.execute(plan, context(), durability).toCompletableFuture().join()); + assertEquals(List.of("delay:1"), events); + } + + @Test + void nestedCompletionIsAwaitedBeforeParentCheckpoint() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + FakeDurability durability = new FakeDurability(events, true); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardPlan child = SharedRewardPlan.immediate("child", List.of(step("child-command", false, events))); + SharedRewardStep nested = new SharedRewardStep("nested", false, + (ctx, path) -> orchestrator.executeNested(child, ctx, durability, path)); + SharedRewardPlan parent = SharedRewardPlan.immediate("parent", List.of(nested, step("after", false, events))); + + orchestrator.execute(parent, context(), durability).toCompletableFuture().join(); + + assertEquals(List.of("child-command", "checkpoint:parent/nested:0/child:1", "checkpoint:parent:1", "after", + "checkpoint:parent:2"), events); + } + + @Test + void durableResumeSkipsCompletedPrefixAndDoesNotRerollChanceOrDelay() { + ArrayList events = new ArrayList<>(); + FakePlatform platform = new FakePlatform(events); + platform.chanceRoll = 0.99; + FakeDurability durability = new FakeDurability(events, true); + durability.completed.put("resume", 1); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); + SharedRewardPlan plan = new SharedRewardPlan("resume", 0.1, Duration.ofSeconds(4), + List.of(ctx -> { + events.add("requirement"); + return CompletableFuture.completedFuture(Boolean.FALSE); + }), List.of(step("already-done", false, events), step("remaining", false, events))); + + SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); + + assertEquals(SharedRewardResult.COMPLETED, result); + assertEquals(List.of("remaining", "checkpoint:resume:2"), events); + } + + private static SharedRewardContext context() { + return new SharedRewardContext(UUID.randomUUID(), "Ben", Map.of("player", "Ben")); + } + + private static SharedRewardStep step(String id, boolean requiresOnline, List events) { + return new SharedRewardStep(id, requiresOnline, (ctx, path) -> { + events.add(id); + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }); + } + + private static final class FakePlatform implements SharedRewardPlatform { + private final List events; + private boolean online = true; + private boolean shuttingDown; + private boolean shutdownBeforeDelayedCallback; + private double chanceRoll; + + private FakePlatform(List events) { + this.events = events; + } + + @Override + public boolean isOnline(UUID userId) { + return online; + } + + @Override + public double nextChanceRoll() { + return chanceRoll; + } + + @Override + public CompletionStage delay(Duration delay, + Supplier> operation) { + events.add("delay:" + delay.toSeconds()); + if (shutdownBeforeDelayedCallback) shuttingDown = true; + return operation.get(); + } + + @Override + public boolean isShuttingDown() { + return shuttingDown; + } + } + + private static final class FakeDurability implements SharedRewardDurability { + private final List events; + private final boolean durable; + private final HashMap completed = new HashMap<>(); + + private FakeDurability(List events, boolean durable) { + this.events = events; + this.durable = durable; + } + + @Override + public int completedSteps(String executionPath) { + return completed.getOrDefault(executionPath, 0); + } + + @Override + public CompletionStage checkpoint(String executionPath, int completedSteps, + SharedRewardContext context) { + events.add("checkpoint:" + executionPath + ":" + completedSteps); + completed.put(executionPath, completedSteps); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletionStage defer(String executionPath, int nextStep, SharedRewardContext context) { + events.add("defer:" + executionPath + ":" + nextStep); + return CompletableFuture.completedFuture(null); + } + + @Override + public boolean durable() { + return durable; + } + } +} From 3b1782d2aec9c3cb63be2cc9f6d3be5d987e7eb7 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 23:29:05 -0600 Subject: [PATCH 10/14] Persist reward eligibility decisions and bind replay to prepared plans Address Codex 3998759040 and 3998759042. Persist the initial decision, placeholder snapshot and stable definition/ordered-plan fingerprint before step zero, delay or offline deferral. Resume without rerolling requirements or chance, reject incompatible/unversioned progress without changing it, and continue awaiting actual action and checkpoint completion. Keep legacy signatures/non-durable integrations. Durable adapters must implement versioned snapshot and atomic begin through their existing replay owner; unsupported adapters fail closed. No second queue or replay store, no changes to #317 native execution, no Bukkit wiring or loader claims. Add 12 regressions and retain/update all 7 existing orchestration tests. All 19 methods passed locally against JDK21-compiled production classes using a small assertion/annotation harness (not Maven or the JUnit engine). Includes a test-only disk restart and lost-acknowledgement scenarios. Local syntax/whitespace checks passed. Full repository Maven/JUnit and packaged validation must run in Actions; live server/SQL not available. --- .../core/reward/SharedRewardDurability.java | 45 ++- .../core/reward/SharedRewardOrchestrator.java | 104 +++--- .../core/reward/SharedRewardPlan.java | 52 ++- .../core/reward/SharedRewardProgress.java | 24 ++ .../SharedRewardDurableDecisionTest.java | 316 ++++++++++++++++++ .../rewards/SharedRewardOrchestratorTest.java | 31 +- 6 files changed, 520 insertions(+), 52 deletions(-) create mode 100644 AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java index a43d5d4b4c..0007e528ac 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardDurability.java @@ -4,8 +4,10 @@ import java.util.concurrent.CompletionStage; /** - * Adapter over the existing durable replay owner. The orchestrator stores no - * replay queue or cursor itself; #317's replay state can implement this boundary. + * Adapter over the existing replay owner, scoped to ONE logical reward occurrence. + * That owner serializes execution/retries of each path; the orchestrator allocates + * no second queue, replay store or lock registry. Snapshot reads must not block the + * platform thread. Write stages complete only after persistence, never submission. */ public interface SharedRewardDurability { SharedRewardDurability NONE = new SharedRewardDurability() { @@ -18,13 +20,52 @@ public interface SharedRewardDurability { @Override public boolean durable() { return false; } }; + /** Existing API retained; an unversioned nonzero cursor is not safe to resume. */ int completedSteps(String executionPath); + /** Already-loaded snapshot, or null only when this occurrence/path has never begun. */ + default SharedRewardProgress loadProgress(String executionPath) { + if (durable()) throw new UnsupportedOperationException("Durable replay must load versioned decision/progress state"); + return null; + } + + /** + * Atomically persist the initial decision, fingerprint and placeholders at cursor + * zero, or return the already-persisted state. Never overwrite an earlier decision. + * A lost acknowledgement must still be recoverable by loadProgress on retry. + */ + default CompletionStage begin(String executionPath, SharedRewardProgress proposed) { + if (durable()) return CompletableFuture.failedFuture( + new UnsupportedOperationException("Durable replay must persist the initial execution decision")); + return CompletableFuture.completedFuture(proposed); + } + /** Completes only after progress and placeholder state are durable. */ CompletionStage checkpoint(String executionPath, int completedSteps, SharedRewardContext context); + /** Bound form used by the shared orchestrator; adapters persist the same binding with the cursor. */ + default CompletionStage checkpoint(String executionPath, String fingerprint, int completedSteps, + SharedRewardContext context) { + requireFingerprint(executionPath, fingerprint); + return checkpoint(executionPath, completedSteps, context); + } + /** Completes only after the still-pending reward occurrence is durable. */ CompletionStage defer(String executionPath, int nextStep, SharedRewardContext context); + default CompletionStage defer(String executionPath, String fingerprint, int nextStep, + SharedRewardContext context) { + requireFingerprint(executionPath, fingerprint); + return defer(executionPath, nextStep, context); + } + + private void requireFingerprint(String executionPath, String fingerprint) { + if (!durable()) return; + SharedRewardProgress state = loadProgress(executionPath); + if (state == null || !state.planFingerprint().equals(fingerprint)) { + throw new IllegalStateException("Reward checkpoint belongs to a different or unbound plan: " + executionPath); + } + } + boolean durable(); } diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java index 05de1d5b55..7a6d05447b 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -28,6 +28,8 @@ public CompletionStage execute(SharedRewardPlan plan, Shared public CompletionStage executeNested(SharedRewardPlan plan, SharedRewardContext context, SharedRewardDurability durability, String parentPath) { + Objects.requireNonNull(plan, "plan"); + Objects.requireNonNull(context, "context"); Objects.requireNonNull(parentPath, "parentPath"); String path = parentPath.isBlank() ? plan.id() : parentPath + "/" + plan.id(); return execute(plan, context, durability == null ? SharedRewardDurability.NONE : durability, path); @@ -35,44 +37,69 @@ public CompletionStage executeNested(SharedRewardPlan plan, private CompletionStage execute(SharedRewardPlan plan, SharedRewardContext context, SharedRewardDurability durability, String executionPath) { - if (platform.isShuttingDown()) { - return failed("Reward platform is shutting down"); + try { + if (platform.isShuttingDown()) return failed("Reward platform is shutting down"); + String fingerprint = durability.durable() ? plan.fingerprint() : "non-durable"; + SharedRewardProgress saved = durability.durable() ? durability.loadProgress(executionPath) : null; + CompletionStage started; + if (saved != null) { + validateProgress(plan, fingerprint, saved, executionPath); + started = CompletableFuture.completedFuture(saved); + } else { + int legacyCursor = durability.completedSteps(executionPath); + if (legacyCursor < 0 || legacyCursor > plan.steps().size()) { + return failed("Invalid durable reward cursor " + legacyCursor + " for " + executionPath); + } + if (durability.durable() && legacyCursor != 0) { + return failed("Cannot resume an unversioned reward cursor: " + executionPath); + } + CompletionStage eligible = legacyCursor > 0 + ? CompletableFuture.completedFuture(Boolean.TRUE) + : evaluateRequirements(plan.requirements(), context, 0).thenApply(passed -> + passed.booleanValue() && (plan.chance() >= 1.0 + || platform.nextChanceRoll() < plan.chance())); + started = eligible.thenCompose(passed -> { + SharedRewardProgress decision = new SharedRewardProgress(fingerprint, passed.booleanValue(), + legacyCursor, context.placeholders()); + CompletionStage persisted = durability.begin(executionPath, decision); + return persisted == null ? CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null begin stage")) : persisted; + }); + } + return started.thenCompose(progress -> { + validateProgress(plan, fingerprint, progress, executionPath); + context.placeholders().clear(); + context.placeholders().putAll(progress.placeholders()); + if (!progress.eligible()) return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); + return executeEligible(plan, context, durability, executionPath, fingerprint, progress.completedSteps()); + }); + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); } + } - int resume = durability.completedSteps(executionPath); - if (resume < 0 || resume > plan.steps().size()) { - return CompletableFuture.failedFuture(new IllegalStateException( - "Invalid durable reward cursor " + resume + " for " + executionPath)); + private void validateProgress(SharedRewardPlan plan, String fingerprint, SharedRewardProgress progress, + String executionPath) { + if (progress == null || !fingerprint.equals(progress.planFingerprint())) { + throw new IllegalStateException("Reward plan changed or has no durable binding: " + executionPath); } + if (progress.completedSteps() > plan.steps().size()) { + throw new IllegalStateException("Invalid durable reward cursor for " + executionPath); + } + } - CompletionStage eligible = resume > 0 - ? CompletableFuture.completedFuture(Boolean.TRUE) - : evaluateRequirements(plan.requirements(), context, 0); - return eligible.thenCompose(requirementsPassed -> { - if (!requirementsPassed.booleanValue()) { - return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); - } - if (resume == 0 && plan.chance() < 1.0 && platform.nextChanceRoll() >= plan.chance()) { - return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); - } - - java.util.function.Supplier> work = - () -> executeSteps(plan, context, durability, executionPath, resume); - Duration delay = resume == 0 ? plan.delay() : Duration.ZERO; - if (delay.isZero()) { - return work.get(); - } - try { - CompletionStage delayed = platform.delay(delay, work); - if (delayed == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward platform returned null delay stage")); - } - return delayed; - } catch (Throwable failure) { - return CompletableFuture.failedFuture(failure); - } - }); + private CompletionStage executeEligible(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability, String executionPath, String fingerprint, int resume) { + java.util.function.Supplier> work = + () -> executeSteps(plan, context, durability, executionPath, fingerprint, resume); + Duration delay = resume == 0 ? plan.delay() : Duration.ZERO; + if (delay.isZero()) return work.get(); + try { + CompletionStage delayed = platform.delay(delay, work); + return delayed == null ? failed("Reward platform returned null delay stage") : delayed; + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } } private CompletionStage evaluateRequirements(List requirements, @@ -96,7 +123,7 @@ private CompletionStage evaluateRequirements(List executeSteps(SharedRewardPlan plan, SharedRewardContext context, - SharedRewardDurability durability, String executionPath, int index) { + SharedRewardDurability durability, String executionPath, String fingerprint, int index) { if (platform.isShuttingDown()) { return failed("Reward platform shut down before execution completed"); } @@ -111,7 +138,7 @@ private CompletionStage executeSteps(SharedRewardPlan plan, } CompletionStage deferred; try { - deferred = durability.defer(executionPath, index, context); + deferred = durability.defer(executionPath, fingerprint, index, context); if (deferred == null) { return CompletableFuture.failedFuture( new IllegalStateException("Reward durability adapter returned null deferral stage")); @@ -144,7 +171,7 @@ private CompletionStage executeSteps(SharedRewardPlan plan, } CompletionStage checkpoint; try { - checkpoint = durability.checkpoint(executionPath, index + 1, context); + checkpoint = durability.checkpoint(executionPath, fingerprint, index + 1, context); if (checkpoint == null) { return CompletableFuture.failedFuture( new IllegalStateException("Reward durability adapter returned null checkpoint stage")); @@ -152,7 +179,8 @@ private CompletionStage executeSteps(SharedRewardPlan plan, } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } - return checkpoint.thenCompose(ignored -> executeSteps(plan, context, durability, executionPath, index + 1)); + return checkpoint.thenCompose(ignored -> executeSteps(plan, context, durability, + executionPath, fingerprint, index + 1)); }); } diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java index aad234889a..5b55a2bc83 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlan.java @@ -1,15 +1,20 @@ package com.bencodez.advancedcore.core.reward; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.time.Duration; +import java.util.HexFormat; import java.util.List; import java.util.Objects; /** - * Platform-neutral plan prepared by the existing reward configuration layer. - * Parsing and native item/sound/effect behavior intentionally remain outside it. + * A prepared configuration snapshot. Durable plans supply a stable definition + * fingerprint covering requirements, native payloads and injection-registry + * versions; lambda identities are deliberately not used as persistent identity. */ public record SharedRewardPlan(String id, double chance, Duration delay, - List requirements, List steps) { + List requirements, List steps, String definitionFingerprint) { public SharedRewardPlan { Objects.requireNonNull(id, "id"); Objects.requireNonNull(delay, "delay"); @@ -22,7 +27,48 @@ public record SharedRewardPlan(String id, double chance, Duration delay, if (delay.isNegative()) throw new IllegalArgumentException("delay cannot be negative"); } + /** Retained for synchronous/non-durable callers; bind a definition before durable execution. */ + public SharedRewardPlan(String id, double chance, Duration delay, + List requirements, List steps) { + this(id, chance, delay, requirements, steps, null); + } + public static SharedRewardPlan immediate(String id, List steps) { return new SharedRewardPlan(id, 1.0, Duration.ZERO, List.of(), steps); } + + public SharedRewardPlan withDefinitionFingerprint(String fingerprint) { + if (fingerprint == null || fingerprint.isBlank()) { + throw new IllegalArgumentException("Definition fingerprint must not be blank"); + } + return new SharedRewardPlan(id, chance, delay, requirements, steps, fingerprint); + } + + /** Versioned, length-delimited identity includes ordered step IDs and execution policy. */ + public String fingerprint() { + if (definitionFingerprint == null || definitionFingerprint.isBlank()) { + throw new IllegalStateException("Durable reward plans require a definition fingerprint"); + } + StringBuilder value = new StringBuilder("shared-reward-plan-v1"); + field(value, id); + field(value, definitionFingerprint); + field(value, Double.toHexString(chance)); + field(value, delay.toString()); + field(value, Integer.toString(requirements.size())); + field(value, Integer.toString(steps.size())); + for (SharedRewardStep step : steps) { + field(value, step.id()); + field(value, Boolean.toString(step.requiresOnlinePlayer())); + } + try { + return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256") + .digest(value.toString().getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException impossible) { + throw new IllegalStateException("SHA-256 is unavailable", impossible); + } + } + + private static void field(StringBuilder target, String value) { + target.append(':').append(value.length()).append(':').append(value); + } } diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java new file mode 100644 index 0000000000..308a2bd679 --- /dev/null +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java @@ -0,0 +1,24 @@ +package com.bencodez.advancedcore.core.reward; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +/** Durable decision and cursor for one occurrence/path, including cursor zero. */ +public record SharedRewardProgress(String planFingerprint, boolean eligible, int completedSteps, + Map placeholders) { + public SharedRewardProgress { + Objects.requireNonNull(planFingerprint, "planFingerprint"); + if (planFingerprint.isBlank()) throw new IllegalArgumentException("Plan fingerprint is blank"); + if (completedSteps < 0 || (!eligible && completedSteps != 0)) { + throw new IllegalArgumentException("Invalid reward progress"); + } + placeholders = Collections.unmodifiableMap(new HashMap<>(Objects.requireNonNull(placeholders, "placeholders"))); + } + + public SharedRewardProgress advance(int nextStep, SharedRewardContext context) { + if (nextStep < completedSteps) throw new IllegalArgumentException("Reward progress cannot move backwards"); + return new SharedRewardProgress(planFingerprint, eligible, nextStep, context.placeholders()); + } +} diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java new file mode 100644 index 0000000000..caaf8ecaf8 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java @@ -0,0 +1,316 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.*; + +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import com.bencodez.advancedcore.core.reward.*; + +/** Headless orchestration with test replay adapters; no replacement production replay store. */ +class SharedRewardDurableDecisionTest { + @TempDir Path directory; + private static final UUID USER = UUID.fromString("fef273b7-aa45-42f9-ac14-cb047533afde"); + + @Test + void firstFailureAndReconstructedReplayRetainTheDecisionAtStepZero() throws Exception { + Platform platform = new Platform(); + platform.roll = 0.1; + AtomicInteger requirements = new AtomicInteger(), attempts = new AtomicInteger(); + SharedRewardPlan plan = plan(0.5, List.of(ctx -> { + requirements.incrementAndGet(); + ctx.placeholders().put("token", "original"); + return CompletableFuture.completedFuture(true); + }), List.of(new SharedRewardStep("command", false, (ctx, path) -> { + if (attempts.incrementAndGet() == 1) return CompletableFuture.failedFuture(new IllegalStateException("offline service")); + assertEquals("original", ctx.placeholders().get("token")); + return done(); + }))); + Path file = directory.resolve("replay.properties"); + DiskReplay first = new DiskReplay(file); + assertThrows(CompletionException.class, () -> execute(platform, plan, first).join()); + assertEquals(0, first.loadProgress("vote").completedSteps()); + platform.roll = 0.99; + DiskReplay reopened = new DiskReplay(file); + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, reopened).join()); + assertEquals(1, requirements.get()); + assertEquals(1, platform.rolls); + assertEquals(1, new DiskReplay(file).loadProgress("vote").completedSteps()); + } + + @Test + void declinedChanceDecisionDoesNotBecomeEligibleOnRetry() { + Platform platform = new Platform(); + platform.roll = 0.9; + AtomicInteger requirements = new AtomicInteger(), actions = new AtomicInteger(); + SharedRewardPlan plan = plan(0.5, List.of(ctx -> { + requirements.incrementAndGet(); + return CompletableFuture.completedFuture(true); + }), List.of(action("command", actions))); + Replay replay = new Replay(); + assertEquals(SharedRewardResult.NOT_ELIGIBLE, execute(platform, plan, replay).join()); + platform.roll = 0.1; + assertEquals(SharedRewardResult.NOT_ELIGIBLE, execute(platform, plan, replay).join()); + assertEquals(1, requirements.get()); + assertEquals(1, platform.rolls); + assertEquals(0, actions.get()); + } + + @Test + void offlineZeroCursorResumesWithoutReevaluatingRequirements() { + Platform platform = new Platform(); + platform.online = false; + AtomicInteger requirements = new AtomicInteger(), actions = new AtomicInteger(); + SharedRewardPlan plan = plan(0.5, List.of(ctx -> { + requirements.incrementAndGet(); + return CompletableFuture.completedFuture(true); + }), List.of(new SharedRewardStep("message", true, (ctx, path) -> { + actions.incrementAndGet(); + return done(); + }))); + Replay replay = new Replay(); + assertEquals(SharedRewardResult.DEFERRED, execute(platform, plan, replay).join()); + assertEquals(0, replay.loadProgress("vote").completedSteps()); + platform.online = true; + platform.roll = 0.99; + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, replay).join()); + assertEquals(1, requirements.get()); + assertEquals(1, platform.rolls); + assertEquals(1, actions.get()); + } + + @Test + void initialDecisionPersistenceCompletesBeforeDelayAndAction() { + Platform platform = new Platform(); + Replay replay = new Replay(); + replay.beginAck = new CompletableFuture<>(); + AtomicInteger actions = new AtomicInteger(); + SharedRewardPlan plan = new SharedRewardPlan("vote", 1, Duration.ofSeconds(1), List.of(), + List.of(action("command", actions))).withDefinitionFingerprint("config-v1"); + CompletableFuture result = execute(platform, plan, replay); + assertFalse(result.isDone()); + assertEquals(0, platform.delays); + assertEquals(0, actions.get()); + replay.beginAck.complete(null); + assertEquals(SharedRewardResult.COMPLETED, result.join()); + assertEquals(1, platform.delays); + assertEquals(1, actions.get()); + } + + @Test + void lostBeginAcknowledgementRecoversThePersistedDecision() { + Platform platform = new Platform(); + Replay replay = new Replay(); + replay.beginAck = CompletableFuture.failedFuture(new IllegalStateException("lost acknowledgement")); + AtomicInteger actions = new AtomicInteger(); + SharedRewardPlan plan = plan(0.5, List.of(), List.of(action("command", actions))); + assertThrows(CompletionException.class, () -> execute(platform, plan, replay).join()); + assertEquals(0, actions.get()); + platform.roll = 0.99; + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, replay).join()); + assertEquals(1, actions.get()); + assertEquals(1, platform.rolls); + } + + @Test + void operationAndCheckpointAcknowledgementBothPrecedeTheNextStep() { + Platform platform = new Platform(); + Replay replay = new Replay(); + replay.checkpointAck = new CompletableFuture<>(); + CompletableFuture operation = new CompletableFuture<>(); + AtomicInteger next = new AtomicInteger(); + SharedRewardPlan plan = plan(1, List.of(), List.of( + new SharedRewardStep("first", false, (ctx, path) -> operation), action("next", next))); + CompletableFuture result = execute(platform, plan, replay); + assertFalse(result.isDone()); + assertEquals(0, replay.loadProgress("vote").completedSteps()); + operation.complete(SharedRewardResult.COMPLETED); + assertFalse(result.isDone()); + assertEquals(0, next.get()); + replay.checkpointAck.complete(null); + assertEquals(SharedRewardResult.COMPLETED, result.join()); + assertEquals(1, next.get()); + } + + @Test + void lostCheckpointAcknowledgementDoesNotRepeatADurablePrefix() { + Platform platform = new Platform(); + Replay replay = new Replay(); + replay.checkpointAck = CompletableFuture.failedFuture(new IllegalStateException("lost acknowledgement")); + AtomicInteger first = new AtomicInteger(), second = new AtomicInteger(); + SharedRewardPlan plan = plan(1, List.of(), List.of(action("first", first), action("second", second))); + assertThrows(CompletionException.class, () -> execute(platform, plan, replay).join()); + assertEquals(1, first.get()); + assertEquals(0, second.get()); + replay.checkpointAck = CompletableFuture.completedFuture(null); + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, replay).join()); + assertEquals(1, first.get()); + assertEquals(1, second.get()); + } + + @Test + void changedPlansAreRejectedEvenWhenTheOldCursorIsWithinBounds() { + Platform platform = new Platform(); + AtomicInteger actions = new AtomicInteger(); + SharedRewardStep a = action("a", actions), b = action("b", actions), c = action("c", actions); + SharedRewardPlan original = plan(1, List.of(), List.of(a, b)); + Replay replay = new Replay(); + SharedRewardProgress progress = new SharedRewardProgress(original.fingerprint(), true, 1, Map.of()); + replay.progress.put("vote", progress); + List edited = List.of( + plan(1, List.of(), List.of(b, a)), + plan(1, List.of(), List.of(c, a, b)), + plan(1, List.of(), List.of(b)), + plan(0.5, List.of(), List.of(a, b)), + new SharedRewardPlan("vote", 1, Duration.ofSeconds(2), List.of(), List.of(a, b), "config-v1"), + original.withDefinitionFingerprint("different-native-payload-or-requirement-v2"), + plan(1, List.of(), List.of(new SharedRewardStep("a", true, a.action()), b))); + for (SharedRewardPlan changed : edited) { + assertThrows(CompletionException.class, () -> execute(platform, changed, replay).join()); + assertSame(progress, replay.loadProgress("vote")); + } + assertEquals(0, actions.get()); + assertEquals(0, platform.rolls); + } + + @Test + void stableDefinitionsDoNotDependOnLambdaObjectIdentity() { + assertEquals(plan(1, List.of(), List.of(action("a", new AtomicInteger()))).fingerprint(), + plan(1, List.of(), List.of(action("a", new AtomicInteger()))).fingerprint()); + } + + @Test + void unversionedCursorIsNotInterpretedAsCurrentPlanProgress() { + Platform platform = new Platform(); + Replay replay = new Replay() { + @Override public int completedSteps(String path) { return 1; } + }; + AtomicInteger actions = new AtomicInteger(); + assertThrows(CompletionException.class, + () -> execute(platform, plan(1, List.of(), List.of(action("a", actions))), replay).join()); + assertEquals(0, actions.get()); + } + + @Test + void oldConstructorRemainsUsableWithoutDurabilityButCannotResumeUnboundWork() { + Platform platform = new Platform(); + AtomicInteger actions = new AtomicInteger(); + SharedRewardPlan unbound = SharedRewardPlan.immediate("vote", List.of(action("a", actions))); + assertThrows(CompletionException.class, () -> execute(platform, unbound, new Replay()).join()); + assertEquals(0, actions.get()); + assertEquals(SharedRewardResult.COMPLETED, execute(platform, unbound, SharedRewardDurability.NONE).join()); + assertEquals(1, actions.get()); + } + + @Test + void unsupportedLegacyDurabilityFailsBeforeRunningActions() { + Platform platform = new Platform(); + AtomicInteger actions = new AtomicInteger(); + SharedRewardDurability legacy = new SharedRewardDurability() { + public boolean durable() { return true; } + public int completedSteps(String path) { return 0; } + public CompletionStage checkpoint(String p, int n, SharedRewardContext c) { return CompletableFuture.completedFuture(null); } + public CompletionStage defer(String p, int n, SharedRewardContext c) { return CompletableFuture.completedFuture(null); } + }; + assertThrows(CompletionException.class, + () -> execute(platform, plan(1, List.of(), List.of(action("a", actions))), legacy).join()); + assertEquals(0, actions.get()); + } + + private static SharedRewardPlan plan(double chance, List requirements, List steps) { + return new SharedRewardPlan("vote", chance, Duration.ZERO, requirements, steps).withDefinitionFingerprint("config-v1"); + } + + private static SharedRewardStep action(String id, AtomicInteger calls) { + return new SharedRewardStep(id, false, (ctx, path) -> { calls.incrementAndGet(); return done(); }); + } + + private static CompletableFuture done() { return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); } + + private static CompletableFuture execute(Platform p, SharedRewardPlan plan, SharedRewardDurability replay) { + return new SharedRewardOrchestrator(p).execute(plan, new SharedRewardContext(USER, "Ben", Map.of()), replay).toCompletableFuture(); + } + + private static final class Platform implements SharedRewardPlatform { + boolean online = true; + double roll; + int rolls, delays; + public boolean isOnline(UUID uuid) { return online; } + public boolean isShuttingDown() { return false; } + public double nextChanceRoll() { rolls++; return roll; } + public CompletionStage delay(Duration d, Supplier> work) { + delays++; + return work.get(); + } + } + + private static class Replay implements SharedRewardDurability { + final Map progress = new HashMap<>(); + CompletableFuture beginAck = CompletableFuture.completedFuture(null); + CompletableFuture checkpointAck = CompletableFuture.completedFuture(null); + public boolean durable() { return true; } + public int completedSteps(String path) { return progress.containsKey(path) ? progress.get(path).completedSteps() : 0; } + public SharedRewardProgress loadProgress(String path) { return progress.get(path); } + public CompletionStage begin(String path, SharedRewardProgress proposed) { + progress.putIfAbsent(path, proposed); + persist(); + return beginAck.thenApply(ignored -> progress.get(path)); + } + public CompletionStage checkpoint(String path, int completed, SharedRewardContext context) { + progress.put(path, progress.get(path).advance(completed, context)); + persist(); + return checkpointAck; + } + public CompletionStage defer(String path, int cursor, SharedRewardContext context) { + assertEquals(cursor, completedSteps(path)); + return CompletableFuture.completedFuture(null); + } + void persist() {} + } + + /** Test-only disk snapshot simulates restart; not a production SQL adapter or power-loss test. */ + private static final class DiskReplay extends Replay { + private final Path file; + DiskReplay(Path file) throws Exception { + this.file = file; + if (Files.exists(file)) { + Properties p = new Properties(); + try (InputStream in = Files.newInputStream(file)) { p.load(in); } + progress.put("vote", new SharedRewardProgress(p.getProperty("fingerprint"), + Boolean.parseBoolean(p.getProperty("eligible")), Integer.parseInt(p.getProperty("cursor")), + Map.of("token", p.getProperty("token", "")))); + } + } + @Override void persist() { + SharedRewardProgress state = progress.get("vote"); + Properties p = new Properties(); + p.setProperty("fingerprint", state.planFingerprint()); + p.setProperty("eligible", Boolean.toString(state.eligible())); + p.setProperty("cursor", Integer.toString(state.completedSteps())); + p.setProperty("token", state.placeholders().getOrDefault("token", "")); + Path pending = file.resolveSibling(file.getFileName() + ".tmp"); + try { + try (OutputStream out = Files.newOutputStream(pending)) { p.store(out, "test replay"); } + Files.move(pending, file, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (Exception failure) { throw new IllegalStateException(failure); } + } + } +} diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java index 32581cc33e..5b4fb0c3cb 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java @@ -21,6 +21,7 @@ import com.bencodez.advancedcore.core.reward.SharedRewardOrchestrator; import com.bencodez.advancedcore.core.reward.SharedRewardPlan; import com.bencodez.advancedcore.core.reward.SharedRewardPlatform; +import com.bencodez.advancedcore.core.reward.SharedRewardProgress; import com.bencodez.advancedcore.core.reward.SharedRewardResult; import com.bencodez.advancedcore.core.reward.SharedRewardStep; @@ -37,7 +38,7 @@ void runsRequirementsDelayActionsAndDurableCheckpointsInOrder() { events.add("requirement"); return CompletableFuture.completedFuture(Boolean.TRUE); }), - List.of(step("command", true, events), step("message", true, events))); + List.of(step("command", true, events), step("message", true, events))).withDefinitionFingerprint("fixture-v1"); SharedRewardResult result = orchestrator.execute(plan, context, durability).toCompletableFuture().join(); @@ -53,7 +54,7 @@ void defersBeforePlayerBoundWorkWhenOffline() { platform.online = false; FakeDurability durability = new FakeDurability(events, true); SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); - SharedRewardPlan plan = SharedRewardPlan.immediate("offline", List.of(step("player-command", true, events))); + SharedRewardPlan plan = SharedRewardPlan.immediate("offline", List.of(step("player-command", true, events))).withDefinitionFingerprint("fixture-v1"); SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); @@ -73,7 +74,7 @@ void disconnectAfterCompletedStepDefersOnlyRemainingSuffix() { return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); }); SharedRewardPlan plan = SharedRewardPlan.immediate("disconnect", - List.of(first, step("second", true, events))); + List.of(first, step("second", true, events))).withDefinitionFingerprint("fixture-v1"); SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); @@ -92,7 +93,7 @@ void partialFailureNeverCheckpointsOrRunsLaterWork() { return CompletableFuture.failedFuture(new IllegalStateException("boom")); }); SharedRewardPlan plan = SharedRewardPlan.immediate("partial", - List.of(step("first", false, events), failure, step("third", false, events))); + List.of(step("first", false, events), failure, step("third", false, events))).withDefinitionFingerprint("fixture-v1"); assertThrows(CompletionException.class, () -> orchestrator.execute(plan, context(), durability).toCompletableFuture().join()); @@ -107,7 +108,7 @@ void delayedWorkFailsClosedWhenShutdownStartsBeforeCallback() { FakeDurability durability = new FakeDurability(events, true); SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); SharedRewardPlan plan = new SharedRewardPlan("shutdown", 1.0, Duration.ofSeconds(1), List.of(), - List.of(step("never", false, events))); + List.of(step("never", false, events))).withDefinitionFingerprint("fixture-v1"); assertThrows(CompletionException.class, () -> orchestrator.execute(plan, context(), durability).toCompletableFuture().join()); @@ -120,10 +121,10 @@ void nestedCompletionIsAwaitedBeforeParentCheckpoint() { FakePlatform platform = new FakePlatform(events); FakeDurability durability = new FakeDurability(events, true); SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); - SharedRewardPlan child = SharedRewardPlan.immediate("child", List.of(step("child-command", false, events))); + SharedRewardPlan child = SharedRewardPlan.immediate("child", List.of(step("child-command", false, events))).withDefinitionFingerprint("fixture-v1"); SharedRewardStep nested = new SharedRewardStep("nested", false, (ctx, path) -> orchestrator.executeNested(child, ctx, durability, path)); - SharedRewardPlan parent = SharedRewardPlan.immediate("parent", List.of(nested, step("after", false, events))); + SharedRewardPlan parent = SharedRewardPlan.immediate("parent", List.of(nested, step("after", false, events))).withDefinitionFingerprint("fixture-v1"); orchestrator.execute(parent, context(), durability).toCompletableFuture().join(); @@ -137,14 +138,15 @@ void durableResumeSkipsCompletedPrefixAndDoesNotRerollChanceOrDelay() { FakePlatform platform = new FakePlatform(events); platform.chanceRoll = 0.99; FakeDurability durability = new FakeDurability(events, true); - durability.completed.put("resume", 1); SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform); SharedRewardPlan plan = new SharedRewardPlan("resume", 0.1, Duration.ofSeconds(4), List.of(ctx -> { events.add("requirement"); return CompletableFuture.completedFuture(Boolean.FALSE); - }), List.of(step("already-done", false, events), step("remaining", false, events))); + }), List.of(step("already-done", false, events), step("remaining", false, events))).withDefinitionFingerprint("fixture-v1"); + durability.completed.put("resume", 1); + durability.progress.put("resume", new SharedRewardProgress(plan.fingerprint(), true, 1, Map.of())); SharedRewardResult result = orchestrator.execute(plan, context(), durability).toCompletableFuture().join(); assertEquals(SharedRewardResult.COMPLETED, result); @@ -201,12 +203,22 @@ private static final class FakeDurability implements SharedRewardDurability { private final List events; private final boolean durable; private final HashMap completed = new HashMap<>(); + private final HashMap progress = new HashMap<>(); private FakeDurability(List events, boolean durable) { this.events = events; this.durable = durable; } + @Override + public SharedRewardProgress loadProgress(String path) { return progress.get(path); } + + @Override + public CompletionStage begin(String path, SharedRewardProgress state) { + progress.putIfAbsent(path, state); + return CompletableFuture.completedFuture(progress.get(path)); + } + @Override public int completedSteps(String executionPath) { return completed.getOrDefault(executionPath, 0); @@ -217,6 +229,7 @@ public CompletionStage checkpoint(String executionPath, int completedSteps SharedRewardContext context) { events.add("checkpoint:" + executionPath + ":" + completedSteps); completed.put(executionPath, completedSteps); + progress.put(executionPath, progress.get(executionPath).advance(completedSteps, context)); return CompletableFuture.completedFuture(null); } From ac51398a5e7fb8e174b25417a0a96b2b97e41088 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 00:00:28 -0600 Subject: [PATCH 11/14] Preserve the original reward deadline across cursor-zero retries Address Codex 3998844331. Persist an absolute notBefore deadline with the initial eligibility decision and retain it on every progress advance. Recovered work waits only the remaining duration, including after first-step failure, offline deferral or a lost initial persistence acknowledgement. Reject delayed legacy cursor-zero snapshots without timing proof instead of restarting or bypassing the configured delay. Keep old constructors and provide an overridable platform clock without breaking existing adapters. Add four deterministic clock/restart regressions and retain the existing nineteen orchestration/decision tests. All twenty-three methods passed with JDK21-compiled production classes and the local assertion/annotation harness; this is not Maven or the JUnit engine. Incremental whitespace checks passed. Full repository tests and packaged validation remain through GitHub Actions; no live SQL/server test or native loader support is claimed. --- .../core/reward/SharedRewardOrchestrator.java | 18 ++++- .../core/reward/SharedRewardPlatform.java | 4 + .../core/reward/SharedRewardProgress.java | 16 +++- .../SharedRewardDurableDecisionTest.java | 73 ++++++++++++++++++- .../rewards/SharedRewardOrchestratorTest.java | 5 ++ 5 files changed, 107 insertions(+), 9 deletions(-) diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java index 7a6d05447b..0073da1abc 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -60,7 +60,8 @@ private CompletionStage execute(SharedRewardPlan plan, Share || platform.nextChanceRoll() < plan.chance())); started = eligible.thenCompose(passed -> { SharedRewardProgress decision = new SharedRewardProgress(fingerprint, passed.booleanValue(), - legacyCursor, context.placeholders()); + legacyCursor, context.placeholders(), + passed.booleanValue() ? platform.now().plus(plan.delay()) : platform.now()); CompletionStage persisted = durability.begin(executionPath, decision); return persisted == null ? CompletableFuture.failedFuture( new IllegalStateException("Reward durability adapter returned null begin stage")) : persisted; @@ -71,7 +72,7 @@ private CompletionStage execute(SharedRewardPlan plan, Share context.placeholders().clear(); context.placeholders().putAll(progress.placeholders()); if (!progress.eligible()) return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); - return executeEligible(plan, context, durability, executionPath, fingerprint, progress.completedSteps()); + return executeEligible(plan, context, durability, executionPath, fingerprint, progress); }); } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); @@ -83,16 +84,25 @@ private void validateProgress(SharedRewardPlan plan, String fingerprint, SharedR if (progress == null || !fingerprint.equals(progress.planFingerprint())) { throw new IllegalStateException("Reward plan changed or has no durable binding: " + executionPath); } + if (progress.eligible() && progress.completedSteps() == 0 && !plan.delay().isZero() + && progress.notBefore() == null) { + throw new IllegalStateException("Durable reward progress has no delay deadline: " + executionPath); + } if (progress.completedSteps() > plan.steps().size()) { throw new IllegalStateException("Invalid durable reward cursor for " + executionPath); } } private CompletionStage executeEligible(SharedRewardPlan plan, SharedRewardContext context, - SharedRewardDurability durability, String executionPath, String fingerprint, int resume) { + SharedRewardDurability durability, String executionPath, String fingerprint, SharedRewardProgress progress) { + int resume = progress.completedSteps(); java.util.function.Supplier> work = () -> executeSteps(plan, context, durability, executionPath, fingerprint, resume); - Duration delay = resume == 0 ? plan.delay() : Duration.ZERO; + Duration delay = Duration.ZERO; + if (resume == 0 && !plan.delay().isZero()) { + delay = durability.durable() ? Duration.between(platform.now(), progress.notBefore()) : plan.delay(); + if (delay.isNegative()) delay = Duration.ZERO; + } if (delay.isZero()) return work.get(); try { CompletionStage delayed = platform.delay(delay, work); diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java index f69dff7ba2..f8bea6a17f 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardPlatform.java @@ -1,6 +1,7 @@ package com.bencodez.advancedcore.core.reward; import java.time.Duration; +import java.time.Instant; import java.util.UUID; import java.util.concurrent.CompletionStage; import java.util.function.Supplier; @@ -9,6 +10,9 @@ public interface SharedRewardPlatform { boolean isOnline(UUID userId); + /** Wall-clock time for durable absolute deadlines; adapters/tests may supply their clock. */ + default Instant now() { return Instant.now(); } + /** Returns a value in the range [0, 1). */ double nextChanceRoll(); diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java index 308a2bd679..7a21c372c5 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardProgress.java @@ -1,24 +1,32 @@ package com.bencodez.advancedcore.core.reward; +import java.time.Instant; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Objects; -/** Durable decision and cursor for one occurrence/path, including cursor zero. */ +/** Durable state for one path in one logical reward occurrence, including cursor zero. */ public record SharedRewardProgress(String planFingerprint, boolean eligible, int completedSteps, - Map placeholders) { + Map placeholders, Instant notBefore) { public SharedRewardProgress { Objects.requireNonNull(planFingerprint, "planFingerprint"); - if (planFingerprint.isBlank()) throw new IllegalArgumentException("Plan fingerprint is blank"); + if (planFingerprint.isBlank()) throw new IllegalArgumentException("Plan fingerprint must not be blank"); if (completedSteps < 0 || (!eligible && completedSteps != 0)) { throw new IllegalArgumentException("Invalid reward progress"); } + // Retain the existing context's support for null placeholder values. placeholders = Collections.unmodifiableMap(new HashMap<>(Objects.requireNonNull(placeholders, "placeholders"))); } + /** Legacy snapshots have no timing proof; delayed cursor-zero recovery rejects them. */ + public SharedRewardProgress(String planFingerprint, boolean eligible, int completedSteps, + Map placeholders) { + this(planFingerprint, eligible, completedSteps, placeholders, null); + } + public SharedRewardProgress advance(int nextStep, SharedRewardContext context) { if (nextStep < completedSteps) throw new IllegalArgumentException("Reward progress cannot move backwards"); - return new SharedRewardProgress(planFingerprint, eligible, nextStep, context.placeholders()); + return new SharedRewardProgress(planFingerprint, eligible, nextStep, context.placeholders(), notBefore); } } diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java index caaf8ecaf8..ca02bee151 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardDurableDecisionTest.java @@ -8,6 +8,7 @@ import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -235,6 +236,69 @@ void unsupportedLegacyDurabilityFailsBeforeRunningActions() { assertEquals(0, actions.get()); } + @Test + void firstStepFailureAfterItsDeadlineDoesNotRestartTheDelay() { + Platform platform = new Platform(); + Replay replay = new Replay(); + AtomicInteger attempts = new AtomicInteger(); + SharedRewardPlan plan = new SharedRewardPlan("vote", 1, Duration.ofHours(12), List.of(), + List.of(new SharedRewardStep("command", false, (ctx, path) -> { + if (attempts.incrementAndGet() == 1) return CompletableFuture.failedFuture(new IllegalStateException("temporary")); + return done(); + })), "config-v1"); + assertThrows(CompletionException.class, () -> execute(platform, plan, replay).join()); + assertEquals(0, replay.loadProgress("vote").completedSteps()); + assertEquals(Instant.EPOCH.plus(Duration.ofHours(12)), replay.loadProgress("vote").notBefore()); + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, replay).join()); + assertEquals(List.of(Duration.ofHours(12)), platform.waited); + } + + @Test + void offlineAfterDelayDefersWithoutChargingTheDelayAgainOnReconnect() { + Platform platform = new Platform(); + platform.online = false; + Replay replay = new Replay(); + SharedRewardPlan plan = new SharedRewardPlan("vote", 1, Duration.ofDays(1), List.of(), + List.of(new SharedRewardStep("message", true, (ctx, path) -> done())), "config-v1"); + assertEquals(SharedRewardResult.DEFERRED, execute(platform, plan, replay).join()); + platform.online = true; + assertEquals(SharedRewardResult.COMPLETED, execute(platform, plan, replay).join()); + assertEquals(List.of(Duration.ofDays(1)), platform.waited); + } + + @Test + void reconstructedReplayWaitsOnlyUntilTheOriginalDeadline() throws Exception { + Platform firstPlatform = new Platform(); + Path file = directory.resolve("deadline.properties"); + DiskReplay first = new DiskReplay(file); + first.beginAck = CompletableFuture.failedFuture(new IllegalStateException("lost acknowledgement")); + SharedRewardPlan plan = new SharedRewardPlan("vote", 1, Duration.ofHours(12), List.of(), + List.of(action("command", new AtomicInteger())), "config-v1"); + assertThrows(CompletionException.class, () -> execute(firstPlatform, plan, first).join()); + assertTrue(firstPlatform.waited.isEmpty()); + Platform restarted = new Platform(); + restarted.time = Instant.EPOCH.plus(Duration.ofHours(10)); + DiskReplay reopened = new DiskReplay(file); + assertEquals(SharedRewardResult.COMPLETED, execute(restarted, plan, reopened).join()); + assertEquals(List.of(Duration.ofHours(2)), restarted.waited); + assertEquals(Instant.EPOCH.plus(Duration.ofHours(12)), new DiskReplay(file).loadProgress("vote").notBefore()); + } + + @Test + void legacyCursorZeroWithoutTimingProofCannotRestartOrBypassTheDelay() { + Platform platform = new Platform(); + Replay replay = new Replay(); + AtomicInteger actions = new AtomicInteger(); + SharedRewardPlan plan = new SharedRewardPlan("vote", 1, Duration.ofDays(1), List.of(), + List.of(action("command", actions)), "config-v1"); + SharedRewardProgress legacy = new SharedRewardProgress(plan.fingerprint(), true, 0, Map.of()); + replay.progress.put("vote", legacy); + assertThrows(CompletionException.class, () -> execute(platform, plan, replay).join()); + assertSame(legacy, replay.loadProgress("vote")); + assertTrue(platform.waited.isEmpty()); + assertEquals(0, actions.get()); + } + private static SharedRewardPlan plan(double chance, List requirements, List steps) { return new SharedRewardPlan("vote", chance, Duration.ZERO, requirements, steps).withDefinitionFingerprint("config-v1"); } @@ -253,11 +317,16 @@ private static final class Platform implements SharedRewardPlatform { boolean online = true; double roll; int rolls, delays; + Instant time = Instant.EPOCH; + final List waited = new ArrayList<>(); + public Instant now() { return time; } public boolean isOnline(UUID uuid) { return online; } public boolean isShuttingDown() { return false; } public double nextChanceRoll() { rolls++; return roll; } public CompletionStage delay(Duration d, Supplier> work) { delays++; + waited.add(d); + time = time.plus(d); return work.get(); } } @@ -296,7 +365,8 @@ private static final class DiskReplay extends Replay { try (InputStream in = Files.newInputStream(file)) { p.load(in); } progress.put("vote", new SharedRewardProgress(p.getProperty("fingerprint"), Boolean.parseBoolean(p.getProperty("eligible")), Integer.parseInt(p.getProperty("cursor")), - Map.of("token", p.getProperty("token", "")))); + Map.of("token", p.getProperty("token", "")), + p.getProperty("notBefore") == null ? null : Instant.parse(p.getProperty("notBefore")))); } } @Override void persist() { @@ -306,6 +376,7 @@ private static final class DiskReplay extends Replay { p.setProperty("eligible", Boolean.toString(state.eligible())); p.setProperty("cursor", Integer.toString(state.completedSteps())); p.setProperty("token", state.placeholders().getOrDefault("token", "")); + if (state.notBefore() != null) p.setProperty("notBefore", state.notBefore().toString()); Path pending = file.resolveSibling(file.getFileName() + ".tmp"); try { try (OutputStream out = Files.newOutputStream(pending)) { p.store(out, "test replay"); } diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java index 5b4fb0c3cb..58a1c61c60 100644 --- a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardOrchestratorTest.java @@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -170,6 +171,9 @@ private static final class FakePlatform implements SharedRewardPlatform { private boolean shuttingDown; private boolean shutdownBeforeDelayedCallback; private double chanceRoll; + private Instant time = Instant.EPOCH; + + @Override public Instant now() { return time; } private FakePlatform(List events) { this.events = events; @@ -189,6 +193,7 @@ public double nextChanceRoll() { public CompletionStage delay(Duration delay, Supplier> operation) { events.add("delay:" + delay.toSeconds()); + time = time.plus(delay); if (shutdownBeforeDelayedCallback) shuttingDown = true; return operation.get(); } From dd469393709788f388540c9772950e74ae64a0c7 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 10:10:33 -0600 Subject: [PATCH 12/14] Make shared reward sequencing stack-safe --- .../core/reward/SharedRewardOrchestrator.java | 59 +++++++++++-------- .../rewards/SharedRewardStackSafetyTest.java | 50 ++++++++++++++++ 2 files changed, 85 insertions(+), 24 deletions(-) create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardStackSafetyTest.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java index 0073da1abc..3bb3aac213 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -55,7 +55,7 @@ private CompletionStage execute(SharedRewardPlan plan, Share } CompletionStage eligible = legacyCursor > 0 ? CompletableFuture.completedFuture(Boolean.TRUE) - : evaluateRequirements(plan.requirements(), context, 0).thenApply(passed -> + : evaluateRequirements(plan.requirements(), context).thenApply(passed -> passed.booleanValue() && (plan.chance() >= 1.0 || platform.nextChanceRoll() < plan.chance())); started = eligible.thenCompose(passed -> { @@ -112,36 +112,48 @@ private CompletionStage executeEligible(SharedRewardPlan pla } } + /** Build requirement sequencing iteratively so completed stages cannot recurse through the JVM stack. */ private CompletionStage evaluateRequirements(List requirements, - SharedRewardContext context, int index) { - if (index >= requirements.size()) { - return CompletableFuture.completedFuture(Boolean.TRUE); - } - CompletionStage stage; - try { - stage = requirements.get(index).test(context); - if (stage == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward requirement returned null completion stage")); - } - } catch (Throwable failure) { - return CompletableFuture.failedFuture(failure); + SharedRewardContext context) { + CompletionStage chain = CompletableFuture.completedFuture(Boolean.TRUE); + for (SharedRewardRequirement requirement : requirements) { + chain = chain.thenCompose(passed -> { + if (!Boolean.TRUE.equals(passed)) return CompletableFuture.completedFuture(Boolean.FALSE); + try { + CompletionStage stage = requirement.test(context); + return stage == null ? CompletableFuture.failedFuture( + new IllegalStateException("Reward requirement returned null completion stage")) : stage; + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + }); } - return stage.thenCompose(passed -> Boolean.TRUE.equals(passed) - ? evaluateRequirements(requirements, context, index + 1) - : CompletableFuture.completedFuture(Boolean.FALSE)); + return chain; } + /** + * Build the step chain iteratively. Already-completed action/checkpoint stages may + * run inline, but no step invokes the next step recursively, so large synchronous + * plans remain stack-safe. + */ private CompletionStage executeSteps(SharedRewardPlan plan, SharedRewardContext context, SharedRewardDurability durability, String executionPath, String fingerprint, int index) { + CompletionStage chain = CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + for (int current = index; current < plan.steps().size(); current++) { + final int stepIndex = current; + chain = chain.thenCompose(previous -> previous == SharedRewardResult.DEFERRED + ? CompletableFuture.completedFuture(SharedRewardResult.DEFERRED) + : executeStep(plan.steps().get(stepIndex), context, durability, + executionPath, fingerprint, stepIndex)); + } + return chain; + } + + private CompletionStage executeStep(SharedRewardStep step, SharedRewardContext context, + SharedRewardDurability durability, String executionPath, String fingerprint, int index) { if (platform.isShuttingDown()) { return failed("Reward platform shut down before execution completed"); } - if (index >= plan.steps().size()) { - return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); - } - - SharedRewardStep step = plan.steps().get(index); if (step.requiresOnlinePlayer() && !platform.isOnline(context.userId())) { if (!durability.durable()) { return failed("Player became unavailable during non-durable reward step " + step.id()); @@ -189,8 +201,7 @@ private CompletionStage executeSteps(SharedRewardPlan plan, } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } - return checkpoint.thenCompose(ignored -> executeSteps(plan, context, durability, - executionPath, fingerprint, index + 1)); + return checkpoint.thenApply(ignored -> SharedRewardResult.COMPLETED); }); } diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardStackSafetyTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardStackSafetyTest.java new file mode 100644 index 0000000000..92f1f75302 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardStackSafetyTest.java @@ -0,0 +1,50 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import com.bencodez.advancedcore.core.reward.SharedRewardContext; +import com.bencodez.advancedcore.core.reward.SharedRewardDurability; +import com.bencodez.advancedcore.core.reward.SharedRewardOrchestrator; +import com.bencodez.advancedcore.core.reward.SharedRewardPlan; +import com.bencodez.advancedcore.core.reward.SharedRewardPlatform; +import com.bencodez.advancedcore.core.reward.SharedRewardResult; +import com.bencodez.advancedcore.core.reward.SharedRewardStep; + +class SharedRewardStackSafetyTest { + @Test + void thousandsOfSynchronousStepsCompleteWithoutRecursiveStackGrowth() { + AtomicInteger executions = new AtomicInteger(); + ArrayList steps = new ArrayList<>(); + for (int i = 0; i < 5_000; i++) { + steps.add(new SharedRewardStep("step-" + i, false, (context, path) -> { + executions.incrementAndGet(); + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + })); + } + SharedRewardPlatform platform = new SharedRewardPlatform() { + public boolean isOnline(UUID userId) { return true; } + public double nextChanceRoll() { return 0.0; } + public CompletionStage delay(Duration delay, + java.util.function.Supplier> operation) { + return operation.get(); + } + public boolean isShuttingDown() { return false; } + }; + SharedRewardResult result = new SharedRewardOrchestrator(platform).execute( + SharedRewardPlan.immediate("large", steps), + new SharedRewardContext(UUID.randomUUID(), "Ben", new HashMap<>()), + SharedRewardDurability.NONE).toCompletableFuture().join(); + assertEquals(SharedRewardResult.COMPLETED, result); + assertEquals(5_000, executions.get()); + } +} From e5ce6370db95b66d258cd64dabe2fb6b8d6174fc Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 10:19:23 -0600 Subject: [PATCH 13/14] Preserve terminal reward semantics in stack-safe chains Retain concurrent commit dd46939's iterative sequencing and existing stack regression rather than overwriting it. Restore the final post-checkpoint shutdown check and normalize a null final requirement to NOT_ELIGIBLE. Add eleven headless regressions for long durable chains/requirements, real stage ordering, callback threads, partial failure, disconnects, checkpoint failure, shutdown and null requirements. No executor, blocking wait or durability contract changes. Local JDK21 syntax and reconstructed incremental whitespace checks passed; full Maven/JUnit and packaged checks require GitHub Actions. No live-server/database validation is claimed. --- .../core/reward/SharedRewardOrchestrator.java | 9 +- .../SharedRewardChainRegressionTest.java | 219 ++++++++++++++++++ .../SharedRewardNullRequirementTest.java | 52 +++++ 3 files changed, 278 insertions(+), 2 deletions(-) create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardChainRegressionTest.java create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardNullRequirementTest.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java index 3bb3aac213..b6077238e3 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -128,7 +128,8 @@ private CompletionStage evaluateRequirements(List executeSteps(SharedRewardPlan plan, : executeStep(plan.steps().get(stepIndex), context, durability, executionPath, fingerprint, stepIndex)); } - return chain; + // Preserve the old terminal shutdown check after the final checkpoint, + // including empty plans and fully checkpointed resumes. Deferrals stay deferred. + return chain.thenCompose(result -> result != SharedRewardResult.DEFERRED && platform.isShuttingDown() + ? failed("Reward platform shut down before execution completed") + : CompletableFuture.completedFuture(result)); } private CompletionStage executeStep(SharedRewardStep step, SharedRewardContext context, diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardChainRegressionTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardChainRegressionTest.java new file mode 100644 index 0000000000..5125b30784 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardChainRegressionTest.java @@ -0,0 +1,219 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.*; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.function.IntFunction; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import com.bencodez.advancedcore.core.reward.*; + +/** Non-Bukkit execution and in-memory checkpoint adapter, not a live server/database. */ +@Timeout(15) +class SharedRewardChainRegressionTest { + private static final int LENGTH = 20_000; + + @Test void synchronousActionsAndCheckpointsDoNotGrowTheCallStack() { + Fixture fixture = new Fixture(); + assertEquals(SharedRewardResult.COMPLETED, fixture.run(fixture.steps(LENGTH)).join()); + assertEquals(LENGTH, fixture.actions); + assertEquals(LENGTH, fixture.progress.completedSteps()); + assertEquals(Integer.toString(LENGTH), fixture.progress.placeholders().get("count")); + } + + @Test void synchronousRequirementsDoNotGrowTheCallStack() { + Fixture fixture = new Fixture(); + List requirements = new ArrayList<>(); + int[] checks = {0}; + for (int i = 0; i < LENGTH; i++) { + int expected = i; + requirements.add(context -> { + assertEquals(expected, checks[0]++); + return CompletableFuture.completedFuture(Boolean.TRUE); + }); + } + var plan = new SharedRewardPlan("chain", 1, Duration.ZERO, requirements, fixture.steps(1)) + .withDefinitionFingerprint("chain-v1"); + assertEquals(SharedRewardResult.COMPLETED, fixture.run(plan).join()); + assertEquals(LENGTH, checks[0]); + assertEquals(1, fixture.actions); + } + + @Test void asynchronousRequirementStillShortCircuitsTheRemainingSuffix() { + Fixture fixture = new Fixture(); + CompletableFuture held = new CompletableFuture<>(); + List requirements = new ArrayList<>(); + int[] checks = {0}; + for (int i = 0; i < LENGTH; i++) { + int index = i; + requirements.add(context -> { + checks[0]++; + return index == 1_000 ? held.minimalCompletionStage() + : CompletableFuture.completedFuture(Boolean.TRUE); + }); + } + var result = fixture.run(new SharedRewardPlan("chain", 1, Duration.ZERO, requirements, fixture.steps(1)) + .withDefinitionFingerprint("chain-v1")); + assertFalse(result.isDone()); + assertEquals(1_001, checks[0]); + assertEquals(0, fixture.actions); + held.complete(false); + assertEquals(SharedRewardResult.NOT_ELIGIBLE, result.join()); + assertEquals(1_001, checks[0]); + assertEquals(0, fixture.actions); + } + + @Test void mixedAsyncActionsAndCheckpointsRemainOrderedOnTheCompletingThread() throws Exception { + Fixture fixture = new Fixture(); + CompletableFuture action = new CompletableFuture<>(); + CompletableFuture checkpoint = new CompletableFuture<>(); + fixture.actionResult = index -> index == 1_000 ? action.minimalCompletionStage() + : CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + fixture.checkpointResult = next -> next == 8_000 ? checkpoint.minimalCompletionStage() + : CompletableFuture.completedFuture(null); + var result = fixture.run(fixture.steps(LENGTH)); + assertFalse(result.isDone()); + assertEquals(1_001, fixture.actions); + assertEquals(1_000, fixture.progress.completedSteps()); + var worker = Executors.newSingleThreadExecutor(); + try { + Thread completing = worker.submit(() -> { + action.complete(SharedRewardResult.COMPLETED); + return Thread.currentThread(); + }).get(5, TimeUnit.SECONDS); + assertFalse(result.isDone()); + assertEquals(8_000, fixture.actions); + assertEquals(7_999, fixture.progress.completedSteps()); + assertSame(completing, fixture.lastActionThread); + worker.submit(() -> checkpoint.complete(null)).get(5, TimeUnit.SECONDS); + assertEquals(SharedRewardResult.COMPLETED, result.get(5, TimeUnit.SECONDS)); + assertEquals(LENGTH, fixture.actions); + assertEquals(LENGTH, fixture.progress.completedSteps()); + assertSame(completing, fixture.lastActionThread); + } finally { + worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test void failedActionDoesNotCheckpointOrExecuteAnyLaterStep() { + Fixture fixture = new Fixture(); + IllegalStateException failure = new IllegalStateException("action failed"); + fixture.actionResult = index -> index == 1_000 ? CompletableFuture.failedFuture(failure) + : CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + assertSame(failure, assertThrows(CompletionException.class, + () -> fixture.run(fixture.steps(LENGTH)).join()).getCause()); + assertEquals(1_001, fixture.actions); + assertEquals(1_000, fixture.progress.completedSteps()); + } + + @Test void failedCheckpointDoesNotAdvanceOrRunLaterSteps() { + Fixture fixture = new Fixture(); + IllegalStateException failure = new IllegalStateException("checkpoint failed"); + fixture.checkpointResult = next -> next == 1_000 ? CompletableFuture.failedFuture(failure) + : CompletableFuture.completedFuture(null); + assertSame(failure, assertThrows(CompletionException.class, + () -> fixture.run(fixture.steps(LENGTH)).join()).getCause()); + assertEquals(1_000, fixture.actions); + assertEquals(999, fixture.progress.completedSteps()); + } + + @Test void disconnectDefersOnlyTheUncompletedSuffixOfALongChain() { + Fixture fixture = new Fixture(); + fixture.actionResult = index -> { + if (index == 999) fixture.online = false; + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }; + assertEquals(SharedRewardResult.DEFERRED, fixture.run(fixture.steps(LENGTH)).join()); + assertEquals(1_000, fixture.actions); + assertEquals(1_000, fixture.progress.completedSteps()); + assertEquals(1_000, fixture.deferredAt); + } + + @Test void anExplicitDeferredActionIsNotCheckpointed() { + Fixture fixture = new Fixture(); + fixture.actionResult = index -> CompletableFuture.completedFuture(index == 1_000 + ? SharedRewardResult.DEFERRED : SharedRewardResult.COMPLETED); + assertEquals(SharedRewardResult.DEFERRED, fixture.run(fixture.steps(LENGTH)).join()); + assertEquals(1_001, fixture.actions); + assertEquals(1_000, fixture.progress.completedSteps()); + } + + @Test void shutdownAfterTheLastCheckpointStillFailsInsteadOfReportingCompletion() { + Fixture fixture = new Fixture(); + CompletableFuture held = new CompletableFuture<>(); + fixture.checkpointResult = next -> held; + var result = fixture.run(fixture.steps(1)); + assertFalse(result.isDone()); + fixture.stopping = true; + held.complete(null); + assertThrows(CompletionException.class, result::join); + assertEquals(1, fixture.actions); + assertEquals(1, fixture.progress.completedSteps()); + } + + private static final class Fixture implements SharedRewardPlatform, SharedRewardDurability { + boolean online = true, stopping; + int actions, deferredAt = -1; + Thread lastActionThread; + SharedRewardProgress progress; + IntFunction> actionResult = index -> + CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + IntFunction> checkpointResult = next -> CompletableFuture.completedFuture(null); + + List steps(int count) { + List steps = new ArrayList<>(); + for (int i = 0; i < count; i++) { + int index = i; + steps.add(new SharedRewardStep("step-" + index, true, (context, path) -> { + assertEquals(index, actions++); + assertEquals(index, progress.completedSteps(), "action preceded its prior checkpoint"); + assertEquals("chain/step-" + index + ":" + index, path); + lastActionThread = Thread.currentThread(); + context.placeholders().put("count", Integer.toString(index + 1)); + return actionResult.apply(index); + })); + } + return steps; + } + CompletableFuture run(List steps) { + return run(SharedRewardPlan.immediate("chain", steps).withDefinitionFingerprint("chain-v1")); + } + CompletableFuture run(SharedRewardPlan plan) { + return new SharedRewardOrchestrator(this).execute(plan, + new SharedRewardContext(UUID.randomUUID(), "Ben", Map.of()), this).toCompletableFuture(); + } + public boolean isOnline(UUID uuid) { return online; } + public boolean isShuttingDown() { return stopping; } + public double nextChanceRoll() { throw new AssertionError("Unexpected chance roll"); } + public CompletionStage delay(Duration delay, + Supplier> work) { throw new AssertionError("Unexpected delay"); } + public boolean durable() { return true; } + public int completedSteps(String path) { return progress == null ? 0 : progress.completedSteps(); } + public SharedRewardProgress loadProgress(String path) { return progress; } + public CompletionStage begin(String path, SharedRewardProgress proposed) { + if (progress == null) progress = proposed; + return CompletableFuture.completedFuture(progress); + } + public CompletionStage checkpoint(String path, int next, SharedRewardContext context) { + assertEquals(next, actions, "checkpoint preceded action completion"); + return checkpointResult.apply(next).thenRun(() -> progress = progress.advance(next, context)); + } + public CompletionStage defer(String path, int next, SharedRewardContext context) { + deferredAt = next; + return CompletableFuture.completedFuture(null); + } + } +} diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardNullRequirementTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardNullRequirementTest.java new file mode 100644 index 0000000000..4b65d0ec8d --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardNullRequirementTest.java @@ -0,0 +1,52 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.*; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; + +import com.bencodez.advancedcore.core.reward.*; + +class SharedRewardNullRequirementTest { + @Test void completedNullFinalRequirementIsNotEligible() { + assertNullRequirement(false); + } + + @Test void asynchronouslyCompletedNullFinalRequirementIsNotEligible() { + assertNullRequirement(true); + } + + private void assertNullRequirement(boolean asynchronous) { + CompletableFuture requirement = new CompletableFuture<>(); + if (!asynchronous) requirement.complete(null); + SharedRewardPlatform platform = new SharedRewardPlatform() { + public boolean isOnline(UUID uuid) { return true; } + public boolean isShuttingDown() { return false; } + public double nextChanceRoll() { throw new AssertionError("ineligible request rerolled chance"); } + public CompletionStage delay(Duration delay, + Supplier> work) { + throw new AssertionError("ineligible request was delayed"); + } + }; + var plan = new SharedRewardPlan("null-requirement", 0.5, Duration.ofHours(1), + List.of(context -> CompletableFuture.completedFuture(true), context -> requirement), + List.of(new SharedRewardStep("never", false, (context, path) -> { + throw new AssertionError("ineligible reward executed"); + }))); + var result = new SharedRewardOrchestrator(platform).execute(plan, + new SharedRewardContext(UUID.randomUUID(), "Ben", Map.of()), SharedRewardDurability.NONE) + .toCompletableFuture(); + if (asynchronous) { + assertFalse(result.isDone()); + requirement.complete(null); + } + assertEquals(SharedRewardResult.NOT_ELIGIBLE, result.join()); + } +} From 026ece90413a72837560383d312c3f405a6689ac Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 10:30:13 -0600 Subject: [PATCH 14/14] Preserve retryable requirements and collision-free reward paths --- .../core/reward/SharedRewardOrchestrator.java | 152 +++++++++--------- .../core/reward/SharedRewardRequirement.java | 32 ++++ .../SharedRewardReviewFollowupTest.java | 105 ++++++++++++ 3 files changed, 217 insertions(+), 72 deletions(-) create mode 100644 AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardReviewFollowupTest.java diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java index b6077238e3..cc3d7670fc 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardOrchestrator.java @@ -1,11 +1,15 @@ package com.bencodez.advancedcore.core.reward; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.Base64; import java.util.List; import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; +import com.bencodez.advancedcore.core.reward.SharedRewardRequirement.Outcome; + /** * Sequences already-configured reward work without Bukkit dependencies. Native * actions remain in platform adapters, while durability remains owned by the @@ -23,7 +27,7 @@ public CompletionStage execute(SharedRewardPlan plan, Shared Objects.requireNonNull(plan, "plan"); Objects.requireNonNull(context, "context"); SharedRewardDurability replay = durability == null ? SharedRewardDurability.NONE : durability; - return execute(plan, context, replay, plan.id()); + return execute(plan, context, replay, pathSegment(plan.id())); } public CompletionStage executeNested(SharedRewardPlan plan, SharedRewardContext context, @@ -31,7 +35,8 @@ public CompletionStage executeNested(SharedRewardPlan plan, Objects.requireNonNull(plan, "plan"); Objects.requireNonNull(context, "context"); Objects.requireNonNull(parentPath, "parentPath"); - String path = parentPath.isBlank() ? plan.id() : parentPath + "/" + plan.id(); + String segment = pathSegment(plan.id()); + String path = parentPath.isBlank() ? segment : parentPath + "/" + segment; return execute(plan, context, durability == null ? SharedRewardDurability.NONE : durability, path); } @@ -41,44 +46,66 @@ private CompletionStage execute(SharedRewardPlan plan, Share if (platform.isShuttingDown()) return failed("Reward platform is shutting down"); String fingerprint = durability.durable() ? plan.fingerprint() : "non-durable"; SharedRewardProgress saved = durability.durable() ? durability.loadProgress(executionPath) : null; - CompletionStage started; if (saved != null) { validateProgress(plan, fingerprint, saved, executionPath); - started = CompletableFuture.completedFuture(saved); - } else { - int legacyCursor = durability.completedSteps(executionPath); - if (legacyCursor < 0 || legacyCursor > plan.steps().size()) { - return failed("Invalid durable reward cursor " + legacyCursor + " for " + executionPath); + return continueFromProgress(plan, context, durability, executionPath, fingerprint, saved); + } + + int legacyCursor = durability.completedSteps(executionPath); + if (legacyCursor < 0 || legacyCursor > plan.steps().size()) { + return failed("Invalid durable reward cursor " + legacyCursor + " for " + executionPath); + } + if (durability.durable() && legacyCursor != 0) { + return failed("Cannot resume an unversioned reward cursor: " + executionPath); + } + + CompletionStage requirements = legacyCursor > 0 + ? CompletableFuture.completedFuture(Outcome.PASS) + : evaluateRequirements(plan.requirements(), context); + return requirements.thenCompose(outcome -> { + if (outcome == Outcome.RETRY) { + if (!durability.durable()) return failed("Retryable reward requirement needs durable deferral: " + executionPath); + CompletionStage deferred; + try { + deferred = durability.defer(executionPath, legacyCursor, context); + if (deferred == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null requirement deferral stage")); + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + return deferred.thenApply(ignored -> SharedRewardResult.DEFERRED); } - if (durability.durable() && legacyCursor != 0) { - return failed("Cannot resume an unversioned reward cursor: " + executionPath); + + boolean passed = outcome == Outcome.PASS + && (plan.chance() >= 1.0 || platform.nextChanceRoll() < plan.chance()); + SharedRewardProgress decision = new SharedRewardProgress(fingerprint, passed, legacyCursor, + context.placeholders(), passed ? platform.now().plus(plan.delay()) : platform.now()); + CompletionStage persisted; + try { + persisted = durability.begin(executionPath, decision); + if (persisted == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null begin stage")); + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); } - CompletionStage eligible = legacyCursor > 0 - ? CompletableFuture.completedFuture(Boolean.TRUE) - : evaluateRequirements(plan.requirements(), context).thenApply(passed -> - passed.booleanValue() && (plan.chance() >= 1.0 - || platform.nextChanceRoll() < plan.chance())); - started = eligible.thenCompose(passed -> { - SharedRewardProgress decision = new SharedRewardProgress(fingerprint, passed.booleanValue(), - legacyCursor, context.placeholders(), - passed.booleanValue() ? platform.now().plus(plan.delay()) : platform.now()); - CompletionStage persisted = durability.begin(executionPath, decision); - return persisted == null ? CompletableFuture.failedFuture( - new IllegalStateException("Reward durability adapter returned null begin stage")) : persisted; + return persisted.thenCompose(progress -> { + validateProgress(plan, fingerprint, progress, executionPath); + return continueFromProgress(plan, context, durability, executionPath, fingerprint, progress); }); - } - return started.thenCompose(progress -> { - validateProgress(plan, fingerprint, progress, executionPath); - context.placeholders().clear(); - context.placeholders().putAll(progress.placeholders()); - if (!progress.eligible()) return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); - return executeEligible(plan, context, durability, executionPath, fingerprint, progress); }); } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } } + private CompletionStage continueFromProgress(SharedRewardPlan plan, SharedRewardContext context, + SharedRewardDurability durability, String executionPath, String fingerprint, SharedRewardProgress progress) { + context.placeholders().clear(); + context.placeholders().putAll(progress.placeholders()); + if (!progress.eligible()) return CompletableFuture.completedFuture(SharedRewardResult.NOT_ELIGIBLE); + return executeEligible(plan, context, durability, executionPath, fingerprint, progress); + } + private void validateProgress(SharedRewardPlan plan, String fingerprint, SharedRewardProgress progress, String executionPath) { if (progress == null || !fingerprint.equals(progress.planFingerprint())) { @@ -112,15 +139,14 @@ private CompletionStage executeEligible(SharedRewardPlan pla } } - /** Build requirement sequencing iteratively so completed stages cannot recurse through the JVM stack. */ - private CompletionStage evaluateRequirements(List requirements, + private CompletionStage evaluateRequirements(List requirements, SharedRewardContext context) { - CompletionStage chain = CompletableFuture.completedFuture(Boolean.TRUE); + CompletionStage chain = CompletableFuture.completedFuture(Outcome.PASS); for (SharedRewardRequirement requirement : requirements) { - chain = chain.thenCompose(passed -> { - if (!Boolean.TRUE.equals(passed)) return CompletableFuture.completedFuture(Boolean.FALSE); + chain = chain.thenCompose(previous -> { + if (previous != Outcome.PASS) return CompletableFuture.completedFuture(previous); try { - CompletionStage stage = requirement.test(context); + CompletionStage stage = requirement.evaluate(context); return stage == null ? CompletableFuture.failedFuture( new IllegalStateException("Reward requirement returned null completion stage")) : stage; } catch (Throwable failure) { @@ -128,15 +154,9 @@ private CompletionStage evaluateRequirements(List outcome == null ? Outcome.FAIL : outcome); } - /** - * Build the step chain iteratively. Already-completed action/checkpoint stages may - * run inline, but no step invokes the next step recursively, so large synchronous - * plans remain stack-safe. - */ private CompletionStage executeSteps(SharedRewardPlan plan, SharedRewardContext context, SharedRewardDurability durability, String executionPath, String fingerprint, int index) { CompletionStage chain = CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); @@ -144,11 +164,8 @@ private CompletionStage executeSteps(SharedRewardPlan plan, final int stepIndex = current; chain = chain.thenCompose(previous -> previous == SharedRewardResult.DEFERRED ? CompletableFuture.completedFuture(SharedRewardResult.DEFERRED) - : executeStep(plan.steps().get(stepIndex), context, durability, - executionPath, fingerprint, stepIndex)); + : executeStep(plan.steps().get(stepIndex), context, durability, executionPath, fingerprint, stepIndex)); } - // Preserve the old terminal shutdown check after the final checkpoint, - // including empty plans and fully checkpointed resumes. Deferrals stay deferred. return chain.thenCompose(result -> result != SharedRewardResult.DEFERRED && platform.isShuttingDown() ? failed("Reward platform shut down before execution completed") : CompletableFuture.completedFuture(result)); @@ -156,53 +173,39 @@ private CompletionStage executeSteps(SharedRewardPlan plan, private CompletionStage executeStep(SharedRewardStep step, SharedRewardContext context, SharedRewardDurability durability, String executionPath, String fingerprint, int index) { - if (platform.isShuttingDown()) { - return failed("Reward platform shut down before execution completed"); - } + if (platform.isShuttingDown()) return failed("Reward platform shut down before execution completed"); if (step.requiresOnlinePlayer() && !platform.isOnline(context.userId())) { - if (!durability.durable()) { - return failed("Player became unavailable during non-durable reward step " + step.id()); - } + if (!durability.durable()) return failed("Player became unavailable during non-durable reward step " + step.id()); CompletionStage deferred; try { deferred = durability.defer(executionPath, fingerprint, index, context); - if (deferred == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward durability adapter returned null deferral stage")); - } + if (deferred == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null deferral stage")); } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } return deferred.thenApply(ignored -> SharedRewardResult.DEFERRED); } - String stepPath = executionPath + "/" + step.id() + ":" + index; + String stepPath = executionPath + "/" + pathSegment(step.id()) + ":" + index; CompletionStage action; try { action = step.action().execute(context, stepPath); - if (action == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward step returned null completion stage: " + step.id())); - } + if (action == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward step returned null completion stage: " + step.id())); } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } return action.thenCompose(result -> { - if (result == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward step returned null result: " + step.id())); - } - if (result == SharedRewardResult.DEFERRED) { - return CompletableFuture.completedFuture(SharedRewardResult.DEFERRED); - } + if (result == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward step returned null result: " + step.id())); + if (result == SharedRewardResult.DEFERRED) return CompletableFuture.completedFuture(SharedRewardResult.DEFERRED); CompletionStage checkpoint; try { checkpoint = durability.checkpoint(executionPath, fingerprint, index + 1, context); - if (checkpoint == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("Reward durability adapter returned null checkpoint stage")); - } + if (checkpoint == null) return CompletableFuture.failedFuture( + new IllegalStateException("Reward durability adapter returned null checkpoint stage")); } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } @@ -210,6 +213,11 @@ private CompletionStage executeStep(SharedRewardStep step, S }); } + private static String pathSegment(String id) { + if (id.indexOf('/') < 0 && id.indexOf(':') < 0 && id.indexOf('%') < 0) return id; + return "%" + Base64.getUrlEncoder().withoutPadding().encodeToString(id.getBytes(StandardCharsets.UTF_8)); + } + private CompletionStage failed(String message) { return CompletableFuture.failedFuture(new IllegalStateException(message)); } diff --git a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java index db1698cb3c..fe7f053d30 100644 --- a/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java +++ b/AdvancedCore/src/main/java/com/bencodez/advancedcore/core/reward/SharedRewardRequirement.java @@ -1,8 +1,40 @@ package com.bencodez.advancedcore.core.reward; +import java.util.Objects; import java.util.concurrent.CompletionStage; @FunctionalInterface public interface SharedRewardRequirement { + enum Outcome { + PASS, + FAIL, + RETRY + } + + /** Existing boolean contract: false is a permanent requirement failure. */ CompletionStage test(SharedRewardContext context); + + /** Adapters with retryable requirements override this outcome instead of collapsing retry into false. */ + default CompletionStage evaluate(SharedRewardContext context) { + CompletionStage stage = test(context); + return stage == null ? null : stage.thenApply(passed -> Boolean.TRUE.equals(passed) ? Outcome.PASS : Outcome.FAIL); + } + + /** Wrap an existing boolean requirement whose false result must remain pending for a later retry. */ + static SharedRewardRequirement retryable(SharedRewardRequirement delegate) { + Objects.requireNonNull(delegate, "delegate"); + return new SharedRewardRequirement() { + @Override + public CompletionStage test(SharedRewardContext context) { + return delegate.test(context); + } + + @Override + public CompletionStage evaluate(SharedRewardContext context) { + CompletionStage stage = delegate.test(context); + return stage == null ? null : stage.thenApply(passed -> + Boolean.TRUE.equals(passed) ? Outcome.PASS : Outcome.RETRY); + } + }; + } } diff --git a/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardReviewFollowupTest.java b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardReviewFollowupTest.java new file mode 100644 index 0000000000..1fea2d76b0 --- /dev/null +++ b/AdvancedCore/src/test/java/com/bencodez/advancedcore/tests/rewards/SharedRewardReviewFollowupTest.java @@ -0,0 +1,105 @@ +package com.bencodez.advancedcore.tests.rewards; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import com.bencodez.advancedcore.core.reward.*; + +class SharedRewardReviewFollowupTest { + @Test + void retryableRequirementDefersWithoutPersistingARejectDecisionAndIsRetested() { + AtomicBoolean ready = new AtomicBoolean(false); + AtomicInteger executions = new AtomicInteger(); + MemoryDurability durability = new MemoryDurability(); + SharedRewardPlan plan = new SharedRewardPlan("retry", 1.0, Duration.ZERO, + List.of(SharedRewardRequirement.retryable(ctx -> CompletableFuture.completedFuture(ready.get()))), + List.of(new SharedRewardStep("work", false, (ctx, path) -> { + executions.incrementAndGet(); + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }))).withDefinitionFingerprint("retry-v1"); + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform()); + SharedRewardContext context = context(); + + assertEquals(SharedRewardResult.DEFERRED, orchestrator.execute(plan, context, durability).toCompletableFuture().join()); + assertNull(durability.progress); + assertEquals(1, durability.deferrals); + assertEquals(0, executions.get()); + ready.set(true); + assertEquals(SharedRewardResult.COMPLETED, orchestrator.execute(plan, context, durability).toCompletableFuture().join()); + assertEquals(1, executions.get()); + } + + @Test + void reservedCharactersInPlanAndStepIdsCannotProduceTheSameNestedPath() { + SharedRewardOrchestrator orchestrator = new SharedRewardOrchestrator(platform()); + SharedRewardContext context = context(); + ArrayList childPaths = new ArrayList<>(); + SharedRewardPlan childA = SharedRewardPlan.immediate("0/y:1/z", List.of( + new SharedRewardStep("leaf", false, (ctx, path) -> { + childPaths.add(path); + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }))); + SharedRewardPlan childB = SharedRewardPlan.immediate("z", List.of( + new SharedRewardStep("leaf", false, (ctx, path) -> { + childPaths.add(path); + return CompletableFuture.completedFuture(SharedRewardResult.COMPLETED); + }))); + SharedRewardPlan parent = SharedRewardPlan.immediate("P", List.of( + new SharedRewardStep("x", false, + (ctx, path) -> orchestrator.executeNested(childA, ctx, SharedRewardDurability.NONE, path)), + new SharedRewardStep("x:0/0/y", false, + (ctx, path) -> orchestrator.executeNested(childB, ctx, SharedRewardDurability.NONE, path)))); + + assertEquals(SharedRewardResult.COMPLETED, + orchestrator.execute(parent, context, SharedRewardDurability.NONE).toCompletableFuture().join()); + assertEquals(2, childPaths.size()); + assertNotEquals(childPaths.get(0), childPaths.get(1)); + } + + private static SharedRewardContext context() { + return new SharedRewardContext(UUID.randomUUID(), "Ben", new HashMap<>()); + } + + private static SharedRewardPlatform platform() { + return new SharedRewardPlatform() { + public boolean isOnline(UUID userId) { return true; } + public double nextChanceRoll() { return 0.0; } + public CompletionStage delay(Duration delay, + java.util.function.Supplier> operation) { return operation.get(); } + public boolean isShuttingDown() { return false; } + }; + } + + private static final class MemoryDurability implements SharedRewardDurability { + SharedRewardProgress progress; + int deferrals; + public int completedSteps(String path) { return progress == null ? 0 : progress.completedSteps(); } + public SharedRewardProgress loadProgress(String path) { return progress; } + public CompletionStage begin(String path, SharedRewardProgress proposed) { + if (progress == null) progress = proposed; + return CompletableFuture.completedFuture(progress); + } + public CompletionStage checkpoint(String path, int completed, SharedRewardContext context) { + progress = progress.advance(completed, context); + return CompletableFuture.completedFuture(null); + } + public CompletionStage defer(String path, int next, SharedRewardContext context) { + deferrals++; + return CompletableFuture.completedFuture(null); + } + public boolean durable() { return true; } + } +}