feat: retry WHIP/WHEP offers after Retry-After

docs/guide/client-api.md

@@ -598,7 +598,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. Cancelling the future, say once `get` gave up waiting, ends the session the same way. `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. An endpoint that is busy, with status 503, or a WHEP endpoint whose stream has yet to start, with status 409, gets the offer again after the wait its `Retry-After` names, at least a second, as long as that fits within the timeout; without `Retry-After` the session fails at once. Cancelling the future, say once `get` gave up waiting, ends the session the same way. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none.
 
 The timeout bounds the requests as a whole, not each read, so a server that sends its answer a byte at a time cannot stretch it. A request still out when the session times out or is cancelled is cut off at once; one still waiting for the head of its answer only from Java 16 on, and before that at its timeout. Once the server has named the session, in the head of its answer, it is deleted all the same; a session the server created for an answer that had not come yet is left for the server to time out.
 

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

@@ -31,13 +31,21 @@
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import javax.net.ssl.SSLContext;
@@ -54,7 +62,8 @@
  * such a request ends with its timeout.
  * <p>
  * Redirects are followed only when asked for, and only those with status 307
- * and 308, which keep the method and body.
+ * and 308, which keep the method and body. Likewise, a request is only sent
+ * again when asked for, after the wait a server names in Retry-After.
  */
 final class HttpCall {
 
@@ -64,6 +73,12 @@ final class HttpCall {
 	/** How much of an error response is read, for a message. */
 	static final int MAX_ERROR_BODY = 1024;
 
+	/**
+	 * The shortest wait before a request is sent again, so that a server that
+	 * says to come back at once is not asked many times a second.
+	 */
+	static final long MIN_RETRY_MILLIS = 1000;
+
 	private final HttpClient client;
 
 	private final String method;
@@ -84,9 +99,16 @@ final class HttpCall {
 
 	private boolean followRedirects;
 
+	private ScheduledExecutorService retryTimer;
+
+	private Set<Integer> retryStatuses = Collections.emptySet();
+
 	/** The exchange under way, which an abort cancels. */
 	private CompletableFuture<?> exchange;
 
+	/** The wait before the request is sent again, which an abort ends. */
+	private Future<?> retry;
+
 	/** The body being read, which an abort ends. */
 	private Body body;
 
@@ -144,6 +166,24 @@ static HttpClient newClient(boolean authenticates) {
 		return builder.build();
 	}
 
+	/**
+	 * Sends the request again when the server answers with one of the given
+	 * statuses and a Retry-After header field, after the time it names, at
+	 * least {@link #MIN_RETRY_MILLIS}, and as often as that fits within the
+	 * timeout. Without the field, or once the wait would not fit, the answer
+	 * is the response.
+	 *
+	 * @param timer    the timer the waits are on.
+	 * @param statuses the statuses to send the request again for.
+	 *
+	 * @return this request.
+	 */
+	HttpCall retry(ScheduledExecutorService timer, Set<Integer> statuses) {
+		this.retryTimer = timer;
+		this.retryStatuses = statuses;
+		return this;
+	}
+
 	/**
 	 * Follows redirects with status 307 and 308, up to {@link #MAX_REDIRECTS}
 	 * of them, with the bearer token. One from https to plain http fails.
@@ -187,6 +227,7 @@ void abort() {
 				: new IOException("The request to " + uri + " was aborted");
 		CompletableFuture<?> exchange;
 		Body body;
+		Future<?> retry;
 
 		synchronized (this) {
 			if (aborted) {
@@ -196,10 +237,14 @@ void abort() {
 			aborted = true;
 			exchange = this.exchange;
 			body = this.body;
+			retry = this.retry;
 		}
 
 		result.completeExceptionally(reason);
 
+		if (retry != null) {
+			retry.cancel(false);
+		}
 		if (body != null) {
 			body.stop(reason);
 		}
@@ -263,7 +308,15 @@ private void exchange(URI target, String contentType, byte[] content, int maxBod
 			String location = value.headers().firstValue("Location").orElse(null);
 
 			if (!isRedirect(value.statusCode(), location)) {
-				result.complete(new Response(target, value.statusCode(), value.headers(), value.body()));
+				long wait = retryStatuses.contains(value.statusCode())
+						? retryDelayMillis(value.headers().firstValue("Retry-After").orElse(null),
+								System.currentTimeMillis())
+						: -1;
+
+				if (wait < 0 || !retryLater(Math.max(wait, MIN_RETRY_MILLIS),
+						() -> exchange(target, contentType, content, maxBody, onHead, redirects))) {
+					result.complete(new Response(target, value.statusCode(), value.headers(), value.body()));
+				}
 				return;
 			}
 			if (redirects == MAX_REDIRECTS) {
@@ -316,6 +369,72 @@ private HttpResponse.BodySubscriber<String> head(HttpResponse.ResponseInfo info,
 		return body;
 	}
 
+	/**
+	 * Sends the request again after the wait, if that leaves time for it.
+	 *
+	 * @return false if the request is not sent again.
+	 */
+	private boolean retryLater(long waitMillis, Runnable send) {
+		if (TimeUnit.MILLISECONDS.toNanos(waitMillis) >= remainingNanos()) {
+			return false;
+		}
+
+		synchronized (this) {
+			if (aborted) {
+				// Failed already.
+				return true;
+			}
+
+			try {
+				retry = retryTimer.schedule(send, waitMillis, TimeUnit.MILLISECONDS);
+			}
+			catch (RejectedExecutionException e) {
+				// The runtime is closing.
+				return false;
+			}
+
+			exchange = null;
+			body = null;
+		}
+
+		return true;
+	}
+
+	/**
+	 * Reads a Retry-After header field, in seconds or as an HTTP date.
+	 *
+	 * @param value     the field, or null.
+	 * @param nowMillis the time, as {@link System#currentTimeMillis()}.
+	 *
+	 * @return how long to wait in milliseconds, or -1 if the field says
+	 * nothing that can be read.
+	 */
+	static long retryDelayMillis(String value, long nowMillis) {
+		if (value == null) {
+			return -1;
+		}
+
+		String field = value.trim();
+
+		if (!field.isEmpty() && field.chars().allMatch(Character::isDigit)) {
+			try {
+				return Math.multiplyExact(Long.parseLong(field), 1000);
+			}
+			catch (NumberFormatException | ArithmeticException e) {
+				return Long.MAX_VALUE;
+			}
+		}
+
+		try {
+			long date = ZonedDateTime.parse(field, DateTimeFormatter.RFC_1123_DATE_TIME).toInstant().toEpochMilli();
+
+			return Math.max(0, date - nowMillis);
+		}
+		catch (DateTimeParseException e) {
+			return -1;
+		}
+	}
+
 	private boolean isRedirect(int status, String location) {
 		return followRedirects && (status == 307 || status == 308) && location != null;
 	}

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

@@ -21,6 +21,7 @@
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.RejectedExecutionException;
@@ -133,15 +134,18 @@ public void close() {
 	 * a server that does not answer holds no thread; what it creates for an
 	 * offer it had yet to answer is left for it to time out.
 	 *
+	 * @param retried the statuses of an answer to the offer, together with a
+	 *                Retry-After header field, after which the offer is sent
+	 *                again, as long as the wait fits within the timeout.
 	 * @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) {
+			String bearerToken, Duration timeout, Set<Integer> retried, 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)),
+		CompletableFuture<S> session = peer.toPublic(negotiate(rtc, peer, endpoint, bearerToken, timeout, retried,
+				abandoned).thenApply(resource -> factory.create(rtc, peer, resource, bearerToken, timeout)),
 				HttpSession::close);
 
 		// Once the session was handed out, this finds the negotiation finished.
@@ -160,7 +164,7 @@ static <S extends HttpSession> CompletableFuture<S> start(WebRtc rtc, PeerImpl p
 	 * @return a future with the resource the server named, or null.
 	 */
 	private static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken,
-			Duration timeout, CompletableFuture<?> abandoned) {
+			Duration timeout, Set<Integer> retried, 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.
@@ -169,7 +173,7 @@ private static CompletableFuture<URI> negotiate(WebRtc rtc, Peer peer, URI endpo
 		// Made now, so that it ends with the session's deadline at the latest,
 		// however long the offer takes.
 		HttpCall offerRequest = new HttpCall(rtc.httpClient(bearerToken != null), endpoint, "POST", bearerToken,
-				timeout).followRedirects();
+				timeout).followRedirects().retry(rtc.timer, retried);
 
 		// Either way the session ends, once: connected, failed or timed out.
 		Consumer<Throwable> fail = error -> {

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

@@ -49,6 +49,14 @@
  */
 public final class Whep {
 
+	/**
+	 * An endpoint under load answers 503, and one whose stream has yet to
+	 * start 409, both with Retry-After (draft-ietf-wish-whep-05, 4.7 and
+	 * 4.3.9).
+	 */
+	private static final Set<Integer> RETRIED = Set.of(409, 503);
+
+
 	private Whep() {
 	}
 
@@ -172,7 +180,7 @@ public CompletableFuture<Session> play(WebRtc rtc) {
 			PeerImpl peer = (PeerImpl) rtc.createPeer(options.build());
 
 			try {
-				return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new);
+				return HttpSession.start(rtc, peer, endpoint, token, limit, RETRIED, 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

@@ -22,6 +22,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
 import dev.onvoid.webrtc.internal.Timeouts;
@@ -47,6 +48,10 @@
  */
 public final class Whip {
 
+	/** An endpoint under load answers 503, with Retry-After (RFC 9725, 4.5). */
+	private static final Set<Integer> RETRIED = Set.of(503);
+
+
 	private Whip() {
 	}
 
@@ -181,7 +186,7 @@ public CompletableFuture<Session> publish(WebRtc rtc) {
 					peer.addTrack(track);
 				}
 
-				return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new);
+				return HttpSession.start(rtc, peer, endpoint, token, limit, RETRIED, 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/HttpCallTests.java

@@ -32,6 +32,9 @@
 import java.net.http.HttpClient;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
@@ -218,6 +221,24 @@ void tunnelRefusesBasicByDefault() throws Exception {
 		}
 	}
 
+	@Test
+	void retryAfterRead() {
+		long now = 1_000_000_000_000L;
+		String inFive = DateTimeFormatter.RFC_1123_DATE_TIME.format(
+				Instant.ofEpochMilli(now + 5000).atZone(ZoneOffset.UTC));
+		String past = DateTimeFormatter.RFC_1123_DATE_TIME.format(
+				Instant.ofEpochMilli(now - 5000).atZone(ZoneOffset.UTC));
+
+		assertEquals(3000, HttpCall.retryDelayMillis("3", now));
+		assertEquals(0, HttpCall.retryDelayMillis(" 0 ", now));
+		assertEquals(5000, HttpCall.retryDelayMillis(inFive, now));
+		assertEquals(0, HttpCall.retryDelayMillis(past, now));
+		assertEquals(Long.MAX_VALUE, HttpCall.retryDelayMillis("99999999999999999999", now));
+		assertEquals(-1, HttpCall.retryDelayMillis("soon", now));
+		assertEquals(-1, HttpCall.retryDelayMillis("", now));
+		assertEquals(-1, HttpCall.retryDelayMillis(null, now));
+	}
+
 	private static HttpCall post(HttpClient client, URI uri) {
 		return new HttpCall(client, uri, "POST", null, TIMEOUT);
 	}