Data streams v2 by 1egoman · Pull Request #1075 · livekit/client-sdk-swift

1egoman · 2026-07-30T18:11:55Z

Reimplements data streams on the Rust livekit-data-stream core over UniFFI (livekit-uniffi-xcframework 0.1.11), replacing the Swift v1 implementation while keeping every public signature. Originally an LLM-driven migration; since rebased onto the current main and reworked through review.

Architecture

DataStreams owns the topic→handler registry and packet routing. It is Room-scoped — handlers are registrable before connect and must survive a reconnect — so it is staged as IdleDependencies, the room tier of DependencyStage.

Both FFI managers are connection-scoped, held by ConnectionDependencies: the incoming one because its payload cap is fixed at construction from the options connect was given, the outgoing one because releasing it closes the writers opened on that session.

Three weakly-linked delegate shims bridge the core: incoming (stream opened/closed → the registry), outgoing (encoded packets → Room.send(dataPacket:), so E2EE, reliable sequencing and identity stamping are unchanged), and a participant registry the core queries to pick framing. Public types — StreamInfo, StreamOptions, readers, writers, StreamError — are thin bridges over internal import LiveKitUniFFI, and the SDK's own consumers take a TextStreamReading protocol so the reader ships no test-only backing.

Wire behaviour

One-shot sends are DEFLATE-compressed and inlined into a single packet when every recipient advertises CAP_COMPRESSION_DEFLATE_RAW and protocol v2, falling back to legacy multi-packet framing otherwise; incremental writers are never compressed or inlined, since their content is unknown up front. Text handlers may opt into wire order, where a stream that opened after another closed runs after it while overlapping streams stay concurrent — transcription needs that, RPC must not pay for it.

Public API

New surface: RoomOptions.dataStreamOptions (maxPayloadByteLength), compress on both option types, Participant.capabilities, and ClientProtocol.v2. Every change is source-compatible for Swift callers; the three @objc initializers that gained a defaulted parameter do change selector, because Objective-C has no default arguments.

Fixes made during review

Sends fail once the room starts tearing down, instead of parking on a drain no channel will ever attach to.
Ordered dispatch is keyed by a per-handler token, so a sender reusing a stream id can neither evict its predecessor nor be evicted by it.
Participant.capabilities and clientProtocol are mirrored into participant state, rather than read off the mutable signal message from FFI threads.
Inbound packets reuse the bytes they were decoded from instead of being re-encoded.

Verification

417 tests in 85 suites pass against a local server, all five platforms build, and the library-evolution build is clean. api-check reports the four widened signatures and no breakage.

1egoman · 2026-08-04T21:06:57Z

I've tested this and everything seems to work for me as best as I can tell. I think it's ready for a proper review cc @pblazej

pblazej · 2026-08-05T08:38:28Z

@1egoman some of the AI comments apply to the data tracks as well (e.g. out-of-order things), I'll try to address them in some systematic way in both.

pblazej · 2026-08-07T12:10:12Z

@1egoman tl;dr it's fine if you leave it now.

I started thinking about oustdanding comments, will focus on test coverage/churn and consistency with data tracks (uniffi in general).

github-actions · 2026-08-17T13:10:02Z

⚠️ This PR does not contain any files in the .changes directory.

…yload cap Addresses review feedback on PR #1075. Every public member of `DataStreamOptions` now carries a docstring, per AGENTS.md. A negative `maxPayloadSize` reached `UInt64(_:)` at the FFI boundary and trapped the process on the first inbound packet. Non-positive values are normalized to `nil` at construction — the built-in cap — so the conversion can't trap, which also keeps the crash out of a consumer-supplied value. Adds the release changeset the PR was missing. 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>

