From 6e7269fcc8a6d937502a5cf8600cfe64065f8d03 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 20:51:02 +0200 Subject: [PATCH] fix: bound received data channel messages before they are queued Received messages count against a new PeerOptions channelReceiveLimit (16 MiB by default) from the moment they arrive until the handlers are done with them, including those kept while a channel has no handler. A channel that goes over it closes and reports why, instead of growing the event queue or dropping messages. --- design/client-api.md | 5 +- docs/guide/client-api.md | 5 +- .../dev/onvoid/webrtc/client/DataChannel.java | 9 +- .../onvoid/webrtc/client/DataChannelImpl.java | 81 +++++++++--- .../dev/onvoid/webrtc/client/PeerImpl.java | 4 + .../dev/onvoid/webrtc/client/PeerOptions.java | 29 +++++ .../dev/onvoid/webrtc/client/PeerTests.java | 116 ++++++++++++++---- 7 files changed, 198 insertions(+), 51 deletions(-) diff --git a/design/client-api.md b/design/client-api.md index a7d13f09..91ab9028 100644 --- a/design/client-api.md +++ b/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. diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 7cc4a362..16c2b760 100644 --- a/docs/guide/client-api.md +++ b/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)` | 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 diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java index eafd4d6b..aa1937b0 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java +++ b/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. + *

+ * 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. * diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java index 297e3906..84357572 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java +++ b/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. + *

+ * 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 messageHandlers; private final Handlers 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 handler) { if (!early.isEmpty()) { List 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)); } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index aef8b725..14785f1e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/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. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java index 24765841..6e11791d 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java +++ b/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. + *

+ * 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. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java index 4002429c..9686b49d 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -36,6 +36,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -103,43 +104,85 @@ void earlyMessagesKept() throws Exception { @Test void earlyMessagesBounded() throws Exception { - // The sender holds more than the receiver keeps. + // The sender holds more than the receiver keeps while it has no + // message handler, so the receiving channel closes. Peer[] peers = pair(options -> options.channelSendLimit(32 * 1024 * 1024)); - CompletableFuture remote = new CompletableFuture<>(); - CompletableFuture dropped = new CompletableFuture<>(); - - peers[1].onChannel(remote::complete); - peers[1].onError(dropped::complete); + CompletableFuture error = new CompletableFuture<>(); + peers[1].onError(error::complete); - // About 20 MiB while the remote channel has no message handler. DataChannel channel = peers[0].createChannel("bulk"); - int count = 100; + CompletableFuture closed = whenClosed(channel); byte[] payload = new byte[200 * 1024]; - for (int i = 0; i < count; i++) { - payload[0] = (byte) i; + // About 20 MiB, beyond the default limit of 16 MiB. + for (int i = 0; i < 100; i++) { channel.send(payload); } - await(channel.send(new byte[] { (byte) 0xFF })); - assertTrue(await(dropped).getMessage().contains("dropped received messages")); + assertTrue(await(error).getMessage().contains("fell more than"), () -> error.join().getMessage()); + await(closed); + } - List received = Collections.synchronizedList(new ArrayList<>()); - CompletableFuture last = new CompletableFuture<>(); - await(remote).onMessage(message -> { - byte[] bytes = message.bytes(); - if (bytes.length == 1) { - last.complete(null); - } - else { - received.add(bytes[0] & 0xFF); + @Test + void slowHandlerClosesChannel() throws Exception { + // The handler blocks on the first message while more arrive than the + // channel may hold. + int limit = 64 * 1024; + Peer[] peers = pair(options -> options.channelReceiveLimit(limit)); + CountDownLatch release = new CountDownLatch(1); + List handled = new CopyOnWriteArrayList<>(); + CompletableFuture error = new CompletableFuture<>(); + + peers[1].onError(error::complete); + peers[1].onChannel(remote -> remote.onMessage(message -> { + handled.add(message.bytes()[0] & 0xFF); + awaitQuietly(release); + })); + + DataChannel channel = peers[0].createChannel("bulk"); + CompletableFuture closed = whenClosed(channel); + byte[] payload = new byte[16 * 1024]; + + try { + for (int i = 0; i < 10; i++) { + payload[0] = (byte) i; + channel.send(payload); } - }); - await(last); - // The newest messages were kept, in order. - assertTrue(received.size() < count, () -> "Kept " + received.size() + " messages"); - assertEquals(count - 1, (int) received.get(received.size() - 1)); + // The receiver closed the channel while its handler still blocked. + await(closed); + } + finally { + release.countDown(); + } + + assertTrue(await(error).getMessage().contains("fell more than"), () -> error.join().getMessage()); + + // The one in the handler and three queued fit the limit; the fifth + // went over it, and nothing after it was delivered. The error came + // after those queued before it. + assertEquals(Arrays.asList(0, 1, 2, 3), handled); + } + + @Test + void handledMessagesFreeLimit() throws Exception { + // Far more than the limit passes through, a message at a time. + int limit = 64 * 1024; + Peer[] peers = pair(options -> options.channelReceiveLimit(limit)); + Semaphore handled = new Semaphore(0); + + peers[1].onChannel(remote -> remote.onMessage(message -> handled.release())); + + DataChannel channel = peers[0].createChannel("bulk"); + byte[] payload = new byte[16 * 1024]; + + for (int i = 0; i < 64; i++) { + await(channel.send(payload)); + assertTrue(handled.tryAcquire(TIMEOUT_SECONDS, TimeUnit.SECONDS), "Message " + i + " was not handled"); + } + + // Nothing reached the error handler either; the base class checks. + assertEquals(ChannelState.OPEN, channel.state()); } @Test @@ -590,4 +633,25 @@ private static Throwable cause(CompletableFuture future) throws Exception { } } + private static CompletableFuture whenClosed(DataChannel channel) { + CompletableFuture closed = new CompletableFuture<>(); + + channel.onStateChange(state -> { + if (state == ChannelState.CLOSED) { + closed.complete(null); + } + }); + + return closed; + } + + private static void awaitQuietly(CountDownLatch latch) { + try { + latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }