From 555cabe3915ce017e2a0804a6fb16be89b449ef1 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:46:40 -0600 Subject: [PATCH 01/19] Add platform-neutral vote input --- .../core/vote/SharedVoteInput.java | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java new file mode 100644 index 000000000..c4a99575b --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java @@ -0,0 +1,27 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; +import java.util.UUID; + +/** + * An already-accepted vote entering the shared processing path. Native ingress, + * proxy/global duplicate filtering, service-site validation and security checks + * remain upstream and are intentionally not reimplemented here. + */ +public record SharedVoteInput(UUID voteId, String playerName, String serviceSite, long voteTime, + boolean realVote, boolean addTotals, boolean proxyVote, boolean wasOnline) { + public SharedVoteInput { + Objects.requireNonNull(voteId, "voteId"); + Objects.requireNonNull(playerName, "playerName"); + Objects.requireNonNull(serviceSite, "serviceSite"); + if (playerName.isBlank()) { + throw new IllegalArgumentException("playerName cannot be blank"); + } + if (serviceSite.isBlank()) { + throw new IllegalArgumentException("serviceSite cannot be blank"); + } + if (voteTime < 0) { + throw new IllegalArgumentException("voteTime cannot be negative"); + } + } +} From 2396efc0ce8ed285aa770ca643f93bf6973a5bdc Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:46:45 -0600 Subject: [PATCH 02/19] Add shared vote identity --- .../votingplugin/core/vote/SharedVoteIdentity.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentity.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentity.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentity.java new file mode 100644 index 000000000..b5f835317 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentity.java @@ -0,0 +1,14 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; +import java.util.UUID; + +public record SharedVoteIdentity(UUID uuid, String playerName, boolean online) { + public SharedVoteIdentity { + Objects.requireNonNull(uuid, "uuid"); + Objects.requireNonNull(playerName, "playerName"); + if (playerName.isBlank()) { + throw new IllegalArgumentException("playerName cannot be blank"); + } + } +} From 7c745354f320af5593eb83c90123a3318ea90a7c Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:46:49 -0600 Subject: [PATCH 03/19] Add shared vote identity resolver --- .../core/vote/SharedVoteIdentityResolver.java | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentityResolver.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentityResolver.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentityResolver.java new file mode 100644 index 000000000..bbb95d2f4 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteIdentityResolver.java @@ -0,0 +1,9 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.concurrent.CompletionStage; + +/** Adapter to the existing AdvancedCore/user identity services. */ +@FunctionalInterface +public interface SharedVoteIdentityResolver { + CompletionStage resolve(SharedVoteInput input); +} From 83305bfd87b3c9bdc3bdcf1e7707c469887e521c Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:46:53 -0600 Subject: [PATCH 04/19] Add shared vote persistence mutation --- .../core/vote/SharedVoteMutation.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteMutation.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteMutation.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteMutation.java new file mode 100644 index 000000000..bec276506 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteMutation.java @@ -0,0 +1,16 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; +import java.util.UUID; + +/** + * Describes the logical vote mutation. The AdvancedCore-facing storage adapter + * owns actual keys, cache/queue ordering, point hooks/caps, and persistence. + */ +public record SharedVoteMutation(UUID voteId, String serviceSite, long voteTime, + boolean countTotals, boolean awardConfiguredPoints) { + public SharedVoteMutation { + Objects.requireNonNull(voteId, "voteId"); + Objects.requireNonNull(serviceSite, "serviceSite"); + } +} From ea53c5c42e8b77a4df94ac5f0eadca5ad75977b3 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:46:57 -0600 Subject: [PATCH 05/19] Add shared vote user snapshot --- .../votingplugin/core/vote/SharedVoteUserSnapshot.java | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserSnapshot.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserSnapshot.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserSnapshot.java new file mode 100644 index 000000000..c1ac9bace --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserSnapshot.java @@ -0,0 +1,6 @@ +package com.bencodez.votingplugin.core.vote; + +/** Durable state returned after the vote mutation has been applied. */ +public record SharedVoteUserSnapshot(int allTimeTotal, int monthTotal, int weeklyTotal, + int dailyTotal, int points) { +} From d0c185c4762e5bbec2ff2844f59278bee9402e21 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:47:03 -0600 Subject: [PATCH 06/19] Add AdvancedCore-facing vote user services port --- .../core/vote/SharedVoteUserServices.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java new file mode 100644 index 000000000..604b88be3 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java @@ -0,0 +1,16 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.UUID; +import java.util.concurrent.CompletionStage; + +/** + * Port for AdvancedCore's shared user/cache/storage runtime. VotingPlugin does + * not own another SQL layer, user cache, or queued-write implementation here. + */ +public interface SharedVoteUserServices { + /** Completes only after the logical vote mutation has been durably flushed. */ + CompletionStage persistVote(SharedVoteIdentity identity, SharedVoteMutation mutation); + + /** Used by restart/replay consumers to inspect the same persisted user. */ + CompletionStage load(UUID uuid); +} From 6bed0f0857cd1e925a86ded1c7db119f5c3c90c5 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:47:07 -0600 Subject: [PATCH 07/19] Add AdvancedCore-facing reward services port --- .../core/vote/SharedVoteRewardServices.java | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java new file mode 100644 index 000000000..f3b025533 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java @@ -0,0 +1,15 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.concurrent.CompletionStage; + +/** + * Port for AdvancedCore shared reward orchestration. Implementations own native + * command/message/item/effect adapters and durable offline delivery. + */ +public interface SharedVoteRewardServices { + CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot persistedState); + + CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot persistedState); +} From e57adc8b107bc4f44fb9dac1fb6106dde3c4209b Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:47:22 -0600 Subject: [PATCH 08/19] Add shared vote processing policy --- .../core/vote/SharedVotePolicy.java | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVotePolicy.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVotePolicy.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVotePolicy.java new file mode 100644 index 000000000..824bcf9e2 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVotePolicy.java @@ -0,0 +1,29 @@ +package com.bencodez.votingplugin.core.vote; + +/** + * Platform-neutral subset of the existing vote-processing configuration. + * Native adapters should populate it from the current VotingPlugin config. + */ +public record SharedVotePolicy(boolean countFakeVotes, boolean addTotals, + boolean addTotalsOffline, boolean processRewards, boolean giveOfflineRewards) { + + boolean shouldApplyConfiguredVoteMutation(SharedVoteInput input) { + return input.addTotals() && (input.realVote() || countFakeVotes); + } + + boolean shouldCountTotals(SharedVoteInput input, boolean online) { + return shouldApplyConfiguredVoteMutation(input) && addTotals && (addTotalsOffline || online); + } + + boolean shouldAwardConfiguredPoints(SharedVoteInput input) { + // Existing Bukkit behavior awards configured vote points when the accepted + // vote is countable even if Config.AddTotals itself is disabled. + return shouldApplyConfiguredVoteMutation(input); + } + + boolean shouldExecuteRewardsNow(SharedVoteInput input, boolean online) { + // Proxy votes preserve the existing force-processing behavior. For native + // votes, ProcessRewards and per-site offline eligibility remain authoritative. + return input.proxyVote() || (processRewards && (online || giveOfflineRewards)); + } +} From 3047c9faa2de08d2be4ff65e7a7a107a164ad35c Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:47:26 -0600 Subject: [PATCH 09/19] Add shared vote processing result --- .../core/vote/SharedVoteProcessingResult.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessingResult.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessingResult.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessingResult.java new file mode 100644 index 000000000..4262fc005 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessingResult.java @@ -0,0 +1,17 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; + +public record SharedVoteProcessingResult(SharedVoteIdentity identity, SharedVoteUserSnapshot persistedState, + RewardDisposition rewardDisposition) { + public SharedVoteProcessingResult { + Objects.requireNonNull(identity, "identity"); + Objects.requireNonNull(persistedState, "persistedState"); + Objects.requireNonNull(rewardDisposition, "rewardDisposition"); + } + + public enum RewardDisposition { + EXECUTED, + DEFERRED + } +} From 429e624d9459ca2e29eba73bf0e876901e7a4d80 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:47:35 -0600 Subject: [PATCH 10/19] Add platform-neutral vote processor --- .../core/vote/SharedVoteProcessor.java | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java new file mode 100644 index 000000000..4b0d5aa81 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java @@ -0,0 +1,68 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +/** + * Headless vote -> identity -> durable user mutation -> reward path. + * + *

Ingress validation, duplicate suppression, proxy/global coordination and + * VoteSite lookup remain in their existing adapters. This processor starts only + * after a vote has been accepted by those contracts.

+ */ +public final class SharedVoteProcessor { + private final SharedVoteIdentityResolver identities; + private final SharedVoteUserServices users; + private final SharedVoteRewardServices rewards; + + public SharedVoteProcessor(SharedVoteIdentityResolver identities, SharedVoteUserServices users, + SharedVoteRewardServices rewards) { + this.identities = Objects.requireNonNull(identities, "identities"); + this.users = Objects.requireNonNull(users, "users"); + this.rewards = Objects.requireNonNull(rewards, "rewards"); + } + + public CompletionStage process(SharedVoteInput input, SharedVotePolicy policy) { + Objects.requireNonNull(input, "input"); + Objects.requireNonNull(policy, "policy"); + + CompletionStage resolution = identities.resolve(input); + if (resolution == null) { + return CompletableFuture.failedFuture(new IllegalStateException("Identity resolver returned null stage")); + } + + return resolution.thenCompose(identity -> { + if (identity == null) { + return CompletableFuture.failedFuture(new IllegalStateException("Identity resolver returned null identity")); + } + boolean online = input.proxyVote() ? input.wasOnline() : identity.online(); + SharedVoteMutation mutation = new SharedVoteMutation(input.voteId(), input.serviceSite(), input.voteTime(), + policy.shouldCountTotals(input, online), policy.shouldAwardConfiguredPoints(input)); + + CompletionStage persistence = users.persistVote(identity, mutation); + if (persistence == null) { + return CompletableFuture.failedFuture(new IllegalStateException("User services returned null persistence stage")); + } + + return persistence.thenCompose(persisted -> { + if (persisted == null) { + return CompletableFuture.failedFuture( + new IllegalStateException("User services returned null persisted state")); + } + boolean executeNow = policy.shouldExecuteRewardsNow(input, online); + CompletionStage rewardCompletion = executeNow + ? rewards.executeVoteRewards(input, identity, persisted) + : rewards.deferVoteRewards(input, identity, persisted); + if (rewardCompletion == null) { + return CompletableFuture.failedFuture(new IllegalStateException( + "Reward services returned null " + (executeNow ? "execution" : "deferral") + " stage")); + } + SharedVoteProcessingResult.RewardDisposition disposition = executeNow + ? SharedVoteProcessingResult.RewardDisposition.EXECUTED + : SharedVoteProcessingResult.RewardDisposition.DEFERRED; + return rewardCompletion.thenApply(ignored -> new SharedVoteProcessingResult(identity, persisted, disposition)); + }); + }); + } +} From f4d39a76b115c756ff562f6a37a303156d88f048 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 18:48:04 -0600 Subject: [PATCH 11/19] Test headless vote persistence and rewards across restart --- .../vote/SharedVoteProcessorEndToEndTest.java | 244 ++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java new file mode 100644 index 000000000..5d1dcf53b --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java @@ -0,0 +1,244 @@ +package com.bencodez.votingplugin.core.vote; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class SharedVoteProcessorEndToEndTest { + @TempDir + Path tempDir; + + @Test + void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { + UUID uuid = UUID.randomUUID(); + Path storeFile = tempDir.resolve("users.properties"); + ArrayList rewards = new ArrayList<>(); + SharedVotePolicy policy = new SharedVotePolicy(false, true, true, true, true); + + FileBackedUserServices firstStore = new FileBackedUserServices(storeFile, 10); + SharedVoteProcessor firstRuntime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), + firstStore, new RecordingRewards(rewards)); + + SharedVoteProcessingResult first = firstRuntime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 1000L, true, true, false, true), + policy).toCompletableFuture().join(); + + assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 10), first.persistedState()); + assertEquals(SharedVoteProcessingResult.RewardDisposition.EXECUTED, first.rewardDisposition()); + assertTrue(Files.isRegularFile(storeFile)); + assertEquals(List.of("command:say Thanks Ben:1", "message:Thanks Ben:10"), rewards); + + // Simulate a process restart: construct new storage/core objects and reload the + // same durable file before processing a vote while the user is offline. + FileBackedUserServices restartedStore = new FileBackedUserServices(storeFile, 10); + assertEquals(first.persistedState(), restartedStore.load(uuid).toCompletableFuture().join()); + SharedVoteProcessor restartedRuntime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), + restartedStore, new RecordingRewards(rewards)); + + SharedVoteProcessingResult second = restartedRuntime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 2000L, true, true, false, false), + policy).toCompletableFuture().join(); + + assertEquals(new SharedVoteUserSnapshot(2, 2, 2, 2, 20), second.persistedState()); + assertEquals(SharedVoteProcessingResult.RewardDisposition.EXECUTED, second.rewardDisposition()); + assertEquals(List.of( + "command:say Thanks Ben:1", "message:Thanks Ben:10", + "command:say Thanks Ben:2", "message:Thanks Ben:20"), rewards); + + FileBackedUserServices secondRestart = new FileBackedUserServices(storeFile, 10); + assertEquals(second.persistedState(), secondRestart.load(uuid).toCompletableFuture().join()); + } + + @Test + void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { + UUID uuid = UUID.randomUUID(); + ArrayList rewards = new ArrayList<>(); + FileBackedUserServices store = new FileBackedUserServices(tempDir.resolve("deferred.properties"), 3); + SharedVoteProcessor runtime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), + store, new RecordingRewards(rewards)); + + SharedVoteProcessingResult result = runtime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3000L, true, true, false, false), + new SharedVotePolicy(false, true, true, true, false)).toCompletableFuture().join(); + + assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 3), result.persistedState()); + assertEquals(SharedVoteProcessingResult.RewardDisposition.DEFERRED, result.rewardDisposition()); + assertEquals(List.of("defer:ExampleSite:1"), rewards); + } + + @Test + void persistenceFailurePreventsRewardExecution() { + ArrayList rewards = new ArrayList<>(); + SharedVoteUserServices failing = new SharedVoteUserServices() { + @Override + public CompletionStage persistVote(SharedVoteIdentity identity, + SharedVoteMutation mutation) { + return CompletableFuture.failedFuture(new IllegalStateException("storage failed")); + } + + @Override + public CompletionStage load(UUID uuid) { + return CompletableFuture.failedFuture(new IllegalStateException("storage failed")); + } + }; + SharedVoteProcessor runtime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(UUID.randomUUID(), "Ben", true)), + failing, new RecordingRewards(rewards)); + + assertThrows(CompletionException.class, () -> runtime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 4000L, true, true, false, true), + new SharedVotePolicy(false, true, true, true, true)).toCompletableFuture().join()); + assertTrue(rewards.isEmpty()); + } + + @Test + void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { + UUID uuid = UUID.randomUUID(); + FileBackedUserServices store = new FileBackedUserServices(tempDir.resolve("policy.properties"), 5); + SharedVoteProcessor runtime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), + store, new RecordingRewards(new ArrayList<>())); + + SharedVoteProcessingResult ignoredFake = runtime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 5000L, false, true, false, true), + new SharedVotePolicy(false, true, true, true, true)).toCompletableFuture().join(); + assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 0), ignoredFake.persistedState()); + + SharedVoteProcessingResult countedFake = runtime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 6000L, false, true, false, true), + new SharedVotePolicy(true, false, true, true, true)).toCompletableFuture().join(); + // Config.AddTotals=false suppresses totals, but current Bukkit behavior still + // awards configured points when the event itself allows totals. + assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 5), countedFake.persistedState()); + } + + private static final class RecordingRewards implements SharedVoteRewardServices { + private final List events; + + private RecordingRewards(List events) { + this.events = events; + } + + @Override + public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot persistedState) { + events.add("command:say Thanks " + identity.playerName() + ":" + persistedState.allTimeTotal()); + events.add("message:Thanks " + identity.playerName() + ":" + persistedState.points()); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot persistedState) { + events.add("defer:" + input.serviceSite() + ":" + persistedState.allTimeTotal()); + return CompletableFuture.completedFuture(null); + } + } + + /** Test-only durable implementation; production persistence remains an AdvancedCore adapter. */ + private static final class FileBackedUserServices implements SharedVoteUserServices { + private final Path file; + private final int configuredPoints; + + private FileBackedUserServices(Path file, int configuredPoints) { + this.file = file; + this.configuredPoints = configuredPoints; + } + + @Override + public synchronized CompletionStage persistVote(SharedVoteIdentity identity, + SharedVoteMutation mutation) { + try { + Properties properties = read(); + String prefix = identity.uuid() + "."; + int all = integer(properties, prefix + "all"); + int month = integer(properties, prefix + "month"); + int week = integer(properties, prefix + "week"); + int day = integer(properties, prefix + "day"); + int points = integer(properties, prefix + "points"); + if (mutation.countTotals()) { + all++; + month++; + week++; + day++; + } + if (mutation.awardConfiguredPoints()) { + points += configuredPoints; + } + properties.setProperty(prefix + "all", Integer.toString(all)); + properties.setProperty(prefix + "month", Integer.toString(month)); + properties.setProperty(prefix + "week", Integer.toString(week)); + properties.setProperty(prefix + "day", Integer.toString(day)); + properties.setProperty(prefix + "points", Integer.toString(points)); + properties.setProperty(prefix + "lastSite", mutation.serviceSite()); + properties.setProperty(prefix + "lastTime", Long.toString(mutation.voteTime())); + properties.setProperty(prefix + "lastVoteId", mutation.voteId().toString()); + write(properties); + return CompletableFuture.completedFuture(new SharedVoteUserSnapshot(all, month, week, day, points)); + } catch (IOException e) { + return CompletableFuture.failedFuture(e); + } + } + + @Override + public synchronized CompletionStage load(UUID uuid) { + try { + Properties properties = read(); + String prefix = uuid + "."; + return CompletableFuture.completedFuture(new SharedVoteUserSnapshot( + integer(properties, prefix + "all"), integer(properties, prefix + "month"), + integer(properties, prefix + "week"), integer(properties, prefix + "day"), + integer(properties, prefix + "points"))); + } catch (IOException e) { + return CompletableFuture.failedFuture(e); + } + } + + private Properties read() throws IOException { + Properties properties = new Properties(); + if (Files.isRegularFile(file)) { + try (InputStream input = Files.newInputStream(file)) { + properties.load(input); + } + } + return properties; + } + + private void write(Properties properties) throws IOException { + Files.createDirectories(file.getParent()); + Path temporary = file.resolveSibling(file.getFileName() + ".tmp"); + try (OutputStream output = Files.newOutputStream(temporary)) { + properties.store(output, "shared vote test store"); + } + try { + Files.move(temporary, file, java.nio.file.StandardCopyOption.REPLACE_EXISTING, + java.nio.file.StandardCopyOption.ATOMIC_MOVE); + } catch (java.nio.file.AtomicMoveNotSupportedException e) { + Files.move(temporary, file, java.nio.file.StandardCopyOption.REPLACE_EXISTING); + } + } + + private static int integer(Properties properties, String key) { + return Integer.parseInt(properties.getProperty(key, "0")); + } + } +} From e231f62ee3a0d87d4d86746c8430c4f2543b4677 Mon Sep 17 00:00:00 2001 From: Ben Date: Sat, 12 Sep 2026 23:43:06 -0600 Subject: [PATCH 12/19] Make shared vote persistence and reward handoff recoverable by vote ID Replace the mutation-only processing boundary with an atomic vote mutation and pending receipt contract. Retry from the persisted receipt without recounting or re-resolving identity; recover bounded pending batches and acknowledge only after keyed reward completion or durable offline handoff. Preserve existing public methods, but fail closed for unsafe legacy adapters rather than silently recreating the persistence-to-reward crash window. Production storage/replay ownership remains with the AdvancedCore-facing adapters; no second SQL stack/cache/queue, native loader or Bukkit entry-path wiring is added. Arbitrary external command exactly-once execution is not claimed. Update the four existing end-to-end tests and add thirteen recovery, lost-acknowledgement and concurrent-retry regressions using test-only atomic file snapshots for the user transaction and keyed reward owner. Local JDK21 production compilation and all seventeen regression methods passed with a small assertion/annotation harness, not Maven/JUnit discovery. Full repository build, real JUnit and fresh packaged artifact validation must run in GitHub Actions. No live SQL/game-server integration result is claimed. --- .../core/vote/SharedVoteProcessor.java | 142 ++++++++---- .../core/vote/SharedVoteReceipt.java | 47 ++++ .../core/vote/SharedVoteRewardServices.java | 27 ++- .../core/vote/SharedVoteUserServices.java | 49 ++++- .../vote/SharedVoteProcessorEndToEndTest.java | 162 +++----------- .../vote/SharedVoteProcessorRecoveryTest.java | 177 +++++++++++++++ .../core/vote/SharedVoteTestStore.java | 207 ++++++++++++++++++ 7 files changed, 635 insertions(+), 176 deletions(-) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java create mode 100644 VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorRecoveryTest.java create mode 100644 VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java index 4b0d5aa81..197d13fd2 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java @@ -1,17 +1,24 @@ package com.bencodez.votingplugin.core.vote; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; import java.util.Objects; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; +import java.util.function.Supplier; + +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; /** - * Headless vote -> identity -> durable user mutation -> reward path. - * - *

