Add new data channel high / low water mark approach for packet congestion control by 1egoman · Pull Request #2013 · livekit/client-sdk-js

1egoman · 2026-07-15T14:51:45Z

(Started by Ryan -> Lukas took over this change and wrote the below)

closes #1995

the gist of this PR is to add a kind dependent mutex around data channel sends.
Previously the waitForBufferStatusLow method could be stalling lots of data requests in parallel and once it resolved all of them got flushed to the data channel at once potentially overflowing its buffer.

To generalise this solution without any performance/throughput hits the PR additionally introduces high and low watermarks for the data channels so that there's a designated "headroom" under which data packets are allowed to be pushed to the data channel buffer immediately (avoiding a constant back and forth between high and low DC buffer status)

…congestion control

changeset-bot · 2026-07-15T14:51:50Z

🦋 Changeset detected

Latest commit: facfe90

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 1 package

Name	Type
livekit-client	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

devin-ai-integration

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no potential bugs to report.

View in Devin Review to see 1 additional finding.

github-actions · 2026-07-15T15:01:07Z

size-limit report 📦

Path	Size
dist/livekit-client.esm.mjs	101.87 KB (+0.39% 🔺)
dist/livekit-client.umd.js	110.91 KB (+0.5% 🔺)

devin-ai-integration

Devin Review found 1 new potential issue.

View 1 additional finding in Devin Review.

devin-ai-integration · 2026-07-16T09:14:01Z

+        case 'wait':
+          if (!this.isBelowHighWaterMark(kind)) {
+            await this.waitForBufferHeadroom(kind);
+          }
+          break;


🟡 Data-track sends can raise unhandled promise rejections during reconnects

The wait-for-space path (waitForBufferHeadroom at src/room/RTCEngine.ts:1699) can now reject while the channel is replaced or torn down during a reconnect, and the data-track sender that drives it (src/room/Room.ts:311-312) only attaches a .finally, so the rejection has no handler and surfaces as an unhandled promise rejection.
Impact: During reconnects, ongoing data-track streaming can spew unhandled rejection errors into the console/error monitoring even though the connection otherwise recovers.

How the widened rejection contract reaches an unhandled caller

Previously the lossy 'wait' path awaited waitForBufferStatusLow, which only rejected on engine close. The new waitForBufferHeadroom/waitForBufferHeadroomLocked (src/room/RTCEngine.ts:1868-1915) additionally reject a parked waiter when the data channel fires close (onDCClose) or when invalidateDataChannelWaiters aborts the per-kind epoch (onEpochAbort) — both of which happen on the normal reconnect/createDataChannels path, not just on close.

Room's packetAvailable handler calls this.engine.sendLossyBytes(bytes, DataChannelKind.DATA_TRACK_LOSSY, 'wait').finally(() => this.outgoingDataTrackManager.handlePacketSendComplete(handle)) (src/room/Room.ts:310-312). A .finally re-rejects with the same reason and there is no .catch, so when a data-track send is parked above the high-water mark and the channels are recreated mid-reconnect, the resulting UnexpectedConnectionState rejection is unhandled.

Prompt for agents

sendLossyBytes's 'wait' behavior (src/room/RTCEngine.ts around line 1697-1701) now awaits waitForBufferHeadroom, which rejects with UnexpectedConnectionState not only on engine close but also when the data channel closes or the per-kind epoch is aborted by invalidateDataChannelWaiters (called from createDataChannels and cleanupPeerConnections during reconnects). The data-track caller in src/room/Room.ts:310-312 attaches only a .finally() (no .catch), so these transient rejections become unhandled promise rejections during reconnect. Consider either catching/swallowing the transient teardown rejection inside sendLossyBytes for the lossy/data-track kinds (they can't meaningfully be replayed, so dropping is acceptable) or adding a .catch to the Room packetAvailable handler so the rejection is handled while still invoking handlePacketSendComplete.

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

1egoman

I can't approve officially because I originally opened this pull request, but this looks good ✅

Probably as expected due to our fairly extensive conversation on slack about this most of my comments are naming / code clarity related.

1egoman · 2026-07-16T15:32:00Z

+      const buffer = (engine as unknown as { reliableMessageBuffer: DataPacketBuffer })
+        .reliableMessageBuffer;


nitpick: it might be nice to have some test specific helpers to avoid all the casting to access these private members. Maybe something like getReliableMessageBuffer(engine), resendReliableMessagesForResume(engine), etc

At one point during the work on this PR I was already half way through changing those casts to ts-ignores when I realised that this pattern just emerged because it's how the other tests are doing it as well.
I'm not convinced that we should make public accessors for all of them, but definitely aligned on changing this current pattern.
Let's do it in a follow up with a clear concept of how to tackle it as a more general solution

