Retain VoteParty eligibility when stored recipient preparation fails

VotingPlugin/src/main/java/com/bencodez/votingplugin/specialrewards/voteparty/VoteParty.java

@@ -4,6 +4,10 @@ package com.bencodez.votingplugin.specialrewards.voteparty;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
+import java.util.function.BooleanSupplier;
 
 import org.bukkit.Bukkit;
 import org.bukkit.command.CommandSender;
@@ -29,6 +33,12 @@ public class VoteParty implements Listener {
 	private VotingPluginMain plugin;
 	private final VotePartyState state;
 	private final VotePartyRewardHandler rewardHandler;
+	private final AtomicReference<Object> storedPreparation = new AtomicReference<>();
+	// Workers may hold this permit across storage; the platform only tries it.
+	// Order: worker monitor -> stateCommit -> scalarCommit; platform only tries the last two.
+	// Scalar commands take scalarCommit alone and release it before user storage.
+	private final ReentrantLock stateCommit = new ReentrantLock();
+	private final ReentrantLock scalarCommit = new ReentrantLock();
 
 	/**
 	 * Constructs a new VoteParty.
@@ -46,10 +56,15 @@ public class VoteParty implements Listener {
 	 *
 	 * @param user the voting plugin user
 	 */
-	public void addTotal(VotingPluginUser user) {
-		setTotalVotes(getTotalVotes() + 1);
-		user.setVotePartyVotes(user.getVotePartyVotes() + 1);
-		plugin.getPlaceholders().onVotePartyUpdate();
+	public synchronized void addTotal(VotingPluginUser user) {
+		stateCommit.lock();
+		try {
+			scalarCommit.lock();
+			try { state.setTotalVotes(getTotalVotes() + 1); }
+			finally { scalarCommit.unlock(); }
+			user.setVotePartyVotes(user.getVotePartyVotes() + 1);
+			plugin.getPlaceholders().onVotePartyUpdate();
+		} finally { stateCommit.unlock(); }
 	}
 
 	/**
@@ -58,12 +73,15 @@ public class VoteParty implements Listener {
 	 * @param user the user
 	 */
 	public void addVotePlayer(VotingPluginUser user) {
-		String uuid = user.getUUID();
-		List<String> voted = getVotedUsers();
-		if (!voted.contains(uuid)) {
-			voted.add(uuid);
-			setVotedUsers(voted);
-		}
+		scalarCommit.lock();
+		try {
+			String uuid = user.getUUID();
+			List<String> voted = getVotedUsers();
+			if (!voted.contains(uuid)) {
+				voted.add(uuid);
+				setVotedUsers(voted);
+			}
+		} finally { scalarCommit.unlock(); }
 	}
 
 	/**
@@ -72,44 +90,78 @@ public class VoteParty implements Listener {
 	 * @param user        the voting plugin user
 	 * @param forceBungee whether to force Bungee processing
 	 */
-	public void check(VotingPluginUser user, boolean forceBungee) {
+	public synchronized void check(VotingPluginUser user, boolean forceBungee) {
+		stateCommit.lock();
+		try {
+			if (storedPreparation.get() != null) return;
+			if (getTotalVotes() >= getVotesRequired()
+					&& prepareStoredParty((ids, finished) -> checkPrepared(user, forceBungee, ids, finished))) return;
+			checkPrepared(user, forceBungee, null, null);
+		} finally { stateCommit.unlock(); }
+	}
+
+	private void checkPrepared(VotingPluginUser user, boolean forceBungee, VotePartyRewardHandler.StoredRecipients storedIds,
+			Object attempt) {
 		if (getTotalVotes() < getVotesRequired()) {
 			plugin.extraDebug("Not enough votes for vote party: " + getTotalVotes() + " / " + getVotesRequired());
+			finishStoredParty(attempt);
 			return;
 		}
 
 		if (plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerDay()
 				&& plugin.getServerData().isLastVotePartySameDay()) {
 			plugin.extraDebug("Already had vote party today");
+			finishStoredParty(attempt);
 			return;
 		}
 		if (plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerWeek()
 				&& plugin.getServerData().isLastVotePartySameWeek()) {
 			plugin.extraDebug("Already had vote party this week");
+			finishStoredParty(attempt);
 			return;
 		}
-		if (plugin.getSpecialRewardsConfig().isVotePartyResetCount()) {
+		if (storedIds == null && plugin.getSpecialRewardsConfig().isVotePartyResetCount()) {
 			setTotalVotes(getTotalVotes() - getVotesRequired());
 		}
 
 		VotePartyEvent event = new VotePartyEvent();
 		Bukkit.getPluginManager().callEvent(event);
 		if (event.isCancelled()) {
+			finishStoredParty(attempt);
 			return;
 		}
 
-		giveRewards(user, forceBungee);
-
-		if (plugin.getSpecialRewardsConfig().getVotePartyIncreaseVotesRequired() > 0) {
-			plugin.getServerData().setVotePartyExtraRequired(plugin.getServerData().getVotePartyExtraRequired()
-					+ plugin.getSpecialRewardsConfig().getVotePartyIncreaseVotesRequired());
+		if (storedIds != null) {
+			dispatchPreparedRewards(user, forceBungee, storedIds, () -> getTotalVotes() >= getVotesRequired()
+					&& (!plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerDay()
+							|| !plugin.getServerData().isLastVotePartySameDay())
+					&& (!plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerWeek()
+							|| !plugin.getServerData().isLastVotePartySameWeek()), () -> {
+				if (plugin.getSpecialRewardsConfig().isVotePartyResetCount())
+					state.setTotalVotes(getTotalVotes() - getVotesRequired());
+				completePartyMetadata();
+			}, attempt);
+		} else {
+			giveRewards(user, forceBungee);
+			completePartyMetadata();
+			finishStoredParty(attempt);
 		}
+	}
 
-		plugin.getServerData().updateLastVoteParty();
+	private void completePartyMetadata() {
+		scalarCommit.lock();
+		try {
+			if (plugin.getSpecialRewardsConfig().getVotePartyIncreaseVotesRequired() > 0) {
+				plugin.getServerData().setVotePartyExtraRequired(plugin.getServerData().getVotePartyExtraRequired()
+						+ plugin.getSpecialRewardsConfig().getVotePartyIncreaseVotesRequired());
+			}
 
-		if (plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerWeek()) {
-			plugin.getServerData().updateLastVotePartyWeek();
-		}
+			plugin.getServerData().updateLastVoteParty();
+
+			if (plugin.getSpecialRewardsConfig().isVotePartyOnlyOncePerWeek()) {
+				plugin.getServerData().updateLastVotePartyWeek();
+			}
+		} finally { scalarCommit.unlock(); }
 	}
 
 	/**
@@ -243,19 +295,106 @@ public class VoteParty implements Listener {
 	 * @param forceBungee whether to force Bungee processing
 	 */
 	public void giveRewards(VotingPluginUser orgUser, boolean forceBungee) {
+		if (prepareStoredParty((ids, finished) ->
+				dispatchPreparedRewards(orgUser, forceBungee, ids, () -> true, () -> { }, finished))) return;
 		rewardHandler.giveRewards(orgUser, forceBungee);
-		// Clear this party's recipient list immediately, as before. Delaying it
-		// would risk wiping voters recorded for the next party.
 		setVotedUsers(new ArrayList<>());
-		// Shared-MySQL recipients capture eligibility on AdvancedCore's ordered
-		// storage worker. Only the persisted per-user counter wipe must run
-		// behind those reads; it must not overtake deferred eligibility checks.
 		com.bencodez.advancedcore.api.user.usercache.UserDataManager manager =
 				plugin.getUserManager().getDataManager();
 		if (manager != null && manager.deferSharedStorageWork(this::resetVotePartyCount)) return;
 		resetVotePartyCount();
 	}
 
+	/** One pending preparation/reset at a time; its native query is always queued. */
+	private boolean prepareStoredParty(BiConsumer<VotePartyRewardHandler.StoredRecipients, Object> action) {
+		if (storedPreparation.get() != null) return true;
+		if (!rewardHandler.needsStoredPreparation()) return false;
+		Object attempt = new Object();
+		if (!storedPreparation.compareAndSet(null, attempt)) return true;
+		try {
+			if (rewardHandler.prepareStoredRecipients(ids -> {
+				try {
+					plugin.getVoteTimer().execute(() -> {
+						synchronized (VoteParty.this) {
+							stateCommit.lock();
+							try {
+								if (storedPreparation.get() != attempt) return;
+								if (!ids.isCurrent()) { finishStoredParty(attempt); return; }
+								try { action.accept(ids, attempt); }
+								catch (RuntimeException | Error error) { finishStoredParty(attempt); throw error; }
+							} finally { stateCommit.unlock(); }
+						}
+					});
+				} catch (RuntimeException | Error error) {
+					finishStoredParty(attempt);
+					rewardHandler.logPreparationFailure(error);
+				}
+			}, error -> {
+				if (storedPreparation.compareAndSet(attempt, null)) rewardHandler.logPreparationFailure(error);
+			})) return true;
+		} catch (RuntimeException | Error error) {
+			finishStoredParty(attempt);
+			throw error;
+		}
+		finishStoredParty(attempt);
+		return false;
+	}
+
+	private void finishStoredParty(Object attempt) {
+		if (attempt != null) storedPreparation.compareAndSet(attempt, null);
+	}
+
+	/** Commit scalar party state with vote(), then capture platform input on its owner. */
+	private void dispatchPreparedRewards(VotingPluginUser user, boolean forceBungee,
+			VotePartyRewardHandler.StoredRecipients storedIds, BooleanSupplier canCommit,
+			Runnable commitState, Object attempt) {
+		try {
+			plugin.getBukkitScheduler().runTask(plugin, () -> {
+				VotePartyRewardHandler.StoredRecipients dispatch;
+				if (!stateCommit.tryLock()) {
+					finishStoredParty(attempt);
+					return;
+				}
+				try {
+					if (!scalarCommit.tryLock()) {
+						finishStoredParty(attempt);
+						return;
+					}
+					try {
+						if (storedPreparation.get() != attempt || !storedIds.isCurrent()
+								|| !rewardHandler.needsStoredPreparation() || !canCommit.getAsBoolean()) {
+							finishStoredParty(attempt);
+							return;
+						}
+						dispatch = storedIds.forDispatch();
+						state.setVotedUsers(new ArrayList<>());
+						commitState.run();
+					} finally { scalarCommit.unlock(); }
+				} finally { stateCommit.unlock(); }
+				try {
+					rewardHandler.giveRewards(user, forceBungee, dispatch);
+					Runnable reset = () -> {
+						try {
+							dispatch.withCurrentOwner(() -> {
+								if (storedPreparation.get() == attempt) resetVotePartyCount();
+								return null;
+							});
+						} finally { finishStoredParty(attempt); }
+					};
+					if (!plugin.getUserManager().getDataManager().deferSharedStorageWork(reset)) {
+						throw new IllegalStateException("Shared VoteParty reset was not admitted to storage worker");
+					}
+				} catch (RuntimeException | Error error) {
+					finishStoredParty(attempt);
+					rewardHandler.logPreparationFailure(error);
+				}
+			});
+		} catch (RuntimeException | Error error) {
+			finishStoredParty(attempt);
+			rewardHandler.logPreparationFailure(error);
+		}
+	}
+
 	/**
 	 * Handles day change events to reset vote party if configured.
 	 *
@@ -325,10 +464,12 @@ public class VoteParty implements Listener {
 	 * @param override whether to override total votes to zero
 	 */
 	public void reset(boolean override) {
-		if (override) {
-			setTotalVotes(0);
-		}
-		setVotedUsers(new ArrayList<>());
+		scalarCommit.lock();
+		try {
+			storedPreparation.set(null);
+			if (override) state.setTotalVotes(0);
+			state.setVotedUsers(new ArrayList<>());
+		} finally { scalarCommit.unlock(); }
 		resetVotePartyCount();
 	}
 
@@ -345,7 +486,10 @@ public class VoteParty implements Listener {
 	 * @param value the new total votes
 	 */
 	public void setTotalVotes(int value) {
-		state.setTotalVotes(value);
+		scalarCommit.lock();
+		try {
+			state.setTotalVotes(value);
+		} finally { scalarCommit.unlock(); }
 	}
 
 	/**
@@ -354,7 +498,10 @@ public class VoteParty implements Listener {
 	 * @param value the new voted users
 	 */
 	public void setVotedUsers(List<String> value) {
-		state.setVotedUsers(value);
+		scalarCommit.lock();
+		try {
+			state.setVotedUsers(value);
+		} finally { scalarCommit.unlock(); }
 	}
 
 	/**
@@ -365,7 +512,10 @@ public class VoteParty implements Listener {
 	 * @param forceBungee whether to force Bungee processing
 	 */
 	public synchronized void vote(VotingPluginUser user, boolean realVote, boolean forceBungee) {
-		vote(user, realVote, forceBungee, user.isOnline());
+		stateCommit.lock();
+		try {
+			vote(user, realVote, forceBungee, user.isOnline());
+		} finally { stateCommit.unlock(); }
 	}
 
 	/**
@@ -381,15 +531,18 @@ public class VoteParty implements Listener {
 	 */
 	public synchronized void vote(VotingPluginUser user, boolean realVote, boolean forceBungee,
 			boolean onlineAtVoteTime) {
-		if (plugin.getSpecialRewardsConfig().isVotePartyEnabled()) {
-			if (plugin.getSpecialRewardsConfig().isVotePartyCountFakeVotes() || realVote) {
-				if (plugin.getSpecialRewardsConfig().isVotePartyCountOfflineVotes() || onlineAtVoteTime) {
-					addTotal(user);
-					addVotePlayer(user);
-					check(user, forceBungee);
-					checkVoteReminder(user);
+		stateCommit.lock();
+		try {
+			if (plugin.getSpecialRewardsConfig().isVotePartyEnabled()) {
+				if (plugin.getSpecialRewardsConfig().isVotePartyCountFakeVotes() || realVote) {
+					if (plugin.getSpecialRewardsConfig().isVotePartyCountOfflineVotes() || onlineAtVoteTime) {
+						addTotal(user);
+						addVotePlayer(user);
+						check(user, forceBungee);
+						checkVoteReminder(user);
+					}
 				}
 			}
-		}
+		} finally { stateCommit.unlock(); }
 	}
 }

VotingPlugin/src/main/java/com/bencodez/votingplugin/specialrewards/voteparty/VotePartyRewardHandler.java

@@ -2,6 +2,10 @@ package com.bencodez.votingplugin.specialrewards.voteparty;
 
 import java.util.ArrayList;
 import java.util.UUID;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import com.bencodez.advancedcore.AdvancedCorePlugin.UserStorageOwner;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -13,6 +17,8 @@ import com.bencodez.advancedcore.api.messages.PlaceholderUtils;
 import com.bencodez.advancedcore.api.misc.MiscUtils;
 import com.bencodez.advancedcore.api.rewards.RewardBuilder;
 import com.bencodez.advancedcore.api.user.UserDataFetchMode;
+import com.bencodez.advancedcore.api.user.UserStorage;
+import com.bencodez.simpleapi.sql.Column;
 import com.bencodez.advancedcore.api.user.usercache.UserDataManager;
 import com.bencodez.simpleapi.array.ArrayUtils;
 import com.bencodez.votingplugin.VotingPluginMain;
@@ -53,11 +59,109 @@ public final class VotePartyRewardHandler {
 	}
 
 	public void giveRewards(VotingPluginUser triggeringUser, boolean forceBungee) {
+		if (prepareStoredRecipients(ids -> giveRewards(triggeringUser, forceBungee, ids.forDispatch()),
+				this::logPreparationFailure)) return;
+		giveRewards(triggeringUser, forceBungee, null);
+	}
+
+	/** True only for the shared-MySQL path that needs stored recipients. */
+	boolean needsStoredPreparation() {
+		return sharedDataManager() != null && plugin.getStorageType() == UserStorage.MYSQL
+				&& plugin.getSpecialRewardsConfig().isVotePartyGiveAllPlayers()
+				&& !plugin.getSpecialRewardsConfig().isVotePartyGiveOnlinePlayersOnly();
+	}
+
+	/** Prepare a complete recipient list before any party effects or counter reset. */
+	boolean prepareStoredRecipients(Consumer<StoredRecipients> success, Consumer<Throwable> failure) {
+		if (!needsStoredPreparation()) return false;
+		UserDataManager manager = sharedDataManager();
+		AtomicBoolean completed = new AtomicBoolean();
+		AtomicReference<UserStorageOwner> capturedOwner = new AtomicReference<>();
+		Consumer<List<String>> ready = ids -> {
+			if (!completed.compareAndSet(false, true)) return;
+			StoredRecipients prepared = new StoredRecipients(manager, capturedOwner.get(), ids, false, 0);
+			if (prepared.isCurrent()) success.accept(prepared);
+			else failure.accept(new IllegalStateException("Shared VoteParty storage changed after enumeration"));
+		};
+		Consumer<Throwable> failed = error -> {
+			if (completed.compareAndSet(false, true)) failure.accept(error);
+		};
+		Supplier<List<String>> query = () -> manager.withSharedNativeUserStorage(owner -> {
+			if (owner == null || owner.storageType() != UserStorage.MYSQL || owner.mysql() == null) {
+				throw new IllegalStateException("Shared VoteParty storage owner is unavailable");
+			}
+			// The legacy UUID-list API converts this query's null failure to an empty
+			// list. Use the existing public nullable query under current-owner admission.
+			ArrayList<Column> rows = owner.mysql().getRowsQuery();
+			if (rows == null) throw new IllegalStateException("Stored VoteParty recipients could not be loaded");
+			ArrayList<String> ids = new ArrayList<>();
+			for (Column row : rows) {
+				if (row == null || row.getValue() == null || !row.getValue().isString()) {
+					throw new IllegalStateException("Stored VoteParty recipient identity is invalid");
+				}
+				ids.add(UUID.fromString(row.getValue().getString()).toString());
+			}
+			capturedOwner.set(owner);
+			return List.copyOf(ids);
+		});
+		try {
+			// Always queue preparation, including worker-originated votes. A vote
+			// can hold its coordinator monitor; no native query belongs in that call.
+			if (!manager.deferSharedStorageResultFromPlatform(query, ready, failed)) {
+				failed.accept(new IllegalStateException("Shared VoteParty storage changed before admission"));
+			}
+		} catch (RuntimeException | Error error) {
+			failed.accept(error);
+		}
+		return true;
+	}
+
+	/** The owner identity travels with the list through delivery admission and reset. */
+	final class StoredRecipients {
+		private final UserDataManager manager;
+		private final UserStorageOwner owner;
+		private final List<String> ids;
+		private final boolean firstPartyToday;
+		private final int userVotesRequired;
+
+		private StoredRecipients(UserDataManager manager, UserStorageOwner owner, List<String> ids,
+				boolean firstPartyToday, int userVotesRequired) {
+			this.manager = manager;
+			this.owner = owner;
+			this.ids = ids;
+			this.firstPartyToday = firstPartyToday;
+			this.userVotesRequired = userVotesRequired;
+		}
+
+		StoredRecipients forDispatch() {
+			return new StoredRecipients(manager, owner, ids, !plugin.getServerData().isLastVotePartySameDay(),
+					plugin.getSpecialRewardsConfig().getVotePartyUserVotesRequired());
+		}
+
+		boolean isCurrent() {
+			try {
+				return owner != null && sharedDataManager() == manager && manager.sharedNativeUserStorageOwner() == owner;
+			} catch (RuntimeException unavailable) { return false; }
+		}
+
+		<T> T withCurrentOwner(Supplier<T> work) {
+			return manager.withSharedNativeUserStorage(current -> {
+				if (current != owner) throw new IllegalStateException("Shared VoteParty storage owner changed");
+				return work.get();
+			});
+		}
+	}
+
+	void logPreparationFailure(Throwable error) {
+		plugin.getLogger().warning("VoteParty not started: stored recipients are unavailable: " + error);
+	}
+
+	void giveRewards(VotingPluginUser triggeringUser, boolean forceBungee, StoredRecipients storedRecipients) {
 		MiscUtils.getInstance().broadcast(plugin.getSpecialRewardsConfig().getVotePartyBroadcast());
 		dispatchGlobalCommands(voteParty.getRandomPlayerName());
 
 		if (plugin.getSpecialRewardsConfig().isVotePartyGiveAllPlayers()) {
-			giveAllPlayers(triggeringUser, forceBungee);
+			giveAllPlayers(triggeringUser, forceBungee, storedRecipients);
 		} else {
 			giveVotedPlayers(triggeringUser, forceBungee);
 		}
@@ -92,25 +196,52 @@ public final class VotePartyRewardHandler {
 
 	/** Dispatch to every eligible user, preserving each online player's scheduler owner. */
 	private void giveAllPlayers(VotingPluginUser triggeringUser, boolean forceBungee) {
+		if (prepareStoredRecipients(ids -> giveAllPlayers(triggeringUser, forceBungee, ids.forDispatch()),
+				this::logPreparationFailure)) return;
+		giveAllPlayers(triggeringUser, forceBungee, null);
+	}
+
+	private void giveAllPlayers(VotingPluginUser triggeringUser, boolean forceBungee, StoredRecipients prepared) {
 		plugin.debug("Trying to give all players vote party");
 		ArrayList<String> alreadyGiven = new ArrayList<>();
 		for (Player player : Bukkit.getOnlinePlayers()) {
 			VotingPluginUser user = resolveUser(triggeringUser, player.getUniqueId(), player);
 			if (!plugin.getSpecialRewardsConfig().isVotePartyGiveOnlinePlayersOnly() || user.isOnline()) {
-				giveRewardWhenLoaded(user, forceBungee, player);
+				if (prepared == null) giveRewardWhenLoaded(user, forceBungee, player);
+				else giveRewardWhenLoaded(user, forceBungee, player, prepared.firstPartyToday,
+						prepared.userVotesRequired, prepared);
 			}
 			alreadyGiven.add(player.getUniqueId().toString());
 		}
 
-		for (String uuid : plugin.getVotingPluginUserManager().getAllUUIDs()) {
-			if (alreadyGiven.contains(uuid)) {
-				continue;
-			}
-			VotingPluginUser user = resolveUser(triggeringUser, UUID.fromString(uuid), null);
-			if (!plugin.getSpecialRewardsConfig().isVotePartyGiveOnlinePlayersOnly() || user.isOnline()) {
-				giveRewardWhenLoaded(user, forceBungee, null);
-			}
+		UserDataManager manager = sharedDataManager();
+		if (manager != null && plugin.getSpecialRewardsConfig().isVotePartyGiveOnlinePlayersOnly()) {
+			// Every live player was considered above. No stored-user query is needed.
+			return;
 		}
+		// The party timestamp is updated as soon as dispatch returns. Carry its
+		// decisions into the deferred enumeration rather than reading the next state.
+		boolean firstPartyToday = prepared == null ? !plugin.getServerData().isLastVotePartySameDay() : prepared.firstPartyToday;
+		int userVotesRequired = prepared == null ? plugin.getSpecialRewardsConfig().getVotePartyUserVotesRequired() : prepared.userVotesRequired;
+		Runnable recipients = () -> {
+			for (String uuid : prepared != null ? prepared.ids : plugin.getVotingPluginUserManager().getAllUUIDs()) {
+				if (alreadyGiven.contains(uuid)) {
+					continue;
+				}
+				VotingPluginUser user = resolveUser(triggeringUser, UUID.fromString(uuid), null);
+				if (!plugin.getSpecialRewardsConfig().isVotePartyGiveOnlinePlayersOnly() || user.isOnline()) {
+					giveRewardWhenLoaded(user, forceBungee, null, firstPartyToday, userVotesRequired, prepared);
+				}
+			}
+		};
+		Runnable storedRecipients = prepared == null ? recipients : () -> prepared.withCurrentOwner(() -> {
+			recipients.run();
+			return null;
+		});
+		// UUID enumeration itself enters storage. Admit it after the online reads
+		// and before VoteParty queues its counter wipe on this same ordered worker.
+		if (manager != null && manager.deferSharedStorageWork(storedRecipients)) return;
+		storedRecipients.run();
 	}
 
 	/** Dispatch to users recorded for this party before its voter list is reset. */
@@ -160,6 +291,18 @@ public final class VotePartyRewardHandler {
 	 * directly because checking publication takes the cache's synchronized monitor.
 	 */
 	private void giveRewardWhenLoaded(VotingPluginUser user, boolean forceBungee, Player callbackOwner) {
+		giveRewardWhenLoaded(user, forceBungee, callbackOwner,
+				!plugin.getServerData().isLastVotePartySameDay(),
+				plugin.getSpecialRewardsConfig().getVotePartyUserVotesRequired());
+	}
+
+	private void giveRewardWhenLoaded(VotingPluginUser user, boolean forceBungee, Player callbackOwner,
+			boolean firstPartyToday, int userVotesRequired) {
+		giveRewardWhenLoaded(user, forceBungee, callbackOwner, firstPartyToday, userVotesRequired, null);
+	}
+
+	private void giveRewardWhenLoaded(VotingPluginUser user, boolean forceBungee, Player callbackOwner,
+			boolean firstPartyToday, int userVotesRequired, StoredRecipients prepared) {
 		UserDataManager manager = sharedDataManager();
 		if (manager == null) {
 			voteParty.giveReward(user, forceBungee);
@@ -167,21 +310,17 @@ public final class VotePartyRewardHandler {
 		}
 		UUID uuid = user.getJavaUUID();
 
-		// VoteParty clears per-user counters and updates the last-party timestamp
-		// after dispatch. Capture these decisions before that reset, not when the
-		// deferred reward finally runs on the player scheduler.
-		final boolean firstPartyToday = !plugin.getServerData().isLastVotePartySameDay();
-		final int userVotesRequired = plugin.getSpecialRewardsConfig().getVotePartyUserVotesRequired();
-
 		// A mapped cache entry can still be an *unpublished* placeholder, so
 		// cacheUserIfNeeded() alone would not make reward placeholders safe to read.
-		Supplier<Integer> populate = () -> {
+		Supplier<Integer> read = () -> {
 			if (manager.getPublishedCache(uuid) == null) manager.cacheUser(uuid, null);
 			if (manager.getPublishedCache(uuid) == null) return null;
 			// Read eligibility on the storage worker, before VoteParty resets counters.
 			return userVotesRequired > 0 ? user.getVotePartyVotes() : 0;
 		};
+		Supplier<Integer> populate = prepared == null ? read : () -> prepared.withCurrentOwner(read);
 		Consumer<Integer> success = votes -> {
+			if (prepared != null && !prepared.isCurrent()) return;
 			if (votes == null) {
 				plugin.getLogger().warning("VoteParty reward not sent: shared user data is unavailable for " + uuid);
 				return;
@@ -190,7 +329,10 @@ public final class VotePartyRewardHandler {
 
 			// Eligibility was captured on the storage worker before reset. Do not
 			// re-read the cache on a platform thread or re-check reset counters.
-			Runnable deliver = () -> givePreparedReward(user, user.isOnline(), forceBungee, firstPartyToday);
+			Runnable deliver = () -> {
+				if (prepared == null || prepared.isCurrent())
+					givePreparedReward(user, user.isOnline(), forceBungee, firstPartyToday);
+			};
 			if (callbackOwner != null) {
 				deliver.run();
 			} else {

VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/specialrewards/VotePartySharedMysqlRewardTest.java

@@ -19,6 +19,9 @@ import static org.mockito.Mockito.when;
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.ArrayDeque;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.Collections;
 import java.util.UUID;
 import java.util.concurrent.RejectedExecutionException;
@@ -39,6 +42,7 @@ import org.mockito.MockedStatic;
 import org.mockito.ArgumentCaptor;
 
 import com.bencodez.advancedcore.api.user.UserDataFetchMode;
+import com.bencodez.advancedcore.api.misc.MiscUtils;
 import com.bencodez.advancedcore.api.rewards.RewardHandler;
 import com.bencodez.advancedcore.api.rewards.RewardOptions;
 import com.bencodez.advancedcore.api.user.usercache.UserDataCache;
@@ -400,6 +404,169 @@ class VotePartySharedMysqlRewardTest {
         verify(voteParty, never()).giveReward(user, false);
     }
 
+    @Test
+    void sharedStoredRecipientEnumerationWaitsForStorageWorker() throws Exception {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        AtomicReference<Runnable> enumerate = new AtomicReference<>();
+        when(dataManager.deferSharedStorageWork(any())).thenAnswer(call -> {
+            enumerate.set(call.getArgument(0));
+            return true;
+        });
+        when(votingUsers.getAllUUIDs()).thenReturn(new ArrayList<>());
+
+        dispatch("giveAllPlayers", null);
+
+        verify(votingUsers, never()).getAllUUIDs();
+        assertNotNull(enumerate.get());
+        enumerate.get().run();
+        verify(votingUsers).getAllUUIDs();
+    }
+
+    @Test
+    void onlineOnlyPartyNeverEnumeratesStoredUsers() throws Exception {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+
+        dispatch("giveAllPlayers", null);
+
+        verify(votingUsers, never()).getAllUUIDs();
+        verify(dataManager, never()).deferSharedStorageWork(any());
+    }
+
+    @Test
+    void deferredStoredRecipientUsesPartyTimeMetadataAndPreResetEligibility() throws Exception {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        when(config.getVotePartyUserVotesRequired()).thenReturn(2);
+        when(user.getVotePartyVotes()).thenReturn(2);
+        when(dataManager.getPublishedCache(USER_UUID)).thenReturn(mock(UserDataCache.class));
+        when(votingUsers.getAllUUIDs()).thenReturn(new ArrayList<>(Collections.singletonList(USER_UUID.toString())));
+        AtomicReference<Runnable> enumerate = new AtomicReference<>();
+        when(dataManager.deferSharedStorageWork(any())).thenAnswer(call -> {
+            enumerate.set(call.getArgument(0));
+            return true;
+        });
+        BukkitScheduler scheduler = mock(BukkitScheduler.class);
+        when(plugin.getBukkitScheduler()).thenReturn(scheduler);
+
+        dispatch("giveAllPlayers", null);
+        // VoteParty.check updates this immediately after recipient admission.
+        when(serverData.isLastVotePartySameDay()).thenReturn(true);
+        when(config.getVotePartyUserVotesRequired()).thenReturn(3);
+        assertNotNull(enumerate.get());
+        enumerate.get().run();
+        when(user.getVotePartyVotes()).thenReturn(0);
+        ArgumentCaptor<Runnable> completion = ArgumentCaptor.forClass(Runnable.class);
+        verify(scheduler).runTask(eq(plugin), completion.capture());
+        completion.getValue().run();
+
+        ArgumentCaptor<RewardOptions> options = ArgumentCaptor.forClass(RewardOptions.class);
+        verify(apiRewards).giveReward(eq(user), eq(rewardData), eq(REWARD_PATH), options.capture());
+        assertEquals("true", options.getValue().getPlaceholders().get("FirstVotePartyToday"));
+        verify(user, times(1)).getVotePartyVotes();
+    }
+
+    @Test
+    void actualPartyEnumeratesAndChecksStoredRecipientsBeforeQueuedCounterReset() throws Exception {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+        when(config.isVotePartyGiveAllPlayers()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        when(config.getVotePartyUserVotesRequired()).thenReturn(2);
+        when(dataManager.getPublishedCache(USER_UUID)).thenReturn(mock(UserDataCache.class));
+        ArrayDeque<Runnable> storage = new ArrayDeque<>();
+        List<String> order = new ArrayList<>();
+        AtomicInteger votes = new AtomicInteger(2);
+        when(user.getVotePartyVotes()).thenAnswer(call -> {
+            order.add("eligibility");
+            return votes.get();
+        });
+        when(votingUsers.getAllUUIDs()).thenAnswer(call -> {
+            order.add("enumeration");
+            return new ArrayList<>(Collections.singletonList(USER_UUID.toString()));
+        });
+        when(dataManager.deferSharedStorageWork(any())).thenAnswer(call -> {
+            storage.add(call.getArgument(0));
+            return true;
+        });
+        com.bencodez.advancedcore.api.user.UserManager coreUsers = plugin.getUserManager();
+        doAnswer(call -> {
+            order.add("reset");
+            votes.set(0);
+            return null;
+        }).when(coreUsers).removeAllKeyValues(eq("VotePartyVotes"), any());
+        BukkitScheduler scheduler = mock(BukkitScheduler.class);
+        when(plugin.getBukkitScheduler()).thenReturn(scheduler);
+        try (MockedStatic<MiscUtils> misc = mockStatic(MiscUtils.class)) {
+            misc.when(MiscUtils::getInstance).thenReturn(mock(MiscUtils.class));
+            new VoteParty(plugin).giveRewards(user, false);
+        }
+        assertEquals(2, storage.size());
+        assertEquals(Collections.emptyList(), order);
+        when(serverData.isLastVotePartySameDay()).thenReturn(true);
+        storage.remove().run();
+        assertEquals(List.of("enumeration", "eligibility"), order);
+        storage.remove().run();
+        assertEquals(List.of("enumeration", "eligibility", "reset"), order);
+        ArgumentCaptor<Runnable> completion = ArgumentCaptor.forClass(Runnable.class);
+        verify(scheduler).runTask(eq(plugin), completion.capture());
+        completion.getValue().run();
+        ArgumentCaptor<RewardOptions> options = ArgumentCaptor.forClass(RewardOptions.class);
+        verify(apiRewards).giveReward(eq(user), eq(rewardData), eq(REWARD_PATH), options.capture());
+        assertEquals("true", options.getValue().getPlaceholders().get("FirstVotePartyToday"));
+    }
+
+    @Test
+    void rejectedStoredEnumerationDoesNotRunInlineOrSendRewards() {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        when(dataManager.deferSharedStorageWork(any()))
+                .thenThrow(new RejectedExecutionException("storage worker stopped"));
+
+        java.lang.reflect.InvocationTargetException failure = org.junit.jupiter.api.Assertions.assertThrows(
+                java.lang.reflect.InvocationTargetException.class, () -> dispatch("giveAllPlayers", null));
+        org.junit.jupiter.api.Assertions.assertInstanceOf(RejectedExecutionException.class, failure.getCause());
+        verify(votingUsers, never()).getAllUUIDs();
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewardData), eq(REWARD_PATH), any(RewardOptions.class));
+    }
+
+    @Test
+    void storedEnumerationFailureIsNotRetriedOrTreatedAsEmptyRecipients() throws Exception {
+        when(dataManager.hasSharedSqlBackend()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        AtomicReference<Runnable> enumeration = new AtomicReference<>();
+        when(dataManager.deferSharedStorageWork(any())).thenAnswer(call -> {
+            enumeration.set(call.getArgument(0));
+            return true;
+        });
+        when(votingUsers.getAllUUIDs()).thenThrow(new IllegalStateException("storage unavailable"));
+
+        dispatch("giveAllPlayers", null);
+        org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException.class, () -> enumeration.get().run());
+
+        verify(votingUsers, times(1)).getAllUUIDs();
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewardData), eq(REWARD_PATH), any(RewardOptions.class));
+    }
+
+    @Test
+    void nonsharedStoredRecipientsRemainImmediateAndLiveRecipientIsNotDuplicated() throws Exception {
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        Player online = mock(Player.class);
+        when(online.getUniqueId()).thenReturn(USER_UUID);
+        when(votingUsers.getVotingPluginUser(online)).thenReturn(user);
+        bukkit.when(Bukkit::getOnlinePlayers).thenReturn(Collections.singletonList(online));
+        UUID offlineId = UUID.fromString("71bc126a-0323-42bb-8334-f31f5fb7f455");
+        VotingPluginUser offline = mock(VotingPluginUser.class);
+        when(votingUsers.getVotingPluginUser(offlineId, true)).thenReturn(offline);
+        when(votingUsers.getAllUUIDs()).thenReturn(new ArrayList<>(List.of(USER_UUID.toString(), offlineId.toString())));
+
+        dispatch("giveAllPlayers", null);
+
+        verify(voteParty, times(1)).giveReward(user, false);
+        verify(voteParty, times(1)).giveReward(offline, false);
+        verify(offline).userDataFetechMode(UserDataFetchMode.NO_CACHE);
+        verify(dataManager, never()).deferSharedStorageWork(any());
+    }
+
     private Deferred deferSharedRead() {
         Deferred deferred = new Deferred();
         doAnswer(call -> {

VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/specialrewards/VotePartyStoredRecipientFailureTest.java

@@ -0,0 +1,557 @@
+package com.bencodez.votingplugin.tests.specialrewards;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
+import org.bukkit.Server;
+import org.bukkit.command.SimpleCommandMap;
+import org.bukkit.plugin.SimplePluginManager;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+
+import org.bukkit.Bukkit;
+import org.bukkit.entity.Entity;
+import org.bukkit.entity.Player;
+import org.bukkit.configuration.file.YamlConfiguration;
+import org.bukkit.plugin.PluginManager;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import com.bencodez.advancedcore.AdvancedCorePlugin.UserStorageOwner;
+import com.bencodez.advancedcore.api.misc.MiscUtils;
+import com.bencodez.advancedcore.api.rewards.RewardHandler;
+import com.bencodez.advancedcore.api.rewards.RewardOptions;
+import com.bencodez.advancedcore.api.user.UserStorage;
+import com.bencodez.advancedcore.api.user.usercache.UserDataCache;
+import com.bencodez.advancedcore.api.user.usercache.UserDataManager;
+import com.bencodez.advancedcore.api.user.userstorage.mysql.MySQL;
+import com.bencodez.simpleapi.scheduler.BukkitScheduler;
+import com.bencodez.simpleapi.sql.Column;
+import com.bencodez.simpleapi.sql.DataType;
+import com.bencodez.simpleapi.sql.data.DataValueString;
+import com.bencodez.votingplugin.VotingPluginMain;
+import com.bencodez.votingplugin.config.SpecialRewardsConfig;
+import com.bencodez.votingplugin.data.ServerData;
+import com.bencodez.votingplugin.specialrewards.voteparty.VoteParty;
+import com.bencodez.votingplugin.user.UserManager;
+import com.bencodez.votingplugin.user.VotingPluginUser;
+
+class VotePartyStoredRecipientFailureTest {
+    private static final UUID ID = UUID.fromString("9345fd57-4cc0-4492-ad39-0bbea6ca7525");
+    private final ArrayDeque<Runnable> storage = new ArrayDeque<>();
+    private final ArrayDeque<Runnable> voteTasks = new ArrayDeque<>();
+    private final AtomicReference<UserStorageOwner> currentOwner = new AtomicReference<>();
+    private final ArrayDeque<Runnable> platform = new ArrayDeque<>();
+    private final AtomicInteger votes = new AtomicInteger(2);
+    private volatile boolean worker;
+    private PluginManager events;
+    private final java.util.concurrent.atomic.AtomicBoolean lastParty = new java.util.concurrent.atomic.AtomicBoolean();
+    private VotingPluginMain plugin;
+    private UserDataManager manager;
+    private MySQL mysql;
+    private SpecialRewardsConfig config;
+    private ServerData serverData;
+    private YamlConfiguration state;
+    private YamlConfiguration rewards;
+    private RewardHandler apiRewards;
+    private VotingPluginUser user;
+    private VoteParty party;
+    private com.bencodez.advancedcore.api.user.UserManager coreUsers;
+    private MockedStatic<Bukkit> bukkit;
+    private MockedStatic<MiscUtils> misc;
+    private Consumer<Object> firstSuccess;
+    private Consumer<Throwable> firstFailure;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() {
+        plugin = mock(VotingPluginMain.class);
+        coreUsers = mock(com.bencodez.advancedcore.api.user.UserManager.class);
+        manager = mock(UserDataManager.class);
+        mysql = mock(MySQL.class);
+        config = mock(SpecialRewardsConfig.class);
+        serverData = mock(ServerData.class);
+        user = mock(VotingPluginUser.class);
+        UserManager votingUsers = mock(UserManager.class);
+        apiRewards = mock(RewardHandler.class);
+        state = new YamlConfiguration();
+        state.set("VoteParty.Total", 3);
+        state.set("VoteParty.Voted", List.of(ID.toString()));
+        rewards = new YamlConfiguration();
+        when(plugin.getUserManager()).thenReturn(coreUsers);
+        when(coreUsers.getDataManager()).thenReturn(manager);
+        when(plugin.getStorageType()).thenReturn(UserStorage.MYSQL);
+        when(plugin.getSpecialRewardsConfig()).thenReturn(config);
+        when(plugin.getServerData()).thenReturn(serverData);
+        when(plugin.getLogger()).thenReturn(mock(Logger.class));
+        when(plugin.getVotingPluginUserManager()).thenReturn(votingUsers);
+        when(plugin.getRewardHandler()).thenReturn(apiRewards);
+        when(plugin.getPlaceholders()).thenReturn(mock(com.bencodez.votingplugin.placeholders.PlaceHolders.class));
+        doAnswer(c -> { votes.set(c.getArgument(0)); return null; }).when(user).setVotePartyVotes(anyInt());
+        when(config.isVotePartyGiveAllPlayers()).thenReturn(true);
+        when(config.isVotePartyGiveOnlinePlayersOnly()).thenReturn(false);
+        when(config.getVotePartyUserVotesRequired()).thenReturn(2);
+        when(config.getVotePartyVotesRequired()).thenReturn(3);
+        when(config.isVotePartyResetCount()).thenReturn(true);
+        when(config.getData()).thenReturn(rewards);
+        when(config.getVotePartyRewardsPath()).thenReturn("VoteParty.Rewards");
+        when(serverData.getData()).thenReturn(state);
+        when(serverData.isLastVotePartySameDay()).thenAnswer(c -> lastParty.get());
+        doAnswer(c -> { lastParty.set(true); return null; }).when(serverData).updateLastVoteParty();
+        doAnswer(c -> { state.set(c.getArgument(0), c.getArgument(1)); return null; })
+                .when(serverData).setStringList(anyString(), anyList());
+        doAnswer(c -> { state.set(c.getArgument(0), c.getArgument(1)); return null; })
+                .when(serverData).setInt(anyString(), anyInt());
+        when(manager.hasSharedSqlBackend()).thenReturn(true);
+        when(manager.isPlatformOwnedThread()).thenAnswer(c -> !worker);
+        when(manager.getPublishedCache(ID)).thenReturn(mock(UserDataCache.class));
+        UserStorageOwner owner = new UserStorageOwner(UserStorage.MYSQL, mysql, null);
+        currentOwner.set(owner);
+        when(manager.sharedNativeUserStorageOwner()).thenAnswer(c -> currentOwner.get());
+        when(manager.withSharedNativeUserStorage(any())).thenAnswer(c -> {
+            assertTrue(worker, "native storage admission must stay off the platform");
+            return ((Function<UserStorageOwner, ?>) c.getArgument(0)).apply(currentOwner.get());
+        });
+        org.mockito.stubbing.Answer<Boolean> deferResult = c -> {
+            if (worker && c.getArguments().length == 4) return false;
+            Supplier<Object> work = c.getArgument(0);
+            Consumer<Object> success = c.getArgument(1);
+            Consumer<Throwable> failure = c.getArgument(2);
+            if (firstSuccess == null) { firstSuccess = success; firstFailure = failure; }
+            storage.add(() -> {
+                try { Object result = work.get(); platform.add(() -> success.accept(result)); }
+                catch (RuntimeException problem) { platform.add(() -> failure.accept(problem)); }
+            });
+            return true;
+        };
+        when(manager.deferSharedStorageResult(any(), any(), any(), nullable(Entity.class))).thenAnswer(deferResult);
+        when(manager.deferSharedStorageResultFromPlatform(any(), any(), any())).thenAnswer(deferResult);
+        when(manager.deferSharedStorageWork(any())).thenAnswer(c -> {
+            if (worker) return false;
+            storage.add(c.getArgument(0));
+            return true;
+        });
+        doAnswer(c -> { votes.set(0); return null; }).when(coreUsers)
+                .removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        when(user.getJavaUUID()).thenReturn(ID);
+        when(user.getVotePartyVotes()).thenAnswer(c -> votes.get());
+        when(user.getPlugin()).thenReturn(plugin);
+        when(user.isOnline()).thenReturn(true);
+        when(votingUsers.getVotingPluginUser(ID, true)).thenReturn(user);
+        // This is the resolved legacy API's failed-empty result. The nullable
+        // native query below still distinguishes failure from successful-empty.
+        when(votingUsers.getAllUUIDs()).thenReturn(new ArrayList<>());
+        BukkitScheduler scheduler = mock(BukkitScheduler.class);
+        when(plugin.getBukkitScheduler()).thenReturn(scheduler);
+        doAnswer(c -> { platform.add(c.getArgument(1)); return null; }).when(scheduler)
+                .runTask(eq(plugin), any(Runnable.class));
+        doAnswer(c -> { platform.add(c.getArgument(1)); return null; }).when(scheduler)
+                .runTask(eq(plugin), any(Runnable.class), any(Entity.class));
+        bukkit = mockStatic(Bukkit.class);
+        Player online = mock(Player.class);
+        when(online.getUniqueId()).thenReturn(ID);
+        when(online.getName()).thenReturn("SyntheticVoter");
+        when(votingUsers.getVotingPluginUser(online)).thenReturn(user);
+        bukkit.when(Bukkit::getOnlinePlayers).thenReturn(List.of(online));
+        Server server = mock(Server.class);
+        when(server.isPrimaryThread()).thenAnswer(c -> !worker);
+        events = new SimplePluginManager(server, mock(SimpleCommandMap.class));
+        bukkit.when(Bukkit::getPluginManager).thenReturn(events);
+        bukkit.when(Bukkit::isPrimaryThread).thenAnswer(c -> !worker);
+        ScheduledExecutorService voteTimer = mock(ScheduledExecutorService.class);
+        when(plugin.getVoteTimer()).thenReturn(voteTimer);
+        doAnswer(c -> { voteTasks.add(c.getArgument(0)); return null; }).when(voteTimer).execute(any());
+        misc = mockStatic(MiscUtils.class);
+        misc.when(MiscUtils::getInstance).thenReturn(mock(MiscUtils.class));
+        party = new VoteParty(plugin);
+    }
+
+    @AfterEach
+    void closeStatics() { misc.close(); bukkit.close(); }
+
+    private void drain() {
+        for (int phase = 0; phase < 12 && (!storage.isEmpty() || !platform.isEmpty() || !voteTasks.isEmpty()); phase++) {
+            while (!voteTasks.isEmpty()) {
+                worker = true;
+                try { voteTasks.remove().run(); } finally { worker = false; }
+            }
+            while (!storage.isEmpty()) {
+                worker = true;
+                try { storage.remove().run(); } finally { worker = false; }
+            }
+            while (!platform.isEmpty()) platform.remove().run();
+        }
+        assertTrue(storage.isEmpty() && platform.isEmpty() && voteTasks.isEmpty(), "bounded party work must settle");
+    }
+
+    private void assertUnconsumed() {
+        assertEquals(List.of(ID.toString()), party.getVotedUsers());
+        assertEquals(2, votes.get());
+        verify(coreUsers, never()).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+        verify(serverData, never()).updateLastVoteParty();
+    }
+
+    @Test
+    void failedEmptyRetainsEligibilityAndRetryDoesNotRepeatOnlineReward() {
+        when(mysql.getRowsQuery()).thenReturn(null);
+        party.giveRewards(user, false);
+        drain();
+        assertUnconsumed();
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>(List.of(new Column("uuid", new DataValueString(ID.toString())))));
+        party.giveRewards(user, false);
+        drain();
+        assertEquals(List.of(), party.getVotedUsers());
+        assertEquals(0, votes.get());
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), eq("VoteParty.Rewards"), any(RewardOptions.class));
+        verify(coreUsers, times(1)).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        verify(mysql, times(2)).getRowsQuery();
+        verify(plugin.getVotingPluginUserManager(), never()).getAllUUIDs();
+    }
+
+    @Test
+    void successfulEmptyCompletesAndStillRewardsTheLiveRecipient() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        drain();
+        assertEquals(List.of(), party.getVotedUsers());
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+        verify(coreUsers, times(1)).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+    }
+
+    @Test
+    void failedCheckDoesNotSpendGlobalCountOrFirstPartyTimestamp() {
+        when(mysql.getRowsQuery()).thenReturn(null);
+        party.check(user, false);
+        drain();
+        assertUnconsumed();
+        assertEquals(3, party.getTotalVotes());
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.check(user, false);
+        drain();
+        assertEquals(0, party.getTotalVotes());
+        verify(serverData, times(1)).updateLastVoteParty();
+        org.mockito.ArgumentCaptor<RewardOptions> options = org.mockito.ArgumentCaptor.forClass(RewardOptions.class);
+        verify(apiRewards).giveReward(eq(user), eq(rewards), anyString(), options.capture());
+        assertEquals("true", options.getValue().getPlaceholders().get("FirstVotePartyToday"));
+    }
+
+    @Test
+    void thrownEnumerationFailureAlsoLeavesPartyAvailableForRetry() {
+        when(mysql.getRowsQuery()).thenThrow(new IllegalStateException("synthetic storage failure"));
+        party.giveRewards(user, false);
+        drain();
+        assertUnconsumed();
+        doReturn(new ArrayList<>()).when(mysql).getRowsQuery();
+        party.giveRewards(user, false);
+        drain();
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void lateCallbacksFromFailedAttemptCannotConsumeAnotherParty() {
+        when(mysql.getRowsQuery()).thenReturn(null);
+        party.giveRewards(user, false);
+        drain();
+        assertUnconsumed();
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        firstSuccess.accept(null);
+        firstFailure.accept(new IllegalStateException("late failure"));
+        assertEquals(List.of(ID.toString()), party.getVotedUsers());
+        drain();
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+        verify(coreUsers, times(1)).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+    }
+    @Test
+    void rejectedAdmissionDoesNotRunQueryInlineOrConsumeParty() {
+        when(manager.deferSharedStorageResultFromPlatform(any(), any(), any()))
+                .thenThrow(new RejectedExecutionException("synthetic worker stopped"));
+        party.giveRewards(user, false);
+        assertUnconsumed();
+        verify(mysql, never()).getRowsQuery();
+        doReturn(false).when(manager).deferSharedStorageResultFromPlatform(any(), any(), any());
+        party.giveRewards(user, false);
+        assertUnconsumed();
+        verify(mysql, never()).getRowsQuery();
+    }
+
+    @Test
+    void pendingPreparationAndResetExcludeAnotherForceRequest() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        party.giveRewards(user, false);
+        assertEquals(1, storage.size());
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        int admissions = storage.size();
+        party.giveRewards(user, false);
+        assertEquals(admissions, storage.size());
+        drain();
+        verify(mysql, times(1)).getRowsQuery();
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void manualResetInvalidatesPreparedResultBeforeItCanTouchNewParty() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        party.reset(true);
+        votes.set(7);
+        party.setVotedUsers(List.of("new-party"));
+        drain();
+        assertEquals(7, votes.get());
+        assertEquals(List.of("new-party"), party.getVotedUsers());
+        verify(coreUsers, times(1)).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void workerOriginatedPreparationNeverQueriesWhileVoteMonitorIsHeld() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        synchronized (party) {
+            worker = true;
+            try { party.giveRewards(user, false); } finally { worker = false; }
+            verify(mysql, never()).getRowsQuery();
+            assertEquals(1, storage.size());
+        }
+        drain();
+        verify(mysql, times(1)).getRowsQuery();
+    }
+
+    @Test
+    void invalidIdentityDoesNotTurnAnIncompleteEnumerationIntoSuccessfulEmpty() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>(List.of(new Column("uuid", new DataValueString("invalid")))));
+        party.giveRewards(user, false);
+        drain();
+        assertUnconsumed();
+    }
+
+    @Test
+    void queuedOldCounterWipeCannotClearCountersAfterManualPartyReset() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        platform.remove().run(); // Recipient reads and the old wipe are admitted.
+        party.reset(true);
+        votes.set(7);
+        party.setVotedUsers(List.of("new-party"));
+        drain();
+        assertEquals(7, votes.get());
+        assertEquals(List.of("new-party"), party.getVotedUsers());
+        verify(coreUsers, times(1)).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+    }
+
+    @Test
+    void ownerReplacementAfterQueryCannotCommitOrResetReplacementStorage() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.check(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        currentOwner.set(new UserStorageOwner(UserStorage.MYSQL, mock(MySQL.class), null));
+        drain();
+        assertUnconsumed();
+        assertEquals(3, party.getTotalVotes());
+    }
+
+    @Test
+    void ownerReplacementBeforeDeliveryDoesNotReadOrResetReplacementEligibility() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        currentOwner.set(new UserStorageOwner(UserStorage.MYSQL, mock(MySQL.class), null));
+        drain();
+        assertEquals(2, votes.get());
+        verify(coreUsers, never()).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void platformCommitNeverWaitsForVoteStorageAndLeavesStateRetryable() throws Exception {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.check(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        Runnable commit = platform.remove();
+        java.util.concurrent.CountDownLatch storageEntered = new java.util.concurrent.CountDownLatch(1);
+        java.util.concurrent.CountDownLatch releaseStorage = new java.util.concurrent.CountDownLatch(1);
+        java.util.concurrent.CountDownLatch platformDone = new java.util.concurrent.CountDownLatch(1);
+        when(user.getVotePartyVotes()).thenAnswer(c -> {
+            storageEntered.countDown();
+            if (!releaseStorage.await(5, java.util.concurrent.TimeUnit.SECONDS))
+                throw new AssertionError("storage fixture was not released");
+            return votes.get();
+        });
+        AtomicReference<Throwable> voteFailure = new AtomicReference<>();
+        AtomicReference<Throwable> platformFailure = new AtomicReference<>();
+        Thread incomingVote = new Thread(() -> {
+            worker = true;
+            try { party.addTotal(user); } catch (Throwable error) { voteFailure.set(error); }
+            finally { worker = false; }
+        });
+        Thread platformAttempt = new Thread(() -> {
+            worker = false;
+            try (MockedStatic<Bukkit> threadBukkit = mockStatic(Bukkit.class)) { commit.run(); }
+            catch (Throwable error) { platformFailure.set(error); }
+            finally { platformDone.countDown(); }
+        });
+        incomingVote.start();
+        boolean completedWithoutStorage;
+        try {
+            assertTrue(storageEntered.await(5, java.util.concurrent.TimeUnit.SECONDS));
+            platformAttempt.start();
+            completedWithoutStorage = platformDone.await(1, java.util.concurrent.TimeUnit.SECONDS);
+        } finally { releaseStorage.countDown(); }
+        incomingVote.join(5000);
+        platformAttempt.join(5000);
+        assertFalse(incomingVote.isAlive());
+        assertFalse(platformAttempt.isAlive());
+        assertNull(voteFailure.get());
+        assertNull(platformFailure.get());
+        assertTrue(completedWithoutStorage, "the platform must not wait for a storage-held vote monitor");
+        assertEquals(4, party.getTotalVotes());
+        assertEquals(List.of(ID.toString()), party.getVotedUsers());
+        verify(serverData, never()).updateLastVoteParty();
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+        party.check(user, false);
+        drain();
+        assertEquals(1, party.getTotalVotes());
+        verify(apiRewards, times(1)).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void realAsyncEventContractRejectsPrimaryThreadAndAllowsWorkerThread() {
+        assertThrows(IllegalStateException.class,
+                () -> events.callEvent(new com.bencodez.votingplugin.events.VotePartyEvent()));
+        worker = true;
+        try { assertDoesNotThrow(() -> events.callEvent(new com.bencodez.votingplugin.events.VotePartyEvent())); }
+        finally { worker = false; }
+    }
+
+    @Test
+    void replacementAfterAdmissionsCannotReadOrWipeNewOwnersCounters() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.giveRewards(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        currentOwner.set(new UserStorageOwner(UserStorage.MYSQL, mock(MySQL.class), null));
+        votes.set(7);
+        worker = true;
+        try {
+            storage.remove().run(); // The per-user read reports failure, without granting a reward.
+            assertThrows(IllegalStateException.class, () -> storage.remove().run()); // Stored loop.
+            assertThrows(IllegalStateException.class, () -> storage.remove().run()); // Counter wipe.
+        } finally { worker = false; }
+        while (!platform.isEmpty()) platform.remove().run();
+        assertEquals(7, votes.get());
+        verify(coreUsers, never()).removeAllKeyValues("VotePartyVotes", DataType.INTEGER);
+        verify(apiRewards, never()).giveReward(eq(user), eq(rewards), anyString(), any(RewardOptions.class));
+    }
+
+    @Test
+    void rejectedPlatformDispatchRetainsCountsVotersAndFirstPartyMetadata() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        BukkitScheduler scheduler = plugin.getBukkitScheduler();
+        doThrow(new RejectedExecutionException("synthetic platform stopped"))
+                .when(scheduler).runTask(eq(plugin), any(Runnable.class));
+        party.check(user, false);
+        drain();
+        assertUnconsumed();
+        assertEquals(3, party.getTotalVotes());
+    }
+
+    @Test
+    void replacementAfterEventBeforeDispatchLeavesPartyRetryable() {
+        when(mysql.getRowsQuery()).thenReturn(new ArrayList<>());
+        party.check(user, false);
+        worker = true;
+        try { storage.remove().run(); } finally { worker = false; }
+        platform.remove().run();
+        worker = true;
+        try { voteTasks.remove().run(); } finally { worker = false; }
+        currentOwner.set(new UserStorageOwner(UserStorage.MYSQL, mock(MySQL.class), null));
+        drain();
+        assertUnconsumed();
+        assertEquals(3, party.getTotalVotes());
+    }
+
+    @Test
+    void countCommandDoesNotWaitForInFlightVoteStorage() throws Exception {
+        assertMetadataCommandDoesNotWait(false);
+    }
+
+    @Test
+    void resetCommandDoesNotWaitForInFlightVoteStorage() throws Exception {
+        assertMetadataCommandDoesNotWait(true);
+    }
+
+    private void assertMetadataCommandDoesNotWait(boolean reset) throws Exception {
+        java.util.concurrent.CountDownLatch entered = new java.util.concurrent.CountDownLatch(1);
+        java.util.concurrent.CountDownLatch release = new java.util.concurrent.CountDownLatch(1);
+        java.util.concurrent.CountDownLatch changed = new java.util.concurrent.CountDownLatch(1);
+        when(user.getVotePartyVotes()).thenAnswer(c -> {
+            entered.countDown();
+            if (!release.await(5, java.util.concurrent.TimeUnit.SECONDS)) throw new AssertionError("storage not released");
+            return votes.get();
+        });
+        AtomicReference<Throwable> failure = new AtomicReference<>();
+        Thread incoming = new Thread(() -> {
+            try { party.addTotal(user); } catch (Throwable error) { failure.set(error); }
+        });
+        Thread command = new Thread(() -> {
+            try {
+                if (reset) party.reset(true); else party.setTotalVotes(8);
+            } catch (Throwable error) { failure.set(error); }
+            finally { changed.countDown(); }
+        });
+        incoming.start();
+        boolean immediate;
+        try {
+            assertTrue(entered.await(5, java.util.concurrent.TimeUnit.SECONDS));
+            command.start();
+            immediate = changed.await(1, java.util.concurrent.TimeUnit.SECONDS);
+        } finally { release.countDown(); }
+        incoming.join(5000);
+        command.join(5000);
+        assertFalse(incoming.isAlive());
+        assertFalse(command.isAlive());
+        assertNull(failure.get());
+        assertTrue(immediate, "scalar commands must not wait for user storage");
+        assertEquals(reset ? 0 : 8, party.getTotalVotes());
+    }
+
+}