Skip to content

test: make the client concurrency tests deterministic #411

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
devopvoid merged 1 commit into from
Oct 22, 2026
+1,091 −113
Merged

test: make the client concurrency tests deterministic #411

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 63 additions & 27 deletions webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,9 +128,11 @@ private WebRtc(Builder builder) {
this.closeTimeout = builder.closeTimeout;
this.errorHandler = builder.errorHandler != null ? builder.errorHandler : WebRtc::uncaught;

this.controlPool = Executors.newCachedThreadPool(threads("webrtc-client-control-" + runtime + "-"));
this.controlPool = builder.controlPool != null ? builder.controlPool
: Executors.newCachedThreadPool(threads("webrtc-client-control-" + runtime + "-"));
this.framePool = Executors.newCachedThreadPool(threads("webrtc-client-frames-" + runtime + "-"));
this.timer = Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-"));
this.timer = builder.timer != null ? builder.timer
: Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-"));

if (builder.eventExecutor != null) {
this.eventExecutor = builder.eventExecutor;
Expand Down Expand Up @@ -237,31 +239,7 @@ public Peer createPeer(PeerOptions options) {
int number = peerCount.incrementAndGet();
String name = options.name != null ? options.name : "peer-" + number;

PeerImpl peer = new PeerImpl(this, options, name);

// Registered before the runtime is checked again: a runtime that
// begins to close either finds the peer among those it closes and
// waits for it before freeing the factory, or is seen closing here.
peers.add(peer);

if (closing.get()) {
peers.remove(peer);
throw new PeerClosedException("The runtime is closed");
}

try {
peer.open();
}
catch (RuntimeException e) {
// Closed before the error goes out, so that nothing of the failed
// peer still runs when the runtime closes and frees WebRTC's threads.
// Closing removes the peer. Once the runtime has stopped its threads
// the close cannot run any more, and there is nothing left to wait for.
CompletableFuture.anyOf(peer.closeInternal(), closed).join();
throw e;
}

return peer;
return open(register(new PeerImpl(this, options, name)));
}

/**
Expand Down Expand Up @@ -335,6 +313,41 @@ public void close() {
}
}

/**
* Registers a peer before the runtime is checked again: a runtime that
* begins to close either finds the peer among those it closes and waits
* for it before freeing the factory, or is seen closing here.
*/
PeerImpl register(PeerImpl peer) {
peers.add(peer);

if (closing.get()) {
peers.remove(peer);
throw new PeerClosedException("The runtime is closed");
}

return peer;
}

/**
* Creates the connection of a registered peer.
*/
PeerImpl open(PeerImpl peer) {
try {
peer.open();
}
catch (RuntimeException e) {
// Closed before the error goes out, so that nothing of the failed
// peer still runs when the runtime closes and frees WebRTC's threads.
// Closing removes the peer. Once the runtime has stopped its threads
// the close cannot run any more, and there is nothing left to wait for.
CompletableFuture.anyOf(peer.closeInternal(), closed).join();
throw e;
}

return peer;
}

void reportError(Throwable error) {
try {
errorHandler.accept(error);
Expand Down Expand Up @@ -545,6 +558,10 @@ public static final class Builder {

private Consumer<PeerConnectionFactory.Builder> factoryCustomizer;

private ExecutorService controlPool;

private ScheduledExecutorService timer;


private Builder() {
}
Expand Down Expand Up @@ -661,6 +678,25 @@ public Builder configureFactory(Consumer<PeerConnectionFactory.Builder> customiz
return this;
}

/**
* Replaces the pool the control tasks run on, so that tests decide
* when each of them runs. The runtime shuts it down when it closes.
*/
Builder controlPool(ExecutorService pool) {
this.controlPool = requireNonNull(pool, "pool");
return this;
}

/**
* Replaces the timer of timeouts and periodic tasks, so that tests
* decide when each of them fires. The runtime shuts it down when it
* closes.
*/
Builder timer(ScheduledExecutorService timer) {
this.timer = requireNonNull(timer, "timer");
return this;
}

/**
* @return the runtime.
*
Expand Down
130 changes: 110 additions & 20 deletions webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,12 @@
import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.UnaryOperator;

import dev.onvoid.webrtc.RTCSdpType;
import dev.onvoid.webrtc.RTCSessionDescription;
Expand All @@ -44,6 +46,9 @@

class HttpSessionTests extends ClientTestBase {

/** Longer than any test takes: the session times out only when a test fires it. */
private static final Duration SESSION_TIMEOUT = Duration.ofSeconds(25);

private ScheduledExecutorService feeder;

private CustomVideo video;
Expand Down Expand Up @@ -136,17 +141,82 @@ void refusedOfferLeavesNothing() throws Exception {

@Test
void lateAnswerDeleted() throws Exception {
// The answer arrives after the session timed out. The server created
// a resource for it all the same, which must not be left there.
// The session times out while the offer is out. The server created a
// resource for it all the same, which must not be left there.
ManualTimer timer = new ManualTimer();
WebRtc client = clientRuntime(builder -> builder.timer(timer));

try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) {
endpoint.stallMillis = 700;
CountDownLatch answer = endpoint.holdAnswers();
CompletableFuture<Whep.Session> session = Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client);

ExecutionException failed = assertThrows(ExecutionException.class,
() -> Whep.from(endpoint.uri()).timeout(Duration.ofSeconds(1)).play(rtc)
.get(TIMEOUT_SECONDS, TimeUnit.SECONDS));
endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS);
assertEquals(1, timer.fire(SESSION_TIMEOUT));

// Failed before the answer arrives, which then meets a closed peer.
ExecutionException failed = assertThrows(ExecutionException.class, () -> await(session));
assertTrue(failed.getCause().getMessage().contains("within"), failed.getCause().getMessage());
awaitRequest(endpoint, "DELETE /live/session/1");

answer.countDown();
endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS);

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.
// Who owns the resource then is the timeout, not the request.
ManualExecutor control = new ManualExecutor();
ManualTimer timer = new ManualTimer();
WebRtc client = clientRuntime(builder -> builder.controlPool(control).timer(timer));

try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> answer(rtc.createPeer(PeerOptions.builder()
.negotiation(Negotiation.MANUAL)
.build()), offer), 201)) {
CompletableFuture<CompletableFuture<Whep.Session>> playing = CompletableFuture.supplyAsync(
() -> Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client));

// The client's side up to the answer: the peer is created, the
// offer is created and posted, and the answer read. The request
// runs on this thread, within the task that sends it.
control.runUntil(() -> endpoint.answered() > 0, TIMEOUT_SECONDS);
assertEquals(1, timer.fire(SESSION_TIMEOUT));
control.runAll();

ExecutionException failed = assertThrows(ExecutionException.class, () -> await(await(playing)));
assertTrue(failed.getCause().getMessage().contains("within"), failed.getCause().getMessage());
// Deleted on this thread, by the last tasks run.
assertEquals(1, endpoint.count("DELETE /live/session/1"));
}
finally {
closeStepping(client, control);
}
}

