Defer journal sealing until callbacks stop

Author: BenCodez

SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpBackendTransportConnector.java

@@ -343,10 +343,25 @@ public boolean flushOutgoing(long deadlineNanos) {
 		finishClose();
 	}
 	private void finishClose() {
+		if (!shutdownCallbacks()) {
+			// A callback may ignore interruption while still able to durably complete.
+			// Keep journal ownership until it really terminates; sealing here would strand RUNNING.
+			Thread reaper = new Thread(this::finishCloseAfterCallbacks, "SimpleAPI-HTTP-callback-reaper");
+			reaper.setDaemon(true);
+			reaper.start();
+			return;
+		}
+		sealAfterCallbacks();
+	}
+	private void finishCloseAfterCallbacks() {
+		boolean interrupted = false;
+		while (!callbackExecutor.isTerminated()) try { callbackExecutor.awaitTermination(1, TimeUnit.DAYS); }
+		catch (InterruptedException ignored) { interrupted = true; }
+		if (interrupted) Thread.currentThread().interrupt();
+		sealAfterCallbacks();
+	}
+	private void sealAfterCallbacks() {
 		try {
-			shutdownCallbacks();
-		// No poller can enqueue more callbacks and every running journal transition has
-		// finished, so ownership can now be revoked without stranding completed work.
 			if (inboundDeliveries != null) inboundDeliveries.seal();
 			if (acknowledgementConfirmationStore != null) acknowledgementConfirmationStore.seal();
 		} finally { closed.countDown(); }
@@ -357,13 +372,13 @@ private void awaitClosed() {
 		catch (InterruptedException stopRequested) { interrupted = true; }
 		if (interrupted) Thread.currentThread().interrupt();
 	}
-	private void shutdownCallbacks() {
+	private boolean shutdownCallbacks() {
 		callbackExecutor.shutdown();
 		boolean interrupted = false;
 		try {
 			if (!callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
 				callbackExecutor.shutdownNow();
-				callbackExecutor.awaitTermination(1, TimeUnit.SECONDS);
+				return callbackExecutor.awaitTermination(1, TimeUnit.SECONDS);
 			}
 		} catch (InterruptedException stopRequested) {
 			interrupted = true;
@@ -372,6 +387,7 @@ private void shutdownCallbacks() {
 			catch (InterruptedException repeated) { interrupted = true; }
 		}
 		if (interrupted) Thread.currentThread().interrupt();
+		return callbackExecutor.isTerminated();
 	}
 	boolean pollerAlive() { Thread current = poller; return current != null && current.isAlive(); }
 	private static boolean joinPoller(Thread poller, long deadlineNanos) {

SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java

@@ -285,20 +285,32 @@ private boolean send(String serverId, String deliveryId, JsonEnvelope envelope,
 	}
 
 	private void finishClose() {
-		try {
-			try { shutdown(handlerExecutor); }
-			finally { shutdown(listenerExecutor); }
-		} finally {
-			try {
-				synchronized (backends) {
-					for (BackendState backend : backends.values()) { backend.seal(); backend.signal(); }
-					backends.clear();
-				}
-			} finally { try { if (durableOutgoing != null) durableOutgoing.close(); }
-			finally {
-				synchronized (closeMonitor) { closeFinalizing = false; closeFinalized = true; closeMonitor.notifyAll(); }
-			} }
+		boolean handlersTerminated;
+		try { handlersTerminated = shutdown(handlerExecutor); }
+		finally { shutdown(listenerExecutor); }
+		if (!handlersTerminated) {
+			Thread reaper = new Thread(this::finishCloseAfterHandlers, "SimpleAPI-HTTP-proxy-handler-reaper");
+			reaper.setDaemon(true);
+			reaper.start();
+			return;
 		}
+		sealAfterHandlers();
+	}
+	private void finishCloseAfterHandlers() {
+		boolean interrupted = false;
+		while (!handlerExecutor.isTerminated()) try { handlerExecutor.awaitTermination(1, TimeUnit.DAYS); }
+		catch (InterruptedException ignored) { interrupted = true; }
+		if (interrupted) Thread.currentThread().interrupt();
+		sealAfterHandlers();
+	}
+	private void sealAfterHandlers() {
+		try {
+			synchronized (backends) {
+				for (BackendState backend : backends.values()) { backend.seal(); backend.signal(); }
+				backends.clear();
+			}
+		} finally { try { if (durableOutgoing != null) durableOutgoing.close(); }
+		finally { synchronized (closeMonitor) { closeFinalizing = false; closeFinalized = true; closeMonitor.notifyAll(); } } }
 	}
 
 	private void awaitClose() {
@@ -472,13 +484,13 @@ private static ThreadPoolExecutor executor(String name, int threads, int queue,
 		return new ThreadPoolExecutor(threads, threads, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queue), factory, new ThreadPoolExecutor.AbortPolicy());
 	}
 	private static void setDefault(String name, String value) { if (System.getProperty(name) == null) System.setProperty(name, value); }
-	static void shutdown(ExecutorService executor) {
+	static boolean shutdown(ExecutorService executor) {
 		executor.shutdown();
 		boolean interrupted = false;
 		try {
 			if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
 				executor.shutdownNow();
-				executor.awaitTermination(1, TimeUnit.SECONDS);
+				return executor.awaitTermination(1, TimeUnit.SECONDS);
 			}
 		} catch (InterruptedException stopRequested) {
 			interrupted = true;
@@ -487,6 +499,7 @@ static void shutdown(ExecutorService executor) {
 			catch (InterruptedException repeated) { interrupted = true; }
 		}
 		if (interrupted) Thread.currentThread().interrupt();
+		return executor.isTerminated();
 	}
 
 	public record ReceivedEnvelope(String serverId, String messageId, JsonEnvelope envelope) { }

SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java

@@ -721,9 +721,15 @@ void interruptedCloseRemainsBoundedWhenACallbackIgnoresInterruption() throws Exc
 			release.countDown();
 			assertTrue(stopped.await(2, TimeUnit.SECONDS));
 		} finally { release.countDown(); }
-		assertEquals(HttpInboundDeliveryStore.State.RUNNING,
-				HttpInboundDeliveryStore.inspect(clientDirectory).state(id),
-				"an ambiguous callback must remain fail-closed after bounded shutdown");
+		long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+		HttpInboundDeliveryStore.State state;
+		do {
+			state = HttpInboundDeliveryStore.inspect(clientDirectory).state(id);
+			if (state == HttpInboundDeliveryStore.State.COMPLETED) break;
+			Thread.sleep(10);
+		} while (System.nanoTime() < deadline);
+		assertEquals(HttpInboundDeliveryStore.State.COMPLETED, state,
+				"the deferred journal seal must allow the late callback completion to become durable");
 	}
 
 	@Test