fix: bound received data channel messages before they are queued

design/client-api.md

@@ -774,8 +774,9 @@ Where the code differs from the API above:
   `CustomAudio.queuedMillis()`/`clear()`, `Message.size()`, `Negotiator.answer(offer, gathering)`.
 - An impolite peer that ignored an offer reports the two-impolite-peers collision after 5 seconds
   without an answer.
-- Messages a data channel keeps while it has no message handler are capped at 16 MiB; past that the
-  oldest are dropped and the peer's `onError` handlers hear of it once.
+- Received messages a data channel holds, waiting for its handlers or kept while it has none, are
+  capped by `PeerOptions.Builder.channelReceiveLimit` (16 MiB by default), counted before they are
+  queued; past that the channel closes and the peer's `onError` handlers hear why.
 - `Gathering.COMPLETE` waits at most `PeerOptions.Builder.gatheringTimeout` (3 seconds by default),
   since an unreachable STUN or TURN server holds up the end of gathering far longer.
 

docs/guide/client-api.md

@@ -103,6 +103,7 @@ Peer peer = rtc.createPeer(PeerOptions.builder()
 | `receive(MediaKind...)` | Receives audio or video without sending any, as a viewer does. |
 | `gatheringTimeout(Duration)` | How long `Gathering.COMPLETE` waits for ICE gathering, 3 seconds by default. See [Manual Negotiation](#manual-negotiation). |
 | `channelSendLimit(long)` | How many bytes each data channel may hold waiting to be sent, 16 MiB by default. See [Data Channels](#data-channels). |
+| `channelReceiveLimit(long)` | How many bytes of received messages each data channel may hold that its handlers are not done with, 16 MiB by default. See [Data Channels](#data-channels). |
 | `configure(Consumer<RTCConfiguration>)` | Anything else the [peer connection configuration](/guide/networking/peer-connection-config) has. |
 
 ### Signaling
@@ -496,7 +497,7 @@ peer.onChannel(channel -> {                  // channels the remote peer opens
 ```
 
 - A channel may be used right away: messages sent before it opens are held until it does.
-- Messages that arrive while a channel has no `onMessage` handler are kept for the next one, so none are lost while a handler is being set up. Up to 16 MiB are kept; beyond that the oldest are dropped, and the peer's `onError` handlers are told.
+- Messages that arrive while a channel has no `onMessage` handler are kept for the next one, so none are lost while a handler is being set up.
 - `send` takes a `String`, a `byte[]` or a `ByteBuffer`, and copies the data before it returns.
 - A `Message` holds a copy of the data. `isText()` tells text from binary, `text()` decodes it as UTF-8, and `bytes()` returns a copy.
 
@@ -510,6 +511,8 @@ for (byte[] chunk : chunks) {
 
 A sender that does not wait still cannot fill Java memory: each channel holds at most 16 MiB waiting to be sent, and a `send` beyond that fails at once with a `ChannelFullException`. `PeerOptions.builder().channelSendLimit(bytes)` changes the limit for every channel of a peer. `bufferedAmount()` tells how much a channel holds.
 
+Received messages are bounded as well. WebRTC cannot be told to stop receiving, so messages that arrive faster than the handlers take them, or while a channel has no handler, wait in Java. Each channel lets at most 16 MiB of them wait. When more arrive, the channel closes instead of dropping messages unnoticed: the peer's `onError` handlers hear why, the messages after the one over the limit are not delivered, and both sides see the channel close. `PeerOptions.builder().channelReceiveLimit(bytes)` changes the limit. A handler that takes long should hand the work to a thread of its own.
+
 `ChannelOptions` set how a channel delivers:
 
 ```java

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

@@ -89,9 +89,12 @@ public interface DataChannel extends AutoCloseable {
 
 	/**
 	 * Calls the handler for each message received. Messages that arrive while
-	 * the channel has no handler are kept for the next one, up to 16 MiB;
-	 * beyond that the oldest are dropped, which is reported to the peer's
-	 * {@link Peer#onError} handlers.
+	 * the channel has no handler are kept for the next one.
+	 * <p>
+	 * Received messages that wait for the handlers, or are kept for one, are
+	 * bounded by {@link PeerOptions.Builder#channelReceiveLimit(long)}. When
+	 * they would exceed it, the channel closes, which the peer's {@link
+	 * Peer#onError} handlers hear of, rather than drop messages unnoticed.
 	 *
 	 * @param handler called on the peer's event thread.
 	 *

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

@@ -24,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
@@ -39,15 +40,18 @@
  * while the channel is not open or holds more than {@link #HIGH_WATER_MARK}
  * bytes. What waits is bounded by the peer's channel send limit, beyond which
  * sends fail at once. Handlers run on the peer's event executor.
+ * <p>
+ * Received messages are counted as they arrive, before they are queued for
+ * the handlers, and until the handlers are done with them. When they reach
+ * the peer's channel receive limit, the channel closes: WebRTC cannot be told
+ * to stop receiving, and dropping messages of a reliable channel would lose
+ * them unnoticed.
  */
 final class DataChannelImpl implements DataChannel {
 
 	/** Bytes the native channel may hold before sends wait in Java. */
 	static final long HIGH_WATER_MARK = 1024 * 1024;
 
-	/** Bytes of received messages kept while there is no message handler. */
-	static final long EARLY_LIMIT = 16 * 1024 * 1024;
-
 	private final PeerImpl peer;
 
 	private final String label;
@@ -58,6 +62,15 @@ final class DataChannelImpl implements DataChannel {
 	/** Bytes sent and not yet handed to the transport. */
 	private final AtomicLong held = new AtomicLong();
 
+	/** The most bytes {@link #received} may reach. */
+	private final long receiveLimit;
+
+	/** Bytes received that the handlers are not done with, or that wait for one. */
+	private final AtomicLong received = new AtomicLong();
+
+	/** Set once received messages went over the limit, after which none is queued. */
+	private final AtomicBoolean overflowed = new AtomicBoolean();
+
 	private volatile ChannelState state = ChannelState.CONNECTING;
 
 	/** What the native channel holds, as last read on the control executor. */
@@ -81,8 +94,6 @@ final class DataChannelImpl implements DataChannel {
 
 	private long earlyBytes;
 
-	private boolean earlyDropReported;
-
 	private final Handlers<Message> messageHandlers;
 
 	private final Handlers<ChannelState> stateHandlers;
@@ -98,6 +109,7 @@ final class DataChannelImpl implements DataChannel {
 		this.peer = peer;
 		this.label = label;
 		this.sendLimit = peer.channelSendLimit();
+		this.receiveLimit = peer.channelReceiveLimit();
 		this.messageHandlers = new Handlers<>(peer.rtc::reportError);
 		this.stateHandlers = new Handlers<>(peer.rtc::reportError);
 
@@ -186,12 +198,15 @@ public Subscription onMessage(Consumer<Message> handler) {
 
 			if (!early.isEmpty()) {
 				List<Message> kept = new ArrayList<>(early);
+				long keptBytes = earlyBytes;
 				early.clear();
 				earlyBytes = 0;
 
 				for (Message message : kept) {
 					peer.call(handler, message);
 				}
+
+				received.addAndGet(-keptBytes);
 			}
 
 			return subscription;
@@ -376,6 +391,19 @@ private void bufferChanged() {
 		}
 	}
 
+	/** Closes a channel whose received messages went over the limit. */
+	private void receiveLimitExceeded() {
+		if (released) {
+			return;
+		}
+
+		WebRtcException error = new WebRtcException("The data channel " + label + " closed: its message "
+				+ "handlers fell more than " + receiveLimit + " bytes of received messages behind");
+
+		peer.deliverError(error);
+		closeFromPeer(error);
+	}
+
 	// On the event executor.
 
 	/**
@@ -405,32 +433,29 @@ private void deliverState(ChannelState newState, WebRtcException error) {
 			messageHandlers.clear();
 			stateHandlers.clear();
 			early.clear();
+			received.addAndGet(-earlyBytes);
 			earlyBytes = 0;
 		}
 	}
 
 	private void deliverMessage(Message message) {
 		if (deliveredState == ChannelState.CLOSED) {
+			received.addAndGet(-message.size());
 			return;
 		}
 
 		if (messageHandlers.isEmpty()) {
+			// Kept for the next handler, and counted until it is done with them.
 			early.add(message);
 			earlyBytes += message.size();
-
-			// Kept for a handler that may never come, so their memory is bounded.
-			while (earlyBytes > EARLY_LIMIT && early.size() > 1) {
-				earlyBytes -= early.poll().size();
-
-				if (!earlyDropReported) {
-					earlyDropReported = true;
-					peer.deliverError(new WebRtcException("The data channel " + label + " dropped received messages: "
-							+ "more than " + (EARLY_LIMIT >> 20) + " MiB arrived while it had no message handler"));
-				}
-			}
 		}
 		else {
-			messageHandlers.accept(message);
+			try {
+				messageHandlers.accept(message);
+			}
+			finally {
+				received.addAndGet(-message.size());
+			}
 		}
 	}
 
@@ -465,12 +490,30 @@ public void onStateChange() {
 		@Override
 		public void onMessage(RTCDataChannelBuffer buffer) {
 			// The buffer is valid only during this call.
-			byte[] data = new byte[buffer.data.remaining()];
+			if (overflowed.get()) {
+				// The channel is closing; nothing after the message over the limit is delivered.
+				return;
+			}
+
+			int size = buffer.data.remaining();
+
+			// Counted before it is queued, since that queue is what grows while
+			// the handlers fall behind.
+			if (received.addAndGet(size) > receiveLimit) {
+				received.addAndGet(-size);
+
+				if (overflowed.compareAndSet(false, true)) {
+					peer.control.execute(DataChannelImpl.this::receiveLimitExceeded);
+				}
+				return;
+			}
+
+			byte[] data = new byte[size];
 			buffer.data.duplicate().get(data);
 
 			Message message = new Message(data, buffer.binary);
 
-			peer.events.execute(() -> deliverMessage(message));
+			peer.events.execute(() -> deliverMessage(message), () -> received.addAndGet(-size));
 		}
 	}
 

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

@@ -555,6 +555,10 @@ long channelSendLimit() {
 		return options.channelSendLimit;
 	}
 
+	long channelReceiveLimit() {
+		return options.channelReceiveLimit;
+	}
+
 	/**
 	 * Hands a future out to the application: it completes on the event
 	 * executor, so its callbacks run there like the handlers.

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

@@ -63,6 +63,8 @@ public final class PeerOptions {
 
 	final long channelSendLimit;
 
+	final long channelReceiveLimit;
+
 
 	private PeerOptions(Builder builder) {
 		this.name = builder.name;
@@ -76,6 +78,7 @@ private PeerOptions(Builder builder) {
 		this.customizer = builder.customizer;
 		this.gatheringTimeout = builder.gatheringTimeout;
 		this.channelSendLimit = builder.channelSendLimit;
+		this.channelReceiveLimit = builder.channelReceiveLimit;
 	}
 
 	/**
@@ -137,6 +140,8 @@ public static final class Builder {
 
 		private long channelSendLimit = 16 * 1024 * 1024;
 
+		private long channelReceiveLimit = 16 * 1024 * 1024;
+
 
 		private Builder() {
 		}
@@ -279,6 +284,30 @@ public Builder channelSendLimit(long bytes) {
 			return this;
 		}
 
+		/**
+		 * Sets how many bytes of received messages each data channel of the
+		 * peer may hold that its message handlers have not finished with yet,
+		 * including those kept while it has no handler; 16 MiB by default.
+		 * <p>
+		 * WebRTC cannot be told to stop receiving, so a channel whose handlers
+		 * fall behind the remote sender by more than that closes rather than
+		 * let received messages fill memory, or drop some unnoticed. The
+		 * peer's {@code onError} handlers hear why, and the messages after the
+		 * one that went over the limit are not delivered.
+		 *
+		 * @param bytes the limit, at least 1.
+		 *
+		 * @return this builder.
+		 */
+		public Builder channelReceiveLimit(long bytes) {
+			if (bytes < 1) {
+				throw new IllegalArgumentException("The limit must be at least 1 byte, got " + bytes);
+			}
+
+			this.channelReceiveLimit = bytes;
+			return this;
+		}
+
 		/**
 		 * Changes anything else of the {@link RTCConfiguration} the peer is
 		 * created with, after this builder applied its own values.