… session The FFI emits one packet per `onPacketsAvailable` call, synchronously and in order. Answering each with `AsyncSerialDelegate.notifyDetached` spawned a task per packet that raced to the serial runner, so emission order was preserved only by timing: measured on the primitive, ordering breaks in 20/20 runs at zero inter-call spacing and 3/20 at ~50us. The receiver drops a chunk that arrives before its header and fails the stream on a non-consecutive index, so drain the callbacks through a single ordered task instead. The incoming manager's payload cap is fixed at construction, so memoizing the manager for the Room's lifetime pinned it to the first connect's value. Discard it in `reset()` and let the next session rebuild it; it holds no handler state. Also take the read fast path when it already exists, keeping the exclusive lock off the per-packet inbound path. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…ror bridging The `ordered` text-stream contract is still implemented in Swift — only chunk assembly moved to the Rust core — so its five specs are re-pointed at the `DataStreams` coordinator rather than dropped. Four pass. `orderedTopicDoesNotDelayOverlappingStreams` does not, and is kept disabled as the specification of the difference: v1 chained a newly opened stream behind handlers of streams that had already closed, while `DataStreams` chains on the order streams opened in, so a stream that stays open head-of-line-blocks later streams from the same sender. Restoring that needs a stream-closed signal the FFI does not surface. `ByteStreamInfoTests`/`TextStreamInfoTests` covered protobuf to `StreamInfo` conversions that no longer exist; their FFI replacements ran untested. Pin every field mapping, the millisecond timestamp scaling, the empty-name-to-nil rule, the operation-type cases, and the twelve-case error mapping. The ObjC options suite is dropped: adding `dataStreamOptions:` changes RoomOptions's ObjC initializer selector, and that break is accepted rather than pinned by a test. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…review The outgoing delegate's stored properties are all immutable and Sendable, so it conforms plainly rather than `@unchecked` — no invariant left for a reviewer to take on trust. Same for the incoming test suite. Document why the pump is unstructured, and drop the `defer`-in-`mutate` trick in `reset()` for a plain take, which does not need evaluation-order reasoning to read. `orderedTopicDoesNotDelayOverlappingStreams` moves from `.disabled` to `withKnownIssue`: it now compiles, runs, records the two divergences with their actual values, and fails if the behavior is ever fixed, instead of silently rotting. Bounded to a 3s wait since the first expectation is meant to time out. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The Rust core already emits `incoming::OutputEvent::TrailerReceived` with the stream id, sender and topic; the UniFFI layer drops it and forwards only `StreamOpened`. Surfacing that event is the fix for the ordered-topic divergence, not re-parsing trailer packets on the Swift side. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

