From e1d3e124a1c86455f6b020c9e5d878c6f3b452e6 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Sat, 24 Oct 2026 03:09:55 +0200 Subject: [PATCH] fix: run queued peer operations after the runtime stopped its timer A closing runtime stops its timer while control tasks queued before still run. An operation reaching a peer then still scheduled its timeout, even when the closed peer had failed it already, and the rejection went to the error handler, which made HttpSessionTests.lateAnswerDeleted fail now and then. Operations that are done need no timeout, and the operation, gathering and collision timers are skipped once the timer rejects. --- .../dev/onvoid/webrtc/client/PeerImpl.java | 68 ++++++++++++++----- .../webrtc/client/RuntimeCloseTests.java | 64 +++++++++++++++++ 2 files changed, 114 insertions(+), 18 deletions(-) diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index 62735b49..98e403f8 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/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; @@ -732,13 +733,13 @@ private CompletableFuture enqueue(Supplier> operatio CompletableFuture 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; @@ -753,6 +754,25 @@ private CompletableFuture enqueue(Supplier> 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; @@ -887,11 +907,18 @@ private CompletableFuture 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; } @@ -1009,15 +1036,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. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java index cd6a3b66..c45e7f3e 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java +++ b/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.video.CustomVideoSource; import org.junit.jupiter.api.Test; @@ -319,6 +321,68 @@ void beforeRelease() { } } + @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 created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Peer peer = await(created); + + try { + CompletableFuture peerClosed = peer.closeAsync(); + control.runUntil(peerClosed::isDone, TIMEOUT_SECONDS); + + CompletableFuture 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 created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Peer peer = await(created); + + try { + timer.shutdown(); + + CompletableFuture 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)