fix: end a WHIP or WHEP start whose future was cancelled

docs/guide/client-api.md

@@ -543,7 +543,7 @@ Whep.Session viewer = Whep.from(URI.create("https://example.com/live/whep"))
 viewer.peer().onTrack(track -> System.out.println("Receiving " + track.kind()));
 ```
 
-The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none.
+The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. Cancelling the future, say once `get` gave up waiting, ends the session the same way, even when the answer is still on its way. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none.
 
 `session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else. A deletion the server refuses or that fails goes to the runtime's error handler.
 

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

@@ -26,6 +26,7 @@
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
@@ -122,14 +123,41 @@ public void close() {
 		closeAsync();
 	}
 
+	/**
+	 * Negotiates the peer over HTTP and hands the session out once the peer is
+	 * connected. A caller that stops waiting, by cancelling the future or
+	 * completing it itself, ends the negotiation, and a session made too late
+	 * for the future is closed: either way the peer is closed and what the
+	 * server created is deleted.
+	 *
+	 * @param factory makes the session of the connected peer.
+	 *
+	 * @return the future handed to the application.
+	 */
+	static <S extends HttpSession> CompletableFuture<S> start(WebRtc rtc, PeerImpl peer, URI endpoint,
+			String bearerToken, Duration timeout, Factory<S> factory) {
+		CompletableFuture<Void> abandoned = new CompletableFuture<>();
+		CompletableFuture<S> session = peer.toPublic(negotiate(rtc, peer, endpoint, bearerToken, timeout, abandoned)
+				.thenApply(resource -> factory.create(rtc, peer, resource, bearerToken, timeout)),
+				HttpSession::close);
+
+		// Once the session was handed out, this finds the negotiation finished.
+		session.whenComplete((value, error) -> abandoned.complete(null));
+
+		return session;
+	}
+
 	/**
 	 * Negotiates a manual peer over HTTP: creates the offer with all candidates,
 	 * posts it, applies the answer and waits until the peer is connected.
 	 *
+	 * @param abandoned completes when no one waits for the session anymore,
+	 *                  which then ends as if it had failed.
+	 *
 	 * @return a future with the resource the server named, or null.
 	 */
-	static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken,
-			Duration timeout) {
+	private static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken,
+			Duration timeout, CompletableFuture<?> abandoned) {
 		CompletableFuture<URI> result = new CompletableFuture<>();
 		// The resource the server created. Whoever takes it out owns it: the
 		// session that connected, or whoever deletes it.
@@ -164,6 +192,11 @@ static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, Str
 				"The session with " + endpoint + " was not connected within " + timeout)),
 				timeout.toMillis(), TimeUnit.MILLISECONDS);
 