rust-sdks#1286 addressed the gaps raised in review; take them up: Stream closes are now reported (`onStreamClosed`), so ordered text topics go back to v1 semantics — a newly opened stream waits on handlers whose streams have already *closed*, not on whatever opened before it. A stream that stays open no longer head-of-line-blocks later streams from the same sender, which re-enables `orderedTopicDoesNotDelayOverlappingStreams`. Entries are dropped when a handler returns, so neither map grows without bound. `handlePacketReceived` takes the wire encryption type. Decryption consumes the packet field that carried it (`EncryptedPacket` shares a `oneof` with the stream payload), so `DataChannelPair` captures it beforehand and passes it alongside. That revives the core's header/chunk mismatch guard, which could not fire while every packet was reported as unencrypted, and lets inbound `StreamInfo` report the stream's real encryption type instead of the room's configured one. `EncryptionTypeMismatch` now carries both types, so the public error stops fabricating `.none`/`.none`. `onPacketsAvailable` is throwing. Its contract — return once the packets reach the transport — can't be met from a synchronous callback when every send path is async, so the ordered pump still acknowledges early; noted in place. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…ures Prototypes the async-delegate change on the Rust side (a follow-up to rust-sdks#1286): `on_packets_available` becomes an `async fn` on the foreign trait, which uniffi supports and which generates `func onPacketsAvailable(packets:) async throws` here. That makes the FFI's stated contract implementable. The core awaits the call before pumping the next packet, so emission order holds without a Swift-side pump — the AsyncStream and its drain task are gone. The originating `write`/`send_*` stays pending until the packet reaches the transport, so a producer can no longer outrun it. And a failed send throws `PacketDeliveryError`, which fails that call and closes the stream. Covered by a test that was impossible to write before: after the room disconnects, `write` throws and `isOpen` reports false, where both previously reported success on a stream that could not be written. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The FFI now exposes `open_stream_count`, restoring the introspection v1 had on its manager. The abort-path tests were inferring "stream is open" from their handler having been dispatched, which measures a different thing and would stop being equivalent if dispatch moved relative to registration. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The core now holds trailers to their stream's encryption type as well as chunks (rust-sdks 1da01b49), and the wire type reaches it from `handleIncoming`, so the v1 behavior this PR dropped is reachable again — and now covers the trailer path the v1 Swift implementation also checked. Parameterized over which packet downgrades, since merging trailer attributes is the more interesting of the two: that is how an unencrypted peer could otherwise close someone else's encrypted stream and inject attributes on the way out. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…yload cap Addresses review feedback on PR #1075. Every public member of `DataStreamOptions` now carries a docstring, per AGENTS.md. A negative `maxPayloadSize` reached `UInt64(_:)` at the FFI boundary and trapped the process on the first inbound packet. Non-positive values are normalized to `nil` at construction — the built-in cap — so the conversion can't trap, which also keeps the crash out of a consumer-supplied value. Adds the release changeset the PR was missing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…lers by token Three defects the v2 migration left reachable: - `Room.send(dataPacket:)` parked forever once the room disconnected. The drain queues a submitted write until a channel attaches and nothing else gates it after `cleanUp`, so every outgoing stream `write` after a disconnect hung instead of failing. `OutgoingDeliveryTests` went from a hang to 4s. - Ordered-topic bookkeeping was keyed by stream id, so a sender reusing an id let the first handler's completion erase its successor's entry; later streams then stopped waiting for it and were delivered out of order (`["a", "c", "b"]`). Keyed by a per-handler token now, with the three maps collapsed under one lock since an open, a close and a completion each touch more than one of them. - `StreamByteOptions.totalSize` reached `UInt64(_:)` at the FFI boundary and trapped on a negative value; normalized at construction, matching `DataStreamOptions.maxPayloadByteLength`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The registry resolved a participant by scanning `remoteParticipants` for a matching identity string, on a dictionary already keyed by identity. The outgoing path asks once per send. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…ility `ConnectionParamsTests` only checked that the client emits `capabilities` on both connection paths. Compression engages only if the server mirrors it onto `ParticipantInfo`, which the SDK never asserted — so the whole feature could be dark and every test still green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The three bridging switches over `LiveKitUniFFI` enums were exhaustive over today's cases. Those enums are resilient under `BUILD_LIBRARY_FOR_DISTRIBUTION=YES`, so the Release build every consumer of the xcframework gets — and the api-check job — failed to compile: error: switch covers known cases, but 'OperationType' may have additional unknown values `@unknown default` on each, matching the data-track bridging already on main. An unrecognized encryption type reports as `.custom` rather than `.none`: a scheme this SDK predates is still a scheme. Also ports v1's `reusedStreamIDAfterChunkErrorDeliversNextStream`, the one behavioral spec the migration dropped without a replacement. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…rotocol in RPC Two shapes existed only to serve consumers the production types shouldn't have known about. `compress` was `Bool?` with `nil` meaning "SDK decides", but the core resolves it as `options.compress.unwrap_or(true)` at all three call sites — the third state carried nothing. As a plain `Bool` it is also Objective-C representable, which matters: `Bool?` had removed the `StreamTextOptions` initializer from the generated interface entirely, leaving the type unconstructible there. `compress:` is now in the selector, and since Swift defaults are invisible to ObjC those callers must supply it. `TextStreamReader` carried a second, in-memory backing — a `Backing` enum, a branching `readAll`, and a dual iterator — shipped to every app so that four RPC test factories could inject payloads. The RPC managers only ever read `info` and `readAll()`, so they now take a `TextStreamReading` existential and the stand-in lives in the test target. Deletes the enum, both branches, and `StreamReaderSource`, and collapses the four factories into two plus one stub. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Every inbound stream packet was decoded by `DataChannelPair`, re-encoded by `DataStreams.handleIncoming` to hand the core its wire form, then decoded again in Rust — three passes where v1 did one, and nanopb does not retain the bytes it decoded from. The internal `DataChannelDelegate` now carries them alongside the packet, so the plaintext path forwards what it received. `nil` after decryption: the payload oneof has been rewritten, so the received bytes no longer describe the packet and it has to be re-encoded. That branch is what the unit suites exercise; `DataStreamTests` covers the reuse path with real bytes off the wire. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The subsystem had three lifetimes hiding in one object. The handler registry is the Room's — handlers are registrable before connect and must survive a reconnect — but both FFI managers belong to a single connection, and each was faking that with an idiom the dependency stage exists to replace. The incoming manager's payload cap is fixed at construction from the options passed to `connect`, so it was built lazily on the first inbound packet and nil'd by `reset()`. A packet racing that reset installed a replacement nothing aborted, which then survived into the next connect with the previous session's cap. It is now a `let` on `ConnectionDependencies`: no stage, no manager, so a late packet has nowhere to land. The outgoing manager holds a `DropGuard` on the task draining its packets, so releasing it closes the writers opened on that session. Kept for the Room's lifetime, a writer retained across a reconnect sent its next chunk into a session whose receivers never saw the header — the write succeeded and the payload vanished. Scoped to the connection, that write fails instead. `DataStreams` itself is staged as the room-scoped tier every stage carries, so `Room.dataStreams` loses its implicitly unwrapped optional. Its back-reference moves to `attach(room:)`: the tier is built in `Room.init`'s first phase, where `self` does not exist yet, and nothing can reach the room until init returns. `begin` checks the tier's identity so the copies each payload carries cannot disagree. `cleanUp` also aborts open incoming streams after the transports are down rather than before, so no late header can open a reader that outlives the abort. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Registration published the handler and its ordered-topic flag through two separate locks, so a stream opening between them read the handler but not the policy: it took the unordered path and never entered the ordered maps, letting a later stream overtake it. Both now live in one registry entry, which dispatch resolves with a single read. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