@Test
void closeAfterRuntimeReported() throws Exception {
// A runtime that closed has no threads left to delete the session on.
WebRtc client = clientRuntime(builder -> builder);

try (TestHttpEndpoint endpoint = sendingEndpoint()) {
Whep.Session session = await(Whep.from(endpoint.uri()).play(client));

client.close();
await(session.closeAsync());

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());
errors.clear();
}
finally {
client.close();
}
}

Expand All @@ -161,7 +231,7 @@ void oversizedAnswerRefused() throws Exception {

assertTrue(failed.getCause().getMessage().contains("larger than"), failed.getCause().getMessage());
// The resource was named before the answer was refused.
awaitRequest(endpoint, "DELETE /live/session/1");
endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS);
}
}

Expand Down Expand Up @@ -208,6 +278,38 @@ void resourceKeepsTls() {
() -> HttpSession.resourceOf(URI.create("http://media.example/whep"), "file:/etc/passwd"));
}

/**
* An endpoint whose side of the session sends the video.
*/
private TestHttpEndpoint sendingEndpoint() throws Exception {
return new TestHttpEndpoint(offer -> {
Peer server = rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL).build());
server.addTrack(video);
return answer(server, offer);
}, 201);
}

/**
* A runtime for the client's side of a session, apart from the one the
* endpoint's side runs on, so that its threads can be held or stopped.
*/
private WebRtc clientRuntime(UnaryOperator<WebRtc.Builder> options) {
return options.apply(WebRtc.builder()
.audio(AudioMode.HEADLESS)
.errorHandler(errors::add))
.build();
}

/**
* Closes a runtime whose control tasks run only when the test runs them.
*/
static void closeStepping(WebRtc runtime, ManualExecutor control) throws Exception {
CompletableFuture<Void> closed = runtime.closeAsync();

control.runUntil(closed::isDone, TIMEOUT_SECONDS);
await(closed);
}

/**
* An endpoint whose side of the session receives what is published.
*/
Expand All @@ -227,18 +329,6 @@ private void assertDeleteReported(String status) {
errors.clear();
}

private static void awaitRequest(TestHttpEndpoint endpoint, String request) throws InterruptedException {
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS);

while (endpoint.requests.stream().noneMatch(r -> r.toString().equals(request))
&& System.nanoTime() < deadline) {
Thread.sleep(50);
}

assertTrue(endpoint.requests.stream().anyMatch(r -> r.toString().equals(request)),
() -> "Requests: " + endpoint.requests);
}

private static String answer(Peer peer, String offer) {
try {
return peer.negotiator()
Expand Down
Loading
Loading