Add owner-configured DateVoteMilestones with durable occurrence accounting

AGENTS.md

@@ -324,6 +324,7 @@ Control behavior in production.
 
 - Trace whether the code runs on the connector worker, proxy thread, Bukkit primary thread, or a SQL executor.
 - Preserve queued votes across saturation, reload, runtime replacement, scheduler rejection/cancellation, shutdown, and restart; overflow handling must be bounded, durable when promised, and observable rather than silently dropping work. Scheduler admission is never proof of durable ownership.
+- Local Votifier captures remain owned after STARTED until normal callback completion or confirmed overflow transfer. Transfer execution stays fenced while the original callback runs; successful non-interrupted completion reconciles only its exact occurrence, while interruption/exception or failed transfer retains bounded recovery ownership. A callback completing after the final overflow snapshot is an explicit uncertain reconciliation case; a retired queue must not overwrite a successor, and ordinary partial external effects retain the existing at-least-once boundary.
 - Proxy-to-backend guaranteed delivery is capability negotiated and at least once. Journal a reward-bearing envelope before
   reporting transport acceptance, retain it until the matching backend completion acknowledgement is durable, persist
   completed IDs before acknowledgement for restart-safe deduplication, retire receipts only through the durable

README.md

@@ -52,3 +52,5 @@ or disabling it; inspections gate disabled state immediately and report a newly
 ```
 
 `LATEST` resolves to the latest stable release.
+
+Owner-defined [DateVoteMilestones](docs/date-vote-milestones.md) count qualifying accepted votes within explicit date windows, independently of normal totals. This optional feature is disabled until events are configured.

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

@@ -331,7 +331,12 @@ private synchronized AtomicBoolean basicBungeeUpdateAdmission() {
 	private VotingPluginConfigHealth configHealth;
 	private VotifierIntegration votifierIntegration;
 	@Getter
-	private VotifierVoteOverflowQueue votifierVoteOverflowQueue;
+	private volatile VotifierVoteOverflowQueue votifierVoteOverflowQueue;
+    private VotiferEvent localVotifierListener;
+    private boolean acceptedVoteIngressRegistered;
+    private boolean timeQueueIngressRegistered;
+    private volatile boolean localVotifierIngressReady;
+    private volatile boolean localVotifierIngressClosed;
 	private VoteLogManager voteLogManager;
 	private VotingPluginWebhookManager webhookManager;
 
@@ -613,6 +618,11 @@ private void loadBungeeHandler() {
 	@Getter
 	private VoteStreakHandler voteStreakHandler;
 
+    @Getter
+    private final com.bencodez.votingplugin.specialrewards.datemilestones.DateVoteMilestones dateVoteMilestones =
+            new com.bencodez.votingplugin.specialrewards.datemilestones.DateVoteMilestones(this);
+
+
 	/**
 	 * Registers a directly editable rewards path from SpecialRewards.yml.
 	 *
@@ -626,6 +636,7 @@ public void loadDirectlyDefined() {
 		VoteSiteRewardRegistrar.register(this);
 		VotePartyRewardRegistrar.register(this);
 		VoteMilestoneRewardRegistrar.register(this);
+        dateVoteMilestones.reload();
 		VoteStreakRewardRegistrar.register(this);
 		TopVoterRewardRegistrar.register(this);
 		VoteShopRewardRegistrar.register(this);
@@ -697,8 +708,36 @@ private void metrics() {
 		new VotingPluginMetrics().load(plugin);
 	}
 
+    /** Publish accepted and transport ingress only after the complete vote pipeline is initialized. */
+    void initializeDateVoteIngress(Runnable openProxy) {
+        if (localVotifierIngressClosed) throw new IllegalStateException("Vote ingress is shutting down");
+        if (localVotifierIngressReady) return;
+        if (voteMilestonesManager == null || voteStreakHandler == null || voteParty == null || specialRewards == null
+                || topVoterHandler == null || voteShopManager == null || placeholders == null) {
+            throw new IllegalStateException("Accepted-vote handlers must initialize before transport ingress");
+        }
+        getDateVoteMilestones().reload();
+        if (!acceptedVoteIngressRegistered) {
+            getServer().getPluginManager().registerEvents(new PlayerVoteListener(this), this);
+            acceptedVoteIngressRegistered = true;
+        }
+        // Construction itself schedules persisted replay. Open it only after its consumer.
+        if (!timeQueueIngressRegistered && (!getBungeeSettings().isUseBungeecoord() || !getBungeeSettings().isGloblalDataEnabled())) {
+            timeQueueHandler = new TimeQueueHandler(this);
+            getServer().getPluginManager().registerEvents(timeQueueHandler, this);
+            timeQueueIngressRegistered = true;
+        }
+        if (getBungeeSettings().isUseBungeecoord()) openProxy.run();
+        localVotifierIngressReady = true;
+        VotifierVoteOverflowQueue overflow = votifierVoteOverflowQueue;
+        if (overflow != null) overflow.start();
+    }
+
 	@Override
 	public void onPostLoad() {
+        loadVoteTimer();
+        checkVotifier();
+        registerEarlyVotifierIngress();
 		ensureCommunicationSecret();
 		// auto conversion for Shop.yml
 		if (plugin.getShopFile().isJustCreated()) {
@@ -726,7 +765,6 @@ public void onPostLoad() {
 
 		voteTester = new VoteTester(plugin);
 
-		loadVoteTimer();
 		BoundedScheduledExecutor setupReminderWorker = new BoundedScheduledExecutor(1, 4);
 		setupReminderWorker.setThreadFactory(runnable -> {
 			Thread worker = new Thread(runnable, "VotingPlugin-SiteSetupReminder");
@@ -739,13 +777,8 @@ public void onPostLoad() {
 		autoCreateVoteSitesSetupReminder.start();
 		getVotingPluginUserManager().startSharedPointTransferRecovery();
 
-		if (bungeeSettings.isUseBungeecoord()) {
-			loadBungeeHandler();
-		}
+		// Proxy ingress opens below, after every accepted-vote handler is initialized.
 
-		if (!bungeeSettings.isUseBungeecoord() || !bungeeSettings.isGloblalDataEnabled()) {
-			this.timeQueueHandler = new TimeQueueHandler(this);
-		}
 
 		if (Bukkit.getPluginManager().getPlugin("PlaceholderAPI") != null) {
 			if (getConfigFile().isLoadInteralExpansion()) {
@@ -760,7 +793,6 @@ public void onPostLoad() {
 		voteStreakHandler.reload();
 
 		registerCommands();
-		checkVotifier();
 		registerEvents();
 		refreshPlaceholderPlayerPresence();
 
@@ -850,11 +882,7 @@ public Object getObject(OfflinePlayer player) {
 		}
 
 		VotingPluginRewardRegistrar.register(this);
-		// Recovered Votifier votes may now traverse the fully initialized vote,
-		// reward, placeholder, shop, and vote-party pipeline.
-		if (votifierVoteOverflowQueue != null) {
-			votifierVoteOverflowQueue.start();
-		}
+        initializeDateVoteIngress(this::loadBungeeHandler);
 
 		plugin.getLogger().info("Enabled VotingPlugin " + plugin.getDescription().getVersion());
 		if (plugin.getDescription().getVersion().contains("SNAPSHOT")) {
@@ -1995,6 +2023,9 @@ private void shutdownVoteReminders() {
 
 	@Override
 	public void onUnLoad() {
+        localVotifierIngressClosed = true;
+        localVotifierIngressReady = false;
+        if (localVotifierListener != null) localVotifierListener.stop();
 		stopAutoCreateVoteSitesSetupReminder();
 		placeholderPlayerPresence.clear();
 		stopBackendHostedControlLifecycle();
@@ -2076,19 +2107,24 @@ private void registerCommands() {
 
 	}
 
-	/**
-	 * Register events.
-	 */
+    /** Capture external events early; only the existing bounded workers may buffer/process them. */
+    void registerEarlyVotifierIngress() {
+        if (localVotifierIngressClosed || !isVotifierLoaded() || localVotifierListener != null) return;
+        VotiferEvent listener = new VotiferEvent(this, () -> localVotifierIngressReady);
+        // Publish the memory owner before capture. Its independent worker loads
+        // disk state; vote-executor cancellation cannot discard queue setup.
+        votifierVoteOverflowQueue = VotifierVoteOverflowQueue.initializeConfirmedAsync(this,
+                listener::processVoteConfirmed, () -> localVotifierIngressReady && !localVotifierIngressClosed);
+        localVotifierListener = listener;
+        if (localVotifierIngressReady) votifierVoteOverflowQueue.start();
+        getServer().getPluginManager().registerEvents(listener, this);
+    }
+
 	private void registerEvents() {
 		PluginManager pm = getServer().getPluginManager();
 
 		pm.registerEvents(new PlayerJoinEvent(this), this);
-		if (isVotifierLoaded()) {
-			VotiferEvent votifierEvent = new VotiferEvent(this);
-			votifierVoteOverflowQueue = new VotifierVoteOverflowQueue(this, votifierEvent::processVote);
-			pm.registerEvents(votifierEvent, this);
-		}
-		pm.registerEvents(new PlayerVoteListener(this), this);
+        pm.registerEvents(new com.bencodez.votingplugin.listeners.DateVoteMenuOpenGuard(), this);
 		pm.registerEvents(new PlayerPostVoteLoggerListener(this), this);
 		pm.registerEvents(new PlayerSpecialRewardLoggerListener(this), this);
 		pm.registerEvents(new VoteShopPurchaseLoggerListener(this), this);
@@ -2102,9 +2138,6 @@ private void registerEvents() {
 			pm.registerEvents(new PlayerInteract(this), this);
 		}
 
-		if (timeQueueHandler != null) {
-			pm.registerEvents(timeQueueHandler, plugin);
-		}
 
 		pm.registerEvents(new VotingPluginUpdateEvent(this), this);
 		/*
@@ -2182,6 +2215,8 @@ private void reloadPlugin(boolean userStorage, boolean reconcileHostedControl,
 		// Re-evaluate after storage has reloaded; UserManager keeps this lifecycle task unique.
 		getVotingPluginUserManager().startSharedPointTransferRecovery();
 
+        getOptions().setServer(bungeeSettings.getServer());
+        getDateVoteMilestones().reload();
 		reloadBackendProxyRuntime(updateActiveBackendRuntime, userStorage);
 		checkYMLError();
 

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java

@@ -569,7 +569,8 @@ private void handleVoteBroadcast(JsonEnvelope msg) {
 		}
 
 		String totalsRaw = nvl(fields.get(VotingPluginWire.K_TOTALS));
-		VoteTotalsSnapshot totals = totalsRaw.isEmpty() ? null : VoteTotalsSnapshot.parseStorage(totalsRaw);
+		VoteTotalsSnapshot totals = !com.bencodez.votingplugin.proxy.VoteOccurrenceMetadata.hasPayload(totalsRaw)
+                ? null : VoteTotalsSnapshot.parseStorage(totalsRaw);
 		VoteSite voteSite = plugin.getVoteSiteManager()
 				.getVoteSite(plugin.getVoteSiteManager().getVoteSiteName(true, service), true);
 		if (voteSite == null) {
@@ -715,16 +716,29 @@ private WireVoteResult handleWireVote(JsonEnvelope msg) {
 		user.cache();
 		boolean wasOnline = vote.wasOnlineKnown ? vote.wasOnline : user.isOnline();
 		boolean queuedDelivery = vote.queuedDeliveryKnown ? vote.queuedDelivery : vote.delayValidated;
-		var event = user.bungeeVotePluginMessagingEvent(vote.service, vote.time, totals, !vote.manageTotals,
-				wasOnline, vote.broadcast, vote.num, queuedDelivery, vote.delayValidationKnown,
-				vote.queuedDeliveryKnown, voteId,
-				VotingPluginWire.SUB_VOTE_ONLINE.equals(msg.getSubChannel()));
-		if ((event == null || !event.isCancelled()) && plugin.getBungeeSettings().isPerServerPoints()) {
+		com.bencodez.votingplugin.events.PlayerVoteEvent dispatchedEvent = null;
+		VotingPluginUser.DateMilestoneAccountingException dateAccountingFailure = null;
+		try {
+			// Keep totals correlation for replay fencing, but only the original wire ID
+			// identifies a canonical transport occurrence in the native event.
+			dispatchedEvent = user.bungeeVotePluginMessagingEvent(vote.service, vote.time, totals, !vote.manageTotals,
+					wasOnline, vote.broadcast, vote.num, queuedDelivery, vote.delayValidationKnown,
+					vote.queuedDeliveryKnown, vote.voteId,
+					VotingPluginWire.SUB_VOTE_ONLINE.equals(msg.getSubChannel()), vote.realVote,
+					vote.canonicalOccurrenceTime);
+		} catch (VotingPluginUser.DateMilestoneAccountingException incomplete) {
+			dateAccountingFailure = incomplete;
+		}
+		if ((dispatchedEvent == null || !dispatchedEvent.isCancelled())
+				&& plugin.getBungeeSettings().isPerServerPoints()) {
 			user.addPoints(plugin.getConfigFile().getPointsOnVote());
 		}
 		if (vote.service != null && !vote.service.isEmpty()) {
 			plugin.getServerData().addServiceSite(vote.service);
 		}
+		// Date accounting cannot suppress normal points/service observations. Keep
+		// the reservation incomplete so reconciliation cannot replay normal rewards.
+		if (dateAccountingFailure != null) throw dateAccountingFailure;
 		return new WireVoteResult(voteId, true, false);
 	}
 

VotingPlugin/src/main/java/com/bencodez/votingplugin/commands/CommandLoader.java

@@ -3486,7 +3486,15 @@ public void updateReplacements() {
 	 */
 	private void loadVoteCommand() {
 		plugin.setVoteCommand(new ArrayList<>());
-		plugin.getVoteCommand().add(new CommandHandler(plugin, new String[] { "Help&?" },
+		plugin.getVoteCommand().add(new CommandHandler(plugin, new String[] { "DateEvents" },
+                "VotingPlugin.Commands.Vote.DateEvents|" + playerPerm, "View date-window vote milestone progress") {
+            @Override public void execute(CommandSender sender, String[] args) {
+                if (sender instanceof Player player) plugin.getDateVoteMilestones().progress(player);
+                else sender.sendMessage("Run /vote dateevents as a player.");
+            }
+        });
+
+        plugin.getVoteCommand().add(new CommandHandler(plugin, new String[] { "Help&?" },
 				"VotingPlugin.Commands.Vote.Help|" + playerPerm, "View help page") {
 
 			@Override