fix: discard expired negotiations and serialize ICE restart cleanup

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

@@ -175,7 +175,7 @@ CompletableFuture<Void> restart() {
 			}
 		}
 
-		Fragment[] before = new Fragment[1];
+		Restart restart = new Restart();
 
 		return peer.restartIce(offer -> {
 			HttpCall request;
@@ -186,26 +186,31 @@ CompletableFuture<Void> 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<Void> restart() {
 		});
 	}
 
+	private static final class Restart {
+		Fragment before;
+		HttpCall request;
+	}
+
 	/**
 	 * Stops sending: cuts off the requests under way, and makes no more.
 	 */

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<RTCIceCandidate> candidates, Runnable gathered)
 	 */
 	CompletableFuture<Void> restartIce(
 			Function<RTCSessionDescription, CompletableFuture<UnaryOperator<String>>> exchange) {
+		return restartIce(exchange, (value, error) -> { });
+	}
+
+	CompletableFuture<Void> restartIce(
+			Function<RTCSessionDescription, CompletableFuture<UnaryOperator<String>>> exchange,
+			BiConsumer<Void, Throwable> finished) {
 		return negotiator.run(() -> {
 			RTCSessionDescription remote = connection.getRemoteDescription();
 
@@ -775,15 +787,15 @@ CompletableFuture<Void> restartIce(
 								.thenCompose(ignored -> failed);
 					})
 					.thenCompose(result -> result);
-		});
+		}, finished);
 	}
 
 	/**
 	 * @return a future that completes as the given one, but on the control
 	 * executor, or fails once the peer is closed.
 	 */
 	private <T> CompletableFuture<T> onControl(CompletableFuture<T> future) {
-		CompletableFuture<T> result = new CompletableFuture<>();
+		CompletableFuture<T> 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 <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> operation) {
+		return enqueue(operation, (value, error) -> { });
+	}
+
+	private <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> operation,
+			BiConsumer<T, Throwable> finished) {
 		CompletableFuture<T> result = new CompletableFuture<>();
 
 		operations.add(() -> {
+			Operation scope = new Operation();
+			activeOperation = scope;
 			CompletableFuture<T> running;
 
 			if (closed) {
@@ -1279,18 +1298,37 @@ private <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> 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<Void> 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 <T> CompletableFuture<T> enqueue(Supplier<CompletableFuture<T>> 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<CompletableFuture<?>> 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<Void> 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 <T> CompletableFuture<T> operationStep() {
+		CompletableFuture<T> 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<CompletableFuture<?>> steps = new HashSet<>();
+		boolean expired;
+		ScheduledFuture<?> cleanupTimeout;
+	}
+
 	private void runNextOperation() {
 		if (operationRunning) {
 			return;
@@ -1332,32 +1422,32 @@ private void runNextOperation() {
 	}
 
 	private CompletableFuture<Void> setLocal() {
-		CompletableFuture<Void> future = new CompletableFuture<>();
+		CompletableFuture<Void> future = operationStep();
 		connection.setLocalDescription(descriptionObserver(future, "Applying the local description failed"));
 		return future;
 	}
 
 	private CompletableFuture<Void> setLocal(RTCSessionDescription description) {
-		CompletableFuture<Void> future = new CompletableFuture<>();
+		CompletableFuture<Void> future = operationStep();
 		connection.setLocalDescription(description, descriptionObserver(future, "Applying the local description failed"));
 		return future;
 	}
 
 	private CompletableFuture<Void> setRemote(RTCSessionDescription description) {
-		CompletableFuture<Void> future = new CompletableFuture<>();
+		CompletableFuture<Void> future = operationStep();
 		connection.setRemoteDescription(description, descriptionObserver(future,
 				"Applying the remote " + description.sdpType.name().toLowerCase() + " failed"));
 		return future;
 	}
 
 	private CompletableFuture<RTCSessionDescription> createOffer() {
-		CompletableFuture<RTCSessionDescription> future = new CompletableFuture<>();
+		CompletableFuture<RTCSessionDescription> future = operationStep();
 		connection.createOffer(new RTCOfferOptions(), createObserver(future, "Creating the offer failed"));
 		return future;
 	}
 
 	private CompletableFuture<RTCSessionDescription> createAnswer() {
-		CompletableFuture<RTCSessionDescription> future = new CompletableFuture<>();
+		CompletableFuture<RTCSessionDescription> 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<Void> addRemoteCandidate(RTCIceCandidate candidate) {
-		CompletableFuture<Void> future = new CompletableFuture<>();
+		CompletableFuture<Void> future = operationStep();
 
 		if (connection.getRemoteDescription() == null) {
 			heldCandidates.add(new HeldCandidate(candidate, future));
@@ -1447,7 +1537,7 @@ private CompletableFuture<Void> waitForGathering() {
 			return CompletableFuture.completedFuture(null);
 		}
 
-		CompletableFuture<Void> future = new CompletableFuture<>();
+		CompletableFuture<Void> 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<RTCSessionDescription> gathered(Gathering gathering) {
 		}
 
 		private <T> CompletableFuture<T> run(Supplier<CompletableFuture<T>> operation) {
+			return run(operation, (value, error) -> { });
+		}
+
+		private <T> CompletableFuture<T> run(Supplier<CompletableFuture<T>> operation,
+				BiConsumer<T, Throwable> finished) {
 			CompletableFuture<T> result = new CompletableFuture<>();
 
-			control.execute(() -> enqueue(operation)
+			control.execute(() -> enqueue(operation, finished)
 					.whenComplete((value, error) -> complete(result, value, error)));
 
 			return toPublic(result);