fix: complete futures whose tasks a SerialExecutor drops

docs/guide/client-api.md

@@ -202,6 +202,8 @@ Every method of the client API may be called from any thread, including from ins
 
 The handlers of one peer never run at the same time, and they run in the order of the events. The handlers of different peers may run at the same time. With the default executor, the event thread is named after its peer, such as `webrtc-client-events-alice`, which shows up in thread dumps.
 
+An executor of your own that rejects work, such as a bounded pool that is full, loses the events it rejects, and the rejection goes to the error handler. The futures those events would have completed complete all the same, on the thread that found them rejected. Once the runtime is closed, every future of this API that is not done fails with `PeerClosedException`, including those of calls made after the close.
+
 ### Handlers on the UI Thread
 
 Pass the UI toolkit's executor, and every handler may touch UI components directly:

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

@@ -95,6 +95,8 @@ final class DataChannelImpl implements DataChannel {
 		this.label = label;
 		this.messageHandlers = new Handlers<>(peer.rtc::reportError);
 		this.stateHandlers = new Handlers<>(peer.rtc::reportError);
+
+		peer.rtc.track(opened, peer.events);
 	}
 
 	/**
@@ -255,7 +257,8 @@ void closeFromPeer(WebRtcException error) {
 		state = ChannelState.CLOSED;
 		peer.channelClosed(this);
 
-		peer.events.execute(() -> deliverState(ChannelState.CLOSED, error));
+		peer.events.execute(() -> deliverState(ChannelState.CLOSED, error),
+				() -> settleOpened(ChannelState.CLOSED, error));
 	}
 
 	private CompletableFuture<Void> enqueue(byte[] data, boolean binary) {
@@ -351,7 +354,7 @@ else if (newState == ChannelState.CLOSING) {
 			failPending(new PeerClosedException("The data channel " + label + " is closing"));
 		}
 
-		peer.events.execute(() -> deliverState(newState, null));
+		peer.events.execute(() -> deliverState(newState, null), () -> settleOpened(newState, null));
 	}
 
 	private void bufferChanged() {
@@ -362,21 +365,30 @@ private void bufferChanged() {
 
 	// On the event executor.
 
+	/**
+	 * Completes the opened future for the state, which is all that is left of
+	 * delivering it when the event executor rejects that.
+	 */
+	private void settleOpened(ChannelState newState, WebRtcException error) {
+		if (newState == ChannelState.OPEN) {
+			opened.complete(this);
+		}
+		else if (newState == ChannelState.CLOSED) {
+			opened.completeExceptionally(error != null ? error
+					: new PeerClosedException("The data channel " + label + " is closed"));
+		}
+	}
+
 	private void deliverState(ChannelState newState, WebRtcException error) {
 		if (newState == deliveredState || deliveredState == ChannelState.CLOSED) {
 			return;
 		}
 
 		deliveredState = newState;
 		stateHandlers.accept(newState);
+		settleOpened(newState, error);
 
-		if (newState == ChannelState.OPEN) {
-			opened.complete(this);
-		}
-		else if (newState == ChannelState.CLOSED) {
-			opened.completeExceptionally(error != null ? error
-					: new PeerClosedException("The data channel " + label + " is closed"));
-
+		if (newState == ChannelState.CLOSED) {
 			messageHandlers.clear();
 			stateHandlers.clear();
 			early.clear();

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

@@ -227,6 +227,7 @@ private static List<DesktopSource> desktopSources(DesktopCapturer capturer) {
 
 	private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
 		CompletableFuture<T> result = new CompletableFuture<>();
+		media.rtc.track(result, null);
 
 		media.executor.execute(() -> {
 			try {

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

@@ -54,6 +54,7 @@ public CompletableFuture<Void> switchCamera(VideoDevice device) {
 		requireNonNull(device, "device");
 
 		CompletableFuture<Void> switched = new CompletableFuture<>();
+		media.rtc.track(switched, null);
 
 		// Opening a camera can take a moment, so it happens off the caller's thread.
 		media.executor.execute(() -> {

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

@@ -178,6 +178,9 @@ final class PeerImpl implements Peer {
 		this.trackHandlers = new Handlers<>(rtc::reportError);
 		this.channelHandlers = new Handlers<>(rtc::reportError);
 		this.errorHandlers = new Handlers<>(rtc::reportError);
+
+		rtc.track(connected, events);
+		rtc.track(closedFuture, events);
 	}
 
 	/**
@@ -224,13 +227,12 @@ void open() {
 
 				opened.completeExceptionally(e);
 			}
+		}, () -> {
+			// A runtime that stopped its threads dropped the task.
+			opened.completeExceptionally(new PeerClosedException("The runtime closed before " + name
+					+ " was created"));
 		});
 
-		// 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();
 		}
@@ -556,7 +558,12 @@ boolean isClosing() {
 	<T> CompletableFuture<T> toPublic(CompletableFuture<T> internal) {
 		ClientFuture<T> result = new ClientFuture<>();
 
-		internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error)));
+		rtc.track(result, events);
+
+		// Completed in any case: should the event executor reject the task, on
+		// the thread that found it rejected.
+		internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error),
+				() -> complete(result, value, error)));
 
 		return result;
 	}
@@ -977,7 +984,7 @@ private void connectionChanged(RTCPeerConnectionState nativeState) {
 		PeerState newState = PeerState.of(nativeState);
 		state = newState;
 
-		events.execute(() -> deliverState(newState));
+		events.execute(() -> deliverState(newState), () -> settleConnected(newState));
 	}
 
 	private void gatheringChanged(RTCIceGatheringState gatheringState) {
@@ -1137,6 +1144,9 @@ private void teardown() {
 			channelHandlers.clear();
 			stateHandlers.clear();
 			errorHandlers.clear();
+		}, () -> {
+			connected.completeExceptionally(error);
+			closedFuture.complete(null);
 		});
 	}
 
@@ -1149,7 +1159,14 @@ private void deliverState(PeerState newState) {
 
 		deliveredState = newState;
 		stateHandlers.accept(newState);
+		settleConnected(newState);
+	}
 
+	/**
+	 * Completes the connected future for the state, which is all that is left
+	 * of delivering it when the event executor rejects that.
+	 */
+	private void settleConnected(PeerState newState) {
 		if (newState == PeerState.CONNECTED) {
 			connected.complete(this);
 		}

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

@@ -17,7 +17,10 @@
 package dev.onvoid.webrtc.client;
 
 import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.function.Consumer;
 
@@ -28,6 +31,11 @@
  * <p>
  * A task that throws does not stop the ones after it; its exception goes to
  * the error handler.
+ * <p>
+ * When the executor rejects the tasks, they are dropped, and those submitted
+ * with a fallback have it run instead, on the thread that found them
+ * rejected. A rejection by an executor that was not shut down goes to the
+ * error handler as well, since it loses work the application expects.
  */
 final class SerialExecutor implements Executor {
 
@@ -55,8 +63,6 @@ protected int[] initialValue() {
 
 	private boolean scheduled;
 
-	private boolean shutdown;
-
 	private volatile Thread runner;
 
 
@@ -84,13 +90,17 @@ static boolean onEventThread() {
 		return EVENT_DEPTH.get()[0] > 0;
 	}
 
+	/**
+	 * Runs the task, or the fallback should the executor reject it, which
+	 * keeps a future the task would complete from staying open forever.
+	 */
+	void execute(Runnable task, Runnable dropped) {
+		execute(new Droppable(task, dropped));
+	}
+
 	@Override
 	public void execute(Runnable task) {
 		synchronized (tasks) {
-			if (shutdown) {
-				return;
-			}
-
 			tasks.add(task);
 
 			if (scheduled) {
@@ -110,29 +120,37 @@ boolean inExecutor() {
 		return runner == Thread.currentThread();
 	}
 
-	/**
-	 * Drops the tasks not yet run and refuses new ones. A task running now
-	 * finishes.
-	 */
-	void shutdown() {
-		synchronized (tasks) {
-			shutdown = true;
-			tasks.clear();
-		}
-	}
-
 	private void schedule() {
 		try {
 			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.
+			List<Runnable> dropped;
+
 			synchronized (tasks) {
 				scheduled = false;
+				dropped = new ArrayList<>(tasks);
 				tasks.clear();
 			}
+
+			// The runtime's own pools reject only once it has stopped them,
+			// and then nothing is lost that it does not fail itself.
+			if (!(executor instanceof ExecutorService && ((ExecutorService) executor).isShutdown())) {
+				errorHandler.accept(new WebRtcException("The executor rejected " + dropped.size()
+						+ " tasks, which were dropped", e));
+			}
+
+			// In order, as they would have run.
+			for (Runnable task : dropped) {
+				if (task instanceof Droppable) {
+					try {
+						((Droppable) task).dropped.run();
+					}
+					catch (Throwable error) {
+						errorHandler.accept(error);
+					}
+				}
+			}
 		}
 	}
 
@@ -190,4 +208,24 @@ private void drain() {
 			schedule();
 		}
 	}
+
+
+
+	private static final class Droppable implements Runnable {
+
+		final Runnable task;
+
+		final Runnable dropped;
+
+
+		Droppable(Runnable task, Runnable dropped) {
+			this.task = task;
+			this.dropped = dropped;
+		}
+
+		@Override
+		public void run() {
+			task.run();
+		}
+	}
 }

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

@@ -111,6 +111,12 @@ public final class WebRtc implements AutoCloseable {
 
 	private final CompletableFuture<Void> closed = new CompletableFuture<>();
 
+	/** The futures tracked that are not done, with what fails each. */
+	private final Map<CompletableFuture<?>, Runnable> pending = new ConcurrentHashMap<>();
+
+	/** Set once the control tasks have run, after which none ever runs. */
+	private volatile boolean stopped;
+
 
 	private WebRtc(Builder builder) {
 		int runtime = RUNTIME_COUNT.incrementAndGet();
@@ -342,11 +348,25 @@ boolean isClosing() {
 	}
 
 	/**
-	 * @return a future that completes once the runtime has closed and stopped
-	 * its threads, after which no task given to them runs.
+	 * Fails the future, unless it is done by then, once the runtime has
+	 * stopped the threads that could complete it. A task that would have
+	 * completed it is dropped then, and the future would stay open forever.
+	 *
+	 * @param future the future handed out or waited on.
+	 * @param events the event executor the future completes on, so that the
+	 *               failure comes after what is queued there; or null.
 	 */
-	CompletableFuture<Void> whenShutDown() {
-		return closed;
+	void track(CompletableFuture<?> future, SerialExecutor events) {
+		Runnable fail = () -> future.completeExceptionally(
+				new PeerClosedException("The runtime closed before this completed"));
+		Runnable orphan = events == null ? fail : () -> events.execute(fail, fail);
+
+		pending.put(future, orphan);
+		future.whenComplete((value, error) -> pending.remove(future));
+
+		if (stopped) {
+			orphan.run();
+		}
 	}
 
 	private void shutdown() {
@@ -393,10 +413,32 @@ private void shutdown() {
 		}
 		finally {
 			shutdownPools();
+			failPending();
 			closed.complete(null);
 		}
 	}
 
+	/**
+	 * Fails the futures that nothing completes any more, once the control
+	 * tasks still queued have run.
+	 */
+	private void failPending() {
+		try {
+			controlPool.awaitTermination(closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+
+		// Set before the futures are listed: one tracked meanwhile either is
+		// listed or sees it set.
+		stopped = true;
+
+		for (Runnable orphan : new ArrayList<>(pending.values())) {
+			orphan.run();
+		}
+	}
+
 	/**
 	 * Closes what the supplier names, again and again, until it names nothing.
 	 *
@@ -526,6 +568,11 @@ public Builder fieldTrials(Map<String, String> fieldTrials) {
 		 * SwingUtilities::invokeLater} or {@code Platform::runLater}. The
 		 * order of each peer's events is kept whatever the executor. By
 		 * default the runtime runs them on threads of its own.
+		 * <p>
+		 * An executor that rejects work loses the events it rejects, which
+		 * goes to the error handler. The futures those events would have
+		 * completed complete all the same, on the thread that found them
+		 * rejected.
 		 *
 		 * @param executor runs handlers.
 		 *