fix: close an HTTP session's peer and resource independently, and fail on errors

docs/guide/client-api.md

@@ -598,7 +598,7 @@ 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. 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.
+`session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. Both start at once, so a slow server does not keep the peer open. 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, and also fails the future of `closeAsync()`, which completes once both are done; a peer that did not close fails it too.
 
 The bearer token goes to the endpoint, and along with the deletion to the session the endpoint names in its `Location` header. That session may be on another host, but an `https` endpoint cannot name a plain `http` one. Redirects are not followed, since they would take the token elsewhere: a redirected request fails with its status.
 

webrtc-examples/src/main/java/dev/onvoid/webrtc/examples/WhepExample.java

@@ -97,7 +97,7 @@ public static void main(String[] args) throws Exception {
 
             Thread.sleep(seconds * 1000L);
 
-            // Deletes the session on the server, then closes the peer.
+            // Deletes the session on the server and closes the peer.
             session.closeAsync().get(5, TimeUnit.SECONDS);
 
             System.out.println("Ended the session " + session.resource());

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

@@ -42,7 +42,7 @@
 /**
  * A WHIP or WHEP session: a peer whose offer and answer went over HTTP, and the
  * resource the server created for it. Closing the session deletes the resource
- * and closes the peer.
+ * and closes the peer, independently of each other.
  */
 abstract class HttpSession implements AutoCloseable {
 
@@ -92,25 +92,40 @@ public URI resource() {
 
 	/**
 	 * Deletes the session's resource on the server, which frees what the
-	 * server holds for it, and closes the peer.
+	 * server holds for it, and closes the peer. Both start at once, so that a
+	 * slow server does not keep the peer open. A deletion that fails also goes
+	 * to the runtime's error handler, for those who close without waiting.
 	 *
-	 * @return a future that completes once both are done.
+	 * @return a future that completes once both are done, or fails with a
+	 * {@link WebRtcException} when the resource could not be deleted, or with
+	 * why the peer did not close. A failure of both carries the peer's as a
+	 * suppressed exception.
 	 */
 	public CompletableFuture<Void> closeAsync() {
 		if (closing.compareAndSet(false, true)) {
-			CompletableFuture<Void> deleted;
+			CompletableFuture<Throwable> peerClosed = peer.closeAsync().handle((ignored, error) -> error);
+			CompletableFuture<WebRtcException> deleted;
 
 			try {
-				deleted = CompletableFuture.runAsync(() -> discard(rtc, resource, bearerToken, timeout),
+				deleted = CompletableFuture.supplyAsync(() -> discard(rtc, resource, bearerToken, timeout),
 						rtc.controlPool);
 			}
 			catch (RejectedExecutionException e) {
-				leftOnServer(rtc, resource);
-				deleted = CompletableFuture.completedFuture(null);
+				deleted = CompletableFuture.completedFuture(leftOnServer(rtc, resource));
 			}
 
-			deleted.thenCompose(ignored -> peer.closeAsync())
-					.whenComplete((ignored, error) -> closed.complete(null));
+			deleted.thenAcceptBoth(peerClosed, (deleteError, peerError) -> {
+				Throwable error = PeerImpl.unwrap(peerError);
+
+				if (deleteError != null) {
+					if (error != null) {
+						deleteError.addSuppressed(error);
+					}
+					error = deleteError;
+				}
+
+				PeerImpl.complete(closed, null, error);
+			});
 		}
 
 		return ClientFuture.copyOf(closed);
@@ -301,25 +316,37 @@ private static void delete(URI resource, String bearerToken, Duration timeout) t
 	/**
 	 * Deletes a resource that nothing else will, and reports rather than
 	 * throws if that fails.
+	 *
+	 * @return the failure reported, or null.
 	 */
-	private static void discard(WebRtc rtc, URI resource, String bearerToken, Duration timeout) {
+	private static WebRtcException discard(WebRtc rtc, URI resource, String bearerToken, Duration timeout) {
 		if (resource == null) {
-			return;
+			return null;
 		}
 
 		try {
 			delete(resource, bearerToken, timeout);
+			return null;
 		}
 		catch (Exception e) {
-			rtc.reportError(new WebRtcException("Deleting the session " + resource + " failed", e));
+			WebRtcException error = new WebRtcException("Deleting the session " + resource + " failed", e);
+			rtc.reportError(error);
+			return error;
 		}
 	}
 
-	private static void leftOnServer(WebRtc rtc, URI resource) {
-		if (resource != null) {
-			rtc.reportError(new WebRtcException("The session " + resource
-					+ " was left on the server, since the runtime is closed"));
+	/**
+	 * @return the failure reported, or null if there was no resource.
+	 */
+	private static WebRtcException leftOnServer(WebRtc rtc, URI resource) {
+		if (resource == null) {
+			return null;
 		}
+
+		WebRtcException error = new WebRtcException("The session " + resource
+				+ " was left on the server, since the runtime is closed");
+		rtc.reportError(error);
+		return error;
 	}
 
 	/**

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

@@ -20,9 +20,11 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.lang.reflect.Proxy;
 import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -329,8 +331,9 @@ void closeAfterRuntimeReported() throws Exception {
 			Whep.Session session = await(Whep.from(endpoint.uri()).play(client));
 
 			client.close();
-			await(session.closeAsync());
+			ExecutionException failed = assertThrows(ExecutionException.class, () -> await(session.closeAsync()));
 
+			assertTrue(failed.getCause().getMessage().contains("left on the server"), failed.getCause().getMessage());
 			assertEquals(0, endpoint.count("DELETE /live/session/1"));
 			assertEquals(1, errors.size(), () -> "Errors: " + errors);
 			assertTrue(errors.get(0).getMessage().contains("left on the server"), errors.get(0).getMessage());
@@ -362,8 +365,54 @@ void failedDeleteReported() throws Exception {
 			endpoint.deleteStatus = 500;
 
 			Whip.Session session = await(Whip.to(endpoint.uri()).track(video).publish(rtc));
-			await(session.closeAsync());
 
+			assertCloseFails(session, "500");
+			// Closed all the same.
+			assertEquals(PeerState.CLOSED, session.peer().state());
+			assertDeleteReported("500");
+		}
+	}
+
+	@Test
+	void slowDeleteKeepsNoPeerOpen() throws Exception {
+		// The peer closes while the server has yet to answer the deletion,
+		// and the session is closed once both are done.
+		try (TestHttpEndpoint endpoint = receivingEndpoint()) {
+			Whip.Session session = await(Whip.to(endpoint.uri()).track(video).publish(rtc));
+			CountDownLatch deleted = endpoint.holdDeletes();
+			CompletableFuture<Void> closed = session.closeAsync();
+
+			try {
+				endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS);
+				await(session.peer().whenClosed());
+				assertFalse(closed.isDone());
+			}
+			finally {
+				deleted.countDown();
+			}
+
+			await(closed);
+		}
+	}
+
+	@Test
+	void failedPeerCloseFails() throws Exception {
+		IllegalStateException peerError = new IllegalStateException("The peer did not close");
+
+		// Nothing to delete: the session fails as the peer did.
+		HttpSession session = new Whep.Session(rtc, failingToClose(peerError), null, null, SESSION_TIMEOUT);
+		ExecutionException failed = assertThrows(ExecutionException.class, () -> await(session.closeAsync()));
+		assertSame(peerError, failed.getCause());
+
+		// Both fail: the deletion leads, and the peer's failure goes along.
+		try (TestHttpEndpoint endpoint = receivingEndpoint()) {
+			endpoint.deleteStatus = 500;
+
+			HttpSession both = new Whep.Session(rtc, failingToClose(peerError),
+					endpoint.uri().resolve("/live/session/1"), null, SESSION_TIMEOUT);
+			Throwable error = assertCloseFails(both, "500");
+
+			assertEquals(Arrays.asList(peerError), Arrays.asList(error.getSuppressed()));
 			assertDeleteReported("500");
 		}
 	}
@@ -375,8 +424,8 @@ void redirectNotFollowed() throws Exception {
 			endpoint.deleteStatus = 307;
 
 			Whip.Session session = await(Whip.to(endpoint.uri()).bearerToken("secret").track(video).publish(rtc));
-			await(session.closeAsync());
 
+			assertCloseFails(session, "307");
 			assertDeleteReported("307");
 			assertFalse(endpoint.requests.stream().anyMatch(request -> request.path.equals(endpoint.redirect)),
 					() -> "Requests: " + endpoint.requests);
@@ -500,6 +549,36 @@ private TestHttpEndpoint receivingEndpoint() throws Exception {
 				.build()), offer), 201);
 	}
 
+	/**
+	 * Checks that closing the session fails with the deletion the server
+	 * answered with the given status.
+	 *
+	 * @return the failure.
+	 */
+	private static Throwable assertCloseFails(HttpSession session, String status) {
+		ExecutionException failed = assertThrows(ExecutionException.class, () -> await(session.closeAsync()));
+		Throwable error = failed.getCause();
+
+		assertInstanceOf(WebRtcException.class, error);
+		assertTrue(error.getMessage().contains("Deleting the session"), error::getMessage);
+		assertTrue(error.getCause().getMessage().contains(status), () -> error.getCause().getMessage());
+
+		return error;
+	}
+
+	/**
+	 * A peer whose close fails, as no peer of the runtime can be made to.
+	 */
+	private static Peer failingToClose(Throwable error) {
+		return (Peer) Proxy.newProxyInstance(Peer.class.getClassLoader(), new Class<?>[] { Peer.class },
+				(proxy, method, args) -> {
+					if (method.getName().equals("closeAsync")) {
+						return ClientFuture.failed(error);
+					}
+					throw new UnsupportedOperationException(method.getName());
+				});
+	}
+
 	/**
 	 * Checks that the one error reported is the failed deletion, and takes it
 	 * out of those that fail the test.

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

@@ -76,6 +76,9 @@ public String toString() {
 	/** Holds the answers to POSTs while set, until it is released. */
 	private volatile CountDownLatch answerGate;
 
+	/** Holds the answers to DELETEs while set, until it is released. */
+	private volatile CountDownLatch deleteGate;
+
 	private int answered;
 
 	private final ServerSocket server;
@@ -113,6 +116,16 @@ CountDownLatch holdAnswers() {
 		return gate;
 	}
 
+	/**
+	 * Holds the answer to every DELETE from now on, after the request has been
+	 * recorded, until the returned latch is counted down.
+	 */
+	CountDownLatch holdDeletes() {
+		CountDownLatch gate = new CountDownLatch(1);
+		deleteGate = gate;
+		return gate;
+	}
+
 	/**
 	 * @return how many POSTs have been answered, counted before the answer is
 	 * written: once a client has read one, it is counted.
@@ -201,32 +214,38 @@ else if (name.equals("authorization")) {
 			}
 
 			String answer = answerer.apply(new String(body, StandardCharsets.UTF_8));
-			CountDownLatch gate = answerGate;
-
-			if (gate != null) {
-				try {
-					gate.await();
-				}
-				catch (InterruptedException e) {
-					Thread.currentThread().interrupt();
-					throw new IOException(e);
-				}
-			}
+			pass(answerGate);
 
 			synchronized (this) {
 				answered++;
 			}
 
 			respond(out, "201 Created", "Location: " + location + "\r\n", answer);
+			return;
 		}
-		else if (deleteStatus / 100 == 3) {
+
+		pass(deleteGate);
+
+		if (deleteStatus / 100 == 3) {
 			respond(out, deleteStatus + " Moved", "Location: " + redirect + "\r\n", "");
 		}
 		else {
 			respond(out, deleteStatus + " Status", "", "");
 		}
 	}
 
+	private static void pass(CountDownLatch gate) throws IOException {
+		if (gate != null) {
+			try {
+				gate.await();
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new IOException(e);
+			}
+		}
+	}
+
 	private static void respond(OutputStream out, String status, String headers, String body) throws IOException {
 		byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
 		String head = "HTTP/1.1 " + status + "\r\nContent-Type: application/sdp\r\n" + headers