[cloudwebrtc]

Apply backpressure to native data stream sends

Targets data-streams-v2 (#1166).

Problem

On native, the outbound pump drained every packet batch from the Rust core straight into an unbounded Dart future chain (_sendChain). Nothing waited for the data channel to drain, because the old waitForBufferStatusLow flow control was lost in the move to the core.

The core's polled OutgoingPacketQueue acks a send as soon as the packets are buffered, which had two effects:

• sendText / sendBytes / sendFile resolved before any data was on the wire.
• Transport errors were only logged, so callers saw success for data that was never delivered.

Fix

1. Backpressure. The pump now sends one packet at a time. It awaits engine.waitForBufferStatusLow(Reliability.reliable) before each send, and pulls the next batch only once the current one is out. The backlog stays queued in the core instead of piling up in Dart.
2. Accurate completion. One-shot sends get a Dart-generated stream id, registered before the core call. The pump completes it when the stream's last packet (its trailer, or an inline header) reaches the engine. It fails the send with DataStreamError(AbnormalEnd) if a packet fails to go out, and fails any sends still pending when it exits.

Results

p1 → p2 through a local SFU: 64 MB sendBytes, uncompressed, both clients in one macOS process.

Metric	Before	After
sendBytes resolves	532 ms (~0 bytes on wire)	2815 ms (~64 MB on wire)
Receiver handler invoked	29.7 s	3.7 s
Receiver complete	36.4 s	4.2 s
Throughput	1.76 MB/s	15.3 MB/s
Payload integrity	OK	OK

Key changes (lib/src/data_stream/data_streams_native.dart)

Pump: send one packet at a time, instead of chaining every packet unbounded

   for (final encoded in batch) {
-    _enqueueSend(encoded);
+    await _sendOutbound(lk_models.DataPacket.fromBuffer(encoded));
   }

Per-packet send: wait for the channel to drain, then report the stream's outcome

// A stream ends with its trailer, or with its only packet when the content was inlined.
final isLast = packet.hasStreamTrailer() ||
    (packet.hasStreamHeader() && packet.streamHeader.hasInlineContent());
try {
  await room.engine.waitForBufferStatusLow(Reliability.reliable);
  await room.engine.sendDataPacket(packet, reliability: Reliability.reliable);
  if (isLast) _pendingSends[streamId]?.complete();
} catch (e) {
  _pendingSends[streamId]?.completeError(DataStreamError(reason: DataStreamErrorReason.AbnormalEnd, ...));
}

send*: resolve when the data is out, not when the core has queued it

Future<T> _sendTracked<T>(String id, Future<T> Function() send) async {
  final sent = Completer<void>();
  sent.future.ignore(); // the pump may fail this before we await it
  _pendingSends[id] = sent; // registered first: the pump may finish the stream before send() returns
  try {
    final result = await mappingFfiErrors(send);
    await sent.future;
    return result;
  } finally {
    _pendingSends.remove(id);
  }
}

sendText, sendBytes and sendFile (including attachments) pass their generated id to the core and go through _sendTracked.

Behavior changes

• sendText / sendBytes / sendFile now resolve only after the stream's last packet has been handed to the engine. For large payloads they take noticeably longer to return.
• These calls can now throw DataStreamError(AbnormalEnd) on a transport failure, where they used to report success.
• Incremental writers (streamText / streamBytes) get the same pump-level backpressure, but write() / close() still return once the core has queued the data.

Not in this PR

• Peak sender memory is unchanged. The core's outgoing queue (polled.rs) is an unbounded channel, so the whole payload is still buffered on the Rust side.
• The generated livekit_ffi Dart bindings leak memory. Lowered calloc buffers and lifted RustBuffers are never freed. These were confirmed separately and need fixes in rust-sdks.

Test plan

• New regression tests in test/core/data_stream_v2_test.dart. Both fail on the previous implementation and pass with this change:
  • A send pauses while the channel reports a full buffer, and resolves only after its trailer is sent.
  • A transport failure fails the send with DataStreamError(AbnormalEnd).
• MockDataChannel gains fakeBufferedAmount to simulate an undrained channel.
• Existing data stream, RPC and FFI tests pass.
• flutter analyze, dart format, import_sorter are clean.
• Manual: p1 → p2 against a local SFU on macOS (results above), with every byte verified on the receiver.

🤖 Generated with Claude Code

[devin-ai-integration]

Devin Review found 3 potential issues.

1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)

[devin-ai-integration]

🔴 Data streams stall after full reconnect

When a full reconnect replaces a backpressured channel, waitForBufferStatusLow can block every remaining stream packet. The engine retains the old _dcBufferStatus value; the fresh channel has no reason to emit a low-buffer event.

Learn more

The outbound pump now waits on the engine's data-channel buffer before each packet. waitForBufferStatusLow checks the channel's current amount, then polls _dcBufferStatus if the channel is not low. A full reconnect replaces _reliableDCPub without resetting _dcBufferStatus, so an old false survives even when the replacement channel is empty. With no packets sent on the new channel, its low-buffer callback never fires and the pump cannot progress.

Example: A 64 MB send fills the reliable channel above 2 MB, setting _dcBufferStatus to false. During a full reconnect the engine replaces that channel. The new channel has zero bytes buffered, but the blocked wait continues polling false and the remaining chunks and subsequent streams cannot send.

Recommended fix: Make Engine.waitForBufferStatusLow re-evaluate the current channel's buffered amount while waiting, or reset and signal the status when replacing publisher data channels. Ensure disconnect also wakes blocked sends.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🟡 Backpressured streams retain engine listeners

Each backpressured packet calls waitForBufferStatusLow, which retains its closing listener after a normal drain. Large sends accumulate listeners until the engine closes.

Learn more

The engine's waitForBufferStatusLow registers an events.once<EngineClosingEvent> callback when the channel is full. A normal low-buffer wake-up completes the wait without cancelling that listener. The new per-packet call makes a large send create many persistent listeners whenever it repeatedly crosses the buffer threshold.

Example: A long file send fills the channel, drains, then fills it again 50 times. Each drain unblocks the send but leaves one EngineClosingEvent listener alive until the room closes.

Recommended fix: Change Engine.waitForBufferStatusLow to retain the cancellation callback returned by events.once, and cancel it in a finally block on success or error.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🟡 Native send behavior lacks a changeset

The backpressure change affects published sends, but this PR adds no .changes entry. The existing v2 entry predates this change.

Learn more

The repository's release rule requires a changeset for every PR affecting the published package. This PR changes when native sendText, sendBytes and sendFile resolve, and which transport failures they propagate. The existing .changes/data-streams-v2 entry comes from the base and does not document this PR's behavior change.

Example: Users upgrading to the next release see new backpressure and send errors but find no corresponding entry for the change in the changesets contributed by this PR.

Recommended fix: Add a new .changes/ file describing the changed send completion and failure semantics, using the repository's specified changeset format.

---

Was this helpful? React with 👍 or 👎 to provide feedback.