Add secure single-port HTTP proxy transport

VotingPlugin/pom.xml

@@ -156,6 +156,10 @@
                             <shadedPattern>
                                 ${project.groupId}.votingplugin.bstats</shadedPattern>
                         </relocation>
+                        <relocation>
+                            <pattern>org.bouncycastle</pattern>
+                            <shadedPattern>${project.groupId}.votingplugin.bouncycastle</shadedPattern>
+                        </relocation>
                         <relocation>
                             <pattern>xyz.upperlevel.spigot</pattern>
                             <shadedPattern>
@@ -209,12 +213,17 @@
                                     <artifact>com.bencodez:simpleapi</artifact>
                                     <excludes>
                                         <exclude>org/bouncycastle/**</exclude>
-                                        <exclude>com/bencodez/simpleapi/servercomm/http/**</exclude>
                                         <exclude>redis/clients/**</exclude>
                                         <exclude>org/eclipse/paho/**</exclude>
                                         <exclude>META-INF/versions/**</exclude>
                                     </excludes>
                                 </filter>
+                                <filter>
+                                    <artifact>org.bouncycastle:*</artifact>
+                                    <excludes>
+                                        <exclude>META-INF/versions/25/**</exclude>
+                                    </excludes>
+                                </filter>
                                 <filter>
                                     <artifact>*:*</artifact>
                                     <excludes>
@@ -433,6 +442,12 @@
             <version>2.14.0</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.bencodez</groupId>
+            <artifactId>simpleapi</artifactId>
+            <version>1.0.2-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-engine</artifactId>

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

@@ -25,6 +25,7 @@
 import com.bencodez.votingplugin.backendproxy.transport.BackendProxyTransportManager;
 import com.bencodez.votingplugin.backendproxy.voteparty.BackendVotePartySync;
 import com.bencodez.votingplugin.proxy.BungeeMethod;
+import com.bencodez.votingplugin.proxy.VotingPluginWire;
 
 import lombok.Getter;
 
@@ -39,6 +40,11 @@ public class BackendProxyHandler implements Listener {
 	private final BackendGlobalDataSync globalDataSync;
 
 	private BackendPresenceManager presenceManager;
+	private boolean presenceReportingActivated;
+	private final Object inboundPublication = new Object();
+	private boolean inboundPublished;
+	private boolean inboundAborted;
+	private BackendProxyHandler inboundRollbackTarget;
 	private BackendVotePartySync votePartySync;
 	private BackendProxyMessageRouter messageRouter;
 
@@ -62,12 +68,26 @@ public BackendProxyHandler(VotingPluginMain plugin, ProcessedVoteCache processed
 	 * Loads the configured backend/proxy communication components.
 	 */
 	public void load() {
+		load(true);
+	}
+
+	/** Loads a replacement without announcing a new presence generation before publication. */
+	public void loadForReplacement() {
+		load(false);
+	}
+
+	private void load(boolean activatePresenceReporting) {
 		plugin.debug("Loading backend proxy handler");
 		method = BungeeMethod.getByName(plugin.getBungeeSettings().getBungeeMethod());
 		plugin.getLogger().info("Using BungeeMethod: " + method.toString());
 
 		globalDataSync.load();
 		globalMessageHandler = new GlobalMessageHandler() {
+			@Override
+			public void onMessage(JsonEnvelope envelope) {
+				BackendProxyHandler.this.dispatchIncomingAfterPublication(envelope, () -> super.onMessage(envelope));
+			}
+
 			@Override
 			public void sendMessage(JsonEnvelope envelope) {
 				transportManager.send(envelope);
@@ -79,20 +99,40 @@ public void sendMessage(JsonEnvelope envelope) {
 		messageRouter = new BackendProxyMessageRouter(plugin, presenceManager, globalDataSync, votePartySync,
 				processedVoteCache);
 		messageRouter.register(globalMessageHandler, method);
-		transportManager.start(method, globalMessageHandler);
+		transportManager.start(method, globalMessageHandler, activatePresenceReporting);
 
 		if (plugin.getOptions().getServer().equalsIgnoreCase("pleaseset")) {
 			plugin.getLogger().warning("Server name for bungee voting is not set, please set it");
 		}
-		presenceManager.start();
+		if (activatePresenceReporting) {
+			activatePresenceReporting();
+			activateInboundMessages();
+		}
+	}
+
+	/** Starts presence only after a staged handler reaches the atomic publication boundary. */
+	public void activatePresenceReporting() {
+		if (presenceManager != null && !presenceReportingActivated) {
+			presenceManager.start();
+			presenceReportingActivated = true;
+		}
+		// Presence startup can throw while scheduling its heartbeat. Keep inbound
+		// HTTP callbacks behind the publication barrier until every fallible part of
+		// the replacement is active, so rollback cannot race a queued callback.
+		transportManager.activateAfterPublication();
 	}
 
 	/**
 	 * Closes backend/proxy components and persists cached proxy state.
 	 */
 	public void close() {
-		if (presenceManager != null) {
+		synchronized (inboundPublication) {
+			if (!inboundPublished && !inboundAborted) inboundAborted = true;
+			inboundPublication.notifyAll();
+		}
+		if (presenceManager != null && presenceReportingActivated) {
 			presenceManager.stop();
+			presenceReportingActivated = false;
 		}
 		transportManager.close();
 		if (votePartySync != null) {
@@ -101,26 +141,207 @@ public void close() {
 		globalDataSync.close();
 	}
 
-	/** Releases a same-method subscriber/listener before its replacement starts. */
-	public void prepareForReplacement(BungeeMethod replacementMethod) {
-		if (method == replacementMethod && method != BungeeMethod.PLUGINMESSAGING && method != BungeeMethod.REDIS) {
+	/** Opens inbound dispatch only after the replacement and all handoffs are committed. */
+	public void activateInboundMessages() {
+		synchronized (inboundPublication) {
+			if (inboundAborted) return;
+			inboundPublished = true;
+			inboundPublication.notifyAll();
+		}
+	}
+
+	/** Routes an already accepted staged callback through the restored predecessor on rollback. */
+	public void abortStagedInboundTo(BackendProxyHandler previous) {
+		synchronized (inboundPublication) {
+			if (inboundPublished || inboundAborted) return;
+			inboundRollbackTarget = previous;
+			inboundAborted = true;
+			inboundPublication.notifyAll();
+		}
+	}
+
+	void dispatchIncomingAfterPublication(JsonEnvelope envelope, Runnable localDispatch) {
+		BackendProxyHandler rollbackTarget;
+		synchronized (inboundPublication) {
+			while (!inboundPublished && !inboundAborted) {
+				try {
+					inboundPublication.wait();
+				} catch (InterruptedException interrupted) {
+					Thread.currentThread().interrupt();
+					return;
+				}
+			}
+			rollbackTarget = inboundPublished ? null : inboundRollbackTarget;
+			if (!inboundPublished && rollbackTarget == null) return;
+		}
+		if (rollbackTarget != null) {
+			GlobalMessageHandler rollbackHandler = rollbackTarget.globalMessageHandler;
+			if (rollbackHandler != null) rollbackHandler.onMessage(envelope);
+			return;
+		}
+		// Global-data checks perform synchronous SQL and already run on their own
+		// timer worker. Explicitly dispatch an inbound wake-up asynchronously too:
+		// plugin messaging can invoke this method on Bukkit's primary thread.
+		if (VotingPluginWire.SUB_BUNGEE_TIME_CHANGE.equals(envelope.getSubChannel())) {
+			plugin.getBukkitScheduler().runTaskAsynchronously(plugin, localDispatch);
+			return;
+		}
+		plugin.getBukkitScheduler().executeOrScheduleSync(plugin, localDispatch);
+	}
+
+	/** Returns whether replacement preparation must preserve accepted deliveries. */
+	public boolean requiresPreparationForReplacement() {
+		return method == BungeeMethod.HTTP || method == BungeeMethod.PLUGINMESSAGING
+				|| transportManager.hasPendingAsyncHandoff() || transportManager.hasPendingRedisReplay();
+	}
+
+	/** Prepares HTTP state or waits off-thread for an earlier cross-transport handoff. */
+	public boolean prepareForReplacement(BungeeMethod replacementMethod) {
+		return prepareForReplacement(replacementMethod,
+				System.nanoTime() + java.util.concurrent.TimeUnit.SECONDS.toNanos(25));
+	}
+
+	public boolean prepareForReplacement(BungeeMethod replacementMethod, long deadlineNanos) {
+		// SocketHandler swallows a listener bind failure. A same-method replacement
+		// must therefore retire its listener before the staged handler is started;
+		// rollback recreates this prepared transport if validation later fails.
+		if (method == BungeeMethod.SOCKETS && replacementMethod == BungeeMethod.SOCKETS) {
+			transportManager.prepareForReplacement();
+			return true;
+		}
+		if (method == BungeeMethod.MQTT && replacementMethod == BungeeMethod.MQTT) {
+			// A duplicate MQTT ClientID disconnects the live broker session, so stage
+			// only after retiring the predecessor and restore it on validation rollback.
+			transportManager.prepareForReplacement();
+			return true;
+		}
+		if (method == BungeeMethod.HTTP) {
 			transportManager.prepareForReplacement();
+			return true;
 		}
+		if (method == BungeeMethod.PLUGINMESSAGING) {
+			transportManager.prepareAsyncHandoffForReplacement(deadlineNanos);
+			return true;
+		}
+		if (method == BungeeMethod.REDIS && replacementMethod != BungeeMethod.REDIS) {
+			if (transportManager.hasPendingAsyncHandoff())
+				transportManager.prepareAsyncHandoffForReplacement(deadlineNanos);
+			return transportManager.prepareRedisReplayTransition(replacementMethod, deadlineNanos);
+		}
+		if (method == BungeeMethod.REDIS) {
+			// Same-Redis retirement installs its send fence during the bounded
+			// off-thread handoff. Returning true here lets the staged replacement
+			// buffer its own sends until the predecessor FIFO is admitted at Bukkit
+			// publication.
+			return true;
+		}
+		if (!transportManager.hasPendingAsyncHandoff()) return false;
+		transportManager.prepareAsyncHandoffForReplacement(deadlineNanos);
+		return true;
+	}
+
+	/** Atomically fences new sends only when disabling cannot discard prepared HTTP messages. */
+	public boolean commitPreparedDisable() {
+		return transportManager.commitPreparedDisable();
+	}
+
+	/** Publishes the final presence update while fencing other sends before transport preparation. */
+	public void preparePresenceForDisable() {
+		preparePresenceForDisableInternal(null);
+	}
+
+	/** Publishes the final presence update before the caller's validation deadline. */
+	public void preparePresenceForDisable(long deadlineNanos) {
+		preparePresenceForDisableInternal(deadlineNanos);
+	}
+
+	private void preparePresenceForDisableInternal(Long deadlineNanos) {
+		transportManager.beginPreparedDisable();
+		if (presenceManager != null && presenceReportingActivated) {
+			// stopForDisable may reject when the transport cannot accept the final
+			// presence update. Mark this inactive first so rollback can start it again.
+			presenceReportingActivated = false;
+			if (deadlineNanos == null) presenceManager.stopForDisable();
+			else presenceManager.stopForDisable(deadlineNanos);
+		}
+	}
+
+	/** Restores delivery and presence when a prepared disable is abandoned. */
+	public void restorePresenceAfterFailedDisablePreparation() {
+		transportManager.cancelPreparedDisable();
+		activatePresenceReporting();
+	}
+
+	public void beginPreparedHttpHandoff() {
+		transportManager.beginPreparedHttpHandoff();
+	}
+
+	/** Reserves staged HTTP capacity before publication can admit replacement sends. */
+	public void reservePreparedHttpHandoff(BackendProxyHandler replacement) {
+		transportManager.reservePreparedTransportHandoff(replacement.transportManager);
+	}
+
+	/** Restores a prepared HTTP transport when its replacement fails validation. */
+	public void restoreAfterFailedReplacement() {
+		transportManager.restoreAfterFailedReplacement();
+	}
+
+	/** Restores a failed same-Redis predecessor without discarding its replacement's replay FIFO. */
+	public void restoreAfterFailedReplacement(BackendProxyHandler failedReplacement) {
+		transportManager.restoreAfterFailedReplacement(
+				failedReplacement == null ? null : failedReplacement.transportManager);
+	}
+
+	/** Reasserts the old handler with a fresh presence generation after rollback. */
+	public void refreshPresenceAfterFailedReplacement() {
+		if (presenceManager != null && presenceReportingActivated) {
+			presenceManager.stop();
+			presenceManager.start();
+		}
+	}
+
+	public void awaitRestoreAfterFailedReplacement(long deadlineNanos) {
+		transportManager.awaitPreparedTransportRestoration(deadlineNanos);
 	}
 
 	/** Fails a configuration apply when its selected transport did not initialize. */
 	public void validateTransport() {
+		validateTransport(System.nanoTime() + java.util.concurrent.TimeUnit.SECONDS.toNanos(25));
+	}
+
+	/** Validates transport startup without extending the caller's existing deadline. */
+	public void validateTransport(long deadlineNanos) {
 		if (method == null || globalMessageHandler == null || presenceManager == null) {
 			throw new IllegalStateException("Backend proxy handler initialization failed");
 		}
-		transportManager.validate();
+		transportManager.validate(deadlineNanos);
 	}
 
 	/** Completes the no-loss/no-duplicate same-Redis subscriber handoff after validation. */
 	public void completeRedisHandoff(BackendProxyHandler replacement) {
-		if (method != BungeeMethod.REDIS || replacement.method != BungeeMethod.REDIS) return;
-		transportManager.closeRedisForHandoff();
-		replacement.transportManager.activateRedisAfterHandoff();
+		if (!requiresRedisHandoff(replacement)) return;
+		transportManager.completeRedisHandoff(replacement.transportManager);
+	}
+
+	/** Returns whether this handler owns a Redis listener whose shutdown can block. */
+	public boolean requiresRedisRetirement() {
+		return method == BungeeMethod.REDIS;
+	}
+
+	/** Returns whether this replacement needs the bounded same-Redis retirement path. */
+	public boolean requiresRedisHandoff(BackendProxyHandler replacement) {
+		return replacement != null && method == BungeeMethod.REDIS && replacement.method == BungeeMethod.REDIS;
+	}
+
+	/** Replays Redis handoff deliveries only after inbound publication is open. */
+	public void replayRedisAfterHandoffPublication() {
+		transportManager.replayRedisAfterHandoffPublication();
+	}
+
+	/** Forwards messages buffered while the previous transport was fenced. */
+	public void completeHttpHandoff(BackendProxyHandler replacement) {
+		if (replacement == null) return;
+		transportManager.completePreparedTransportHandoff(replacement.transportManager);
 	}
 
 	public void playerOnline(String playerName, String uuid) {
@@ -136,14 +357,15 @@ public void playerOffline(String playerName) {
 	}
 
 	public void reloadPresenceReporting() {
-		if (presenceManager != null) {
+		if (presenceManager != null && presenceReportingActivated) {
 			presenceManager.reload();
 		}
 	}
 
 	public void disablePresenceReporting() {
-		if (presenceManager != null) {
+		if (presenceManager != null && presenceReportingActivated) {
 			presenceManager.stop();
+			presenceReportingActivated = false;
 		}
 	}