1egoman · 2026-07-16T15:39:55Z

+      expect(dc.send.mock.calls[2][0]).not.toBe(replayed2);
+    });
+
+    it('transmits a packet deferred mid-replay instead of marking it sent without sending', async () => {


question: To make sure I understand, this test is exercising the case where a new data packet is enqueued when the data channel's fullness is above the high water mark?

If so, a suggested better test name:

Suggested change

it('transmits a packet deferred mid-replay instead of marking it sent without sending', async () => {

it('transmits a packet deferred mid-replay instead of dropping it due to the data channel buffer being full', async () => {

no, this tests that if another packet is enqueued while resuming (landing in our own intermediate buffer), but the replay has already started, the new packet should either not be marked as sent when the replay finishes

1egoman · 2026-07-16T15:45:45Z

+      // Park the send on a full buffer, then invalidate the channel (reconnect/replacement).
+      dc.bufferedAmount = 2 * 1024 * 1024;
+      const send = engine.sendDataPacket(makePacket(1), DataChannelKind.RELIABLE);
+      await tick();
+      (
+        engine as unknown as { invalidateDataChannelWaiters: (reason: string) => void }
+      ).invalidateDataChannelWaiters('data channels recreated');


question: To make sure I understand properly, this is simulating a full reconnect?

any condition in which the data channels would get recreated, this can also happen on resume, see

client-sdk-js/src/room/RTCEngine.ts

Lines 1511 to 1515 in cc79fd7

// recreate publish datachannel if it's id is null

// (for safari https://bugs.webkit.org/show_bug.cgi?id=184688)

if (this.reliableDC?.readyState === 'open' && this.reliableDC.id === null) {

this.createDataChannels();

}

1egoman · 2026-07-16T15:52:05Z

+  describe('sendLossyBytes', () => {
+    it('ensures the publisher is connected before sending (direct data-track path)', async () => {


question: I only see one place where sendLossyBytes(..., ..., 'drop') is exercised in here (await engine.sendLossyBytes(new Uint8Array(100), DataChannelKind.LOSSY, 'drop');), is there more worth adding? (ie, maybe verifying drops occur when the data channel isn't initialized, etc).

the refactor PR makes this a bit obsolete, so I'll hold off on adding this here now, but happy to add something in that regard on the refactor PR if you still see a gap there

1egoman · 2026-07-16T16:20:58Z

+  /**
+   * Resolves once the caller may send on the `kind` channel: immediately while the send buffer is
+   * at or below its high-water mark, otherwise once the buffer has drained to the low-water mark
+   * (the `bufferedamountlow` event). Callers are serialized through a per-kind mutex so that, when
+   * the buffer drains, they refill it one at a time (up to the high-water mark) rather than all
+   * sending at once and overflowing the SCTP send buffer (see livekit/client-sdk-js#1995). The
+   * closed/buffer checks run inside the lock so queued callers proceed in FIFO order.
+   */
+  async waitForBufferHeadroom(kind: DataChannelKind) {
+    const unlock = await this.lockBufferHeadroom(kind);
+    try {
+      await this.waitForBufferHeadroomLocked(kind);
+    } finally {
+      unlock();
+    }
+  }
+
+  /**
+   * Core wait of {@link waitForBufferHeadroom}. The caller must hold the kind's headroom lock —
+   * batch senders (the resume replay) hold it across all of their sends so no other sender can
+   * interleave, and call this per message to respect flow control within the batch.
+   */
+  private async waitForBufferHeadroomLocked(
+    kind: DataChannelKind,


thought: These two method name I find really confusing - waitForBufferHeadroomLocked to me implies that it is acquiring the lock, and in comparison waitForBufferHeadroom implies that it isn't. Should these names be flipped around?

yeah, it's confusing. There's some (debated) convention that the *locked suffix indicates it needs to be called within a lock.

Changed this to be explicit *withLock and *withoutLock

1egoman · 2026-07-16T16:23:45Z

+      const onEpochAbort = () => {
+        cleanup();
+        reject(
+          new UnexpectedConnectionState(
+            `DataChannel ${kind} was replaced or torn down while waiting for headroom`,
+          ),
+        );
+      };
+      const cleanup = () => {
+        dc.removeEventListener('bufferedamountlow', onBufferedAmountLow);
+        dc.removeEventListener('close', onDCClose);
+        epochSignal.removeEventListener('abort', onEpochAbort);
+      };
+      if (epochSignal.aborted) {
+        onEpochAbort();
+        return;


question: Can you explain what epoch means here - is there something you could call this which makes it more clear this is like "signal which fires when data channel closes"? (I'm assuming that's all this is for, but maybe it's more broad?)

just meant to indicate a generation of data channels. I dropped the epoch entirely and just called it onAbort now and waiterAbortController

1egoman · 2026-07-16T16:30:37Z

+    buffer.push(item(2, 100, false));
+    buffer.push(item(3, 100, false));
+
+    buffer.popToSequence(2);


nitpick (probably out of scope of this pull request since it was pre-existing): Should this be unshiftToSequence? pop usually implies the inverse of push, but this looks like it's offsetting from the front.

makes sense, yeah let's address this in a follow up

main squash-merged the data-channel watermark work (#2013), so its inline RTCEngine changes conflicted with this branch's refactored data-channel module. Resolved RTCEngine.ts / RTCEngine.test.ts in favour of the module version (which already carries all of that work), and took main's other additions cleanly — notably the data-track SID reassignment fix (#2000) this branch didn't yet have. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

feat: add new data channel high / low water mark approach for packet …

3b101cd

…congestion control

1egoman requested review from lukasIO and xianshijing-lk as code owners July 15, 2026 14:51

devin-ai-integration Bot reviewed Jul 15, 2026

View reviewed changes

1egoman mentioned this pull request Jul 15, 2026

Address data channel thundering herd due to concurrent sendFile(...) calls #2010

Closed

This comment was marked as resolved.

Sign in to view

1egoman added 3 commits July 15, 2026 11:27

fix: address typescript errors

c4aefaf

fix: add missing changeset

35215c7

fix: add waitUntilBelowHighWaterMark mock to tests

2b509cd

1egoman force-pushed the data-channel-buffer-range branch from a34512a to 2b509cd Compare July 15, 2026 15:27

This comment was marked as resolved.

Sign in to view

lukasIO added 2 commits July 16, 2026 08:00

tighten lossy control sending, reject on dc close

f7d349d

consistency

2031267

This comment was marked as resolved.

Sign in to view

rename

ccc1669

This comment was marked as resolved.

Sign in to view

lukasIO added 5 commits July 16, 2026 08:17

fix dynamic lossy behaviour

cbd6a13

naming

22ffc35

cleanup

756e8e3

comments

23da009

naming

e6fa0d7

This comment was marked as resolved.

Sign in to view

lukasIO added 3 commits July 16, 2026 09:03

guard against silent package loss on resume

95dd9fa

address resume edge cases

0461bc2

catch throwing buffer check

ed2ac47

This comment was marked as resolved.

Sign in to view

fix test

808bc5a

devin-ai-integration Bot reviewed Jul 16, 2026

View reviewed changes

lukasIO added 2 commits July 16, 2026 16:14

fix throws-check

d6faabc

address devin comment

cc79fd7

1egoman commented Jul 16, 2026

View reviewed changes

lukasIO added 2 commits July 17, 2026 16:35

make naming clearer

3f63fbb

clarify epoch/abort naming

facfe90

lukasIO approved these changes Jul 17, 2026

View reviewed changes

lukasIO merged commit ccb9939 into main Jul 17, 2026
9 checks passed

lukasIO deleted the data-channel-buffer-range branch July 17, 2026 15:08

This was referenced Jul 17, 2026

Version Packages #2006

Merged

Version Packages apnerve/client-sdk-js#123

Open

1egoman mentioned this pull request Jul 20, 2026

chore: extract data channel handling into manager class #2014

Merged

		const buffer = (engine as unknown as { reliableMessageBuffer: DataPacketBuffer })
		.reliableMessageBuffer;

	it('transmits a packet deferred mid-replay instead of marking it sent without sending', async () => {
	it('transmits a packet deferred mid-replay instead of dropping it due to the data channel buffer being full', async () => {

	// recreate publish datachannel if it's id is null
	// (for safari https://bugs.webkit.org/show_bug.cgi?id=184688)
	if (this.reliableDC?.readyState === 'open' && this.reliableDC.id === null) {
	this.createDataChannels();
	}

		describe('sendLossyBytes', () => {
		it('ensures the publisher is connected before sending (direct data-track path)', async () => {

Uh oh!

Conversation

1egoman commented Jul 15, 2026 • edited by lukasIO Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

changeset-bot Bot commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🦋 Changeset detected

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

✅ Devin Review: No Issues Found

Uh oh!

github-actions Bot commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

size-limit report 📦

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

This comment was marked as resolved.

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

1egoman left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukasIO Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

1egoman Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

1egoman Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

1egoman commented Jul 15, 2026 •

edited by lukasIO

Loading

changeset-bot Bot commented Jul 15, 2026 •

edited

Loading

github-actions Bot commented Jul 15, 2026 •

edited

Loading

lukasIO Jul 17, 2026 •

edited

Loading

1egoman Jul 16, 2026 •

edited

Loading

1egoman Jul 16, 2026 •

edited

Loading