fix: close races in MediaReader.getInfo and WebRtc.createPeer

webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/player/MediaReader.java

@@ -111,7 +111,9 @@ public MediaReader(String source, Duration timeout) throws IOException {
 	 *
 	 * @throws IllegalStateException if this reader has been closed.
 	 */
-	public MediaInfo getInfo() {
+	public synchronized MediaInfo getInfo() {
+		// Synchronized like close, which may run on any thread: otherwise the
+		// handle read here could be freed while the native side still uses it.
 		return info(handle);
 	}
 

webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/player/MediaReaderTest.java

@@ -25,6 +25,9 @@
 import java.net.URL;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -130,6 +133,33 @@ void infoAfterCloseFails() throws Exception {
 		assertThrows(IllegalStateException.class, reader::getInfo);
 	}
 
+	@Test
+	void infoRacingClose() throws Exception {
+		// Close may run on any thread, so a getInfo on another one either gets
+		// the info or finds the reader closed, never a reader being freed.
+		for (int round = 0; round < 200; round++) {
+			MediaReader reader = new MediaReader(asset());
+			CountDownLatch started = new CountDownLatch(1);
+
+			CompletableFuture<Void> reading = CompletableFuture.runAsync(() -> {
+				started.countDown();
+
+				try {
+					while (true) {
+						assertNotNull(reader.getInfo());
+					}
+				}
+				catch (IllegalStateException e) {
+					// Closed, which is how the loop ends.
+				}
+			});
+
+			started.await();
+			reader.close();
+			reading.get(10, TimeUnit.SECONDS);
+		}
+	}
+
 	private static Path asset() throws Exception {
 		return asset(ASSET);
 	}

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

@@ -188,6 +188,13 @@ void open() {
 		CompletableFuture<Void> opened = new CompletableFuture<>();
 
 		control.execute(() -> {
+			if (closed) {
+				// The runtime closed the peer before this ran, and may already
+				// be freeing the factory.
+				opened.completeExceptionally(closedError());
+				return;
+			}
+
 			try {
 				connection = rtc.factory.createPeerConnection(options.toConfiguration(), new Observer());
 
@@ -219,6 +226,11 @@ void open() {
 			}
 		});
 
+		// A runtime that stops its threads drops the task if it has not run,
+		// and then nothing else would end the wait.
+		rtc.whenShutDown().thenRun(() -> opened.completeExceptionally(
+				new PeerClosedException("The runtime closed before " + name + " was created")));
+
 		try {
 			opened.get();
 		}

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

@@ -126,6 +126,9 @@ private void schedule() {
 			executor.execute(this::drain);
 		}
 		catch (RejectedExecutionException e) {
+			// The pool rejects only once the runtime has stopped it. A caller
+			// that waits on a dropped task ends its wait with the runtime's
+			// end, as PeerImpl.open does.
 			synchronized (tasks) {
 				scheduled = false;
 				tasks.clear();

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

@@ -230,24 +230,29 @@ public Peer createPeer(PeerOptions options) {
 		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.
-			peers.remove(peer);
-			peer.closeInternal().join();
+			// 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;
 		}
 
-		if (closing.get()) {
-			// The runtime began to close meanwhile and may have missed it.
-			peer.closeInternal();
-		}
-
 		return peer;
 	}
 
@@ -336,6 +341,14 @@ boolean isClosing() {
 		return closing.get();
 	}
 
+	/**
+	 * @return a future that completes once the runtime has closed and stopped
+	 * its threads, after which no task given to them runs.
+	 */
+	CompletableFuture<Void> whenShutDown() {
+		return closed;
+	}
+
 	private void shutdown() {
 		try {
 			// Peers and tracks created while the runtime closes are closed as

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

@@ -20,6 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.time.Duration;
@@ -372,6 +373,70 @@ void runtimeCloseClosesPeers() throws Exception {
 		assertThrows(PeerClosedException.class, this::pair);
 	}
 
+	@Test
+	void openAfterShutdownFails() {
+		// A createPeer that passed its check just before the runtime closed:
+		// the runtime's threads are gone and drop the task that would create
+		// the connection, so opening has to fail rather than wait forever.
+		PeerImpl peer = new PeerImpl(rtc, manualOptions(), "late");
+
+		rtc.close();
+
+		assertThrows(PeerClosedException.class, () -> assertTimeoutPreemptively(
+				Duration.ofSeconds(TIMEOUT_SECONDS), peer::open));
+	}
+
+	@Test
+	void openAfterCloseFails() throws Exception {
+		// The closing runtime found the peer before its connection was created.
+		// It may free the factory once the peer is closed, so the connection
+		// must not be created after that.
+		PeerImpl peer = new PeerImpl(rtc, manualOptions(), "closed-first");
+
+		await(peer.closeInternal());
+
+		assertThrows(PeerClosedException.class, peer::open);
+	}
+
+	@Test
+	void createPeerRacingClose() throws Exception {
+		// A peer created while its runtime closes is either refused or closed
+		// with the runtime, and creating it never waits forever.
+		for (int round = 0; round < 50; round++) {
+			WebRtc runtime = WebRtc.builder()
+					.audio(AudioMode.HEADLESS)
+					.errorHandler(errors::add)
+					.build();
+			CountDownLatch started = new CountDownLatch(1);
+
+			CompletableFuture<List<Peer>> created = CompletableFuture.supplyAsync(() -> {
+				List<Peer> peers = new ArrayList<>();
+
+				try {
+					while (true) {
+						peers.add(runtime.createPeer(manualOptions()));
+						started.countDown();
+					}
+				}
+				catch (PeerClosedException e) {
+					return peers;
+				}
+			});
+
+			// Closed while peers are being created, not before the first.
+			started.await();
+			runtime.closeAsync().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+			for (Peer peer : created.get(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+				assertEquals(PeerState.CLOSED, peer.state());
+			}
+		}
+	}
+
+	private static PeerOptions manualOptions() {
+		return PeerOptions.builder().negotiation(Negotiation.MANUAL).build();
+	}
+
 	@Test
 	void statsAfterConnecting() throws Exception {
 		Peer[] peers = pair();