From 31438ec2da7408d3781daf57af3390e04bd1cfe3 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Sat, 31 Oct 2026 23:24:31 +0100 Subject: [PATCH] fix: discard expired negotiations and serialize ICE restart cleanup --- .../dev/onvoid/webrtc/client/HttpIce.java | 22 +- .../dev/onvoid/webrtc/client/PeerImpl.java | 145 +++++++++++-- .../client/NegotiationTimeoutTests.java | 202 ++++++++++++++++++ 3 files changed, 343 insertions(+), 26 deletions(-) create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/NegotiationTimeoutTests.java diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpIce.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpIce.java index 6f50772f1..5672bdbf6 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpIce.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpIce.java @@ -175,7 +175,7 @@ CompletableFuture restart() { } } - Fragment[] before = new Fragment[1]; + Restart restart = new Restart(); return peer.restartIce(offer -> { HttpCall request; @@ -186,26 +186,31 @@ CompletableFuture restart() { return CompletableFuture.failedFuture(new PeerClosedException("The session is closed")); } - before[0] = local; + restart.before = local; local = fragment; restarting = true; pending.clear(); gathered = false; request = new HttpCall(client, session, "PATCH", authorization, timeout).header("If-Match", "*"); + restart.request = request; restartRequest = request; } return request.send(FRAGMENT_TYPE, fragment.text.getBytes(StandardCharsets.UTF_8), MAX_FRAGMENT, - head -> { }).handle((response, error) -> answered(request, response, error)); - }).whenComplete((restarted, error) -> { + head -> { }).handle((response, error) -> sdp -> answered(request, response, error).apply(sdp)); + }, (restarted, error) -> { synchronized (this) { + if (restart.request == null || restartRequest != restart.request) { + return; + } restarting = false; restartRequest = null; + restart.request.abort(); - if (error != null && before[0] != null) { + if (error != null && restart.before != null) { // Taken back: the candidates of the new offer go nowhere. - local = before[0]; + local = restart.before; pending.clear(); gathered = false; } @@ -215,6 +220,11 @@ CompletableFuture restart() { }); } + private static final class Restart { + Fragment before; + HttpCall request; + } + /** * Stops sending: cuts off the requests under way, and makes no more. */ 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 8470bbae3..df16b7a72 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -44,6 +44,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -149,6 +150,11 @@ final class PeerImpl implements Peer { private CompletableFuture runningOperation; + /** The unfinished steps of the current operation, owned by control. */ + private Operation activeOperation; + + private boolean teardownStarted; + private boolean negotiationQueued; private boolean ignoreOffer; @@ -710,6 +716,12 @@ void onLocalCandidates(Consumer candidates, Runnable gathered) */ CompletableFuture restartIce( Function>> exchange) { + return restartIce(exchange, (value, error) -> { }); + } + + CompletableFuture restartIce( + Function>> exchange, + BiConsumer finished) { return negotiator.run(() -> { RTCSessionDescription remote = connection.getRemoteDescription(); @@ -775,7 +787,7 @@ CompletableFuture restartIce( .thenCompose(ignored -> failed); }) .thenCompose(result -> result); - }); + }, finished); } /** @@ -783,7 +795,7 @@ CompletableFuture restartIce( * executor, or fails once the peer is closed. */ private CompletableFuture onControl(CompletableFuture future) { - CompletableFuture result = new CompletableFuture<>(); + CompletableFuture result = operationStep(); future.whenComplete((value, error) -> control.execute(() -> complete(result, value, error), () -> result.completeExceptionally(closedError()))); @@ -1256,9 +1268,16 @@ private LocalTrackImpl checkTrack(LocalTrack track) { * @return a future that completes on the control executor. */ private CompletableFuture enqueue(Supplier> operation) { + return enqueue(operation, (value, error) -> { }); + } + + private CompletableFuture enqueue(Supplier> operation, + BiConsumer finished) { CompletableFuture result = new CompletableFuture<>(); operations.add(() -> { + Operation scope = new Operation(); + activeOperation = scope; CompletableFuture running; if (closed) { @@ -1279,18 +1298,37 @@ private CompletableFuture enqueue(Supplier> operatio runningOperation = current; // One that is done already, as on a closed peer, needs no timeout. - ScheduledFuture timeout = current.isDone() ? null : timeOut(current); + ScheduledFuture timeout = current.isDone() ? null : timeOut(current, scope); current.whenComplete((value, error) -> control.execute(() -> { if (timeout != null) { timeout.cancel(false); } - runningOperation = null; - operationRunning = false; - - complete(result, value, error); - runNextOperation(); + CompletableFuture stable = scope.expired && !closed ? rollback() : + CompletableFuture.completedFuture(null); + stable.whenComplete((ignored, cleanupError) -> control.execute(() -> { + Throwable failure = cleanupError == null ? error : cleanupError; + if (cleanupError != null && !closed) { + // A peer whose signaling cannot be restored must not run + // another negotiation against the abandoned description. + teardown(); + } + if (scope.cleanupTimeout != null) { + scope.cleanupTimeout.cancel(false); + } + activeOperation = null; + runningOperation = null; + operationRunning = false; + try { + finished.accept(value, failure); + } + catch (Throwable e) { + failure = e; + } + complete(result, value, failure); + runNextOperation(); + })); })); }); @@ -1304,11 +1342,37 @@ private CompletableFuture enqueue(Supplier> operatio * * @return the timeout, or null once the runtime has stopped its timer. */ - private ScheduledFuture timeOut(CompletableFuture operation) { + private ScheduledFuture timeOut(CompletableFuture operation, Operation scope) { try { - return rtc.timer.schedule(() -> control.execute(() -> - operation.completeExceptionally(new TimeoutException("WebRTC did not finish an operation of " - + name + " within " + rtc.operationTimeout))), + return rtc.timer.schedule(() -> control.execute(() -> { + if (operation.isDone()) { + return; + } + scope.expired = true; + // Recovery calls are native operations too. If they stop + // answering, close the peer instead of leaving its queue open. + try { + scope.cleanupTimeout = rtc.timer.schedule(() -> control.execute(() -> { + if (activeOperation == scope && !closed) { + teardown(); + } + }), Timeouts.toMillis(rtc.operationTimeout), TimeUnit.MILLISECONDS); + } + catch (RejectedExecutionException e) { + // Runtime shutdown closes the peer and its unfinished steps. + } + TimeoutException error = new TimeoutException("WebRTC did not finish an operation of " + + name + " within " + rtc.operationTimeout); + // Fail the awaited steps, so dependent stages cannot apply a + // late answer. Their error paths finish before the queue moves. + List> steps = new ArrayList<>(scope.steps); + for (CompletableFuture step : steps) { + step.completeExceptionally(error); + } + if (steps.isEmpty()) { + operation.completeExceptionally(error); + } + }), Timeouts.toMillis(rtc.operationTimeout), TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { @@ -1318,6 +1382,32 @@ private ScheduledFuture timeOut(CompletableFuture operation) { } } + private CompletableFuture rollback() { + RTCSignalingState state = connection.getSignalingState(); + if (state == RTCSignalingState.STABLE) { + return CompletableFuture.completedFuture(null); + } + RTCSessionDescription description = new RTCSessionDescription(RTCSdpType.ROLLBACK, ""); + return state == RTCSignalingState.HAVE_REMOTE_OFFER || state == RTCSignalingState.HAVE_LOCAL_PR_ANSWER + ? setRemote(description) : setLocal(description); + } + + private CompletableFuture operationStep() { + CompletableFuture future = new CompletableFuture<>(); + Operation scope = activeOperation; + if (scope != null) { + scope.steps.add(future); + future.whenComplete((value, error) -> scope.steps.remove(future)); + } + return future; + } + + private static final class Operation { + final Set> steps = new HashSet<>(); + boolean expired; + ScheduledFuture cleanupTimeout; + } + private void runNextOperation() { if (operationRunning) { return; @@ -1332,32 +1422,32 @@ private void runNextOperation() { } private CompletableFuture setLocal() { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); connection.setLocalDescription(descriptionObserver(future, "Applying the local description failed")); return future; } private CompletableFuture setLocal(RTCSessionDescription description) { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); connection.setLocalDescription(description, descriptionObserver(future, "Applying the local description failed")); return future; } private CompletableFuture setRemote(RTCSessionDescription description) { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); connection.setRemoteDescription(description, descriptionObserver(future, "Applying the remote " + description.sdpType.name().toLowerCase() + " failed")); return future; } private CompletableFuture createOffer() { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); connection.createOffer(new RTCOfferOptions(), createObserver(future, "Creating the offer failed")); return future; } private CompletableFuture createAnswer() { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); connection.createAnswer(new RTCAnswerOptions(), createObserver(future, "Creating the answer failed")); return future; } @@ -1398,7 +1488,7 @@ public void onFailure(String error) { * description to add it to. */ private CompletableFuture addRemoteCandidate(RTCIceCandidate candidate) { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); if (connection.getRemoteDescription() == null) { heldCandidates.add(new HeldCandidate(candidate, future)); @@ -1447,7 +1537,7 @@ private CompletableFuture waitForGathering() { return CompletableFuture.completedFuture(null); } - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture future = operationStep(); gatheringWaiters.add(future); // A STUN or TURN server that cannot be reached holds up the end of @@ -1724,10 +1814,20 @@ void channelClosed(DataChannelImpl channel) { } private void teardown() { + if (teardownStarted) { + return; + } + teardownStarted = true; + closing.set(true); closed = true; PeerClosedException error = closedError(); + if (activeOperation != null) { + for (CompletableFuture step : new ArrayList<>(activeOperation.steps)) { + step.completeExceptionally(error); + } + } if (runningOperation != null) { runningOperation.completeExceptionally(error); } @@ -1969,9 +2069,14 @@ private CompletableFuture gathered(Gathering gathering) { } private CompletableFuture run(Supplier> operation) { + return run(operation, (value, error) -> { }); + } + + private CompletableFuture run(Supplier> operation, + BiConsumer finished) { CompletableFuture result = new CompletableFuture<>(); - control.execute(() -> enqueue(operation) + control.execute(() -> enqueue(operation, finished) .whenComplete((value, error) -> complete(result, value, error))); return toPublic(result); diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/NegotiationTimeoutTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/NegotiationTimeoutTests.java new file mode 100644 index 000000000..c8bb75c4e --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/NegotiationTimeoutTests.java @@ -0,0 +1,202 @@ +/* + * Copyright 2026 Alex Andres + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.onvoid.webrtc.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.http.HttpClient; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.UnaryOperator; + +import dev.onvoid.webrtc.RTCSessionDescription; +import dev.onvoid.webrtc.RTCSignalingState; + +import org.junit.jupiter.api.Test; + +class NegotiationTimeoutTests extends ClientTestBase { + + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(37); + + @Test + void lateRestartAnswerDiscarded() throws Exception { + lateRestartAnswer(false); + } + + @Test + void lateCounterOfferDiscarded() throws Exception { + lateRestartAnswer(true); + } + + private void lateRestartAnswer(boolean remoteOffer) throws Exception { + ManualExecutor control = new ManualExecutor(); + ManualExecutor events = new ManualExecutor(); + ManualTimer timer = new ManualTimer(); + WebRtc client = runtime(control, events, timer); + try { + PeerImpl peer = open(client, control, events, remoteOffer); + CompletableFuture> late = new CompletableFuture<>(); + AtomicBoolean firstSent = new AtomicBoolean(); + AtomicBoolean applied = new AtomicBoolean(); + CompletableFuture first = peer.restartIce(offer -> { + firstSent.set(true); + return late; + }); + control.runUntil(firstSent::get, TIMEOUT_SECONDS); + + CompletableFuture> answer = new CompletableFuture<>(); + AtomicBoolean nextSent = new AtomicBoolean(); + CompletableFuture next = peer.restartIce(offer -> { + nextSent.set(true); + return answer; + }); + assertEquals(1, timer.fire(OPERATION_TIMEOUT)); + control.runUntil(nextSent::get, TIMEOUT_SECONDS); + events.runAll(); + assertInstanceOf(TimeoutException.class, + assertThrows(ExecutionException.class, () -> await(first)).getCause()); + + // The old answer arrives while the next restart owns signaling. + late.complete(sdp -> { + applied.set(true); + return sdp; + }); + control.runAll(); + assertFalse(applied.get()); + assertFalse(next.isDone()); + answer.complete(UnaryOperator.identity()); + drive(next, control, events); + assertEquals(RTCSignalingState.STABLE, + drive(peer.withNative(connection -> connection.getSignalingState()), control, events)); + } + finally { + close(client, control, events); + } + } + + @Test + void restartCleanupPrecedesEvents() throws Exception { + ManualExecutor control = new ManualExecutor(); + ManualExecutor events = new ManualExecutor(); + WebRtc client = runtime(control, events, new ManualTimer()); + CountDownLatch release = new CountDownLatch(1); + HttpIce ice = null; + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) { + PeerImpl peer = open(client, control, events, false); + String local = drive(peer.withNative(connection -> connection.getLocalDescription().sdp), control, events); + String remote = drive(peer.withNative(connection -> connection.getRemoteDescription().sdp), control, events); + ice = new HttpIce(client, HttpClient.newHttpClient(), peer, null, Duration.ofSeconds(30)); + ice.attach(endpoint.uri(), "\"initial\"", local); + endpoint.iceRestarts = fragment -> HttpIce.Fragment.of(remote).text; + endpoint.iceRestartStatus = 503; + CompletableFuture first = ice.restart(); + // A second queued operation can start before first's public future + // is delivered. Its request is held while those events are drained. + AtomicBoolean secondEntered = new AtomicBoolean(); + AtomicReference secondOffer = new AtomicReference<>(); + CompletableFuture ready = new CompletableFuture<>(); + peer.withNative(connection -> { + ready.complete(null); + return null; + }); + control.runUntil(ready::isDone, TIMEOUT_SECONDS); + assertFalse(first.isDone()); + endpoint.iceRestartStatus = 200; + endpoint.iceRestarts = fragment -> { + secondOffer.set(fragment); + secondEntered.set(true); + try { + assertTrue(release.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + return HttpIce.Fragment.of(remote).text; + }; + CompletableFuture second = ice.restart(); + control.runUntil(secondEntered::get, TIMEOUT_SECONDS); + events.runAll(); + assertThrows(ExecutionException.class, () -> await(first)); + release.countDown(); + drive(second, control, events); + // The failed restart's event must not restore old credentials or + // clear the candidates gathered for the successful restart. + control.runUntil(() -> endpoint.requests.stream().anyMatch( + request -> "\"restart-1\"".equals(request.ifMatch)), TIMEOUT_SECONDS); + TestHttpEndpoint.Request candidates = endpoint.awaitRequest( + request -> "\"restart-1\"".equals(request.ifMatch), "restart candidates", TIMEOUT_SECONDS); + assertEquals(HttpIce.Fragment.parse(secondOffer.get()).ufrag, + HttpIce.Fragment.parse(candidates.body).ufrag); + } + finally { + release.countDown(); + if (ice != null) { + ice.close(); + } + close(client, control, events); + } + } + + private WebRtc runtime(ManualExecutor control, ManualExecutor events, ManualTimer timer) { + return WebRtc.builder().audio(AudioMode.HEADLESS).controlPool(control).eventExecutor(events) + .timer(timer).operationTimeout(OPERATION_TIMEOUT).errorHandler(errors::add).build(); + } + + private PeerImpl open(WebRtc client, ManualExecutor control, ManualExecutor events, boolean remoteOffer) + throws Exception { + CompletableFuture opening = CompletableFuture.supplyAsync(() -> client.createPeer(PeerOptions.builder() + .negotiation(Negotiation.MANUAL).receive(MediaKind.AUDIO).build())); + PeerImpl peer = (PeerImpl) drive(opening, control, events); + Peer server = rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL) + .receive(MediaKind.AUDIO).build()); + if (remoteOffer) { + RTCSessionDescription offer = await(server.negotiator().createOffer(Gathering.COMPLETE)); + RTCSessionDescription answer = drive(peer.negotiator().answer(offer, Gathering.COMPLETE), control, events); + await(server.negotiator().applyAnswer(answer)); + } + else { + RTCSessionDescription offer = drive(peer.negotiator().createOffer(Gathering.COMPLETE), control, events); + RTCSessionDescription answer = await(server.negotiator().answer(offer, Gathering.COMPLETE)); + drive(peer.negotiator().applyAnswer(answer), control, events); + } + return peer; + } + + private static T drive(CompletableFuture future, ManualExecutor control, ManualExecutor events) + throws Exception { + control.runUntil(() -> { + events.runAll(); + return future.isDone(); + }, TIMEOUT_SECONDS); + return await(future); + } + + private static void close(WebRtc client, ManualExecutor control, ManualExecutor events) throws Exception { + drive(client.closeAsync(), control, events); + } +}