Ingress validation, duplicate suppression, proxy/global coordination and - * VoteSite lookup remain in their existing adapters. This processor starts only - * after a vote has been accepted by those contracts.

+ * Accepted vote -> atomic user mutation/pending reward receipt -> keyed reward + * delivery -> durable acknowledgement. Ingress validation, proxy/global ownership + * and duplicate filtering remain upstream; these receipt checks make local retries + * recoverable and do not replace or bypass those ingress/security contracts. */ public final class SharedVoteProcessor { + private static final int MAX_RECOVERY_BATCH = 100; private final SharedVoteIdentityResolver identities; private final SharedVoteUserServices users; private final SharedVoteRewardServices rewards; @@ -26,43 +33,104 @@ public SharedVoteProcessor(SharedVoteIdentityResolver identities, SharedVoteUser public CompletionStage process(SharedVoteInput input, SharedVotePolicy policy) { Objects.requireNonNull(input, "input"); Objects.requireNonNull(policy, "policy"); - - CompletionStage resolution = identities.resolve(input); - if (resolution == null) { - return CompletableFuture.failedFuture(new IllegalStateException("Identity resolver returned null stage")); - } - - return resolution.thenCompose(identity -> { - if (identity == null) { - return CompletableFuture.failedFuture(new IllegalStateException("Identity resolver returned null identity")); + return call(() -> users.findVote(input.voteId()), "receipt lookup").thenCompose(existing -> { + if (existing != null) { + existing.requireInput(input); + return deliver(existing); } - boolean online = input.proxyVote() ? input.wasOnline() : identity.online(); - SharedVoteMutation mutation = new SharedVoteMutation(input.voteId(), input.serviceSite(), input.voteTime(), - policy.shouldCountTotals(input, online), policy.shouldAwardConfiguredPoints(input)); + return call(() -> identities.resolve(input), "identity resolution").thenCompose(identity -> { + if (identity == null) return failed("Identity resolver returned null identity"); + boolean online = input.proxyVote() ? input.wasOnline() : identity.online(); + SharedVoteMutation mutation = new SharedVoteMutation(input.voteId(), input.serviceSite(), input.voteTime(), + policy.shouldCountTotals(input, online), policy.shouldAwardConfiguredPoints(input)); + boolean executeNow = policy.shouldExecuteRewardsNow(input, online); + return call(() -> users.persistVoteWithReward(input, identity, mutation, executeNow), "atomic persistence") + .thenCompose(receipt -> { + if (receipt == null) return failed("User services returned null vote receipt"); + receipt.requireInput(input); + if (!receipt.identity().uuid().equals(identity.uuid())) { + return failed("Vote receipt belongs to a different resolved identity"); + } + return deliver(receipt); + }); + }); + }); + } - CompletionStage persistence = users.persistVote(identity, mutation); - if (persistence == null) { - return CompletableFuture.failedFuture(new IllegalStateException("User services returned null persistence stage")); + /** Recover one persisted occurrence without recounting or resolving the player again. */ + public CompletionStage recover(UUID voteId) { + Objects.requireNonNull(voteId, "voteId"); + return call(() -> users.findVote(voteId), "receipt lookup").thenCompose(receipt -> { + if (receipt == null || !voteId.equals(receipt.input().voteId())) { + return failed("Pending vote receipt was not found"); } + return deliver(receipt); + }); + } - return persistence.thenCompose(persisted -> { - if (persisted == null) { - return CompletableFuture.failedFuture( - new IllegalStateException("User services returned null persisted state")); - } - boolean executeNow = policy.shouldExecuteRewardsNow(input, online); - CompletionStage rewardCompletion = executeNow - ? rewards.executeVoteRewards(input, identity, persisted) - : rewards.deferVoteRewards(input, identity, persisted); - if (rewardCompletion == null) { - return CompletableFuture.failedFuture(new IllegalStateException( - "Reward services returned null " + (executeNow ? "execution" : "deferral") + " stage")); - } - SharedVoteProcessingResult.RewardDisposition disposition = executeNow - ? SharedVoteProcessingResult.RewardDisposition.EXECUTED - : SharedVoteProcessingResult.RewardDisposition.DEFERRED; - return rewardCompletion.thenApply(ignored -> new SharedVoteProcessingResult(identity, persisted, disposition)); + /** + * Bounded startup batch; individual failures remain pending and do not prevent + * other entries from recovering. The returned failed stage aggregates failures + * after the batch, so callers cannot mistake a partial recovery for success. + */ + public CompletionStage> recoverPending(int limit) { + if (limit < 1 || limit > MAX_RECOVERY_BATCH) throw new IllegalArgumentException("Recovery limit must be 1..100"); + return call(() -> users.pendingVotes(limit), "pending receipt scan").thenCompose(receipts -> { + if (receipts == null || receipts.size() > limit) return failed("Invalid pending receipt batch"); + List batch = List.copyOf(receipts); + HashSet ids = new HashSet<>(); + for (SharedVoteReceipt receipt : batch) { + if (!receipt.pending() || !ids.add(receipt.input().voteId())) return failed("Invalid pending receipt entry"); + } + List results = new ArrayList<>(); + List failures = new ArrayList<>(); + CompletionStage chain = CompletableFuture.completedFuture(null); + for (SharedVoteReceipt receipt : batch) { + chain = chain.thenCompose(ignored -> deliver(receipt).handle((result, failure) -> { + if (failure == null) results.add(result); + else failures.add(failure); + return null; + })); + } + return chain.thenCompose(ignored -> { + if (failures.isEmpty()) return CompletableFuture.completedFuture(List.copyOf(results)); + IllegalStateException failure = new IllegalStateException("Some pending vote rewards could not be recovered"); + failures.forEach(failure::addSuppressed); + return CompletableFuture.failedFuture(failure); }); }); } + + private CompletionStage deliver(SharedVoteReceipt receipt) { + if (!receipt.pending()) return CompletableFuture.completedFuture(result(receipt)); + return call(() -> rewards.deliverOnce(receipt), "keyed reward delivery").thenCompose(disposition -> { + if (disposition == null) return failed("Reward services returned null delivery disposition"); + return call(() -> users.markRewardCompleted(receipt.input().voteId(), disposition), "reward acknowledgement") + .thenApply(completed -> { + if (completed == null) throw new IllegalStateException("Missing acknowledged vote receipt"); + completed.requireSameOrigin(receipt); + if (completed.completedDisposition() != disposition) { + throw new IllegalStateException("Vote reward acknowledgement did not preserve the delivery result"); + } + return result(completed); + }); + }); + } + + private static SharedVoteProcessingResult result(SharedVoteReceipt receipt) { + return new SharedVoteProcessingResult(receipt.identity(), receipt.persistedState(), receipt.completedDisposition()); + } + + private static CompletionStage call(Supplier> operation, String name) { + try { + CompletionStage stage = operation.get(); + return stage == null ? failed("Adapter returned null stage for " + name) : stage; + } catch (Throwable failure) { + return CompletableFuture.failedFuture(failure); + } + } + + private static CompletionStage failed(String message) { + return CompletableFuture.failedFuture(new IllegalStateException(message)); + } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java new file mode 100644 index 000000000..3e5e6988d --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java @@ -0,0 +1,47 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; + +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; + +/** + * Immutable receipt keyed by voteId in the existing persistence owner. Its initial + * form is committed atomically with totals/points and the pending reward intent. + * Keep the original input, resolved UUID, counting policy and per-vote snapshot + * across retries; only the durable reward acknowledgement may change. + */ +public record SharedVoteReceipt(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteMutation mutation, + SharedVoteUserSnapshot persistedState, boolean executeRewardsNow, RewardDisposition completedDisposition) { + public SharedVoteReceipt { + Objects.requireNonNull(input, "input"); + Objects.requireNonNull(identity, "identity"); + Objects.requireNonNull(mutation, "mutation"); + Objects.requireNonNull(persistedState, "persistedState"); + if (!input.voteId().equals(mutation.voteId()) || !input.serviceSite().equals(mutation.serviceSite()) + || input.voteTime() != mutation.voteTime()) { + throw new IllegalArgumentException("Vote receipt input and mutation do not match"); + } + } + + public boolean pending() { return completedDisposition == null; } + + public SharedVoteReceipt completed(RewardDisposition disposition) { + Objects.requireNonNull(disposition, "disposition"); + if (!pending() && disposition != completedDisposition) { + throw new IllegalStateException("A completed vote receipt cannot change its reward result"); + } + return new SharedVoteReceipt(input, identity, mutation, persistedState, executeRewardsNow, disposition); + } + + public void requireInput(SharedVoteInput expected) { + if (!input.equals(expected)) throw new IllegalStateException("voteId is already bound to different vote input"); + } + + public void requireSameOrigin(SharedVoteReceipt expected) { + requireInput(expected.input()); + if (!identity.equals(expected.identity()) || !mutation.equals(expected.mutation()) + || !persistedState.equals(expected.persistedState()) || executeRewardsNow != expected.executeRewardsNow()) { + throw new IllegalStateException("Vote receipt changed while acknowledging reward delivery"); + } + } +} diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java index f3b025533..3a4f259ea 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java @@ -1,15 +1,34 @@ package com.bencodez.votingplugin.core.vote; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; -/** - * Port for AdvancedCore shared reward orchestration. Implementations own native - * command/message/item/effect adapters and durable offline delivery. - */ +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; + +/** Adapter for AdvancedCore reward orchestration and its existing durable replay owner. */ public interface SharedVoteRewardServices { CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot persistedState); CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot persistedState); + + /** + * Admit/deduplicate by receipt.input().voteId() in the existing reward owner. + * Serialize concurrent delivery/recovery for that occurrence, bind its prepared + * reward definition, preserve step checkpoints and durably remember the terminal + * disposition. A repeated call, including after restart or lost acknowledgement, + * resumes pending work or returns the same result without redoing completed work. + * + *

Complete only after actual execution/checkpoints or durable offline handoff, + * never task submission. Recheck live player availability in native adapters; + * receipt.identity().online() describes the original vote, not a current player. + * Native actions retain their documented delivery semantics; this port alone is + * not a claim of exactly-once arbitrary external commands across process death.

+ */ + default CompletionStage deliverOnce(SharedVoteReceipt receipt) { + // Existing unkeyed methods cannot safely implement durable retry implicitly. + return CompletableFuture.failedFuture(new UnsupportedOperationException( + "Shared vote rewards require keyed durable replay support")); + } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java index 604b88be3..fd0ea79ce 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java @@ -1,16 +1,57 @@ package com.bencodez.votingplugin.core.vote; +import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; + /** - * Port for AdvancedCore's shared user/cache/storage runtime. VotingPlugin does - * not own another SQL layer, user cache, or queued-write implementation here. + * Port for AdvancedCore's shared user/cache/storage owner. Implement atomic receipt + * operations there, not in a second VotingPlugin SQL stack/cache/queue. A voteId + * remains bound after completion for the ingress owner's supported replay horizon. */ public interface SharedVoteUserServices { - /** Completes only after the logical vote mutation has been durably flushed. */ + /** Legacy API retained; the recoverable processor never uses this mutation-only operation. */ CompletionStage persistVote(SharedVoteIdentity identity, SharedVoteMutation mutation); - /** Used by restart/replay consumers to inspect the same persisted user. */ CompletionStage load(UUID uuid); + + /** Read-only receipt lookup; null means absent, not an unavailable database. */ + default CompletionStage findVote(UUID voteId) { + return unsupported(); + } + + /** + * One transaction atomically applies the mutation and inserts a pending receipt, + * or returns the existing receipt without applying totals/points again. Enforce + * unique voteId and reject conflicting input or identity in the SAME transaction. + * Concurrent retries and lost commit acknowledgements must be safe. Existing + * receipts retain their original policy/snapshot; do not recompute on replay. + */ + default CompletionStage persistVoteWithReward(SharedVoteInput input, + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean executeRewardsNow) { + return unsupported(); + } + + /** + * Acknowledge only after the reward owner has durably completed delivery or + * accepted offline responsibility. Retain the receipt/deduplication key. A retry + * returns the same terminal receipt rather than changing its disposition. + */ + default CompletionStage markRewardCompleted(UUID voteId, RewardDisposition disposition) { + return unsupported(); + } + + /** Bounded SQL-backed startup/recovery scan of pending receipts, never all user rows. */ + default CompletionStage> pendingVotes(int limit) { + return unsupported(); + } + + private static CompletionStage unsupported() { + // No mutation-only fallback: it would recreate the crash window. + return CompletableFuture.failedFuture(new UnsupportedOperationException( + "Shared vote persistence requires atomic vote/receipt recovery support")); + } } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java index 5d1dcf53b..b7bebdef6 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java @@ -4,14 +4,10 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; -import java.util.Properties; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -31,10 +27,10 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { ArrayList rewards = new ArrayList<>(); SharedVotePolicy policy = new SharedVotePolicy(false, true, true, true, true); - FileBackedUserServices firstStore = new FileBackedUserServices(storeFile, 10); + SharedVoteTestStore firstStore = new SharedVoteTestStore(storeFile, 10); SharedVoteProcessor firstRuntime = new SharedVoteProcessor( input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), - firstStore, new RecordingRewards(rewards)); + firstStore, recording(firstStore, rewards)); SharedVoteProcessingResult first = firstRuntime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 1000L, true, true, false, true), @@ -47,11 +43,11 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { // Simulate a process restart: construct new storage/core objects and reload the // same durable file before processing a vote while the user is offline. - FileBackedUserServices restartedStore = new FileBackedUserServices(storeFile, 10); + SharedVoteTestStore restartedStore = new SharedVoteTestStore(storeFile, 10); assertEquals(first.persistedState(), restartedStore.load(uuid).toCompletableFuture().join()); SharedVoteProcessor restartedRuntime = new SharedVoteProcessor( input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), - restartedStore, new RecordingRewards(rewards)); + restartedStore, recording(restartedStore, rewards)); SharedVoteProcessingResult second = restartedRuntime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 2000L, true, true, false, false), @@ -63,7 +59,7 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { "command:say Thanks Ben:1", "message:Thanks Ben:10", "command:say Thanks Ben:2", "message:Thanks Ben:20"), rewards); - FileBackedUserServices secondRestart = new FileBackedUserServices(storeFile, 10); + SharedVoteTestStore secondRestart = new SharedVoteTestStore(storeFile, 10); assertEquals(second.persistedState(), secondRestart.load(uuid).toCompletableFuture().join()); } @@ -71,10 +67,10 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { UUID uuid = UUID.randomUUID(); ArrayList rewards = new ArrayList<>(); - FileBackedUserServices store = new FileBackedUserServices(tempDir.resolve("deferred.properties"), 3); + SharedVoteTestStore store = new SharedVoteTestStore(tempDir.resolve("deferred.properties"), 3); SharedVoteProcessor runtime = new SharedVoteProcessor( input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), - store, new RecordingRewards(rewards)); + store, recording(store, rewards)); SharedVoteProcessingResult result = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3000L, true, true, false, false), @@ -88,35 +84,31 @@ void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { @Test void persistenceFailurePreventsRewardExecution() { ArrayList rewards = new ArrayList<>(); - SharedVoteUserServices failing = new SharedVoteUserServices() { + SharedVoteTestStore failing = new SharedVoteTestStore(tempDir.resolve("failure.properties"), 1) { @Override - public CompletionStage persistVote(SharedVoteIdentity identity, - SharedVoteMutation mutation) { - return CompletableFuture.failedFuture(new IllegalStateException("storage failed")); - } - - @Override - public CompletionStage load(UUID uuid) { + public CompletionStage persistVoteWithReward(SharedVoteInput input, + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute) { return CompletableFuture.failedFuture(new IllegalStateException("storage failed")); } }; SharedVoteProcessor runtime = new SharedVoteProcessor( input -> CompletableFuture.completedFuture(new SharedVoteIdentity(UUID.randomUUID(), "Ben", true)), - failing, new RecordingRewards(rewards)); + failing, recording(failing, rewards)); - assertThrows(CompletionException.class, () -> runtime.process( + CompletionException failure = assertThrows(CompletionException.class, () -> runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 4000L, true, true, false, true), new SharedVotePolicy(false, true, true, true, true)).toCompletableFuture().join()); + assertEquals("storage failed", failure.getCause().getMessage()); assertTrue(rewards.isEmpty()); } @Test void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { UUID uuid = UUID.randomUUID(); - FileBackedUserServices store = new FileBackedUserServices(tempDir.resolve("policy.properties"), 5); + SharedVoteTestStore store = new SharedVoteTestStore(tempDir.resolve("policy.properties"), 5); SharedVoteProcessor runtime = new SharedVoteProcessor( input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), - store, new RecordingRewards(new ArrayList<>())); + store, recording(store, new ArrayList<>())); SharedVoteProcessingResult ignoredFake = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 5000L, false, true, false, true), @@ -131,114 +123,22 @@ void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 5), countedFake.persistedState()); } - private static final class RecordingRewards implements SharedVoteRewardServices { - private final List events; - - private RecordingRewards(List events) { - this.events = events; - } - - @Override - public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, - SharedVoteUserSnapshot persistedState) { - events.add("command:say Thanks " + identity.playerName() + ":" + persistedState.allTimeTotal()); - events.add("message:Thanks " + identity.playerName() + ":" + persistedState.points()); - return CompletableFuture.completedFuture(null); - } - - @Override - public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, - SharedVoteUserSnapshot persistedState) { - events.add("defer:" + input.serviceSite() + ":" + persistedState.allTimeTotal()); - return CompletableFuture.completedFuture(null); - } - } - - /** Test-only durable implementation; production persistence remains an AdvancedCore adapter. */ - private static final class FileBackedUserServices implements SharedVoteUserServices { - private final Path file; - private final int configuredPoints; - - private FileBackedUserServices(Path file, int configuredPoints) { - this.file = file; - this.configuredPoints = configuredPoints; - } - - @Override - public synchronized CompletionStage persistVote(SharedVoteIdentity identity, - SharedVoteMutation mutation) { - try { - Properties properties = read(); - String prefix = identity.uuid() + "."; - int all = integer(properties, prefix + "all"); - int month = integer(properties, prefix + "month"); - int week = integer(properties, prefix + "week"); - int day = integer(properties, prefix + "day"); - int points = integer(properties, prefix + "points"); - if (mutation.countTotals()) { - all++; - month++; - week++; - day++; - } - if (mutation.awardConfiguredPoints()) { - points += configuredPoints; - } - properties.setProperty(prefix + "all", Integer.toString(all)); - properties.setProperty(prefix + "month", Integer.toString(month)); - properties.setProperty(prefix + "week", Integer.toString(week)); - properties.setProperty(prefix + "day", Integer.toString(day)); - properties.setProperty(prefix + "points", Integer.toString(points)); - properties.setProperty(prefix + "lastSite", mutation.serviceSite()); - properties.setProperty(prefix + "lastTime", Long.toString(mutation.voteTime())); - properties.setProperty(prefix + "lastVoteId", mutation.voteId().toString()); - write(properties); - return CompletableFuture.completedFuture(new SharedVoteUserSnapshot(all, month, week, day, points)); - } catch (IOException e) { - return CompletableFuture.failedFuture(e); - } - } - - @Override - public synchronized CompletionStage load(UUID uuid) { - try { - Properties properties = read(); - String prefix = uuid + "."; - return CompletableFuture.completedFuture(new SharedVoteUserSnapshot( - integer(properties, prefix + "all"), integer(properties, prefix + "month"), - integer(properties, prefix + "week"), integer(properties, prefix + "day"), - integer(properties, prefix + "points"))); - } catch (IOException e) { - return CompletableFuture.failedFuture(e); - } - } - - private Properties read() throws IOException { - Properties properties = new Properties(); - if (Files.isRegularFile(file)) { - try (InputStream input = Files.newInputStream(file)) { - properties.load(input); - } - } - return properties; - } - - private void write(Properties properties) throws IOException { - Files.createDirectories(file.getParent()); - Path temporary = file.resolveSibling(file.getFileName() + ".tmp"); - try (OutputStream output = Files.newOutputStream(temporary)) { - properties.store(output, "shared vote test store"); - } - try { - Files.move(temporary, file, java.nio.file.StandardCopyOption.REPLACE_EXISTING, - java.nio.file.StandardCopyOption.ATOMIC_MOVE); - } catch (java.nio.file.AtomicMoveNotSupportedException e) { - Files.move(temporary, file, java.nio.file.StandardCopyOption.REPLACE_EXISTING); + private static SharedVoteRewardServices recording(SharedVoteTestStore store, List events) { + return new SharedVoteRewardServices() { + @Override + public CompletionStage deliverOnce(SharedVoteReceipt receipt) { + return store.deliverOnce(receipt).thenApply(result -> { + events.clear(); + events.addAll(store.events()); + return result; + }); } - } - - private static int integer(Properties properties, String key) { - return Integer.parseInt(properties.getProperty(key, "0")); - } + @Override + public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot state) { throw new AssertionError("Unkeyed execution"); } + @Override + public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + SharedVoteUserSnapshot state) { throw new AssertionError("Unkeyed deferral"); } + }; } } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorRecoveryTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorRecoveryTest.java new file mode 100644 index 000000000..d9548d17a --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorRecoveryTest.java @@ -0,0 +1,177 @@ +package com.bencodez.votingplugin.core.vote; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +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 org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; + +/** Restart and acknowledgement boundaries with test-only file transaction/replay owners. */ +@Timeout(15) +class SharedVoteProcessorRecoveryTest { + @TempDir Path directory; + private static final UUID USER = UUID.fromString("9837d441-a4d6-461f-aa86-17958c01bc8c"); + private static final SharedVotePolicy POLICY = new SharedVotePolicy(false, true, true, true, true); + + @Test void crashAfterCommitRecoversPendingRewardWithoutAnotherIngressVote() { + SharedVoteTestStore store = store(); + store.failAfterCommit = true; + SharedVoteInput input = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(input, POLICY).toCompletableFuture().join()); + assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 5), store.load(USER).toCompletableFuture().join()); + assertTrue(store.events().isEmpty()); + SharedVoteTestStore reopened = store(); + SharedVoteProcessor recovered = new SharedVoteProcessor(vote -> { throw new AssertionError("Do not resolve a committed vote again"); }, reopened, reopened); + assertEquals(1, recovered.recoverPending(10).toCompletableFuture().join().size()); + assertEquals(2, reopened.events().size()); + assertEquals(1, reopened.load(USER).toCompletableFuture().join().allTimeTotal()); + assertTrue(reopened.pendingVotes(10).toCompletableFuture().join().isEmpty()); + } + + @Test void lostCommitAckRetryUsesOriginalIdentityPolicyAndSnapshot() { + SharedVoteTestStore store = store(); + store.failAfterCommit = true; + SharedVoteInput input = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(input, POLICY).toCompletableFuture().join()); + SharedVoteTestStore reopened = store(); + SharedVoteProcessor retry = new SharedVoteProcessor(vote -> { throw new AssertionError("Do not re-resolve"); }, reopened, reopened); + SharedVoteProcessingResult result = retry.process(input, + new SharedVotePolicy(false, false, false, false, false)).toCompletableFuture().join(); + assertEquals(RewardDisposition.EXECUTED, result.rewardDisposition()); + assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 5), result.persistedState()); + } + + @Test void failureBeforeCommitCreatesNeitherMutationNorRewardWork() { + SharedVoteTestStore store = store(); + store.failBeforeCommit = true; + SharedVoteInput input = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(input, POLICY).toCompletableFuture().join()); + assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 0), store.load(USER).toCompletableFuture().join()); + assertEquals(null, store.findVote(input.voteId()).toCompletableFuture().join()); + assertTrue(store.events().isEmpty()); + runtime(store).process(input, POLICY).toCompletableFuture().join(); + assertEquals(1, store.mutationCount); + } + + @Test void failedRewardDeliveryRemainsPendingForRestart() { + SharedVoteTestStore store = store(); + store.failBeforeDelivery = true; + SharedVoteInput input = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(input, POLICY).toCompletableFuture().join()); + assertTrue(store.findVote(input.voteId()).toCompletableFuture().join().pending()); + SharedVoteTestStore reopened = store(); + runtime(reopened).recover(input.voteId()).toCompletableFuture().join(); + assertEquals(2, reopened.events().size()); + assertEquals(1, reopened.load(USER).toCompletableFuture().join().allTimeTotal()); + } + + @Test void lostRewardOwnerAcknowledgementDoesNotRepeatCompletedEffects() { + lostAck(true, false, false); + } + @Test void failedReceiptAcknowledgementDoesNotRepeatCompletedEffects() { + lostAck(false, true, false); + } + @Test void lostTerminalReceiptAcknowledgementDoesNotRepeatCompletedEffects() { + lostAck(false, false, true); + } + private void lostAck(boolean reward, boolean beforeMark, boolean afterMark) { + SharedVoteTestStore store = store(); + store.failAfterDelivery = reward; + store.failBeforeMark = beforeMark; + store.failAfterMark = afterMark; + SharedVoteInput input = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(input, POLICY).toCompletableFuture().join()); + assertEquals(2, store.events().size()); + SharedVoteTestStore reopened = store(); + SharedVoteProcessingResult result = runtime(reopened).process(input, POLICY).toCompletableFuture().join(); + assertEquals(RewardDisposition.EXECUTED, result.rewardDisposition()); + assertEquals(2, reopened.events().size()); + assertEquals(1, reopened.load(USER).toCompletableFuture().join().allTimeTotal()); + } + + @Test void duplicateVoteIdWithDifferentInputIsRejectedBeforeAnotherMutation() { + SharedVoteTestStore store = store(); + SharedVoteInput input = input(); + runtime(store).process(input, POLICY).toCompletableFuture().join(); + SharedVoteInput conflict = new SharedVoteInput(input.voteId(), "Other", "Example", input.voteTime(), true, true, false, true); + assertThrows(CompletionException.class, () -> runtime(store).process(conflict, POLICY).toCompletableFuture().join()); + assertEquals(1, store.mutationCount); + assertEquals(2, store.events().size()); + } + + @Test void concurrentDuplicateSubmissionsShareThePersistenceAndRewardOwners() throws Exception { + SharedVoteTestStore store = store(); + SharedVoteInput input = input(); + var workers = Executors.newFixedThreadPool(4); + try { + List> tasks = new ArrayList<>(); + for (int i = 0; i < 12; i++) tasks.add(workers.submit(() -> runtime(store).process(input, POLICY).toCompletableFuture().join())); + for (var task : tasks) task.get(5, TimeUnit.SECONDS); + assertEquals(1, store.mutationCount); + assertEquals(2, store.events().size()); + } finally { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test void unacknowledgedAtomicPersistenceDoesNotTriggerTheOriginatingRewardCall() { + SharedVoteTestStore store = store(); + store.commitAck = new CompletableFuture<>(); + var result = runtime(store).process(input(), POLICY).toCompletableFuture(); + assertFalse(result.isDone()); + assertTrue(store.events().isEmpty()); + store.commitAck.complete(null); + assertEquals(RewardDisposition.EXECUTED, result.join().rewardDisposition()); + } + + @Test void offlineHandoffIsIdempotentAcrossRestartAndLostAcknowledgement() { + SharedVoteTestStore store = store(); + store.failBeforeMark = true; + SharedVoteInput input = input(); + SharedVoteProcessor offline = new SharedVoteProcessor(vote -> CompletableFuture.completedFuture(new SharedVoteIdentity(USER, "Ben", false)), store, store); + SharedVotePolicy defer = new SharedVotePolicy(false, true, true, true, false); + assertThrows(CompletionException.class, () -> offline.process(input, defer).toCompletableFuture().join()); + SharedVoteTestStore reopened = store(); + assertEquals(RewardDisposition.DEFERRED, runtime(reopened).recover(input.voteId()).toCompletableFuture().join().rewardDisposition()); + assertEquals(List.of("defer:Example:1"), reopened.events()); + } + + @Test void recoveryRejectsInvalidBoundsWithoutScanningStorage() { + SharedVoteTestStore store = store(); + assertThrows(IllegalArgumentException.class, () -> runtime(store).recoverPending(0)); + assertThrows(IllegalArgumentException.class, () -> runtime(store).recoverPending(101)); + } + + @Test void aFailedRecoveryEntryDoesNotBlockOtherPendingEntries() { + SharedVoteTestStore store = store(); + store.failBeforeDelivery = true; + SharedVoteInput first = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(first, POLICY).toCompletableFuture().join()); + store.failBeforeDelivery = true; + SharedVoteInput second = input(); + assertThrows(CompletionException.class, () -> runtime(store).process(second, POLICY).toCompletableFuture().join()); + store.failBeforeDelivery = true; + assertThrows(CompletionException.class, () -> runtime(store).recoverPending(10).toCompletableFuture().join()); + assertEquals(1, store.pendingVotes(10).toCompletableFuture().join().size()); + assertEquals(2, store.events().size()); + } + + private SharedVoteTestStore store() { return new SharedVoteTestStore(directory.resolve("users.properties"), 5); } + private static SharedVoteProcessor runtime(SharedVoteTestStore store) { + return new SharedVoteProcessor(vote -> CompletableFuture.completedFuture(new SharedVoteIdentity(USER, "Ben", true)), store, store); + } + private static SharedVoteInput input() { return new SharedVoteInput(UUID.randomUUID(), "Ben", "Example", 1000, true, true, false, true); } +} diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java new file mode 100644 index 000000000..774174268 --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java @@ -0,0 +1,207 @@ +package com.bencodez.votingplugin.core.vote; + +import java.io.IOException; +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.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; + +/** + * TEST ONLY: atomic file snapshots model the existing user transaction and a + * separate keyed reward owner. Recorded command/message strings are test effects, + * not live server commands or an exactly-once external-effect implementation. + */ +class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardServices { + final Path file; + final Path rewardFile; + final int pointsPerVote; + boolean failBeforeCommit, failAfterCommit, failBeforeMark, failAfterMark; + boolean failBeforeDelivery, failAfterDelivery; + CompletableFuture commitAck = CompletableFuture.completedFuture(null); + int mutationCount; + + SharedVoteTestStore(Path file, int pointsPerVote) { + this.file = file; + this.rewardFile = file.resolveSibling(file.getFileName() + ".rewards"); + this.pointsPerVote = pointsPerVote; + } + + @Override public CompletionStage persistVote(SharedVoteIdentity user, SharedVoteMutation mutation) { + return CompletableFuture.failedFuture(new AssertionError("Unsafe mutation-only API was called")); + } + + @Override public synchronized CompletionStage load(UUID uuid) { + try { return CompletableFuture.completedFuture(snapshot(read(file), uuid + ".")); } + catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public synchronized CompletionStage findVote(UUID id) { + try { return CompletableFuture.completedFuture(receipt(read(file), id)); } + catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public synchronized CompletionStage persistVoteWithReward(SharedVoteInput input, + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute) { + try { + Properties properties = read(file); + SharedVoteReceipt existing = receipt(properties, input.voteId()); + if (existing != null) { + existing.requireInput(input); + if (!existing.identity().uuid().equals(identity.uuid())) throw new IllegalStateException("Conflicting user"); + return CompletableFuture.completedFuture(existing); + } + if (failBeforeCommit) { failBeforeCommit = false; throw new IOException("before commit"); } + String prefix = identity.uuid() + "."; + SharedVoteUserSnapshot previous = snapshot(properties, prefix); + int increment = mutation.countTotals() ? 1 : 0; + SharedVoteUserSnapshot next = new SharedVoteUserSnapshot(previous.allTimeTotal() + increment, + previous.monthTotal() + increment, previous.weeklyTotal() + increment, previous.dailyTotal() + increment, + previous.points() + (mutation.awardConfiguredPoints() ? pointsPerVote : 0)); + putSnapshot(properties, prefix, next); + SharedVoteReceipt created = new SharedVoteReceipt(input, identity, mutation, next, execute, null); + putReceipt(properties, created); + // A SINGLE replace commits BOTH user state and the pending reward receipt. + write(file, properties); + mutationCount++; + if (failAfterCommit) { failAfterCommit = false; throw new IOException("lost commit acknowledgement"); } + return commitAck.thenApply(ignored -> created); + } catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public synchronized CompletionStage markRewardCompleted(UUID id, RewardDisposition disposition) { + try { + if (failBeforeMark) { failBeforeMark = false; throw new IOException("before receipt acknowledgement"); } + Properties properties = read(file); + SharedVoteReceipt completed = receipt(properties, id).completed(disposition); + putReceipt(properties, completed); + write(file, properties); + if (failAfterMark) { failAfterMark = false; throw new IOException("lost receipt acknowledgement"); } + return CompletableFuture.completedFuture(completed); + } catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public synchronized CompletionStage> pendingVotes(int limit) { + try { + Properties properties = read(file); + List pending = new ArrayList<>(); + for (String key : properties.stringPropertyNames().stream().sorted().toList()) { + if (key.startsWith("receipt.") && key.endsWith(".done") && properties.getProperty(key).isEmpty()) { + UUID id = UUID.fromString(key.substring("receipt.".length(), key.length() - ".done".length())); + pending.add(receipt(properties, id)); + if (pending.size() == limit) break; + } + } + return CompletableFuture.completedFuture(pending); + } catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public synchronized CompletionStage deliverOnce(SharedVoteReceipt receipt) { + try { + if (failBeforeDelivery) { failBeforeDelivery = false; throw new IOException("reward owner unavailable"); } + Properties properties = read(rewardFile); + String prefix = receipt.input().voteId() + "."; + String origin = receipt.input() + "|" + receipt.identity() + "|" + receipt.mutation() + + "|" + receipt.persistedState() + "|" + receipt.executeRewardsNow(); + if (properties.containsKey(prefix + "done")) { + if (!origin.equals(properties.getProperty(prefix + "origin"))) throw new IllegalStateException("Conflicting reward receipt"); + return CompletableFuture.completedFuture(RewardDisposition.valueOf(properties.getProperty(prefix + "done"))); + } + RewardDisposition disposition = receipt.executeRewardsNow() ? RewardDisposition.EXECUTED : RewardDisposition.DEFERRED; + if (receipt.executeRewardsNow()) { + effect(properties, "command:say Thanks " + receipt.identity().playerName() + ":" + receipt.persistedState().allTimeTotal()); + effect(properties, "message:Thanks " + receipt.identity().playerName() + ":" + receipt.persistedState().points()); + } else { + effect(properties, "defer:" + receipt.input().serviceSite() + ":" + receipt.persistedState().allTimeTotal()); + } + properties.setProperty(prefix + "origin", origin); + properties.setProperty(prefix + "done", disposition.name()); + write(rewardFile, properties); + if (failAfterDelivery) { failAfterDelivery = false; throw new IOException("lost durable reward acknowledgement"); } + return CompletableFuture.completedFuture(disposition); + } catch (Exception failure) { return CompletableFuture.failedFuture(failure); } + } + + @Override public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot snapshot) { + return CompletableFuture.failedFuture(new AssertionError("Unkeyed execution API was called")); + } + @Override public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot snapshot) { + return CompletableFuture.failedFuture(new AssertionError("Unkeyed deferral API was called")); + } + + synchronized List events() { + try { + Properties properties = read(rewardFile); + List events = new ArrayList<>(); + for (int i = 0; i < integer(properties, "effects"); i++) events.add(properties.getProperty("effect." + i)); + return events; + } catch (IOException failure) { throw new IllegalStateException(failure); } + } + private static void effect(Properties p, String value) { + int size = integer(p, "effects"); + p.setProperty("effect." + size, value); + p.setProperty("effects", Integer.toString(size + 1)); + } + private static Properties read(Path file) throws IOException { + Properties properties = new Properties(); + if (Files.exists(file)) try (InputStream input = Files.newInputStream(file)) { properties.load(input); } + return properties; + } + private static void write(Path file, Properties properties) throws IOException { + Files.createDirectories(file.getParent()); + Path temporary = file.resolveSibling(file.getFileName() + ".tmp"); + try (OutputStream out = Files.newOutputStream(temporary)) { properties.store(out, "test snapshot"); } + Files.move(temporary, file, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } + private static int integer(Properties p, String key) { return Integer.parseInt(p.getProperty(key, "0")); } + private static boolean bool(Properties p, String key) { return Boolean.parseBoolean(p.getProperty(key)); } + private static SharedVoteUserSnapshot snapshot(Properties p, String prefix) { + return new SharedVoteUserSnapshot(integer(p, prefix + "all"), integer(p, prefix + "month"), integer(p, prefix + "week"), + integer(p, prefix + "day"), integer(p, prefix + "points")); + } + private static void putSnapshot(Properties p, String prefix, SharedVoteUserSnapshot s) { + p.setProperty(prefix + "all", Integer.toString(s.allTimeTotal())); + p.setProperty(prefix + "month", Integer.toString(s.monthTotal())); + p.setProperty(prefix + "week", Integer.toString(s.weeklyTotal())); + p.setProperty(prefix + "day", Integer.toString(s.dailyTotal())); + p.setProperty(prefix + "points", Integer.toString(s.points())); + } + private static void putReceipt(Properties p, SharedVoteReceipt r) { + String k = "receipt." + r.input().voteId() + "."; + p.setProperty(k + "name", r.input().playerName()); + p.setProperty(k + "site", r.input().serviceSite()); + p.setProperty(k + "time", Long.toString(r.input().voteTime())); + p.setProperty(k + "real", Boolean.toString(r.input().realVote())); + p.setProperty(k + "add", Boolean.toString(r.input().addTotals())); + p.setProperty(k + "proxy", Boolean.toString(r.input().proxyVote())); + p.setProperty(k + "wasOnline", Boolean.toString(r.input().wasOnline())); + p.setProperty(k + "uuid", r.identity().uuid().toString()); + p.setProperty(k + "resolvedName", r.identity().playerName()); + p.setProperty(k + "online", Boolean.toString(r.identity().online())); + p.setProperty(k + "count", Boolean.toString(r.mutation().countTotals())); + p.setProperty(k + "award", Boolean.toString(r.mutation().awardConfiguredPoints())); + p.setProperty(k + "execute", Boolean.toString(r.executeRewardsNow())); + p.setProperty(k + "done", r.pending() ? "" : r.completedDisposition().name()); + putSnapshot(p, k + "state.", r.persistedState()); + } + private static SharedVoteReceipt receipt(Properties p, UUID id) { + String k = "receipt." + id + "."; + if (!p.containsKey(k + "done")) return null; + SharedVoteInput input = new SharedVoteInput(id, p.getProperty(k + "name"), p.getProperty(k + "site"), + Long.parseLong(p.getProperty(k + "time")), bool(p, k + "real"), bool(p, k + "add"), bool(p, k + "proxy"), bool(p, k + "wasOnline")); + SharedVoteIdentity identity = new SharedVoteIdentity(UUID.fromString(p.getProperty(k + "uuid")), + p.getProperty(k + "resolvedName"), bool(p, k + "online")); + SharedVoteMutation mutation = new SharedVoteMutation(id, input.serviceSite(), input.voteTime(), bool(p, k + "count"), bool(p, k + "award")); + String done = p.getProperty(k + "done"); + return new SharedVoteReceipt(input, identity, mutation, snapshot(p, k + "state."), bool(p, k + "execute"), + done.isEmpty() ? null : RewardDisposition.valueOf(done)); + } +} From 0a0a7eb1e74f8950840effba2d476383db0d0963 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 10:09:34 -0600 Subject: [PATCH 13/19] Use live online state for proxy vote totals --- .../core/vote/SharedVoteProcessor.java | 9 ++++-- .../vote/SharedVoteProcessorEndToEndTest.java | 28 ++++++++++++++++--- 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java index 197d13fd2..982e9a7b2 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java @@ -40,10 +40,13 @@ public CompletionStage process(SharedVoteInput input } return call(() -> identities.resolve(input), "identity resolution").thenCompose(identity -> { if (identity == null) return failed("Identity resolver returned null identity"); - boolean online = input.proxyVote() ? input.wasOnline() : identity.online(); + // Totals follow the user's live resolved state, matching PlayerVoteListener. + // wasOnline is retained only for proxy reward semantics. + boolean currentOnline = identity.online(); + boolean rewardOnline = input.proxyVote() ? input.wasOnline() : currentOnline; SharedVoteMutation mutation = new SharedVoteMutation(input.voteId(), input.serviceSite(), input.voteTime(), - policy.shouldCountTotals(input, online), policy.shouldAwardConfiguredPoints(input)); - boolean executeNow = policy.shouldExecuteRewardsNow(input, online); + policy.shouldCountTotals(input, currentOnline), policy.shouldAwardConfiguredPoints(input)); + boolean executeNow = policy.shouldExecuteRewardsNow(input, rewardOnline); return call(() -> users.persistVoteWithReward(input, identity, mutation, executeNow), "atomic persistence") .thenCompose(receipt -> { if (receipt == null) return failed("User services returned null vote receipt"); diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java index b7bebdef6..c3ad20cb6 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java @@ -41,8 +41,6 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { assertTrue(Files.isRegularFile(storeFile)); assertEquals(List.of("command:say Thanks Ben:1", "message:Thanks Ben:10"), rewards); - // Simulate a process restart: construct new storage/core objects and reload the - // same durable file before processing a vote while the user is offline. SharedVoteTestStore restartedStore = new SharedVoteTestStore(storeFile, 10); assertEquals(first.persistedState(), restartedStore.load(uuid).toCompletableFuture().join()); SharedVoteProcessor restartedRuntime = new SharedVoteProcessor( @@ -81,6 +79,30 @@ void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { assertEquals(List.of("defer:ExampleSite:1"), rewards); } + @Test + void proxyTotalsUseLiveStateInsteadOfHistoricalWasOnline() { + UUID uuid = UUID.randomUUID(); + SharedVotePolicy policy = new SharedVotePolicy(false, true, false, true, true); + + SharedVoteTestStore offlineStore = new SharedVoteTestStore(tempDir.resolve("proxy-offline.properties"), 2); + SharedVoteProcessor offlineRuntime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), + offlineStore, recording(offlineStore, new ArrayList<>())); + SharedVoteProcessingResult offline = offlineRuntime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3100L, true, true, true, true), + policy).toCompletableFuture().join(); + assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 2), offline.persistedState()); + + SharedVoteTestStore onlineStore = new SharedVoteTestStore(tempDir.resolve("proxy-online.properties"), 2); + SharedVoteProcessor onlineRuntime = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), + onlineStore, recording(onlineStore, new ArrayList<>())); + SharedVoteProcessingResult online = onlineRuntime.process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3200L, true, true, true, false), + policy).toCompletableFuture().join(); + assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 2), online.persistedState()); + } + @Test void persistenceFailurePreventsRewardExecution() { ArrayList rewards = new ArrayList<>(); @@ -118,8 +140,6 @@ void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { SharedVoteProcessingResult countedFake = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 6000L, false, true, false, true), new SharedVotePolicy(true, false, true, true, true)).toCompletableFuture().join(); - // Config.AddTotals=false suppresses totals, but current Bukkit behavior still - // awards configured points when the event itself allows totals. assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 5), countedFake.persistedState()); } From 64697cb4e507ffb49f4fed90dd2ba8ac9935b98b Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 10:35:10 -0600 Subject: [PATCH 14/19] Persist normalized vote time and prepared reward version --- .../core/vote/SharedVoteInput.java | 33 +++++-- .../core/vote/SharedVoteProcessor.java | 60 +++++------ .../core/vote/SharedVoteReceipt.java | 20 ++-- .../core/vote/SharedVoteRewardPlan.java | 17 ++++ .../core/vote/SharedVoteRewardServices.java | 25 ++--- .../core/vote/SharedVoteUserServices.java | 37 +++---- .../vote/SharedVoteProcessorEndToEndTest.java | 99 ++++++------------- .../vote/SharedVoteReviewFollowupTest.java | 56 +++++++++++ .../core/vote/SharedVoteTestStore.java | 29 ++++-- 9 files changed, 210 insertions(+), 166 deletions(-) create mode 100644 VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardPlan.java create mode 100644 VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java index c4a99575b..72a9a904d 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java @@ -14,14 +14,29 @@ public record SharedVoteInput(UUID voteId, String playerName, String serviceSite Objects.requireNonNull(voteId, "voteId"); Objects.requireNonNull(playerName, "playerName"); Objects.requireNonNull(serviceSite, "serviceSite"); - if (playerName.isBlank()) { - throw new IllegalArgumentException("playerName cannot be blank"); - } - if (serviceSite.isBlank()) { - throw new IllegalArgumentException("serviceSite cannot be blank"); - } - if (voteTime < 0) { - throw new IllegalArgumentException("voteTime cannot be negative"); - } + if (playerName.isBlank()) throw new IllegalArgumentException("playerName cannot be blank"); + if (serviceSite.isBlank()) throw new IllegalArgumentException("serviceSite cannot be blank"); + if (voteTime < 0) throw new IllegalArgumentException("voteTime cannot be negative"); + } + + /** PlayerVoteEvent uses zero as "now"; normalize before durable mutation/receipt creation. */ + public SharedVoteInput normalizedVoteTime(long nowEpochMillis) { + if (voteTime != 0) return this; + if (nowEpochMillis <= 0) throw new IllegalArgumentException("normalized vote time must be positive"); + return new SharedVoteInput(voteId, playerName, serviceSite, nowEpochMillis, + realVote, addTotals, proxyVote, wasOnline); + } + + /** A retry carrying the zero sentinel still identifies its already-normalized persisted receipt. */ + public boolean matchesPersisted(SharedVoteInput persisted) { + if (persisted == null) return false; + return voteId.equals(persisted.voteId()) + && playerName.equals(persisted.playerName()) + && serviceSite.equals(persisted.serviceSite()) + && (voteTime == 0 || voteTime == persisted.voteTime()) + && realVote == persisted.realVote() + && addTotals == persisted.addTotals() + && proxyVote == persisted.proxyVote() + && wasOnline == persisted.wasOnline(); } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java index 982e9a7b2..919b20eca 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java @@ -11,12 +11,6 @@ import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; -/** - * Accepted vote -> atomic user mutation/pending reward receipt -> keyed reward - * delivery -> durable acknowledgement. Ingress validation, proxy/global ownership - * and duplicate filtering remain upstream; these receipt checks make local retries - * recoverable and do not replace or bypass those ingress/security contracts. - */ public final class SharedVoteProcessor { private static final int MAX_RECOVERY_BATCH = 100; private final SharedVoteIdentityResolver identities; @@ -38,44 +32,43 @@ public CompletionStage process(SharedVoteInput input existing.requireInput(input); return deliver(existing); } - return call(() -> identities.resolve(input), "identity resolution").thenCompose(identity -> { + SharedVoteInput normalized = input.normalizedVoteTime(System.currentTimeMillis()); + return call(() -> identities.resolve(normalized), "identity resolution").thenCompose(identity -> { if (identity == null) return failed("Identity resolver returned null identity"); - // Totals follow the user's live resolved state, matching PlayerVoteListener. - // wasOnline is retained only for proxy reward semantics. boolean currentOnline = identity.online(); - boolean rewardOnline = input.proxyVote() ? input.wasOnline() : currentOnline; - SharedVoteMutation mutation = new SharedVoteMutation(input.voteId(), input.serviceSite(), input.voteTime(), - policy.shouldCountTotals(input, currentOnline), policy.shouldAwardConfiguredPoints(input)); - boolean executeNow = policy.shouldExecuteRewardsNow(input, rewardOnline); - return call(() -> users.persistVoteWithReward(input, identity, mutation, executeNow), "atomic persistence") - .thenCompose(receipt -> { - if (receipt == null) return failed("User services returned null vote receipt"); - receipt.requireInput(input); - if (!receipt.identity().uuid().equals(identity.uuid())) { - return failed("Vote receipt belongs to a different resolved identity"); - } - return deliver(receipt); + boolean rewardOnline = normalized.proxyVote() ? normalized.wasOnline() : currentOnline; + SharedVoteMutation mutation = new SharedVoteMutation(normalized.voteId(), normalized.serviceSite(), + normalized.voteTime(), policy.shouldCountTotals(normalized, currentOnline), + policy.shouldAwardConfiguredPoints(normalized)); + boolean executeNow = policy.shouldExecuteRewardsNow(normalized, rewardOnline); + return call(() -> rewards.prepareVoteRewards(normalized, identity, executeNow), "reward preparation") + .thenCompose(rewardPlan -> { + if (rewardPlan == null) return failed("Reward services returned null prepared reward plan"); + return call(() -> users.persistVoteWithReward(normalized, identity, mutation, executeNow, rewardPlan), + "atomic persistence").thenCompose(receipt -> { + if (receipt == null) return failed("User services returned null vote receipt"); + receipt.requireInput(input); + if (!receipt.identity().uuid().equals(identity.uuid())) { + return failed("Vote receipt belongs to a different resolved identity"); + } + if (!receipt.rewardPlan().equals(rewardPlan)) { + return failed("Vote receipt belongs to a different prepared reward version"); + } + return deliver(receipt); + }); }); }); }); } - /** Recover one persisted occurrence without recounting or resolving the player again. */ public CompletionStage recover(UUID voteId) { Objects.requireNonNull(voteId, "voteId"); return call(() -> users.findVote(voteId), "receipt lookup").thenCompose(receipt -> { - if (receipt == null || !voteId.equals(receipt.input().voteId())) { - return failed("Pending vote receipt was not found"); - } + if (receipt == null || !voteId.equals(receipt.input().voteId())) return failed("Pending vote receipt was not found"); return deliver(receipt); }); } - /** - * Bounded startup batch; individual failures remain pending and do not prevent - * other entries from recovering. The returned failed stage aggregates failures - * after the batch, so callers cannot mistake a partial recovery for success. - */ public CompletionStage> recoverPending(int limit) { if (limit < 1 || limit > MAX_RECOVERY_BATCH) throw new IllegalArgumentException("Recovery limit must be 1..100"); return call(() -> users.pendingVotes(limit), "pending receipt scan").thenCompose(receipts -> { @@ -90,8 +83,7 @@ public CompletionStage> recoverPending(int limi CompletionStage chain = CompletableFuture.completedFuture(null); for (SharedVoteReceipt receipt : batch) { chain = chain.thenCompose(ignored -> deliver(receipt).handle((result, failure) -> { - if (failure == null) results.add(result); - else failures.add(failure); + if (failure == null) results.add(result); else failures.add(failure); return null; })); } @@ -128,9 +120,7 @@ private static CompletionStage call(Supplier> operatio try { CompletionStage stage = operation.get(); return stage == null ? failed("Adapter returned null stage for " + name) : stage; - } catch (Throwable failure) { - return CompletableFuture.failedFuture(failure); - } + } catch (Throwable failure) { return CompletableFuture.failedFuture(failure); } } private static CompletionStage failed(String message) { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java index 3e5e6988d..492b25666 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteReceipt.java @@ -6,17 +6,19 @@ /** * Immutable receipt keyed by voteId in the existing persistence owner. Its initial - * form is committed atomically with totals/points and the pending reward intent. - * Keep the original input, resolved UUID, counting policy and per-vote snapshot - * across retries; only the durable reward acknowledgement may change. + * form is committed atomically with totals/points, pending reward intent and the + * immutable prepared reward version. Across retries only the acknowledgement changes. */ public record SharedVoteReceipt(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteMutation mutation, - SharedVoteUserSnapshot persistedState, boolean executeRewardsNow, RewardDisposition completedDisposition) { + SharedVoteUserSnapshot persistedState, boolean executeRewardsNow, SharedVoteRewardPlan rewardPlan, + RewardDisposition completedDisposition) { public SharedVoteReceipt { Objects.requireNonNull(input, "input"); Objects.requireNonNull(identity, "identity"); Objects.requireNonNull(mutation, "mutation"); Objects.requireNonNull(persistedState, "persistedState"); + Objects.requireNonNull(rewardPlan, "rewardPlan"); + if (input.voteTime() <= 0) throw new IllegalArgumentException("Persisted vote time must be normalized"); if (!input.voteId().equals(mutation.voteId()) || !input.serviceSite().equals(mutation.serviceSite()) || input.voteTime() != mutation.voteTime()) { throw new IllegalArgumentException("Vote receipt input and mutation do not match"); @@ -30,17 +32,21 @@ public SharedVoteReceipt completed(RewardDisposition disposition) { if (!pending() && disposition != completedDisposition) { throw new IllegalStateException("A completed vote receipt cannot change its reward result"); } - return new SharedVoteReceipt(input, identity, mutation, persistedState, executeRewardsNow, disposition); + return new SharedVoteReceipt(input, identity, mutation, persistedState, executeRewardsNow, rewardPlan, disposition); } public void requireInput(SharedVoteInput expected) { - if (!input.equals(expected)) throw new IllegalStateException("voteId is already bound to different vote input"); + Objects.requireNonNull(expected, "expected"); + if (!expected.matchesPersisted(input)) { + throw new IllegalStateException("voteId is already bound to different vote input"); + } } public void requireSameOrigin(SharedVoteReceipt expected) { requireInput(expected.input()); if (!identity.equals(expected.identity()) || !mutation.equals(expected.mutation()) - || !persistedState.equals(expected.persistedState()) || executeRewardsNow != expected.executeRewardsNow()) { + || !persistedState.equals(expected.persistedState()) || executeRewardsNow != expected.executeRewardsNow() + || !rewardPlan.equals(expected.rewardPlan())) { throw new IllegalStateException("Vote receipt changed while acknowledging reward delivery"); } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardPlan.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardPlan.java new file mode 100644 index 000000000..840f94db3 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardPlan.java @@ -0,0 +1,17 @@ +package com.bencodez.votingplugin.core.vote; + +import java.util.Objects; + +/** + * Immutable reference to the exact reward definition prepared for one vote. + * The version reference must continue resolving to the same archived/snapshotted + * definition after configuration reloads and process restarts. + */ +public record SharedVoteRewardPlan(String planId, String versionReference) { + public SharedVoteRewardPlan { + Objects.requireNonNull(planId, "planId"); + Objects.requireNonNull(versionReference, "versionReference"); + if (planId.isBlank()) throw new IllegalArgumentException("planId cannot be blank"); + if (versionReference.isBlank()) throw new IllegalArgumentException("versionReference cannot be blank"); + } +} diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java index 3a4f259ea..3fb9c6a3e 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteRewardServices.java @@ -13,21 +13,24 @@ CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdenti CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot persistedState); + /** + * Resolve and freeze the exact reward configuration BEFORE the vote transaction. + * versionReference must resolve to this same definition after reload/restart. + */ + default CompletionStage prepareVoteRewards(SharedVoteInput input, + SharedVoteIdentity identity, boolean executeRewardsNow) { + return CompletableFuture.failedFuture(new UnsupportedOperationException( + "Shared vote rewards require a persistable prepared reward version")); + } + /** * Admit/deduplicate by receipt.input().voteId() in the existing reward owner. - * Serialize concurrent delivery/recovery for that occurrence, bind its prepared - * reward definition, preserve step checkpoints and durably remember the terminal - * disposition. A repeated call, including after restart or lost acknowledgement, - * resumes pending work or returns the same result without redoing completed work. - * - *

Complete only after actual execution/checkpoints or durable offline handoff, - * never task submission. Recheck live player availability in native adapters; - * receipt.identity().online() describes the original vote, not a current player. - * Native actions retain their documented delivery semantics; this port alone is - * not a claim of exactly-once arbitrary external commands across process death.

+ * Use receipt.rewardPlan() rather than current configuration. Serialize concurrent + * delivery/recovery for that occurrence, preserve step checkpoints and durably + * remember the terminal disposition. A repeated call resumes pending work or + * returns the same result without redoing completed work. */ default CompletionStage deliverOnce(SharedVoteReceipt receipt) { - // Existing unkeyed methods cannot safely implement durable retry implicitly. return CompletableFuture.failedFuture(new UnsupportedOperationException( "Shared vote rewards require keyed durable replay support")); } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java index fd0ea79ce..899bbc677 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteUserServices.java @@ -7,50 +7,37 @@ import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; -/** - * Port for AdvancedCore's shared user/cache/storage owner. Implement atomic receipt - * operations there, not in a second VotingPlugin SQL stack/cache/queue. A voteId - * remains bound after completion for the ingress owner's supported replay horizon. - */ +/** Port for AdvancedCore's shared user/cache/storage owner. */ public interface SharedVoteUserServices { - /** Legacy API retained; the recoverable processor never uses this mutation-only operation. */ CompletionStage persistVote(SharedVoteIdentity identity, SharedVoteMutation mutation); - CompletionStage load(UUID uuid); - /** Read-only receipt lookup; null means absent, not an unavailable database. */ - default CompletionStage findVote(UUID voteId) { - return unsupported(); - } + default CompletionStage findVote(UUID voteId) { return unsupported(); } /** * One transaction atomically applies the mutation and inserts a pending receipt, - * or returns the existing receipt without applying totals/points again. Enforce - * unique voteId and reject conflicting input or identity in the SAME transaction. - * Concurrent retries and lost commit acknowledgements must be safe. Existing - * receipts retain their original policy/snapshot; do not recompute on replay. + * including the prepared immutable reward version, or returns the existing receipt + * without applying totals/points again. Enforce unique voteId in that transaction. */ default CompletionStage persistVoteWithReward(SharedVoteInput input, - SharedVoteIdentity identity, SharedVoteMutation mutation, boolean executeRewardsNow) { + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean executeRewardsNow, + SharedVoteRewardPlan rewardPlan) { return unsupported(); } - /** - * Acknowledge only after the reward owner has durably completed delivery or - * accepted offline responsibility. Retain the receipt/deduplication key. A retry - * returns the same terminal receipt rather than changing its disposition. - */ - default CompletionStage markRewardCompleted(UUID voteId, RewardDisposition disposition) { + /** Unsafe legacy shape intentionally has no mutation-only fallback. */ + default CompletionStage persistVoteWithReward(SharedVoteInput input, + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean executeRewardsNow) { return unsupported(); } - /** Bounded SQL-backed startup/recovery scan of pending receipts, never all user rows. */ - default CompletionStage> pendingVotes(int limit) { + default CompletionStage markRewardCompleted(UUID voteId, RewardDisposition disposition) { return unsupported(); } + default CompletionStage> pendingVotes(int limit) { return unsupported(); } + private static CompletionStage unsupported() { - // No mutation-only fallback: it would recreate the crash window. return CompletableFuture.failedFuture(new UnsupportedOperationException( "Shared vote persistence requires atomic vote/receipt recovery support")); } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java index c3ad20cb6..26c35b645 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessorEndToEndTest.java @@ -17,25 +17,18 @@ import org.junit.jupiter.api.io.TempDir; class SharedVoteProcessorEndToEndTest { - @TempDir - Path tempDir; + @TempDir Path tempDir; - @Test - void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { + @Test void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { UUID uuid = UUID.randomUUID(); Path storeFile = tempDir.resolve("users.properties"); ArrayList rewards = new ArrayList<>(); SharedVotePolicy policy = new SharedVotePolicy(false, true, true, true, true); - SharedVoteTestStore firstStore = new SharedVoteTestStore(storeFile, 10); SharedVoteProcessor firstRuntime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), - firstStore, recording(firstStore, rewards)); - + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), firstStore, recording(firstStore, rewards)); SharedVoteProcessingResult first = firstRuntime.process( - new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 1000L, true, true, false, true), - policy).toCompletableFuture().join(); - + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 1000L, true, true, false, true), policy).toCompletableFuture().join(); assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 10), first.persistedState()); assertEquals(SharedVoteProcessingResult.RewardDisposition.EXECUTED, first.rewardDisposition()); assertTrue(Files.isRegularFile(storeFile)); @@ -44,79 +37,54 @@ void votePersistsThenExecutesCommandAndMessageAcrossRestartAndOfflineVote() { SharedVoteTestStore restartedStore = new SharedVoteTestStore(storeFile, 10); assertEquals(first.persistedState(), restartedStore.load(uuid).toCompletableFuture().join()); SharedVoteProcessor restartedRuntime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), - restartedStore, recording(restartedStore, rewards)); - + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), restartedStore, recording(restartedStore, rewards)); SharedVoteProcessingResult second = restartedRuntime.process( - new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 2000L, true, true, false, false), - policy).toCompletableFuture().join(); - + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 2000L, true, true, false, false), policy).toCompletableFuture().join(); assertEquals(new SharedVoteUserSnapshot(2, 2, 2, 2, 20), second.persistedState()); assertEquals(SharedVoteProcessingResult.RewardDisposition.EXECUTED, second.rewardDisposition()); - assertEquals(List.of( - "command:say Thanks Ben:1", "message:Thanks Ben:10", + assertEquals(List.of("command:say Thanks Ben:1", "message:Thanks Ben:10", "command:say Thanks Ben:2", "message:Thanks Ben:20"), rewards); - - SharedVoteTestStore secondRestart = new SharedVoteTestStore(storeFile, 10); - assertEquals(second.persistedState(), secondRestart.load(uuid).toCompletableFuture().join()); } - @Test - void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { + @Test void offlineIneligibleRewardIsDurablyDelegatedAfterPersistence() { UUID uuid = UUID.randomUUID(); ArrayList rewards = new ArrayList<>(); SharedVoteTestStore store = new SharedVoteTestStore(tempDir.resolve("deferred.properties"), 3); SharedVoteProcessor runtime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), - store, recording(store, rewards)); - + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), store, recording(store, rewards)); SharedVoteProcessingResult result = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3000L, true, true, false, false), new SharedVotePolicy(false, true, true, true, false)).toCompletableFuture().join(); - assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 3), result.persistedState()); assertEquals(SharedVoteProcessingResult.RewardDisposition.DEFERRED, result.rewardDisposition()); assertEquals(List.of("defer:ExampleSite:1"), rewards); } - @Test - void proxyTotalsUseLiveStateInsteadOfHistoricalWasOnline() { + @Test void proxyTotalsUseLiveStateInsteadOfHistoricalWasOnline() { UUID uuid = UUID.randomUUID(); SharedVotePolicy policy = new SharedVotePolicy(false, true, false, true, true); - SharedVoteTestStore offlineStore = new SharedVoteTestStore(tempDir.resolve("proxy-offline.properties"), 2); - SharedVoteProcessor offlineRuntime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), - offlineStore, recording(offlineStore, new ArrayList<>())); - SharedVoteProcessingResult offline = offlineRuntime.process( - new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3100L, true, true, true, true), - policy).toCompletableFuture().join(); + SharedVoteProcessingResult offline = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", false)), offlineStore, recording(offlineStore, new ArrayList<>())).process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3100L, true, true, true, true), policy).toCompletableFuture().join(); assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 2), offline.persistedState()); - SharedVoteTestStore onlineStore = new SharedVoteTestStore(tempDir.resolve("proxy-online.properties"), 2); - SharedVoteProcessor onlineRuntime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), - onlineStore, recording(onlineStore, new ArrayList<>())); - SharedVoteProcessingResult online = onlineRuntime.process( - new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3200L, true, true, true, false), - policy).toCompletableFuture().join(); + SharedVoteProcessingResult online = new SharedVoteProcessor( + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), onlineStore, recording(onlineStore, new ArrayList<>())).process( + new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 3200L, true, true, true, false), policy).toCompletableFuture().join(); assertEquals(new SharedVoteUserSnapshot(1, 1, 1, 1, 2), online.persistedState()); } - @Test - void persistenceFailurePreventsRewardExecution() { + @Test void persistenceFailurePreventsRewardExecution() { ArrayList rewards = new ArrayList<>(); SharedVoteTestStore failing = new SharedVoteTestStore(tempDir.resolve("failure.properties"), 1) { - @Override - public CompletionStage persistVoteWithReward(SharedVoteInput input, - SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute) { + @Override public CompletionStage persistVoteWithReward(SharedVoteInput input, + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute, SharedVoteRewardPlan rewardPlan) { return CompletableFuture.failedFuture(new IllegalStateException("storage failed")); } }; SharedVoteProcessor runtime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(UUID.randomUUID(), "Ben", true)), - failing, recording(failing, rewards)); - + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(UUID.randomUUID(), "Ben", true)), failing, recording(failing, rewards)); CompletionException failure = assertThrows(CompletionException.class, () -> runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 4000L, true, true, false, true), new SharedVotePolicy(false, true, true, true, true)).toCompletableFuture().join()); @@ -124,19 +92,15 @@ public CompletionStage persistVoteWithReward(SharedVoteInput assertTrue(rewards.isEmpty()); } - @Test - void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { + @Test void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { UUID uuid = UUID.randomUUID(); SharedVoteTestStore store = new SharedVoteTestStore(tempDir.resolve("policy.properties"), 5); SharedVoteProcessor runtime = new SharedVoteProcessor( - input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), - store, recording(store, new ArrayList<>())); - + input -> CompletableFuture.completedFuture(new SharedVoteIdentity(uuid, "Ben", true)), store, recording(store, new ArrayList<>())); SharedVoteProcessingResult ignoredFake = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 5000L, false, true, false, true), new SharedVotePolicy(false, true, true, true, true)).toCompletableFuture().join(); assertEquals(new SharedVoteUserSnapshot(0, 0, 0, 0, 0), ignoredFake.persistedState()); - SharedVoteProcessingResult countedFake = runtime.process( new SharedVoteInput(UUID.randomUUID(), "Ben", "ExampleSite", 6000L, false, true, false, true), new SharedVotePolicy(true, false, true, true, true)).toCompletableFuture().join(); @@ -145,19 +109,16 @@ void fakeVoteAndAddTotalsPolicyPreserveExistingCountingRules() { private static SharedVoteRewardServices recording(SharedVoteTestStore store, List events) { return new SharedVoteRewardServices() { - @Override - public CompletionStage deliverOnce(SharedVoteReceipt receipt) { - return store.deliverOnce(receipt).thenApply(result -> { - events.clear(); - events.addAll(store.events()); - return result; - }); + @Override public CompletionStage prepareVoteRewards(SharedVoteInput input, + SharedVoteIdentity identity, boolean executeRewardsNow) { + return store.prepareVoteRewards(input, identity, executeRewardsNow); + } + @Override public CompletionStage deliverOnce(SharedVoteReceipt receipt) { + return store.deliverOnce(receipt).thenApply(result -> { events.clear(); events.addAll(store.events()); return result; }); } - @Override - public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + @Override public CompletionStage executeVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot state) { throw new AssertionError("Unkeyed execution"); } - @Override - public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, + @Override public CompletionStage deferVoteRewards(SharedVoteInput input, SharedVoteIdentity identity, SharedVoteUserSnapshot state) { throw new AssertionError("Unkeyed deferral"); } }; } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java new file mode 100644 index 000000000..dcc8161b2 --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java @@ -0,0 +1,56 @@ +package com.bencodez.votingplugin.core.vote; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class SharedVoteReviewFollowupTest { + @TempDir Path directory; + private static final UUID USER = UUID.fromString("9837d441-a4d6-461f-aa86-17958c01bc8c"); + private static final SharedVotePolicy POLICY = new SharedVotePolicy(false, true, true, true, true); + + @Test + void zeroTimestampIsNormalizedOnceAndZeroSentinelRetryMatchesTheReceipt() { + SharedVoteTestStore store = new SharedVoteTestStore(directory.resolve("zero.properties"), 5); + SharedVoteProcessor processor = runtime(store); + UUID voteId = UUID.randomUUID(); + SharedVoteInput sentinel = new SharedVoteInput(voteId, "Ben", "Example", 0, true, true, false, true); + processor.process(sentinel, POLICY).toCompletableFuture().join(); + SharedVoteReceipt receipt = store.findVote(voteId).toCompletableFuture().join(); + assertTrue(receipt.input().voteTime() > 0); + long normalized = receipt.input().voteTime(); + processor.process(sentinel, POLICY).toCompletableFuture().join(); + assertEquals(1, store.mutationCount); + assertEquals(normalized, store.findVote(voteId).toCompletableFuture().join().input().voteTime()); + } + + @Test + void restartUsesThePersistedPreparedRewardVersionInsteadOfCurrentConfiguration() { + Path file = directory.resolve("plan.properties"); + SharedVoteTestStore first = new SharedVoteTestStore(file, 5); + first.preparedPlanVersion = "config-v1"; + first.failAfterCommit = true; + SharedVoteInput input = new SharedVoteInput(UUID.randomUUID(), "Ben", "Example", 1000, true, true, false, true); + assertThrows(CompletionException.class, () -> runtime(first).process(input, POLICY).toCompletableFuture().join()); + assertEquals("config-v1", first.findVote(input.voteId()).toCompletableFuture().join().rewardPlan().versionReference()); + + SharedVoteTestStore reopened = new SharedVoteTestStore(file, 5); + reopened.preparedPlanVersion = "config-v2"; + runtime(reopened).recover(input.voteId()).toCompletableFuture().join(); + assertEquals("config-v1", reopened.lastDeliveredPlanVersion); + assertEquals("config-v1", reopened.findVote(input.voteId()).toCompletableFuture().join().rewardPlan().versionReference()); + } + + private static SharedVoteProcessor runtime(SharedVoteTestStore store) { + return new SharedVoteProcessor( + vote -> CompletableFuture.completedFuture(new SharedVoteIdentity(USER, "Ben", true)), store, store); + } +} diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java index 774174268..3931c8f7b 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java @@ -15,11 +15,7 @@ import com.bencodez.votingplugin.core.vote.SharedVoteProcessingResult.RewardDisposition; -/** - * TEST ONLY: atomic file snapshots model the existing user transaction and a - * separate keyed reward owner. Recorded command/message strings are test effects, - * not live server commands or an exactly-once external-effect implementation. - */ +/** TEST ONLY atomic file snapshot/reward owner. */ class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardServices { final Path file; final Path rewardFile; @@ -28,6 +24,8 @@ class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardSer boolean failBeforeDelivery, failAfterDelivery; CompletableFuture commitAck = CompletableFuture.completedFuture(null); int mutationCount; + String preparedPlanVersion = "fixture-v1"; + String lastDeliveredPlanVersion; SharedVoteTestStore(Path file, int pointsPerVote) { this.file = file; @@ -49,8 +47,15 @@ class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardSer catch (Exception failure) { return CompletableFuture.failedFuture(failure); } } + @Override + public CompletionStage prepareVoteRewards(SharedVoteInput input, + SharedVoteIdentity identity, boolean executeRewardsNow) { + return CompletableFuture.completedFuture( + new SharedVoteRewardPlan("vote-site:" + input.serviceSite(), preparedPlanVersion)); + } + @Override public synchronized CompletionStage persistVoteWithReward(SharedVoteInput input, - SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute) { + SharedVoteIdentity identity, SharedVoteMutation mutation, boolean execute, SharedVoteRewardPlan rewardPlan) { try { Properties properties = read(file); SharedVoteReceipt existing = receipt(properties, input.voteId()); @@ -67,9 +72,8 @@ class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardSer previous.monthTotal() + increment, previous.weeklyTotal() + increment, previous.dailyTotal() + increment, previous.points() + (mutation.awardConfiguredPoints() ? pointsPerVote : 0)); putSnapshot(properties, prefix, next); - SharedVoteReceipt created = new SharedVoteReceipt(input, identity, mutation, next, execute, null); + SharedVoteReceipt created = new SharedVoteReceipt(input, identity, mutation, next, execute, rewardPlan, null); putReceipt(properties, created); - // A SINGLE replace commits BOTH user state and the pending reward receipt. write(file, properties); mutationCount++; if (failAfterCommit) { failAfterCommit = false; throw new IOException("lost commit acknowledgement"); } @@ -110,11 +114,13 @@ class SharedVoteTestStore implements SharedVoteUserServices, SharedVoteRewardSer Properties properties = read(rewardFile); String prefix = receipt.input().voteId() + "."; String origin = receipt.input() + "|" + receipt.identity() + "|" + receipt.mutation() - + "|" + receipt.persistedState() + "|" + receipt.executeRewardsNow(); + + "|" + receipt.persistedState() + "|" + receipt.executeRewardsNow() + "|" + receipt.rewardPlan(); if (properties.containsKey(prefix + "done")) { if (!origin.equals(properties.getProperty(prefix + "origin"))) throw new IllegalStateException("Conflicting reward receipt"); + lastDeliveredPlanVersion = receipt.rewardPlan().versionReference(); return CompletableFuture.completedFuture(RewardDisposition.valueOf(properties.getProperty(prefix + "done"))); } + lastDeliveredPlanVersion = receipt.rewardPlan().versionReference(); RewardDisposition disposition = receipt.executeRewardsNow() ? RewardDisposition.EXECUTED : RewardDisposition.DEFERRED; if (receipt.executeRewardsNow()) { effect(properties, "command:say Thanks " + receipt.identity().playerName() + ":" + receipt.persistedState().allTimeTotal()); @@ -189,6 +195,8 @@ private static void putReceipt(Properties p, SharedVoteReceipt r) { p.setProperty(k + "count", Boolean.toString(r.mutation().countTotals())); p.setProperty(k + "award", Boolean.toString(r.mutation().awardConfiguredPoints())); p.setProperty(k + "execute", Boolean.toString(r.executeRewardsNow())); + p.setProperty(k + "planId", r.rewardPlan().planId()); + p.setProperty(k + "planVersion", r.rewardPlan().versionReference()); p.setProperty(k + "done", r.pending() ? "" : r.completedDisposition().name()); putSnapshot(p, k + "state.", r.persistedState()); } @@ -200,8 +208,9 @@ private static SharedVoteReceipt receipt(Properties p, UUID id) { SharedVoteIdentity identity = new SharedVoteIdentity(UUID.fromString(p.getProperty(k + "uuid")), p.getProperty(k + "resolvedName"), bool(p, k + "online")); SharedVoteMutation mutation = new SharedVoteMutation(id, input.serviceSite(), input.voteTime(), bool(p, k + "count"), bool(p, k + "award")); + SharedVoteRewardPlan plan = new SharedVoteRewardPlan(p.getProperty(k + "planId"), p.getProperty(k + "planVersion")); String done = p.getProperty(k + "done"); - return new SharedVoteReceipt(input, identity, mutation, snapshot(p, k + "state."), bool(p, k + "execute"), + return new SharedVoteReceipt(input, identity, mutation, snapshot(p, k + "state."), bool(p, k + "execute"), plan, done.isEmpty() ? null : RewardDisposition.valueOf(done)); } } From 5f25641f2e21101122878b5c3e51869958e46026 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 13:25:50 -0600 Subject: [PATCH 15/19] Preserve vote sentinel and proxy routing semantics --- .../votingplugin/core/vote/SharedVoteInput.java | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java index 72a9a904d..946bc13de 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteInput.java @@ -9,7 +9,7 @@ * remain upstream and are intentionally not reimplemented here. */ public record SharedVoteInput(UUID voteId, String playerName, String serviceSite, long voteTime, - boolean realVote, boolean addTotals, boolean proxyVote, boolean wasOnline) { + boolean realVote, boolean addTotals, boolean proxyVote, boolean forceProxyRouting, boolean wasOnline) { public SharedVoteInput { Objects.requireNonNull(voteId, "voteId"); Objects.requireNonNull(playerName, "playerName"); @@ -19,12 +19,23 @@ public record SharedVoteInput(UUID voteId, String playerName, String serviceSite if (voteTime < 0) throw new IllegalArgumentException("voteTime cannot be negative"); } + /** + * Compatibility constructor for callers that historically had one proxy bit. + * Native adapters should use the full constructor and map isBungee() and + * isForceBungee() independently. + */ + public SharedVoteInput(UUID voteId, String playerName, String serviceSite, long voteTime, + boolean realVote, boolean addTotals, boolean proxyVote, boolean wasOnline) { + this(voteId, playerName, serviceSite, voteTime, realVote, addTotals, + proxyVote, proxyVote, wasOnline); + } + /** PlayerVoteEvent uses zero as "now"; normalize before durable mutation/receipt creation. */ public SharedVoteInput normalizedVoteTime(long nowEpochMillis) { if (voteTime != 0) return this; if (nowEpochMillis <= 0) throw new IllegalArgumentException("normalized vote time must be positive"); return new SharedVoteInput(voteId, playerName, serviceSite, nowEpochMillis, - realVote, addTotals, proxyVote, wasOnline); + realVote, addTotals, proxyVote, forceProxyRouting, wasOnline); } /** A retry carrying the zero sentinel still identifies its already-normalized persisted receipt. */ @@ -37,6 +48,7 @@ public boolean matchesPersisted(SharedVoteInput persisted) { && realVote == persisted.realVote() && addTotals == persisted.addTotals() && proxyVote == persisted.proxyVote() + && forceProxyRouting == persisted.forceProxyRouting() && wasOnline == persisted.wasOnline(); } } From b38e5105b57c9d4810b9b30e630ee6890d8e24da Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 13:26:08 -0600 Subject: [PATCH 16/19] Trust atomic vote receipt winner --- .../votingplugin/core/vote/SharedVoteProcessor.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java index 919b20eca..32ddcac44 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/core/vote/SharedVoteProcessor.java @@ -44,16 +44,16 @@ public CompletionStage process(SharedVoteInput input return call(() -> rewards.prepareVoteRewards(normalized, identity, executeNow), "reward preparation") .thenCompose(rewardPlan -> { if (rewardPlan == null) return failed("Reward services returned null prepared reward plan"); - return call(() -> users.persistVoteWithReward(normalized, identity, mutation, executeNow, rewardPlan), + // Preserve the ingress sentinel for the atomic uniqueness check. The + // mutation carries this caller's normalized candidate; if another + // concurrent caller wins, the returned receipt is authoritative. + return call(() -> users.persistVoteWithReward(input, identity, mutation, executeNow, rewardPlan), "atomic persistence").thenCompose(receipt -> { if (receipt == null) return failed("User services returned null vote receipt"); receipt.requireInput(input); if (!receipt.identity().uuid().equals(identity.uuid())) { return failed("Vote receipt belongs to a different resolved identity"); } - if (!receipt.rewardPlan().equals(rewardPlan)) { - return failed("Vote receipt belongs to a different prepared reward version"); - } return deliver(receipt); }); }); From 24a8dca8aa7a097b9b8e81f166aeafc48dc066fe Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 13:26:42 -0600 Subject: [PATCH 17/19] Model atomic sentinel and proxy-route persistence --- .../core/vote/SharedVoteTestStore.java | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java index 3931c8f7b..aad9f531d 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteTestStore.java @@ -64,6 +64,14 @@ public CompletionStage prepareVoteRewards(SharedVoteInput if (!existing.identity().uuid().equals(identity.uuid())) throw new IllegalStateException("Conflicting user"); return CompletableFuture.completedFuture(existing); } + if (!input.voteId().equals(mutation.voteId()) || !input.serviceSite().equals(mutation.serviceSite())) { + throw new IllegalStateException("Mutation origin does not match vote input"); + } + if (input.voteTime() != 0 && input.voteTime() != mutation.voteTime()) { + throw new IllegalStateException("Explicit vote timestamp changed before persistence"); + } + SharedVoteInput persistedInput = input.voteTime() == 0 + ? input.normalizedVoteTime(mutation.voteTime()) : input; if (failBeforeCommit) { failBeforeCommit = false; throw new IOException("before commit"); } String prefix = identity.uuid() + "."; SharedVoteUserSnapshot previous = snapshot(properties, prefix); @@ -72,7 +80,7 @@ public CompletionStage prepareVoteRewards(SharedVoteInput previous.monthTotal() + increment, previous.weeklyTotal() + increment, previous.dailyTotal() + increment, previous.points() + (mutation.awardConfiguredPoints() ? pointsPerVote : 0)); putSnapshot(properties, prefix, next); - SharedVoteReceipt created = new SharedVoteReceipt(input, identity, mutation, next, execute, rewardPlan, null); + SharedVoteReceipt created = new SharedVoteReceipt(persistedInput, identity, mutation, next, execute, rewardPlan, null); putReceipt(properties, created); write(file, properties); mutationCount++; @@ -188,6 +196,7 @@ private static void putReceipt(Properties p, SharedVoteReceipt r) { p.setProperty(k + "real", Boolean.toString(r.input().realVote())); p.setProperty(k + "add", Boolean.toString(r.input().addTotals())); p.setProperty(k + "proxy", Boolean.toString(r.input().proxyVote())); + p.setProperty(k + "forceProxy", Boolean.toString(r.input().forceProxyRouting())); p.setProperty(k + "wasOnline", Boolean.toString(r.input().wasOnline())); p.setProperty(k + "uuid", r.identity().uuid().toString()); p.setProperty(k + "resolvedName", r.identity().playerName()); @@ -204,7 +213,8 @@ private static SharedVoteReceipt receipt(Properties p, UUID id) { String k = "receipt." + id + "."; if (!p.containsKey(k + "done")) return null; SharedVoteInput input = new SharedVoteInput(id, p.getProperty(k + "name"), p.getProperty(k + "site"), - Long.parseLong(p.getProperty(k + "time")), bool(p, k + "real"), bool(p, k + "add"), bool(p, k + "proxy"), bool(p, k + "wasOnline")); + Long.parseLong(p.getProperty(k + "time")), bool(p, k + "real"), bool(p, k + "add"), + bool(p, k + "proxy"), bool(p, k + "forceProxy"), bool(p, k + "wasOnline")); SharedVoteIdentity identity = new SharedVoteIdentity(UUID.fromString(p.getProperty(k + "uuid")), p.getProperty(k + "resolvedName"), bool(p, k + "online")); SharedVoteMutation mutation = new SharedVoteMutation(id, input.serviceSite(), input.voteTime(), bool(p, k + "count"), bool(p, k + "award")); From de49ab1c136bd12c6df236ef289232130f300b03 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 13 Sep 2026 13:28:24 -0600 Subject: [PATCH 18/19] Cover concurrent vote receipt races --- .../vote/SharedVoteReviewFollowupTest.java | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java index dcc8161b2..04a591b4f 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java @@ -1,6 +1,7 @@ package com.bencodez.votingplugin.core.vote; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -32,6 +33,26 @@ void zeroTimestampIsNormalizedOnceAndZeroSentinelRetryMatchesTheReceipt() { assertEquals(normalized, store.findVote(voteId).toCompletableFuture().join().input().voteTime()); } + @Test + void concurrentZeroSentinelCandidatesUseTheFirstAtomicReceipt() { + SharedVoteTestStore store = new SharedVoteTestStore(directory.resolve("zero-race.properties"), 5); + UUID voteId = UUID.randomUUID(); + SharedVoteInput sentinel = new SharedVoteInput(voteId, "Ben", "Example", 0, true, true, false, true); + SharedVoteIdentity identity = new SharedVoteIdentity(USER, "Ben", true); + SharedVoteRewardPlan firstPlan = new SharedVoteRewardPlan("vote-site:Example", "v1"); + SharedVoteRewardPlan losingPlan = new SharedVoteRewardPlan("vote-site:Example", "v2"); + SharedVoteReceipt first = store.persistVoteWithReward(sentinel, identity, + new SharedVoteMutation(voteId, "Example", 1001, true, true), true, firstPlan) + .toCompletableFuture().join(); + SharedVoteReceipt retry = store.persistVoteWithReward(sentinel, identity, + new SharedVoteMutation(voteId, "Example", 2002, true, true), true, losingPlan) + .toCompletableFuture().join(); + assertEquals(1001, first.input().voteTime()); + assertEquals(first, retry); + assertEquals(firstPlan, retry.rewardPlan()); + assertEquals(1, store.mutationCount); + } + @Test void restartUsesThePersistedPreparedRewardVersionInsteadOfCurrentConfiguration() { Path file = directory.resolve("plan.properties"); @@ -49,6 +70,23 @@ void restartUsesThePersistedPreparedRewardVersionInsteadOfCurrentConfiguration() assertEquals("config-v1", reopened.findVote(input.voteId()).toCompletableFuture().join().rewardPlan().versionReference()); } + @Test + void proxyOriginAndForcedRoutingRemainIndependentAcrossPersistence() { + SharedVoteTestStore store = new SharedVoteTestStore(directory.resolve("proxy-flags.properties"), 5); + UUID voteId = UUID.randomUUID(); + SharedVoteInput input = new SharedVoteInput(voteId, "Ben", "Example", 1234, + true, true, true, false, true); + SharedVoteIdentity identity = new SharedVoteIdentity(USER, "Ben", true); + SharedVoteReceipt receipt = store.persistVoteWithReward(input, identity, + new SharedVoteMutation(voteId, "Example", 1234, true, true), true, + new SharedVoteRewardPlan("vote-site:Example", "v1")).toCompletableFuture().join(); + assertTrue(receipt.input().proxyVote()); + assertFalse(receipt.input().forceProxyRouting()); + SharedVoteReceipt reopened = new SharedVoteTestStore(store.file, 5).findVote(voteId).toCompletableFuture().join(); + assertTrue(reopened.input().proxyVote()); + assertFalse(reopened.input().forceProxyRouting()); + } + private static SharedVoteProcessor runtime(SharedVoteTestStore store) { return new SharedVoteProcessor( vote -> CompletableFuture.completedFuture(new SharedVoteIdentity(USER, "Ben", true)), store, store); From 4949ac96209389c452e7d6304aab934fe6d4c32c Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Sun, 13 Sep 2026 15:46:47 -0600 Subject: [PATCH 19/19] Exercise atomic vote receipt concurrency --- .../vote/SharedVoteReviewFollowupTest.java | 54 +++++++++++++++---- 1 file changed, 43 insertions(+), 11 deletions(-) diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java index 04a591b4f..c081ad233 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/core/vote/SharedVoteReviewFollowupTest.java @@ -9,6 +9,9 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -34,23 +37,42 @@ void zeroTimestampIsNormalizedOnceAndZeroSentinelRetryMatchesTheReceipt() { } @Test - void concurrentZeroSentinelCandidatesUseTheFirstAtomicReceipt() { + void concurrentZeroSentinelCandidatesUseTheFirstAtomicReceipt() throws Exception { SharedVoteTestStore store = new SharedVoteTestStore(directory.resolve("zero-race.properties"), 5); UUID voteId = UUID.randomUUID(); SharedVoteInput sentinel = new SharedVoteInput(voteId, "Ben", "Example", 0, true, true, false, true); SharedVoteIdentity identity = new SharedVoteIdentity(USER, "Ben", true); SharedVoteRewardPlan firstPlan = new SharedVoteRewardPlan("vote-site:Example", "v1"); SharedVoteRewardPlan losingPlan = new SharedVoteRewardPlan("vote-site:Example", "v2"); - SharedVoteReceipt first = store.persistVoteWithReward(sentinel, identity, - new SharedVoteMutation(voteId, "Example", 1001, true, true), true, firstPlan) - .toCompletableFuture().join(); - SharedVoteReceipt retry = store.persistVoteWithReward(sentinel, identity, - new SharedVoteMutation(voteId, "Example", 2002, true, true), true, losingPlan) - .toCompletableFuture().join(); - assertEquals(1001, first.input().voteTime()); - assertEquals(first, retry); - assertEquals(firstPlan, retry.rewardPlan()); - assertEquals(1, store.mutationCount); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + var workers = Executors.newFixedThreadPool(2); + try { + CompletableFuture firstAttempt = CompletableFuture.supplyAsync(() -> { + awaitStart(ready, start); + return store.persistVoteWithReward(sentinel, identity, + new SharedVoteMutation(voteId, "Example", 1001, true, true), true, firstPlan) + .toCompletableFuture().join(); + }, workers); + CompletableFuture retryAttempt = CompletableFuture.supplyAsync(() -> { + awaitStart(ready, start); + return store.persistVoteWithReward(sentinel, identity, + new SharedVoteMutation(voteId, "Example", 2002, true, true), true, losingPlan) + .toCompletableFuture().join(); + }, workers); + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + SharedVoteReceipt first = firstAttempt.join(); + SharedVoteReceipt retry = retryAttempt.join(); + assertEquals(first, retry); + SharedVoteRewardPlan winningPlan = first.input().voteTime() == 1001 ? firstPlan : losingPlan; + assertEquals(winningPlan, first.rewardPlan()); + assertTrue(first.input().voteTime() == 1001 || first.input().voteTime() == 2002); + assertEquals(1, store.mutationCount); + } finally { + start.countDown(); + workers.shutdownNow(); + } } @Test @@ -91,4 +113,14 @@ private static SharedVoteProcessor runtime(SharedVoteTestStore store) { return new SharedVoteProcessor( vote -> CompletableFuture.completedFuture(new SharedVoteIdentity(USER, "Ben", true)), store, store); } + + private static void awaitStart(CountDownLatch ready, CountDownLatch start) { + ready.countDown(); + try { + if (!start.await(5, TimeUnit.SECONDS)) throw new IllegalStateException("Timed out awaiting concurrent start"); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted awaiting concurrent start", interrupted); + } + } }