From 5c59a0303a34c5a4def0386ab4e8a4290752ab8b Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Tue, 27 Oct 2026 08:09:37 +0100 Subject: [PATCH] feat: retry WHIP/WHEP offers after Retry-After A 503 answer, and for WHEP a 409 one, with Retry-After sends the offer again after that wait, at least a second, as long as it fits within the session's timeout. Cancelling the session ends the wait. --- docs/guide/client-api.md | 2 +- .../dev/onvoid/webrtc/client/HttpCall.java | 123 +++++++++++++++++- .../dev/onvoid/webrtc/client/HttpSession.java | 14 +- .../java/dev/onvoid/webrtc/client/Whep.java | 10 +- .../java/dev/onvoid/webrtc/client/Whip.java | 7 +- .../onvoid/webrtc/client/HttpCallTests.java | 21 +++ .../webrtc/client/HttpSessionTests.java | 114 ++++++++++++++++ .../webrtc/client/TestHttpEndpoint.java | 37 ++++++ 8 files changed, 318 insertions(+), 10 deletions(-) diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index eb6be5ad..eece8452 100644 --- a/docs/guide/client-api.md +++ b/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. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpCall.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpCall.java index 9b51d5fd..4eea4d56 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpCall.java +++ b/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. *

* 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 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 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 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; } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java index d4661f51..3ee9dd6e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java +++ b/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 CompletableFuture start(WebRtc rtc, PeerImpl peer, URI endpoint, - String bearerToken, Duration timeout, Factory factory) { + String bearerToken, Duration timeout, Set retried, Factory factory) { CompletableFuture abandoned = new CompletableFuture<>(); - CompletableFuture session = peer.toPublic(negotiate(rtc, peer, endpoint, bearerToken, timeout, abandoned) - .thenApply(resource -> factory.create(rtc, peer, resource, bearerToken, timeout)), + CompletableFuture 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 CompletableFuture start(WebRtc rtc, PeerImpl p * @return a future with the resource the server named, or null. */ private static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken, - Duration timeout, CompletableFuture abandoned) { + Duration timeout, Set retried, CompletableFuture abandoned) { CompletableFuture 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 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 fail = error -> { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java index 59761bee..3f823f7f 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java +++ b/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 RETRIED = Set.of(409, 503); + + private Whep() { } @@ -172,7 +180,7 @@ public CompletableFuture 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. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java index 53bd11a3..9668ce10 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java +++ b/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 RETRIED = Set.of(503); + + private Whip() { } @@ -181,7 +186,7 @@ public CompletableFuture 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. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpCallTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpCallTests.java index 742fb6bc..3e47fcb5 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpCallTests.java +++ b/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); } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java index caf05c78..50647fe8 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java @@ -235,6 +235,104 @@ void tricklingAnswerTimesOut() throws Exception { } } + @Test + void conflictRetriedForWhep() throws Exception { + // The stream has yet to start, and the endpoint says to come back at + // once. The offer goes again after the shortest wait, on the timer. + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.timer(timer)); + Duration wait = Duration.ofMillis(HttpCall.MIN_RETRY_MILLIS); + + try (TestHttpEndpoint endpoint = sendingEndpoint()) { + endpoint.refuseAnswers(1, 409, "0"); + CompletableFuture session = Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client); + + awaitScheduled(timer, wait); + assertEquals(1, endpoint.count("POST /live/endpoint")); + assertEquals(1, timer.fire(wait)); + + await(await(session).closeAsync()); + assertEquals(2, endpoint.count("POST /live/endpoint")); + } + finally { + client.close(); + } + } + + @Test + void unavailableRetriedForWhip() throws Exception { + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.timer(timer)); + Duration wait = Duration.ofSeconds(2); + + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + endpoint.refuseAnswers(1, 503, "2"); + CompletableFuture session = Whip.to(endpoint.uri()) + .track(client.media().customVideo("video")) + .timeout(SESSION_TIMEOUT) + .publish(client); + + awaitScheduled(timer, wait); + assertEquals(1, timer.fire(wait)); + + await(await(session).closeAsync()); + assertEquals(2, endpoint.count("POST /live/endpoint")); + } + finally { + client.close(); + } + } + + @Test + void conflictFailsWhip() throws Exception { + // 409 means nothing particular to WHIP, so it is no reason to retry. + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + endpoint.refuseAnswers(1, 409, "1"); + + ExecutionException failed = assertThrows(ExecutionException.class, + () -> await(Whip.to(endpoint.uri()).track(video).publish(rtc))); + + assertTrue(failed.getCause().getMessage().contains("409"), failed.getCause().getMessage()); + assertEquals(1, endpoint.count("POST /live/endpoint")); + } + } + + @Test + void retryBeyondTimeoutFails() throws Exception { + // Coming back in a minute would miss the session's timeout. + try (TestHttpEndpoint endpoint = sendingEndpoint()) { + endpoint.refuseAnswers(1, 503, "60"); + + ExecutionException failed = assertThrows(ExecutionException.class, + () -> await(Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(rtc))); + + assertTrue(failed.getCause().getMessage().contains("503"), failed.getCause().getMessage()); + assertEquals(1, endpoint.count("POST /live/endpoint")); + } + } + + @Test + void cancelEndsRetryWait() throws Exception { + // Cancelled while it waits to send the offer again: it never does. + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.timer(timer)); + Duration wait = Duration.ofSeconds(2); + + try (TestHttpEndpoint endpoint = sendingEndpoint()) { + endpoint.refuseAnswers(1, 503, "2"); + CompletableFuture session = Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client); + + awaitScheduled(timer, wait); + assertTrue(session.cancel(true)); + + assertEquals(0, timer.fire(wait)); + assertEquals(1, endpoint.count("POST /live/endpoint")); + } + finally { + client.close(); + } + } + @Test void redirectFollowed() throws Exception { // The endpoint sends the offer on to another, as WHIP and WHEP allow, @@ -623,6 +721,22 @@ private WebRtc clientRuntime(UnaryOperator options) { .build(); } + /** + * Waits until a task with the delay is scheduled on the timer, which an + * answer read on a thread of the HTTP client does. + */ + private static void awaitScheduled(ManualTimer timer, Duration delay) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); + + while (timer.scheduled(delay) == 0) { + if (System.nanoTime() > deadline) { + throw new AssertionError("Nothing was scheduled in " + delay + " within " + TIMEOUT_SECONDS + " s"); + } + + Thread.sleep(10); + } + } + /** * Starts a session and cancels it while the server holds the answer, * which must cut off the request. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java index eaaf9c9f..1e05a78c 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java @@ -88,6 +88,14 @@ public String toString() { /** Whether the head of the answer to a DELETE trickles in. */ private volatile boolean trickleDeletes; + /** How many POSTs are still refused with {@link #refusalStatus}. */ + private int refusals; + + private volatile int refusalStatus; + + /** The Retry-After of a refusal, or null for none. */ + private volatile String refusalRetryAfter; + /** Where POSTs are redirected to, or null. */ private volatile String redirectLocation; @@ -137,6 +145,19 @@ URI uri() { return URI.create(scheme + "://127.0.0.1:" + server.getLocalPort() + "/live/endpoint"); } + /** + * Refuses the next POSTs, then answers as before. + * + * @param times how many POSTs to refuse. + * @param status the status, such as 503. + * @param retryAfter the Retry-After to send along, or null for none. + */ + synchronized void refuseAnswers(int times, int status, String retryAfter) { + refusals = times; + refusalStatus = status; + refusalRetryAfter = retryAfter; + } + /** * Answers every POST from now on with a redirect. * @@ -307,6 +328,22 @@ else if (name.equals("authorization")) { OutputStream out = socket.getOutputStream(); if (requestLine[0].equals("POST")) { + boolean refused; + + synchronized (this) { + refused = refusals > 0; + + if (refused) { + refusals--; + } + } + + if (refused) { + String retryAfter = refusalRetryAfter; + respond(out, refusalStatus + " Refused", retryAfter != null ? "Retry-After: " + retryAfter + "\r\n" : "", + "come back later"); + return; + } if (redirectLocation != null) { respond(out, redirectStatus + " Redirect", "Location: " + redirectLocation + "\r\n", ""); return;