+		abandoned.thenRun(() -> {
+			deadline.cancel(false);
+			fail.accept(new CancellationException("The session with " + endpoint + " was abandoned"));
+		});
+
 		peer.negotiator().createOffer(Gathering.COMPLETE)
 				.thenApplyAsync(offer -> {
 					try {
@@ -238,6 +271,16 @@ private static String post(URI endpoint, String sdp, String bearerToken, Duratio
 		}
 	}
 
+	/**
+	 * Makes the session of a peer that connected.
+	 *
+	 * @param <S> the session type.
+	 */
+	interface Factory<S extends HttpSession> {
+
+		S create(WebRtc rtc, Peer peer, URI resource, String bearerToken, Duration timeout);
+	}
+
 	private static void delete(URI resource, String bearerToken, Duration timeout) throws IOException {
 		HttpURLConnection connection = open(resource, "DELETE", bearerToken, timeout);
 

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

@@ -564,14 +564,32 @@ long channelReceiveLimit() {
 	 * executor, so its callbacks run there like the handlers.
 	 */
 	<T> CompletableFuture<T> toPublic(CompletableFuture<T> internal) {
+		return toPublic(internal, value -> { });
+	}
+
+	/**
+	 * Hands a future out to the application, as {@link #toPublic(CompletableFuture)}
+	 * does, and frees a result that comes too late for it: the caller may have
+	 * cancelled or completed the future meanwhile, and then no one else would.
+	 *
+	 * @param unclaimed frees a result the future did not take.
+	 */
+	<T> CompletableFuture<T> toPublic(CompletableFuture<T> internal, Consumer<? super T> unclaimed) {
 		ClientFuture<T> result = new ClientFuture<>();
 
 		rtc.track(result, events);
 
-		// Completed in any case: should the event executor reject the task, on
-		// the thread that found it rejected.
-		internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error),
-				() -> complete(result, value, error)));
+		internal.whenComplete((value, error) -> {
+			Runnable deliver = () -> {
+				if (!complete(result, value, error) && error == null && value != null) {
+					unclaimed.accept(value);
+				}
+			};
+
+			// Completed in any case: should the event executor reject the
+			// task, on the thread that found it rejected.
+			events.execute(deliver, deliver);
+		});
 
 		return result;
 	}
@@ -681,13 +699,15 @@ static WebRtcException asWebRtcException(Throwable error, String message) {
 		return new WebRtcException(message + ": " + error, error);
 	}
 
-	static <T> void complete(CompletableFuture<T> future, T value, Throwable error) {
+	/**
+	 * @return whether this completed the future, which was not done before.
+	 */
+	static <T> boolean complete(CompletableFuture<T> future, T value, Throwable error) {
 		if (error != null) {
-			future.completeExceptionally(unwrap(error));
-		}
-		else {
-			future.complete(value);
+			return future.completeExceptionally(unwrap(error));
 		}
+
+		return future.complete(value);
 	}
 
 	private LocalTrackImpl checkTrack(LocalTrack track) {

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

@@ -154,7 +154,8 @@ public Builder timeout(Duration timeout) {
 		 *
 		 * @return a future with the session once its peer is connected. It
 		 * fails if the endpoint refuses the offer or the peer does not connect
-		 * in time, with nothing left behind on the server.
+		 * in time, with nothing left behind on the server. Cancelling it ends
+		 * the session the same way.
 		 */
 		public CompletableFuture<Session> play(WebRtc rtc) {
 			requireNonNull(rtc, "rtc");
@@ -175,8 +176,7 @@ public CompletableFuture<Session> play(WebRtc rtc) {
 			PeerImpl peer = (PeerImpl) rtc.createPeer(options.build());
 
 			try {
-				return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, token, limit)
-						.thenApply(resource -> new Session(rtc, peer, resource, token, limit)));
+				return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new);
 			}
 			catch (RuntimeException | Error e) {
 				// Until the session has it, no one else would close the peer.

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

@@ -145,7 +145,8 @@ public Builder timeout(Duration timeout) {
 		 *
 		 * @return a future with the session once its peer is connected. It
 		 * fails if the endpoint refuses the offer or the peer does not connect
-		 * in time, with nothing left behind on the server.
+		 * in time, with nothing left behind on the server. Cancelling it ends
+		 * the session the same way.
 		 *
 		 * @throws IllegalStateException    if no track was given.
 		 * @throws IllegalArgumentException if a track was not created by the
@@ -184,8 +185,7 @@ public CompletableFuture<Session> publish(WebRtc rtc) {
 					peer.addTrack(track);
 				}
 
-				return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, token, limit)
-						.thenApply(resource -> new Session(rtc, peer, resource, token, limit)));
+				return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new);
 			}
 			catch (RuntimeException | Error e) {
 				// Until the session has it, no one else would close the peer.

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

@@ -34,6 +34,7 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
 import dev.onvoid.webrtc.RTCSdpType;
@@ -211,6 +212,54 @@ void lateAnswerDeleted() throws Exception {
 		}
 	}
 
+	@Test
+	void cancelledStartDeletes() throws Exception {
+		// The caller cancels while the offer is out. The peer must not go on to
+		// connect, and what the server creates must not be left there.
+		WebRtc client = clientRuntime(builder -> builder);
+
+		try {
+			CustomVideo track = client.media().customVideo("video");
+
+			assertCancelDeletes(client, uri -> Whip.to(uri).track(track).publish(client));
+			assertCancelDeletes(client, uri -> Whep.from(uri).play(client));
+		}
+		finally {
+			client.close();
+		}
+	}
+
+	@Test
+	void unclaimedSessionClosed() throws Exception {
+		// The caller cancels once the peer is connected, as the session is
+		// made or right before. A session the future cannot take anymore is
+		// closed, rather than left connected with no one to close it.
+		WebRtc client = clientRuntime(builder -> builder);
+
+		try (TestHttpEndpoint endpoint = receivingEndpoint()) {
+			CountDownLatch answer = endpoint.holdAnswers();
+			CompletableFuture<Whip.Session> session = Whip.to(endpoint.uri())
+					.track(client.media().customVideo("video"))
+					.publish(client);
+
+			endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS);
+			Peer peer = client.peers().get(0);
+			CompletableFuture<Void> cancelled = peer.whenConnected().thenRun(() -> session.cancel(true));
+
+			answer.countDown();
+			await(cancelled);
+			await(peer.whenClosed());
+			endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS);
+
+			assertTrue(session.isCancelled());
+			client.close();
+			assertEquals(1, endpoint.count("DELETE /live/session/1"));
+		}
+		finally {
+			client.close();
+		}
+	}
+
 	@Test
 	void timeoutAfterAnswerDeletes() throws Exception {
 		// The session times out once the answer was read, before it is applied.
@@ -377,6 +426,32 @@ private WebRtc clientRuntime(UnaryOperator<WebRtc.Builder> options) {
 				.build();
 	}
 
+	/**
+	 * Starts a session, cancels it while the answer is held and releases the
+	 * answer, whose resource must then be deleted.
+	 */
+	private static void assertCancelDeletes(WebRtc client,
+			Function<URI, CompletableFuture<? extends HttpSession>> start) throws Exception {
+		try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) {
+			CountDownLatch answer = endpoint.holdAnswers();
+			CompletableFuture<? extends HttpSession> session = start.apply(endpoint.uri());
+
+			endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS);
+			PeerImpl peer = (PeerImpl) client.peers().get(0);
+
+			assertTrue(session.cancel(true));
+			// Closed by the cancellation itself, so it never connects.
+			assertTrue(peer.isClosing(), () -> "Not closing: " + peer);
+
+			answer.countDown();
+			endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS);
+			await(peer.whenClosed());
+
+			assertTrue(session.isCancelled());
+			assertEquals(1, endpoint.count("DELETE /live/session/1"));
+		}
+	}
+
 	/**
 	 * Closes a runtime whose control tasks run only when the test runs them.
 	 */

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

@@ -64,6 +64,28 @@ void channelRoundTrip() throws Exception {
 		assertTrue(errors.isEmpty(), () -> "Errors: " + errors);
 	}
 
+	@Test
+	void lateResultFreed() throws Exception {
+		// A result that comes after the caller cancelled has no one else to
+		// free it. One the caller takes is left alone.
+		PeerImpl peer = (PeerImpl) rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL).build());
+		CompletableFuture<String> late = new CompletableFuture<>();
+		CompletableFuture<String> freed = new CompletableFuture<>();
+		CompletableFuture<String> cancelled = peer.toPublic(late, freed::complete);
+
+		assertTrue(cancelled.cancel(true));
+		late.complete("late");
+		assertEquals("late", await(freed));
+
+		CompletableFuture<String> taken = new CompletableFuture<>();
+		CompletableFuture<String> kept = new CompletableFuture<>();
+		CompletableFuture<String> handed = peer.toPublic(taken, kept::complete);
+
+		taken.complete("taken");
+		assertEquals("taken", await(handed));
+		assertFalse(kept.isDone());
+	}
+
 	@Test
 	void stateHandlerGetsCurrentState() throws Exception {
 		Peer[] peers = pair();