Extract per-channel data channel flow control - #1093
Merged
Conversation
Extracts the flow-control concept that DataChannelPair and DataTrackFrameSender each grew independently into one type owning a single channel's buffered-amount mirror, its low-water gate, and a latch producers can await. It owns no queue and no send policy — those differ per consumer (DataChannelPair parks senders in FIFO order, DataTrackFrameSender drops the oldest frame) — and no channel reference, so pair readiness can stay a single locked read in DataChannelPair. The amount is mirrored locally rather than read back from the channel. `didChangeBufferedAmount:` reports bytes *drained* since the last report rather than the current level (SctpDataChannel:: MaybeSendOnBufferedAmountChanged sends a diff, and only once >=100 KiB has drained or the buffer empties), and `bufferedAmount` is a PROXY_SECONDARY_CONSTMETHOD0 getter, so reading it blocks on WebRTC's network thread. `waitForHeadroom()` is the capability the SDK currently lacks: both client-sdk-js and client-sdk-android let a producer await headroom before committing work rather than handing bytes to a drain that will park them. No in-tree caller yet; the intended one is the data stream writers. Built on existing primitives only, per AGENTS.md: StateSync for the mirror, AsyncCompleter for the gate (its rearm/resume pair is exactly the sticky-latch semantics needed). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Replaces SendBuffer.rtcAmount / canSend(threshold:) / updateTarget with one BufferedDataChannel per channel. The drain keeps its queue, its retry buffer, its sequence stamping and its event loop; only the byte accounting moves. Behaviour-preserving by construction: - `hasHeadroom` is `pending <= lowWaterMark`, exactly the old `canSend(threshold:)`. Readiness is deliberately *not* folded into it, so `channel(for:)` remains the sole authority on whether a dequeued request can ship and the park-and-replay path is untouched. - `didDrain` keeps the old self-heal and its "Unexpected buffer size detected" error log. - The reliable retry buffer still trims against the raw report, as before. - `reportReadiness` derives from the same `State.isOpen` snapshot the wakeup already used, and only gates `waitForHeadroom`. Deliberately not fixed here: the mirror is still not cleared by `reset(throwing:)`, so it stays biased against a replacement channel after a full reconnect. That is a behaviour change and gets its own commit. One cost: `hasHeadroom` is a locked read per drain iteration where the old counter was a plain struct field. Uncontended, and ~0.03% of a measured 127 us small-packet send. Verified: 15 tests across DataChannelPairTests, RealiableDataChannelTests (all 6 reconnect/replay modes), EncryptedDataChannelTests and SDPMaxMessageSizeParserTests, matching the pre-change baseline. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
reset(throwing:) dropped the channel references but left the mirror holding whatever was outstanding. Room keeps one DataChannelPair for its lifetime (`lazy var publisherDataChannel`) and swaps fresh channels in on a full reconnect, so that stale count carries over to a channel whose outbound buffer is actually empty. Below the low-water mark it self-corrected on the next drain report. Above it, reliable sends stalled permanently: `hasHeadroom` was false, so the drain sent nothing, so WebRTC reported no drain, so nothing ever lowered the mirror. Reachable whenever a full reconnect lands while more than 2 MB is buffered. didDrain no longer logs drift when the mirror is already empty. That case is now expected — a closing channel can flush bytes the mirror has just forgotten — and it is not actionable by the consumer, which per AGENTS.md is the bar for .error. The regression test asserts through DataChannelPair rather than the mirror alone, since the bug was in the wiring; it fails on both channels without the fix. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
PublishDataRequest served three phases at once — not yet dispatched, in
flight, and retained for replay — and distinguished the last one by a
convention: `withoutContinuation()` at the retry-buffer boundary, guarded
downstream by
assert(request.continuation == nil,
"Continuation may fire multiple times while retrying causing crash")
Replay hands the same bytes over again, so a retained write that could
still resume a caller resumes it more than once. That is now
unrepresentable rather than asserted: RetainedWrite has no continuation
field, and re-entering the send buffer goes through `replayed`, which
states that a replayed write has no waiter.
ReadyWrite serialized, stamped, size-checked; what sendData accepts
RetainedWrite dispatched and retained for resume; nothing to resume
No behaviour change: the same writes reach sendData in the same order, and
the assert only ever fired in debug builds if the invariant broke.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
channel(for:) required *both* channels to be open before either would send, so a lossy channel briefly down parked reliable writes, and vice versa. The two are independent SCTP streams with no cross-stream ordering to preserve, and the SFU's dedup gate is per-kind, so the coupling bought nothing. client-sdk-js gates per kind (dataChannelForKind) and rust keeps per-kind channels; Swift was the outlier. The pre-flight requirement is unchanged: openCompleter still resolves only once both channels have reached .open, and ensurePublisherConnected still awaits it before any send. Only the live gate is now per channel. .wakeup follows the gate — a channel reaching .open now wakes its own parked writes rather than waiting for its sibling, which is what makes the new gate reachable. reportReadiness likewise reports per channel. Covered by the reconnect/replay E2E matrix (6 modes), which exercises transient channel closes; channel(for:) itself can't be unit-tested since LKRTCDataChannel needs a live peer connection. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
DataTrackFrameSender read `channel.bufferedAmount` on every pump iteration, which is a PROXY_SECONDARY_CONSTMETHOD0 getter — a BlockingCall onto WebRTC's network thread. Two blocking hops per packet where one will do, and the second implementation of an accounting rule that DataChannelPair already had. It now mirrors the drained-byte deltas the callback already carried and threw away. `bufferedAmount` leaves the DataTrackSendChannel seam entirely, so nothing reads the level back. This is a behaviour change, deliberately: the absolute read was self-correcting, so if `send` reported success for bytes libwebrtc then dropped, the next callback resynced. The mirror instead carries that error until the next `attach` (transport swap). Chosen because the drift path needs a lost packet libwebrtc reported as sent, whereas the blocking hop was paid on every single packet. The queue and the drop-oldest policy stay here — they are what differs from DataChannelPair's park-in-FIFO — matching how rust-sdks keeps dc_sender.rs separate from data_channel_task. Unit tests pin the same semantics: FakeSendChannel now reports drained counts through `flush()` instead of exposing a level, and the three tests that pre-set a full buffer fill it by sending, since the mirror only counts what the sender hands over. 8/8 plus 44 data-track tests across all 8 suites, including the stress suite. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
DataChannelPair was one object driving two channels, so every callback and
every event had to say which channel it concerned: 21 references to a
ChannelKind enum, 18 branch sites, and an extension that recovered the
kind by sniffing the channel's label. That axis existed only because one
object held both delegate slots.
Now each channel has a DataChannelDrain<Stage> that is its own
LKRTCDataChannelDelegate, so buffered-amount and state callbacks land on
the object owning the affected state. ChannelKind is gone. Following
client-sdk-android's DataChannelManager, the drain forwards what it does
not own back to its creator — received bytes and readiness — via two
closures set at construction.
DataChannelWrite.swift write phases + the SendStage protocol
DataChannelSendStages.swift LossyStage, ReliableStage, DataTrackStage
DataChannelDrain.swift queue, overflow policy, event loop, delegate
DataChannelPair.swift 769 -> 294 lines: encryption, receive dedup,
the open latch, and nothing else
SendStage.Input is what lets one drain serve both shapes of work: a
Livekit_DataPacket for the pair, a frame's packets for data tracks. One
input becomes one group of writes, dispatched in order and never
interleaved, which is the frame-atomicity invariant the data-track sender
needs — so the same queue serves park-in-FIFO and drop-oldest, differing
only in what happens on overflow.
Preserved deliberately:
- Sequence stamping stays inside the single consumer (ReliableStage.
prepare), so the sequence on the wire still matches the order writes
reach sendData.
- The size guard still runs at submit time, so an oversized packet still
fails its caller immediately rather than when it reaches the head.
- Only the last write of a group carries the continuation, so a submitter
resumes once its whole input is handed over. For the pair, where a group
is always one write, that is exactly the old behaviour.
- The retry buffer still survives reset() and the sequence still restarts
at 1 — a pre-existing bug, left alone so this stays a refactor.
Verified: 23 unit tests plus the reconnect/replay E2E matrix (6 modes) and
concurrent-exactly-once, 12 E2EE data-channel tests, and 66 tests across
streams, RPC and room suites.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The frame sender was down to a queue plus its overflow policy, which is exactly what DataChannelDrain owns. Data tracks now use a third drain — DataChannelDrain<DataTrackStage> with .dropOldest — and the sender, its DataTrackSendChannel seam, and the liveKitWebRTC queue confinement all go away with it. The drain also owns the publisher channel's delegate slot, so DataTracks' own LKRTCDataChannelDelegate now only ever sees the subscriber channel and the `dataChannel === publisherChannel` checks in all three callbacks are gone. onPacketsAvailable submits straight to the drain instead of hopping onto liveKitWebRTC first: the drain's event loop already serializes, so the hop was doing the same work twice. SendStage.Input is what makes one drain serve both: a frame's packets are one input, so they become one group, dispatched in order and never interleaved — the frame-atomicity invariant, now a property of the queue rather than of the data-track sender specifically. Data tracks keep maxMessageSize at 0 (no guard), as before: frames arrive from Rust already packetized to fit. Tests: DataTrackFrameSenderTests becomes DataChannelDrainTests, all 8 cases carried over against the drain — drop-oldest eviction, whole-group atomicity, no-interleaving, attach-clears-stale, rejected-send, empty batch, closed-channel queueing, and metered streaming of an oversized group. Two changes there worth noting. The fake channel keeps its state in StateSync rather than a lock of its own, per AGENTS.md. And the assertions poll to a deadline instead of sleeping a fixed 30 ms: the drain's loop is a Task, and a fixed sleep flaked once when the E2E suites had the cooperative pool busy. Verified: 8 drain tests, 44 data-track tests across 8 suites (twice, for the flake), and 57 across the data-channel, stream, RPC and room suites. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
reset() restarted the sequence at 1 but kept the retained writes, so a resume could replay writes stamped 1…N into a session that had never seen them. Only a resume asks for a replay, and reset() runs on teardown and full reconnect — never on a resume — so those entries could not be legitimately used again. Checked against the siblings before picking a rule, since both are self-consistent and Swift was neither: - client-sdk-js clears the buffer, the sequence and the receive state together in cleanupPeerConnections(), called from close() and from restartConnection() — the full-reconnect path — but not on a resume. - client-sdk-android keeps all three across a full reconnect (closeResources touches only the channels) and clears them in close(), at session end. Swift already resets the sequence and the receive state on teardown, so following js is the smaller change: the buffer and the sequence now share a lifetime. ReliableStageTests covers the stage directly — no channel needed for either sequence stamping or the replay set. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
dispatch() called flow.didSend() after channel.send(), leaving a window where the transport holds bytes the mirror has not counted. A drain report landing in that window subtracts them from a mirror that has never seen them (clamped to zero), and didSend then adds them back — so the mirror ends up holding bytes the transport no longer has, and every later report reads zero. The gate stays closed with nothing left to drain: the same permanent stall as an uncleared mirror, from the opposite direction. Counting first keeps the mirror a conservative over-estimate, which is the only safe direction for a gate. A rejected send returns the bytes, since they never reached the transport. Surfaced by the ported drain tests: replacing their fixed 30 ms sleeps with deadline polling made them report drains fast enough to hit the window, and two of them stalled outright. The window is narrower against a real channel — libwebrtc would have to report a drain between sendData returning and the next line — but it is the same race. Also removes a `DispatchQueue.liveKitWebRTC.sync` hop per write: RTC.createDataBuffer blocks a cooperative-pool thread once per write, and the deleted DataTrackFrameSender avoided it for exactly this reason. The buffer is a plain container; the helper's queue serializes factory access, which constructing one does not need. While here: the queue's manipulations move onto WriteQueue itself (promoteIfIdle, advance, removeAll, dropInFlight), which is what the group-atomicity invariant actually is, and shortens dispatch() and the discard path to match. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…s that asserted nothing Simplification: - DataChannelDrain.label was stored but never read; the unexpected-close log uses the channel's own label. Dropped the field, kept the init parameter that names the flow. - wake() had one caller, in tests. A zero-byte drain report re-runs the queue just as well. - ReliableStage.retryFloor was held only to rebuild the buffer in reset(); RetryBuffer.removeAll() instead. - Event.discard(error:failing:) selected behaviour with a Bool. Split into .fail(Error?) and .discardStale, so "discarding without failing, but with an error" stops being expressible. Concurrency: - Two paths dropped writes without settling their continuations: eviction under .dropOldest, and .discardStale on a channel swap. No submitter passes a continuation to a drop-oldest drain today, but that is a convention rather than something enforced, and a dropped continuation strands its caller forever. Both settle now. - eventLoopTask documents why it cannot be a `let` (subscribe needs a fully initialized self, which rules out pre-super.init assignment) and that it is assigned once before the drain escapes. Tests: - `#expect(!waiter.isCancelled)` asserted nothing in two tests — nothing cancels those tasks, so it was always false. They now track resumption through a StateSync flag, which actually pins "a waiter stays suspended". - Positive headroom tests no longer sleep before draining: the latch is sticky, so a drain landing before the waiter suspends resolves it anyway. Deterministic and faster. - until() throws on timeout instead of only recording, so a polling loop stops at the first stall rather than reporting one failure per remaining round (the earlier stall produced 100 issues and ran for 148s). - The poll deadline is named rather than a bare iteration count. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…veat The previous commit closed two paths that dropped writes without settling their continuations but left no check behind. Three tests now pin them: eviction by a newer group, a channel swap, and teardown. All three go through `submit(_:continuation:)` on a drop-oldest drain — the combination no production caller uses, which is exactly why it needed a test rather than a convention. Also restores a limitation the rewrite dropped: cancelling the submitting task does not withdraw an already-queued write. DataChannelPair documented this before the drain existed; the note now lives on submit(), where the continuation is taken. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
waitForHeadroom had no production caller — #1075's stream writers were the intended one, and until they exist it is speculative surface. Removing it removes the only reason the mirror needed cross-thread reads, and with them the lock, the readiness flag and the AsyncCompleter that backed the gate. BufferedDataChannel (a lock-guarded class) becomes BufferedAmountMeter (a plain struct) living in the drain's event-loop state, so the mirror is loop-confined again — structurally, the way `Buffers` was before this branch — and the drain went from two locked reads per write to one, the remaining one being the channel reference. That reference stays locked and re-read per write on purpose: hoisting it out of the loop would send to a stale channel after a mid-loop swap, and skipping the liveness re-check would turn a parked write into a failed one when a channel closes mid-drain. Both are reconnect-window regressions not worth ~20 ns. Drain reports now flow through the event loop rather than mutating the mirror synchronously, which also restores the ordering the pre-branch code had between a drain report and the retry buffer's trim. setLowWaterMark arrives the same way, so a future retune stays ordered with the writes it governs. Tests: BufferedDataChannelTests becomes BufferedAmountMeterTests (the arithmetic, now on a struct); the pair-level mirror-reset regression moves to the drain, where `attachClearsQueuedGroups` already covers it behaviourally. `#expect` cannot call a mutating member — the macro captures the value immutably — so didDrain's result is bound first. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The lossy channel queued every write and parked its sender, so a stalled lossy channel accumulated parked continuations, each holding its payload — on the channel whose whole premise is that loss is acceptable. It now uses the drop-oldest policy the data-track channel already uses. A dropped write resolves its sender rather than failing it, matching `sendLossyBytes` with bufferStatusLowBehavior 'drop' in client-sdk-js, which returns normally and counts the drop. Failing instead would make every existing `try await room.send(..., reliable: false)` start throwing under load, which is the worse break. Drops are counted and logged every 100, as in js. The threshold drops from 2 MB to 256 KiB. Since the channel now drops rather than queues, the threshold bounds send *latency* rather than memory, and 2 MB of lossy buffering is a long time to sit behind. 256 KiB is the ceiling client-sdk-js clamps its lossy threshold to (8 KiB…256 KiB, tuned to roughly 100 ms of measured throughput) — the 2 MB figure was eight times that ceiling. Reliable is untouched: it still parks in FIFO order, because its writes must arrive. BEHAVIOUR CHANGE — for the PR description: a lossy send under sustained backpressure now returns without having reached the wire, where before it blocked until the buffer drained. Callers that relied on `send` blocking as flow control on the lossy channel will see it return sooner. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds RoomDelegate.room(_:didUpdateBufferStatus:of:), the analogue of DCBufferStatusChanged in client-sdk-js and of Android's observable bufferedAmount: an app publishing data has had no way to see that a channel's send buffer is full, which is the one thing it needs in order to back off. Only transitions are reported, not every change in the amount buffered — the same contract updateAndEmitDCBufferStatus gives in js. DataChannelKind names which channel changed; the three are independent SCTP streams with different behaviour when full, so which one is congested matters. Only the publisher's channels report, since the subscriber's are receive-only. The lossy threshold stays at 2 MB rather than tracking measured throughput. Tuning it down the way js does (byterate/10, floored at 8 KiB) interacts badly with dropping: a burst of small publishes would drop packets the transport would have flushed in microseconds, and since `DataPublishOptions.reliable` defaults to false, that burst is the SDK's default publish path. Dropping stays reserved for genuine sustained overload, and the tuning machinery is left out rather than added unused. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…tach-time readiness Four review findings shared one root cause: DataChannelDrain never detached from or identity-checked a replaced channel, and never told its owner the state of a channel that was already open when the delegate landed. - The three delegate hooks now act only for the channel the drain currently owns (isCurrent), and setChannel/reset detach the replaced channel's delegate. Without this, a stale channel's late .closed — delivered from WebRTC's network thread with no ordering against a reconnect — could re-arm DataTracks' open gate after the replacement was already open, wedging every subsequent publish; and a stale drain report was subtracted from the successor's freshly-reset meter. Main had this check as `guard dataChannel === publisherChannel`; the delegate migration lost it. - setChannel now reports the channel's current state through onStateChange, so owners stop probing readyState to cover the already-open-at-attach case: DataChannelPair's manual handleStateChange() calls and DataTracks' readyState probe are gone. Any owner that forgot such a probe would hang its open latch whenever the channel opened before its delegate landed. - DataTracks teardown now resets its publisher drain, so a routine full reconnect no longer logs "closed unexpectedly" at .error for the data-track channel (nothing set wasReset on that drain before). - Buffer status now takes one route: a closure handed to the publisher DataChannelPair at construction, arriving at the same Room.notify funnel the data-track drain uses. The subscriber pair — whose channels never send — is constructed without reporting instead of being filtered by object identity downstream, and the internal DataChannelDelegate protocol loses the extra required method (also friendlier to data-streams-v2's rebase). State also drops the duplicate channel/sendTarget fields (one object, two names) and the drain's log lines carry its label, since one session runs three drains and "Dropped N writes" needs to say where. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Three review findings, one commit because they are all "an event-loop path that forgets to settle something": - Replays now append synchronously inside the .commanded event instead of routing back through the stream. The round-trip left a window where a concurrent reset()'s .fail landed between two replay yields, re-queueing writes stamped with pre-reset sequences into the next session — whose fresh packets (stamped from 1) the receiver's dedup gate would then drop as duplicates. - A rejected sendData mid-group now fails the discarded remainder of the group. A group's continuation rides its last write, so dropping the remainder unsettled would leave a caller awaiting a multi-write group suspended forever. Latent (no such caller yet), but submit is the drain's public seam and the doc comment claimed the opposite. - .fail resets the meter and no longer skips the buffer-status publication, so a drain torn down while its status was "not low" publishes the recovering transition. Without it, an app that backed off on isLow == false waited forever on a permanent disconnect — the exact backoff pattern the new delegate API documents. Also: the dispatch comment referenced a `.teardown` event that was renamed to `.fail`, and WriteQueue gained a single-write append so replays don't wrap each write in a throwaway array. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Four efficiency findings, all on the send path the DC-throughput work says is Swift-overhead-bound: - Writes now carry Data until dispatch; the LKRTCDataBuffer — whose init memcpys the payload into a CopyOnWriteBuffer, and whose .data getter copies it back out on every read — is built inside the channel seam at send time. byteCount becomes payload.count (O(1)), so the meter, the retry buffer and the trim loop stop re-copying whole payloads to read a length (~3–5 MB/s of hidden copies for a busy reliable publisher), and an evicted drop-oldest write costs nothing, as the deleted DataTrackFrameSender's queue already ensured. Supersedes the DataChannelPair hunk of #1092, which fixes the same finding on main. - dispatch() and enqueue() no longer take the _state lock per write: the send target and max-message-size are mirrored into the loop's state via .attached/.configured events, FIFO-ordered with the writes they govern. The locked copy remains for isOpen and the delegate identity guard, which read from arbitrary threads. Liveness is still re-checked per write (channel.isOpen — a cheap bypass-proxy read), so a channel closing mid-drain still parks instead of failing. - makeWrites fills a reused scratch instead of returning a fresh array, so the park (reliable) hot path allocates no group array per submit. - RetryBuffer.removeAll clears by reassignment instead of dequeuing element-by-element — a teardown at the 2.5 MB retry floor no longer does per-element bookkeeping just to discard everything. The file_length lint disable is carried over from the predecessor DataChannelPair.swift, which had the same at nearly twice the length. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Every encrypted data message reached RoomDelegate's didReceiveData(encryptionType:) as .none. Decryption applies the decrypted payload into the packet's oneof, which clears `encryptedPacket` — and the type was read off the decrypted packet, i.e. off the field decryption had just erased. Found by Devin's review of #1075, where the same read moves but the flaw predates it (present on main). The type is now captured before decryption and rides the internal delegate alongside the packet, so Room's dispatch uses what actually arrived on the wire instead of re-deriving it from a cleared field. The E2EE suite never caught this because its RoomDelegate discarded the encryptionType parameter. It now asserts every received message reports a non-.none type, which fails on all 12 cases without the fix. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…cking-call rationale - DataChannelDrain gains `send(_:) async throws`, the one audited wrapping of the resume-exactly-once contract; DataChannelPair uses it instead of re-deriving withCheckedThrowingContinuation at the call site. The awaited self also keeps the drain alive for the write's whole lifetime, which the raw continuation entry point requires of its callers. - The IUO drain properties in DataChannelPair and DataTracks document the init-ordering invariant their unsynchronized cross-thread reads depend on (assigned before self escapes to the WebRTC/FFI callback threads), so a reordering doesn't silently turn them into foreign-thread traps. - dispatch() documents why the blocking sendData proxy call is accepted on a cooperative-pool thread, and what the eventual fix is if pool starvation is ever observed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…reading exception - The lossy changeset claimed the threshold "tracks roughly 100ms of measured throughput (matching client-sdk-js)" — text left over from a tuning iteration that was deliberately dropped (an 8 KiB floor would make a burst of small publishes drop packets the transport would flush in microseconds, on the SDK's default publish path). Rewritten to describe what ships: drop-oldest past a fixed 2 MB, return-normally, counted. - The eviction comment claimed to match sendLossyBytes' 'drop' behaviour in client-sdk-js. The *outcome* matches (return normally, count the drop); which packet dies does not — js drops the incoming payload, this drain keeps the freshest. Stated as the design choice it is: freshest-wins suits the supersede-style data (cursor, presence, state) lossy carries. - publish(data:) and DataPublishOptions.reliable now say that the default lossy path may drop under sustained backpressure and point at the buffer-status delegate for backing off. Doc-only additions. - ReliableStage.reset()'s docstring kept a three-SDK release-history narrative; trimmed to the invariant (replay set and sequence share a lifetime), the history lives in the commit that fixed it. - AGENTS.md documents the data-channel exception to the liveKitWebRTC-queue rule: sendData/readyState/LKRTCDataBuffer-init are internally proxied by libwebrtc, so the queue would only serialize what is already safe. This codifies what the pre-rewrite code (and main) already did. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…onnect E2E
- LiveKitTestSupport gains poll(for:until:), a throwing deadline-based
condition wait — the drain tests' `until` was at least the sixth private
copy of that loop in the test tree, and this one also *throws* on
timeout, so a missed condition stops the test at the first stall instead
of cascading (a non-throwing copy once turned one failure into ~100
redundant issues over 148s). Older suites' private copies are left in
place; migrating them is unrelated churn.
- The drain suites replace every fixed sleep with flushEvents(), an awaited
FIFO barrier built on an existing property: an empty submitted input
resolves inside the loop after all prior events are processed. Negative
assertions ("nothing was reported") become exact instead of lenient, and
the suites shed ~1.5s of pure sleep per run. emptyBatchIsIgnored pins the
barrier property itself.
- One DrainFixture builds the drain-under-test, one fillBuffer verifies its
fill actually landed (the copy it replaces was a bare 50ms sleep that
verified nothing), and the suites that mirror named upstream behavior now
carry @test(.spec(...)) pins to rust-sdks' dc_sender.rs and
client-sdk-js's RTCEngine.ts at fixed commits, per the repo convention.
- New E2E: a full reconnect mid-burst with the reliable buffer saturated
past its 2 MB mark, then a fresh marked send that must arrive. Covers the
two teardown fixes end to end — a stale mirror stalls the marker's send
forever, and a stale-sequence replay would trip the receiver's dedup gate
against it. Also: a status-reporting case for teardown publishing the
recovering isLow transition, and the eviction/swap settlement tests now
exercise the drain's awaited send() surface.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
pblazej
marked this pull request as ready for review
August 20, 2026 09:20
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
When the transport rejected a send on a drop-oldest channel, the failed
write was settled while still at the queue's head — and dropFailedWrite's
drop-the-rest-of-the-group cleanup then removed and settled it a second
time. Double-resuming a CheckedContinuation traps ("SWIFT TASK
CONTINUATION MISUSE"), and the lossy channel is the default publish path,
so a single rejected awaited send crashed the app. Found by Devin on the
PR; introduced by the settle-what-you-drop fix, whose tests covered
rejection without a waiter and waiters without rejection, but not both.
The failed write is now advanced out of the queue before its continuation
is settled, so the cleanup can only ever reach the remainder. The park
branch's advance moves with it, making the invariant uniform: a write's
continuation is settled only after the write has been removed.
The regression test is the missing combination — a rejected send with an
awaiting caller; a double resume traps the test process, so passing is the
assertion.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
pblazej
changed the title
The double-resume Devin found was fixed positionally (remove before settling); this makes the whole class of it unreachable. A write's waiter now lives in a SendToken whose settlement is first-wins idempotent — a redundant settle attempt is a no-op instead of a "SWIFT TASK CONTINUATION MISUSE" trap — and whose deinit fails a submitter whose write was dropped without settlement, so a forgotten path strands nobody. Chosen over compile-time enforcement via noncopyable writes, which was built and green but required replacing Deque with a hand-rolled UnsafeMutablePointer FIFO (Deque/Array/AsyncStream all require Copyable elements) plus ownership annotations on every future touch of the queue: containment that makes unknown bugs benign was judged worth more than diagnostics for known ones, at ~a tenth of the machinery. The one @unchecked Sendable rides the token, justified by the loop's single-consumer invariant. RetainedWrite still has no token field at all, so the retry buffer's replay path remains structurally unable to resume anyone. SendTokenTests pins both properties: a second settle neither traps nor overrides (the exact shape of the fixed crash), and releasing an unsettled token fails its waiter with .cancelled. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Dropping the last LKRTCDataChannel reference runs its proxy destructor — PROXY_PRIMARY_DESTRUCTOR, a BlockingCall into WebRTC's signaling thread — and close() blocks the same way. The drains held a channel mirror in their loop state and dropped it in-loop on .fail/.attached/reset, so when an audio-engine teardown wedged the worker thread on overloaded CI simulators (signaling blocks on worker), six loop threads blocked inside channel destructors, the width-limited cooperative pool exhausted, and the whole test process deadlocked — every simulator job timed out at 30 minutes while macOS passed. Diagnosed from a process sample of a local full-bundle repro plus the CI artifacts (client log ends in minutes of "HALC_ProxyIOContext: skipping cycle due to overload"; server log shows only sustained high CPU). parkChannelRelease now hops every close and final release to DispatchQueue.liveKitWebRTC *asynchronously* — the queue may block, the pool must not — at all drop sites: drain reset, channel swap, the loop's .attached/.fail mirror replacement, and DataTracks' subscriber-channel displacement (main-inherited, same hazard). A re-sample after the fix shows zero drain frames in the wedge; the residual cycle is pre-existing main code (Transport.close on a cooperative thread, liveKitDefault()'s sync onto the RTC queue) plus the audio-stack root, both out of scope here. AGENTS.md's data-channel exception gains the rule this taught: proxied calls being thread-safe does not make them pool-safe — blocking calls, including the deinit nobody audits as a call, belong on the queue. Main was exposed to the same hazard at smaller width (reset closed and released on the caller's cooperative thread, and channel swaps released inside _state.mutate under the lock); the branch is now stricter than main on this axis, not just restored. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
dispatch() promoted the pending group into the in-flight slot before checking channel readiness, so on an attached-but-still-opening channel a drop-oldest group got stranded where eviction (which only inspects the pending slot) could not reach it — and once the channel opened, the stale group shipped ahead of the newer one that should have replaced it. That violates freshest-wins exactly in the connect/reconnect window, and regressed the deleted DataTrackFrameSender, whose pump returned before touching the evictable slot on a closed channel. Found by Devin on the PR. Readiness is now checked before promotion (a no-op for .park, which never uses the pending slot); the failure path's remove-before-settle ordering is unchanged. The regression test pins it: two groups queued against a closed channel, only the newest ships on open, and the evicted one never trails in. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Resolves the DataChannelPair.reset conflict with #1097: the negotiated max-message-size now lives in the drains, so the per-session re-default is ported as set(maxMessageSize: defaultMaxMessageSize) after the drains reset — same event stream, so it is ordered after the .fail. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Extracts the flow control that
DataChannelPairandDataTrackFrameSendereach grew independently into one per-channelDataChannelDrain<Stage>, then fixes what the extraction and a multi-angle review surfaced.Architecture
Each drain owns one channel's queue, buffered-amount mirror, overflow policy and per-kind
SendStage— and is that channel'sLKRTCDataChannelDelegate(identity-guarded against replaced channels), so nothing dispatches on channel labels anymore.flowchart TB subgraph ROOM["Room"] NOTIFY["Room.notify(bufferStatus:of:) — single funnel"] RD["RoomDelegate.room(_:didUpdateBufferStatus:of:)"] subgraph PAIR["publisherDataChannel: DataChannelPair — E2EE, receive dedup, open latch"] LOSSY["DataChannelDrain<LossyStage><br/>.dropOldest · 2 MB mark<br/>serialize only"] REL["DataChannelDrain<ReliableStage><br/>.park · 2 MB mark<br/>sequence stamping + retry buffer"] end SUB["subscriberDataChannel: DataChannelPair<br/>receive-only — constructed without buffer reporting"] subgraph DT["DataTracks"] TRACK["DataChannelDrain<DataTrackStage><br/>.dropOldest · 8 KiB mark<br/>frames pre-packetized by Rust"] end end LOSSY ---|"is delegate of"| CH1[("_lossy")] REL ---|"is delegate of"| CH2[("_reliable")] TRACK ---|"is delegate of"| CH3[("_data_track")] SUB --- CH4[("_lossy / _reliable (sub)")] LOSSY -. "isLow transitions" .-> NOTIFY REL -. "isLow transitions" .-> NOTIFY TRACK -. "isLow transitions" .-> NOTIFY NOTIFY --> RDInside a drain, every mutation is serialized through one FIFO event loop — no locks on the per-write path:
flowchart LR SUBMIT["submit(input) /<br/>send(input) async"] --> LOOP CB["delegate callbacks<br/>drained bytes (delta) · state"] -->|"isCurrent guard"| LOOP subgraph LOOP["AsyncStream FIFO event loop — single consumer"] PREP["SendStage.prepare<br/>reliable: stamp sequence in-loop<br/>(wire order == FIFO order)"] GUARD["max-message-size guard"] Q["WriteQueue<br/>.park: unbounded FIFO<br/>.dropOldest: newest group wins,<br/>evicted waiter resolved"] METER{"BufferedAmountMeter<br/>pending ≤ low-water mark?"} PREP --> GUARD --> Q --> METER end METER -->|"yes"| SEND["channel.send(Data)<br/>LKRTCDataBuffer built here —<br/>evicted writes never pay the copy"] METER -->|"no"| PARK["wait for the next<br/>drained report"] SEND --> RETAIN["stage.didDispatch<br/>reliable: retain for resume replay"]vs. the other SDKs
DCBufferStatusChangedeventbufferedAmountDrop-oldest over js's drop-incoming is deliberate, not a porting gap: js's behaviour is an artifact of having no app-level queue at all, and it keeps stale data flowing while discarding the fresh update. The lossy channel's typical cargo — cursor positions, presence, game-state deltas — is supersede-style, where the newest payload makes its predecessor worthless, so freshest-wins is the policy that matches (and what rust-sdks' frame sender already does). Either way loss only starts past the threshold; which packet dies is the only difference.
Fixes (each with a regression test)
encryptionTypeas.none, because decryption clears the field the type was read from (found by Devin on Data streams v2 #1075; predates it — the E2EE suite now asserts the type).SendToken(first outcome wins, an unsettled drop fails its waiter fromdeinit), so the double-resume crash Devin caught on the rejected-send path is unreachable, not just avoided.Behaviour changes
openCompleterstill requires both.publish(data:).New public API
RoomDelegate.room(_:didUpdateBufferStatus:of:)+DataChannelKind: transition-only backpressure reporting, the analogue of js'sDCBufferStatusChanged.Notes
DataChannelPairhunk of Skip avoidable copies on the data packet and data stream paths #1092 (same byteCount finding, fixed on main); its other hunks don't overlap.liveKitWebRTC-queue rule (those calls are internally proxied by libwebrtc — codifies what main already did).