`info` is a plain mutable property that signaling replaces wholesale, while the outgoing data stream registry reads `capabilities` and `clientProtocol` from the FFI's own threads on every send, to decide compression and framing. Nothing synchronized the two, and replacing a nanopb value can free the storage the reader is walking. Both are mirrored into `Participant.State` inside the mutate that already applies the rest of the message, and the getters read that instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

0.1.11 is published in livekit/podspecs now, so the podspec no longer has to lag the package at 0.1.9. The SDK's app delta measures 17.43 MB — the Rust data stream core takes RustLiveKitUniFFI to 1.40 MB — so the budget moves to the next half-megabyte. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

`disconnect()` moves the room to `.disconnecting` before `cleanUpRTC` resets the drains, so a send that raced it passed the `.disconnected`-only guard and parked on a drain no channel would ever attach to. It did not fail — it stayed suspended for the process's lifetime; the test added here hangs without the guard. `isTearingDown` already draws exactly this line, and `.connecting` / `.reconnecting` keep parking by design. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The digester keys a function by its argument labels, so adding a parameter reads as a decl removed and another added — or as a rename, when it manages to pair them itself. Neither is a break a Swift caller can hit when every gained parameter carries a default: the existing call compiles unchanged. Those pairs move to their own section instead of failing the job, matched structurally rather than by allowlist: same context and base name, and dropping the parameters marked `hasDefaultArg` has to leave the old signature exactly, labels, types and result included. Anything else the digester reports still fails, a widening whose new parameter has no default included. What the pair does cost is stated with it: the mangled symbol moves, so a consumer linking a prebuilt binary rebuilds, and an `@objc` member's selector changes with it — Obj-C has no default arguments, so its callers must pass the new one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

swift-sh compiles one file, so the tool can't be split across several, and it is mostly prose about what the digester reports and why. Contorting it to fit costs the explanations, not complexity — so it says so and opts out. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

`isOpen` is an `async` property, and under Swift 6.1 — the floor, and one half of the CI matrix — `#expect` expands its argument into a context that cannot await, so `await #expect(writer.isOpen)` fails to compile there. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

1egoman force-pushed the data-streams-v2 branch from 907406e to f0590aa Compare July 30, 2026 19:41

1egoman marked this pull request as ready for review August 4, 2026 21:06

1egoman requested review from hiroshihorie, pblazej and xianshijing-lk as code owners August 4, 2026 21:06