Make period changes recoverable across shutdown

VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java

@@ -736,12 +736,12 @@ public void onPostLoad() {
 
 		loadVoteBroadcast();
 
+		specialRewards = new SpecialRewards(this);
 		loadVoteShopManager();
 
 		loadDirectlyDefined();
 		checkUpdate = new CheckUpdate(this);
 		checkUpdate.startUp();
-		specialRewards = new SpecialRewards(this);
 		signs = new Signs(this);
 
 		coolDownCheck.checkEnabled();
@@ -2016,7 +2016,8 @@ private void registerEvents() {
 		pm.registerEvents(new PlayerJoinEvent(this), this);
 		if (isVotifierLoaded()) {
 			VotiferEvent votifierEvent = new VotiferEvent(this);
-			votifierVoteOverflowQueue = new VotifierVoteOverflowQueue(this, votifierEvent::processVote);
+			votifierVoteOverflowQueue = new VotifierVoteOverflowQueue(this,
+					(serviceSite, username, voteId) -> votifierEvent.processQueuedVoteOutcome(serviceSite, username, voteId));
 			pm.registerEvents(votifierEvent, this);
 		}
 		pm.registerEvents(new PlayerVoteListener(this), this);

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendOrderedVoteOverflowQueue.java

@@ -341,11 +341,26 @@ PendingEnvelope peekDurable() {
 
 	/** Persists completion off the platform thread before removing the durable head. */
 	void acknowledgeAsync(PendingEnvelope expected, Consumer<Boolean> completion) {
+		completeHeadAsync(expected, null, completion);
+	}
+
+	/** Atomically replaces the durable head with follow-up work. */
+	void replaceAsync(PendingEnvelope expected, JsonEnvelope replacement, Consumer<Boolean> completion) {
+		PendingEnvelope pendingReplacement = pending(replacement);
+		if (pendingReplacement == null) {
+			completion.accept(false);
+			return;
+		}
+		completeHeadAsync(expected, pendingReplacement, completion);
+	}
+
+	private void completeHeadAsync(PendingEnvelope expected, PendingEnvelope replacement,
+			Consumer<Boolean> completion) {
 		PendingAcknowledgement request = null;
 		synchronized (lock) {
 			if (!closeRequested.get() && !closed && !loadFailed && expected != null && entries.peekFirst() == expected
 					&& pendingAcknowledgement == null && pendingFailure == null) {
-				request = new PendingAcknowledgement(expected, completion);
+				request = new PendingAcknowledgement(expected, replacement, completion);
 				pendingAcknowledgement = request;
 			}
 		}
@@ -382,13 +397,15 @@ private Boolean acknowledge(PendingAcknowledgement request) {
 				}
 				List<String> active = payloadSnapshotLocked();
 				active.remove(0);
+				if (request.replacement != null) active.add(0, request.replacement.payload);
 				try {
 					writeSnapshotLocked(active, failedSnapshotLocked());
 				} catch (IOException failure) {
 					warn("Unable to acknowledge ordered proxy vote overflow", failure);
 					return false;
 				}
 				entries.removeFirst();
+				if (request.replacement != null) entries.addFirst(request.replacement);
 				durableVersion = ++stateVersion;
 				request.stored = true;
 				pendingAcknowledgement = null;
@@ -464,7 +481,8 @@ private static boolean isOrderedVoteMessage(JsonEnvelope envelope) {
 		return VotingPluginWire.SUB_VOTE.equals(subChannel)
 				|| VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel)
 				|| VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel)
-				|| VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(subChannel);
+				|| VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(subChannel)
+				|| BackendProxyHandler.isLegacyDeliveryRetirement(envelope);
 	}
 
 	private void requestPersistenceLocked() {
@@ -689,7 +707,12 @@ private void closeBlocking() {
 					&& (closingFailure.expected == null || entries.peekFirst() == closingFailure.expected);
 			canStoreAcknowledgement = closingAcknowledgement != null && !closingAcknowledgement.stored
 					&& entries.peekFirst() == closingAcknowledgement.expected;
-			if (canStoreAcknowledgement) snapshot.remove(0);
+			if (canStoreAcknowledgement) {
+				snapshot.remove(0);
+				if (closingAcknowledgement.replacement != null) {
+					snapshot.add(0, closingAcknowledgement.replacement.payload);
+				}
+			}
 			if (canStoreFailure) {
 				if (closingFailure.expected != null) snapshot.remove(0);
 				if (!closingFailureAlreadyStored) failures.add(closingFailure.failed.payload);
@@ -733,6 +756,9 @@ private void closeBlocking() {
 		if (saved && canStoreAcknowledgement) {
 			synchronized (lock) {
 				entries.removeFirst();
+				if (closingAcknowledgement.replacement != null) {
+					entries.addFirst(closingAcknowledgement.replacement);
+				}
 				closingAcknowledgement.stored = true;
 			}
 		}
@@ -756,12 +782,15 @@ enum CloseState { OPEN, CLOSING, WAITING_TO_PERSIST, SNAPSHOT_REPLACED, COMPLETE
 
 	private static final class PendingAcknowledgement {
 		private final PendingEnvelope expected;
+		private final PendingEnvelope replacement;
 		private final Consumer<Boolean> completion;
 		private final AtomicBoolean completed = new AtomicBoolean();
 		private boolean stored;
 
-		private PendingAcknowledgement(PendingEnvelope expected, Consumer<Boolean> completion) {
+		private PendingAcknowledgement(PendingEnvelope expected, PendingEnvelope replacement,
+				Consumer<Boolean> completion) {
 			this.expected = expected;
+			this.replacement = replacement;
 			this.completion = completion;
 		}
 

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java

@@ -31,13 +31,16 @@
 import com.bencodez.votingplugin.backendproxy.voteparty.BackendVotePartySync;
 import com.bencodez.votingplugin.proxy.BungeeMethod;
 import com.bencodez.votingplugin.proxy.VotingPluginWire;
+import com.bencodez.votingplugin.util.VoteTaskAdmission;
+import com.bencodez.votingplugin.voteshop.service.VoteShopPurchaseService;
 
 import lombok.Getter;
 
 /**
  * Coordinates backend/proxy communication components.
  */
 public class BackendProxyHandler implements Listener {
+	private static final String SUB_LEGACY_DELIVERY_RETIREMENT = "LegacyVoteDeliveryRetirement";
 
 	private final VotingPluginMain plugin;
 	static final int MAX_ORDERED_VOTE_QUEUE = 256;
@@ -360,7 +363,8 @@ private void scheduleOrderedVoteDispatchLocked() {
 		}
 		orderedVoteDispatchActive = true;
 		try {
-			plugin.getBukkitScheduler().runTaskAsynchronously(plugin, this::runNextOrderedVoteDispatch);
+			plugin.getBukkitScheduler().runTaskAsynchronously(plugin,
+					VoteTaskAdmission.ownedTask(this::runNextOrderedVoteDispatch));
 		} catch (RuntimeException schedulingFailure) {
 			orderedVoteDispatchActive = false;
 			orderedVoteDispatch.notifyAll();
@@ -433,6 +437,16 @@ private void runNextOrderedVoteDispatch() {
 			}
 		};
 		try {
+			if (isLegacyDeliveryRetirement(next)) {
+				try {
+					completeLegacyDeliveryRetirement(next);
+					complete.accept(OrderedVoteOutcome.COMPLETE_WITHOUT_RETIREMENT);
+				} catch (RuntimeException retirementFailure) {
+					plugin.debug(retirementFailure);
+					complete.accept(OrderedVoteOutcome.RETRY);
+				}
+				return;
+			}
 			if (dispatchDurableReceiptRelease(messageRouter, next)) {
 				complete.accept(OrderedVoteOutcome.COMPLETE);
 				return;
@@ -464,16 +478,30 @@ private void finishOrderedVoteDispatch(BackendOrderedVoteOverflowQueue.PendingEn
 			}
 			return;
 		}
-		boolean successful = outcome == OrderedVoteOutcome.COMPLETE;
+		boolean successful = outcome == OrderedVoteOutcome.COMPLETE
+				|| outcome == OrderedVoteOutcome.COMPLETE_WITHOUT_RETIREMENT;
+		boolean retireDelivery = outcome == OrderedVoteOutcome.COMPLETE;
+		if (successful && retireDelivery && requiresLegacyDeliveryRetirement(envelope)) {
+			JsonEnvelope retirement = legacyDeliveryRetirement(envelope);
+			if (overflowEntry != null && orderedVoteOverflow != null) {
+				orderedVoteOverflow.replaceAsync(overflowEntry, retirement,
+						stored -> completeOrderedVoteAcknowledgement(stored, retirement, false));
+			} else {
+				replaceInMemoryVoteWithRetirement(envelope, retirement);
+			}
+			return;
+		}
 		if (successful && overflowEntry != null && orderedVoteOverflow != null) {
 			orderedVoteOverflow.acknowledgeAsync(overflowEntry,
-					stored -> completeOrderedVoteAcknowledgement(stored, envelope));
+					stored -> completeOrderedVoteAcknowledgement(stored, envelope, retireDelivery));
 			return;
 		}
+		boolean completedInMemory = false;
 		synchronized (orderedVoteDispatch) {
 			if (successful && overflowEntry == null
 					&& orderedVoteDispatchQueue.peekFirst() == orderedVoteDispatchInFlight) {
 				orderedVoteDispatchQueue.removeFirst();
+				completedInMemory = true;
 			}
 			orderedVoteDispatchInFlight = null;
 			orderedVoteOverflowInFlight = null;
@@ -482,10 +510,52 @@ private void finishOrderedVoteDispatch(BackendOrderedVoteOverflowQueue.PendingEn
 			if (successful) scheduleOrderedVoteDispatchLocked();
 			else retryOrderedVoteDispatchLocked();
 		}
-		if (successful) sendVoteDeliveryAcknowledgement(envelope);
+		if (completedInMemory && retireDelivery) completeVoteDelivery(envelope);
 	}
 
-	private void completeOrderedVoteAcknowledgement(boolean stored, JsonEnvelope envelope) {
+	private boolean requiresLegacyDeliveryRetirement(JsonEnvelope envelope) {
+		if (envelope == null || VotingPluginWire.requestsVoteDeliveryAcknowledgement(envelope)) return false;
+		if (!VotingPluginWire.SUB_VOTE.equals(envelope.getSubChannel())
+				&& !VotingPluginWire.SUB_VOTE_ONLINE.equals(envelope.getSubChannel())) return false;
+		return VotingPluginWire.resolveVoteId(VotingPluginWire.readVote(envelope)) != null;
+	}
+
+	static boolean isLegacyDeliveryRetirement(JsonEnvelope envelope) {
+		return envelope != null && SUB_LEGACY_DELIVERY_RETIREMENT.equals(envelope.getSubChannel());
+	}
+
+	private JsonEnvelope legacyDeliveryRetirement(JsonEnvelope envelope) {
+		UUID voteId = VotingPluginWire.resolveVoteId(VotingPluginWire.readVote(envelope));
+		return JsonEnvelope.builder(SUB_LEGACY_DELIVERY_RETIREMENT)
+				.put(VotingPluginWire.K_VOTE_ID, voteId.toString()).build();
+	}
+
+	private void completeLegacyDeliveryRetirement(JsonEnvelope envelope) {
+		VoteShopPurchaseService.completeVoteDelivery(plugin,
+				UUID.fromString(envelope.getFields().get(VotingPluginWire.K_VOTE_ID)));
+	}
+
+	private void replaceInMemoryVoteWithRetirement(JsonEnvelope envelope, JsonEnvelope retirement) {
+		synchronized (orderedVoteDispatch) {
+			if (orderedVoteDispatchInFlight != envelope || orderedVoteDispatchQueue.peekFirst() != envelope) {
+				orderedVoteQuarantineFailed = true;
+				orderedVoteDispatchPaused = true;
+				if (plugin != null && plugin.getLogger() != null) {
+					plugin.getLogger().severe("Unable to retain legacy proxy vote retirement; processing has stopped");
+				}
+			} else {
+				orderedVoteDispatchQueue.removeFirst();
+				orderedVoteDispatchQueue.addFirst(retirement);
+			}
+			orderedVoteDispatchInFlight = null;
+			orderedVoteOverflowInFlight = null;
+			orderedVoteDispatchActive = false;
+			orderedVoteDispatch.notifyAll();
+			scheduleOrderedVoteDispatchLocked();
+		}
+	}
+
+	private void completeOrderedVoteAcknowledgement(boolean stored, JsonEnvelope envelope, boolean retireDelivery) {
 		synchronized (orderedVoteDispatch) {
 			if (stored) {
 				orderedVoteDispatchInFlight = null;
@@ -501,7 +571,16 @@ private void completeOrderedVoteAcknowledgement(boolean stored, JsonEnvelope env
 			orderedVoteDispatch.notifyAll();
 			if (stored) scheduleOrderedVoteDispatchLocked();
 		}
-		if (stored) sendVoteDeliveryAcknowledgement(envelope);
+		if (stored && retireDelivery) completeVoteDelivery(envelope);
+	}
+
+	private void completeVoteDelivery(JsonEnvelope envelope) {
+		String subChannel = envelope == null ? null : envelope.getSubChannel();
+		if (!VotingPluginWire.SUB_VOTE.equals(subChannel)
+				&& !VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel)) return;
+		VoteShopPurchaseService.completeVoteDelivery(plugin,
+				VotingPluginWire.resolveVoteId(VotingPluginWire.readVote(envelope)));
+		sendVoteDeliveryAcknowledgement(envelope);
 	}
 
 	private void sendVoteDeliveryAcknowledgement(JsonEnvelope envelope) {

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java

@@ -99,6 +99,11 @@ public boolean reserve(UUID voteId) {
 		}
 	}
 
+	/** Releases an admission that failed before any vote side effects ran. */
+	public void release(UUID voteId) {
+		if (voteId != null) processedVotes.remove(voteId);
+	}
+
 	/** Persists successful processing before the backend emits a delivery acknowledgement. */
 	public boolean complete(UUID voteId) {
 		if (voteId == null) return true;

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/global/BackendGlobalDataSync.java

@@ -24,6 +24,7 @@
 import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope;
 import com.bencodez.simpleapi.sql.data.DataValue;
 import com.bencodez.simpleapi.sql.data.DataValueBoolean;
+import com.bencodez.simpleapi.sql.data.DataValueString;
 import com.bencodez.simpleapi.sql.mysql.config.MysqlConfigSpigot;
 import com.bencodez.votingplugin.VotingPluginMain;
 import com.bencodez.votingplugin.proxy.VotingPluginWire;
@@ -264,6 +265,10 @@ private boolean checkGlobalDataTime(TimeType type, HashMap<String, DataValue> da
 			expectedHandler.setBoolean(plugin.getBungeeSettings().getServer(), type.toString(), false);
 			return false;
 		}
+		String transitionId = data.containsKey(VotingPluginWire.timeChangeTransitionKey(type.toString()))
+				? data.get(VotingPluginWire.timeChangeTransitionKey(type.toString())).getString() : "";
+		boolean confirmsBoundary = transitionId != null && !transitionId.isBlank()
+				&& !VotingPluginWire.LEGACY_TIME_CHANGE_TRANSITION.equals(transitionId);
 		String serverName = plugin.getBungeeSettings().getServer();
 		GlobalDataHandler transitionHandler = admitTimeChange(type, serverName, expectedHandler);
 		if (transitionHandler == null) return false;
@@ -284,7 +289,7 @@ private boolean checkGlobalDataTime(TimeType type, HashMap<String, DataValue> da
 					return;
 				}
 				try {
-					finishTimeChange(transitionHandler, type, serverName);
+					finishTimeChange(transitionHandler, type, serverName, transitionId, confirmsBoundary);
 				} catch (RuntimeException failure) {
 					// finishTimeChange releases the pinned handler in its finally block.
 					plugin.debug(failure);
@@ -332,10 +337,19 @@ private GlobalDataHandler admitTimeChange(TimeType type, String serverName,
 		}
 	}
 
-	private void finishTimeChange(GlobalDataHandler handler, TimeType type, String serverName) {
+	private void finishTimeChange(GlobalDataHandler handler, TimeType type, String serverName,
+			String transitionId, boolean confirmsBoundary) {
 		boolean completed = false;
 		try {
-			handler.setBoolean(serverName, type.toString(), false);
+			if (confirmsBoundary) {
+				HashMap<String, DataValue> completion = new HashMap<>();
+				completion.put(type.toString(), new DataValueBoolean(false));
+				completion.put(VotingPluginWire.timeChangeBoundaryCapturedKey(type.toString()),
+						new DataValueString(transitionId));
+				handler.setData(serverName, completion);
+			} else {
+				handler.setBoolean(serverName, type.toString(), false);
+			}
 			JsonEnvelope.Builder builder = JsonEnvelope.builder("TimeChangeFinished")
 					.schema(VotingPluginWire.SCHEMA_VERSION);
 			builder.put("server", serverName);
@@ -500,17 +514,41 @@ public void load() {
 				"ForceUpdate", "VARCHAR(5)").entrySet()) {
 			loadedHandler.getGlobalMysql().alterColumnType(column.getKey(), column.getValue());
 		}
+		for (TimeType type : TimeType.values()) {
+			loadedHandler.getGlobalMysql().alterColumnType(
+					VotingPluginWire.timeChangeBoundaryCapturedKey(type.toString()), "VARCHAR(36)");
+			loadedHandler.getGlobalMysql().alterColumnType(
+					VotingPluginWire.timeChangeTransitionKey(type.toString()), "VARCHAR(36)");
+		}
+		loadedHandler.getGlobalMysql().alterColumnType(
+				VotingPluginWire.TIME_CHANGE_BOUNDARY_PROTOCOL_KEY, "VARCHAR(32)");
 		acceptingWork = true;
 		timer = Executors.newSingleThreadScheduledExecutor(task -> {
 			Thread thread = new Thread(task, "VotingPlugin-GlobalData");
 			thread.setDaemon(true);
 			return thread;
 		});
 		timer.scheduleWithFixedDelay(this::checkGlobalData, 60, 10, TimeUnit.SECONDS);
-		timer.scheduleWithFixedDelay(this::updateLastOnline, 1, 60, TimeUnit.MINUTES);
+		publishBoundaryProtocolHeartbeat();
+		timer.scheduleWithFixedDelay(this::publishBoundaryProtocolHeartbeat, 1, 60, TimeUnit.MINUTES);
 		plugin.getTimeChecker().setProcessingEnabled(false);
 	}
 
+	private void publishBoundaryProtocolHeartbeat() {
+		GlobalDataHandler pollingHandler = acquireHandlerReference();
+		if (pollingHandler == null) return;
+		try {
+			String heartbeat = "" + LocalDateTime.now().atZone(ZoneOffset.UTC).toInstant().toEpochMilli();
+			HashMap<String, DataValue> state = new HashMap<>();
+			state.put("LastOnline", new DataValueString(heartbeat));
+			state.put(VotingPluginWire.TIME_CHANGE_BOUNDARY_PROTOCOL_KEY,
+					new DataValueString(VotingPluginWire.timeChangeBoundaryProtocolHeartbeat(heartbeat)));
+			pollingHandler.setData(plugin.getBungeeSettings().getServer(), state);
+		} finally {
+			releaseHandlerReference(pollingHandler);
+		}
+	}
+
 	public void close() {
 		close(CLOSE_GRACE_SECONDS, TimeUnit.SECONDS);
 	}