Add secure single-port HTTP proxy transport

VotingPlugin/pom.xml

@@ -160,6 +160,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>
@@ -339,6 +343,12 @@
             <version>3.8.2-SNAPSHOT</version>
             <scope>compile</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

@@ -39,6 +39,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 +67,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 +98,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 +140,177 @@ 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;
+		}
+		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) {
+		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() {
+		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;
+			presenceManager.stopForDisable();
+		}
+	}
+
+	/** 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 +326,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;
 		}
 	}
 

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

@@ -25,6 +25,7 @@ public class ProcessedVoteCache {
 	private final LinkedHashMap<String, Long> processedRedisDeliveries = new LinkedHashMap<>();
 	private final LinkedHashMap<String, Integer> legacyRedisDeliveries = new LinkedHashMap<>();
 	private long legacyRedisDeliveryBytes;
+	private boolean legacyRedisHandoffOverflowed;
 	private Object activeRedisSubscriber;
 	private Object standbyRedisSubscriber;
 
@@ -82,14 +83,18 @@ public synchronized boolean reserveRedisDelivery(String deliveryId) {
 		return true;
 	}
 
-	public synchronized void registerRedisSubscriber(Object subscriber) {
+	public synchronized boolean registerRedisSubscriber(Object subscriber) {
 		if (activeRedisSubscriber == null) {
 			activeRedisSubscriber = subscriber;
+			return true;
 		} else if (activeRedisSubscriber != subscriber) {
 			standbyRedisSubscriber = subscriber;
 			legacyRedisDeliveries.clear();
 			legacyRedisDeliveryBytes = 0;
+			legacyRedisHandoffOverflowed = false;
+			return false;
 		}
+		return true;
 	}
 
 	/** Returns true only for the active subscriber and counts its legacy delivery during overlap. */
@@ -107,12 +112,16 @@ public synchronized boolean reserveLegacyRedisDelivery(Object subscriber, String
 						&& legacyRedisDeliveryBytes <= MAX_LEGACY_REDIS_TOTAL_BYTES - bytes) {
 					legacyRedisDeliveries.put(signature, 1);
 					legacyRedisDeliveryBytes += bytes;
-				}
+				} else legacyRedisHandoffOverflowed = true;
 			}
 		}
 		return true;
 	}
 
+	public synchronized boolean isLegacyRedisHandoffOverflowed() {
+		return legacyRedisHandoffOverflowed;
+	}
+
 	public synchronized void activateRedisSubscriber(Object subscriber) {
 		if (standbyRedisSubscriber != subscriber) {
 			throw new IllegalStateException("Redis replacement subscriber is not registered");
@@ -121,6 +130,15 @@ public synchronized void activateRedisSubscriber(Object subscriber) {
 		standbyRedisSubscriber = null;
 	}
 
+	/** Restores the retired active listener after a promoted replacement is rolled back. */
+	public synchronized void restoreRedisSubscriber(Object subscriber) {
+		activeRedisSubscriber = subscriber;
+		standbyRedisSubscriber = null;
+		legacyRedisDeliveries.clear();
+		legacyRedisDeliveryBytes = 0;
+		legacyRedisHandoffOverflowed = false;
+	}
+
 	/** Consumes one matching delivery processed by the previous active subscriber. */
 	public synchronized boolean consumeLegacyRedisDelivery(String signature) {
 		Integer count = legacyRedisDeliveries.get(signature);
@@ -135,13 +153,15 @@ public synchronized boolean consumeLegacyRedisDelivery(String signature) {
 	public synchronized void finishRedisHandoff() {
 		legacyRedisDeliveries.clear();
 		legacyRedisDeliveryBytes = 0;
+		legacyRedisHandoffOverflowed = false;
 	}
 
 	public synchronized void unregisterRedisSubscriber(Object subscriber) {
 		if (standbyRedisSubscriber == subscriber) {
 			standbyRedisSubscriber = null;
 			legacyRedisDeliveries.clear();
 			legacyRedisDeliveryBytes = 0;
+			legacyRedisHandoffOverflowed = false;
 		}
 		if (activeRedisSubscriber == subscriber) activeRedisSubscriber = null;
 	}