fix: harden data channel sends and runtime close

docs/guide/client-api.md

@@ -102,6 +102,7 @@ Peer peer = rtc.createPeer(PeerOptions.builder()
 | `negotiation(Negotiation)` | `AUTOMATIC` by default, or `MANUAL`, see [Manual Negotiation](#manual-negotiation). |
 | `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). |
 | `configure(Consumer<RTCConfiguration>)` | Anything else the [peer connection configuration](/guide/networking/peer-connection-config) has. |
 
 ### Signaling
@@ -507,6 +508,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.
+
 `ChannelOptions` set how a channel delivers:
 
 ```java
@@ -617,6 +620,8 @@ You close what you created, and closing an owner closes what it owns:
 
 `WebRtc.close()` is the one close that waits, for up to the close timeout, which makes it fit for try-with-resources. Once it returns, every peer's `state()` is `CLOSED`; the `whenClosed` futures and `CLOSED` events follow on the peers' event threads, since waiting for them could never end when `close()` is called on an event thread.
 
+When a peer or track does not close within the close timeout, the runtime leaves its factory undisposed rather than free what WebRTC's threads may still use. `WebRtc.closeAsync()` then fails with a `WebRtcException` naming the cause, and the error handler hears of it as well, which is how `close()` reports it.
+
 ## Errors
 
 Failures arrive as exceptions, thrown or as the cause of a failed future, all of them a `WebRtcException`:
@@ -625,6 +630,7 @@ Failures arrive as exceptions, thrown or as the cause of a failed future, all of
 |---|---|
 | `NegotiationException` | A description or candidate was rejected, or two impolite peers collided. |
 | `PeerClosedException` | An operation was pending when its peer, channel or runtime closed, or came after that. |
+| `ChannelFullException` | A data channel already holds as many bytes waiting to be sent as its peer's `channelSendLimit` allows. |
 | `MediaException` | A device is missing, busy, refused access, or already open with other options. |
 | `ConfigurationException` | A call does not fit how the runtime was built, or WebRTC refused a configuration. |
 | `BlockingCallException` | A future was waited on from an event thread. |

webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp

@@ -201,39 +201,48 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_dispose
 JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBuffer
 (JNIEnv * env, jobject caller, jobject jBuffer, jboolean isBinary)
 {
-	webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
-	CHECK_HANDLE(channel);
+	try {
+		webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
+		CHECK_HANDLE(channel);
 
-	uint8_t * address = static_cast<uint8_t *>(env->GetDirectBufferAddress(jBuffer));
+		uint8_t * address = static_cast<uint8_t *>(env->GetDirectBufferAddress(jBuffer));
 
-	if (address != NULL) {
-		jlong bufferLength = env->GetDirectBufferCapacity(jBuffer);
+		if (address != NULL) {
+			jlong bufferLength = env->GetDirectBufferCapacity(jBuffer);
 
-		webrtc::CopyOnWriteBuffer data(address, static_cast<size_t>(bufferLength));
+			webrtc::CopyOnWriteBuffer data(address, static_cast<size_t>(bufferLength));
 
-		if (!channel->Send(webrtc::DataBuffer(data, static_cast<bool>(isBinary)))) {
-			env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send"));
+			if (!channel->Send(webrtc::DataBuffer(data, static_cast<bool>(isBinary)))) {
+				env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send"));
+			}
+		}
+		else {
+			env->Throw(jni::JavaError(env, "Non-direct buffer provided"));
 		}
 	}
-	else {
-		env->Throw(jni::JavaError(env, "Non-direct buffer provided"));
+	catch (...) {
+		ThrowCxxJavaException(env);
 	}
 }
 
 JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBuffer
 (JNIEnv * env, jobject caller, jbyteArray jBufferArray, jboolean isBinary)
 {
-	webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
-	CHECK_HANDLE(channel);
-
-	int8_t * arrayPtr = env->GetByteArrayElements(jBufferArray, nullptr);
-	size_t arrayLength = env->GetArrayLength(jBufferArray);
-
-	webrtc::CopyOnWriteBuffer data(arrayPtr, arrayLength);
+	try {
+		webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
+		CHECK_HANDLE(channel);
 
-	env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT);
+		auto releaseArray = [env, jBufferArray](jbyte * bytes) {
+			env->ReleaseByteArrayElements(jBufferArray, bytes, JNI_ABORT);
+		};
+		std::unique_ptr<jbyte, decltype(releaseArray)> bytes(
+			env->GetByteArrayElements(jBufferArray, nullptr), releaseArray);
+		if (!bytes) {
+			return;
+		}
+		webrtc::CopyOnWriteBuffer data(bytes.get(), env->GetArrayLength(jBufferArray));
+		bytes.reset();
 
-	try {
 		if (!channel->Send(webrtc::DataBuffer(data, static_cast<bool>(isBinary)))) {
 			env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send"));
 		}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2026 Alex Andres
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.onvoid.webrtc.client;
+
+/**
+ * A message was not sent, since its data channel already holds as many bytes
+ * waiting to be sent as {@link PeerOptions.Builder#channelSendLimit(long)}
+ * allows. Messages sent later are taken again once the backlog shrank.
+ */
+public class ChannelFullException extends WebRtcException {
+
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * Creates the exception with a message that says what went wrong.
+	 *
+	 * @param message the detail message.
+	 */
+	public ChannelFullException(String message) {
+		super(message);
+	}
+
+}

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

@@ -54,7 +54,9 @@ public interface DataChannel extends AutoCloseable {
 	 * @return a future that completes when the message is handed to the
 	 * transport. While a lot of data waits to be sent, the future completes
 	 * only once the backlog shrank, so a loop that waits for each send slows
-	 * down by itself.
+	 * down by itself. The future fails at once with a {@link
+	 * ChannelFullException} when the message would take the backlog beyond
+	 * {@link PeerOptions.Builder#channelSendLimit(long)}.
 	 */
 	CompletableFuture<Void> send(String text);
 
@@ -64,7 +66,7 @@ public interface DataChannel extends AutoCloseable {
 	 * @param data the message.
 	 *
 	 * @return a future that completes when the message is handed to the
-	 * transport.
+	 * transport, or fails as {@link #send(String)} describes.
 	 */
 	CompletableFuture<Void> send(byte[] data);
 
@@ -75,7 +77,7 @@ public interface DataChannel extends AutoCloseable {
 	 * @param data the message.
 	 *
 	 * @return a future that completes when the message is handed to the
-	 * transport.
+	 * transport, or fails as {@link #send(String)} describes.
 	 */
 	CompletableFuture<Void> send(ByteBuffer data);
 

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

@@ -37,7 +37,8 @@
  * <p>
  * The native channel is used on the peer's control executor, where sends wait
  * while the channel is not open or holds more than {@link #HIGH_WATER_MARK}
- * bytes. Handlers run on the peer's event executor.
+ * 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.
  */
 final class DataChannelImpl implements DataChannel {
 
@@ -51,6 +52,9 @@ final class DataChannelImpl implements DataChannel {
 
 	private final String label;
 
+	/** The most bytes {@link #held} may reach. */
+	private final long sendLimit;
+
 	/** Bytes sent and not yet handed to the transport. */
 	private final AtomicLong held = new AtomicLong();
 
@@ -93,6 +97,7 @@ final class DataChannelImpl implements DataChannel {
 	DataChannelImpl(PeerImpl peer, String label) {
 		this.peer = peer;
 		this.label = label;
+		this.sendLimit = peer.channelSendLimit();
 		this.messageHandlers = new Handlers<>(peer.rtc::reportError);
 		this.stateHandlers = new Handlers<>(peer.rtc::reportError);
 
@@ -262,8 +267,16 @@ void closeFromPeer(WebRtcException error) {
 	}
 
 	private CompletableFuture<Void> enqueue(byte[] data, boolean binary) {
+		long size = data.length;
+		long before = held.getAndUpdate(amount -> amount + size <= sendLimit ? amount + size : amount);
+
+		if (before + size > sendLimit) {
+			return peer.toPublic(ClientFuture.failed(new ChannelFullException("The data channel " + label
+					+ " holds " + before + " bytes waiting to be sent, and " + size + " more would exceed "
+					+ "its limit of " + sendLimit)));
+		}
+
 		CompletableFuture<Void> sent = new CompletableFuture<>();
-		held.addAndGet(data.length);
 
 		peer.control.execute(() -> {
 			if (released || state == ChannelState.CLOSING || state == ChannelState.CLOSED) {

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

@@ -551,6 +551,10 @@ boolean isClosing() {
 		return closing.get();
 	}
 
+	long channelSendLimit() {
+		return options.channelSendLimit;
+	}
+
 	/**
 	 * 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

@@ -61,6 +61,8 @@ public final class PeerOptions {
 
 	final Duration gatheringTimeout;
 
+	final long channelSendLimit;
+
 
 	private PeerOptions(Builder builder) {
 		this.name = builder.name;
@@ -73,6 +75,7 @@ private PeerOptions(Builder builder) {
 				? EnumSet.noneOf(MediaKind.class) : EnumSet.copyOf(builder.receive));
 		this.customizer = builder.customizer;
 		this.gatheringTimeout = builder.gatheringTimeout;
+		this.channelSendLimit = builder.channelSendLimit;
 	}
 
 	/**
@@ -132,6 +135,8 @@ public static final class Builder {
 
 		private Duration gatheringTimeout = Duration.ofSeconds(3);
 
+		private long channelSendLimit = 16 * 1024 * 1024;
+
 
 		private Builder() {
 		}
@@ -254,6 +259,26 @@ public Builder gatheringTimeout(Duration timeout) {
 			return this;
 		}
 
+		/**
+		 * Sets how many bytes each data channel of the peer may hold that were
+		 * sent and not yet handed to the transport; 16 MiB by default. A send
+		 * beyond that fails at once with a {@link ChannelFullException}
+		 * instead of letting the backlog grow without bound, and so does a
+		 * single message larger than the limit.
+		 *
+		 * @param bytes the limit, at least 1.
+		 *
+		 * @return this builder.
+		 */
+		public Builder channelSendLimit(long bytes) {
+			if (bytes < 1) {
+				throw new IllegalArgumentException("The limit must be at least 1 byte, got " + bytes);
+			}
+
+			this.channelSendLimit = bytes;
+			return this;
+		}
+
 		/**
 		 * Changes anything else of the {@link RTCConfiguration} the peer is
 		 * created with, after this builder applied its own values.