fix: run queued peer operations after the runtime stopped its timer

webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -752,13 +753,13 @@ private <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> operatio
 			CompletableFuture<T> current = running;
 			runningOperation = current;
 
-			ScheduledFuture<?> timeout = rtc.timer.schedule(() -> control.execute(() ->
-					current.completeExceptionally(new TimeoutException("WebRTC did not finish an operation of "
-							+ name + " within " + rtc.operationTimeout))),
-					rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS);
+			// One that is done already, as on a closed peer, needs no timeout.
+			ScheduledFuture<?> timeout = current.isDone() ? null : timeOut(current);
 
 			current.whenComplete((value, error) -> control.execute(() -> {
-				timeout.cancel(false);
+				if (timeout != null) {
+					timeout.cancel(false);
+				}
 
 				runningOperation = null;
 				operationRunning = false;
@@ -773,6 +774,25 @@ private <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> operatio
 		return result;
 	}
 
+	/**
+	 * Fails a running operation once the operation timeout has passed.
+	 *
+	 * @return the timeout, or null once the runtime has stopped its timer.
+	 */
+	private ScheduledFuture<?> timeOut(CompletableFuture<?> operation) {
+		try {
+			return rtc.timer.schedule(() -> control.execute(() ->
+					operation.completeExceptionally(new TimeoutException("WebRTC did not finish an operation of "
+							+ name + " within " + rtc.operationTimeout))),
+					rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS);
+		}
+		catch (RejectedExecutionException e) {
+			// The runtime is closing: control tasks queued before still run,
+			// and it fails the futures they leave open.
+			return null;
+		}
+	}
+
 	private void runNextOperation() {
 		if (operationRunning) {
 			return;
@@ -907,11 +927,18 @@ private CompletableFuture<Void> waitForGathering() {
 
 		// A STUN or TURN server that cannot be reached holds up the end of
 		// gathering for long; the description then goes with what there is.
-		rtc.timer.schedule(() -> control.execute(() -> {
-			if (gatheringWaiters.remove(future)) {
-				future.complete(null);
-			}
-		}), options.gatheringTimeout.toMillis(), TimeUnit.MILLISECONDS);
+		try {
+			rtc.timer.schedule(() -> control.execute(() -> {
+				if (gatheringWaiters.remove(future)) {
+					future.complete(null);
+				}
+			}), options.gatheringTimeout.toMillis(), TimeUnit.MILLISECONDS);
+		}
+		catch (RejectedExecutionException e) {
+			// The runtime is closing and has no timer left to wait with.
+			gatheringWaiters.remove(future);
+			future.complete(null);
+		}
 
 		return future;
 	}
@@ -1029,15 +1056,20 @@ private void watchCollision() {
 
 		int answers = remoteAnswers;
 
-		collisionCheck = rtc.timer.schedule(() -> control.execute(() -> {
-			collisionCheck = null;
+		try {
+			collisionCheck = rtc.timer.schedule(() -> control.execute(() -> {
+				collisionCheck = null;
 
-			if (!closed && remoteAnswers == answers
-					&& connection.getSignalingState() == RTCSignalingState.HAVE_LOCAL_OFFER) {
-				deliverError(new NegotiationException("Both peers sent an offer at once, and both are impolite, "
-						+ "so neither gives way. Build exactly one of them with PeerOptions.Builder.polite(true)"));
-			}
-		}), COLLISION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+				if (!closed && remoteAnswers == answers
+						&& connection.getSignalingState() == RTCSignalingState.HAVE_LOCAL_OFFER) {
+					deliverError(new NegotiationException("Both peers sent an offer at once, and both are impolite, "
+							+ "so neither gives way. Build exactly one of them with PeerOptions.Builder.polite(true)"));
+				}
+			}), COLLISION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+		}
+		catch (RejectedExecutionException e) {
+			// The runtime is closing, and the peer with it.
+		}
 	}
 
 	// What WebRTC reports, on the control executor.

webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java

@@ -30,6 +30,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import dev.onvoid.webrtc.RTCSdpType;
+import dev.onvoid.webrtc.RTCSessionDescription;
 import dev.onvoid.webrtc.media.audio.AudioDevice;
 import dev.onvoid.webrtc.media.video.CustomVideoSource;
 
@@ -360,6 +362,68 @@ void switchDeviceAfterClose() throws Exception {
 		await(closed);
 	}
 
+	@Test
+	void closedPeerNeedsNoTimer() throws Exception {
+		// An operation reaches a peer the closing runtime has closed, after it
+		// stopped its timer, as control tasks queued before still run.
+		ManualExecutor control = new ManualExecutor();
+		ManualTimer timer = new ManualTimer();
+		WebRtc runtime = runtime(control, timer);
+		CompletableFuture<Peer> created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions()));
+		control.runUntil(created::isDone, TIMEOUT_SECONDS);
+		Peer peer = await(created);
+
+		try {
+			CompletableFuture<Void> peerClosed = peer.closeAsync();
+			control.runUntil(peerClosed::isDone, TIMEOUT_SECONDS);
+
+			CompletableFuture<Object> operation = peer.withNative(connection -> null);
+			timer.shutdown();
+			control.runUntil(operation::isDone, TIMEOUT_SECONDS);
+
+			ExecutionException e = assertThrows(ExecutionException.class, () -> await(operation));
+			assertInstanceOf(PeerClosedException.class, e.getCause());
+			assertTrue(errors.isEmpty(), () -> "Errors: " + errors);
+		}
+		finally {
+			HttpSessionTests.closeStepping(runtime, control);
+		}
+	}
+
+	@Test
+	void operationWithoutTimer() throws Exception {
+		// A peer the runtime could not close still runs what was queued, with
+		// no timer left to bound how long WebRTC or gathering takes.
+		ManualExecutor control = new ManualExecutor();
+		ManualTimer timer = new ManualTimer();
+		WebRtc runtime = runtime(control, timer);
+		CompletableFuture<Peer> created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions()));
+		control.runUntil(created::isDone, TIMEOUT_SECONDS);
+		Peer peer = await(created);
+
+		try {
+			timer.shutdown();
+
+			CompletableFuture<RTCSessionDescription> offer = peer.negotiator().createOffer(Gathering.COMPLETE);
+			control.runUntil(offer::isDone, TIMEOUT_SECONDS);
+
+			assertEquals(RTCSdpType.OFFER, await(offer).sdpType);
+			assertTrue(errors.isEmpty(), () -> "Errors: " + errors);
+		}
+		finally {
+			HttpSessionTests.closeStepping(runtime, control);
+		}
+	}
+
+	private WebRtc runtime(ManualExecutor control, ManualTimer timer) {
+		return WebRtc.builder()
+				.audio(AudioMode.HEADLESS)
+				.errorHandler(errors::add)
+				.controlPool(control)
+				.timer(timer)
+				.build();
+	}
+
 	private WebRtc runtime(ManualExecutor control) {
 		return WebRtc.builder()
 				.audio(AudioMode.HEADLESS)