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

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

Summary

This PR moves data streams from the Swift v1 code to the Rust livekit-data-stream core through LiveKitFFI 0.13.1. It builds on #1162, which renamed the Rust bindings to LiveKitFFI and is now merged into main. Existing public signatures keep working for Swift source, Obj-C callers, and apps that are already built.

Design

DataStreams keeps the topic handlers for the life of the Room, so handlers that you register before connect survive reconnects. Each connection owns its incoming and outgoing Rust managers, and a full reconnect replaces both, so old writers fail. Packets still go through Room.send(dataPacket:), and one-shot sends use single-packet framing only when every recipient supports protocol v2.

Public API

The new API is RoomOptions.dataStreamOptions, one-shot LocalParticipant.sendBytes, Participant.capabilities, TextStreamInfo.mimeType, and ClientProtocol.v2. The stream options add compress, operationType, generated, and senderIdentity, and one-shot sendText can attach files. StreamError gets one new case for each Rust error, and four initializers keep their old signatures, so api-check reports no breaking change.

Hardening

A stream length over Int.max reads as unknown, writeToFile accepts only a plain file name, and an inline stream must match its declared length. When you cancel a task, its read, readAll(), writeToFile, send, write, or performRpc call ends with CancellationError. A receive limit of zero now stays zero, where before it became 5 GB.

Testing and parity

Lint, format, five platform builds, the library-evolution build, api-check, and the size check pass, with SIZE_BASELINE_MB raised from 17.7 to 17.8. All 566 enabled LiveKitCoreTests, with rooms that mix v1 and v2 clients, and all 20 LiveKitObjCTests pass against a local server. Swift matches Android #997 and the other SDKs, except trailer attributes, progress, text stream length, and reader close, which need a new LiveKitFFI.

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>

Android #997 checks that the info reports how a stream arrived, that a plaintext stream in an encrypted room reads as plaintext, and that a stream encrypted throughout reads normally (DataStreamsV2ReceiveTest). The new test feeds an inline stream as plaintext and as GCM, in a room with and without encryption, and expects the info to report how it arrived. A second test feeds a GCM header, chunk and trailer and reads the text. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

A read cancelled while it waits returns at once and leaves the core's read in flight. The new test cancels a readAll() on a chunked stream, then feeds the rest and the trailer, and expects the coordinator to report no open stream. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Android #997 checks the capability list a participant reads from its info (ParticipantTest): empty by default, both known values, unknown values dropped, and cleared by a later update. Swift had no test. The new tests build a RemoteParticipant from info with no capabilities, both known ones, and the placeholder plus an unknown value plus one known one, and expect the empty list, both, and the known one. A second test updates the participant with info that has none and expects the list to clear. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Android #997 checks that readAll of an empty stream returns nothing and that reading past the end does not return more (StreamReaderTest). The new tests, for both readers, read an empty stream with readAll() and expect an empty result, and iterate a stream to its end and expect the next call to return nil again. The text reader's test helper takes the chunks to send. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Android #997 checks three timings of a v2 RPC call against a v2 remote (RpcV2MockE2ETest): a response right after the publish, a response that arrives while the request is still being published, and an ack and response that arrive after the connection timeout. Swift's timing tests used a remote it did not know, which takes the v1 packet path. The new tests install a v2 remote. The first responds from the after-publish hook to two calls in a row and expects each call's own response, two request IDs and nothing pending. The second injects the ack and response from the mocked data channel as soon as the request stream's header goes out, and expects that response. The third fires the ack watchdog from the hook, expects connectionTimeout, then delivers a late ack and response and expects nothing pending. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Android #997 checks that five concurrent v2 calls each get the response to their own request (RpcV2MockE2ETest caller_five_concurrent_calls). Swift's concurrency test used the v1 path and checked only that the request IDs were distinct. The new test installs a v2 remote and records each request stream's payload by request ID from the mocked data channel. After each publish the remote answers with that payload. Each of the five calls expects the response to its own payload, and nothing is left pending. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

The framing tests run against the core with no server. The new e2e test sends a 315 KB repetitive payload from one room to another through the local server, which two v2 SDKs send compressed and in chunks: as text with sendText, and as bytes with sendBytes, sendFile and a byte stream. The receiver reads back exactly what was sent. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

The cap tests feed packets to the coordinator directly. The new e2e test connects a receiver with a 1000-byte cap and sends from another room through the local server: 500 bytes of text read, and 5000 bytes fail with .payloadTooLarge. RoomTestingOptions takes DataStreamOptions, and withRooms passes them to each room. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

App Size measured 17.71 MB at 71abab8 against the 17.70 MB budget. Compiled Swift grew from 4.43 MB to 4.51 MB with the data stream API this PR adds (one-shot sendBytes and sendText attachments, sender identity, operation type, generated flag, and the forwarding option initializers). The frameworks did not change: LiveKitWebRTC 11.78 MB, RustLiveKitFFI 1.41 MB. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

cancelledReadAllLeavesRestOfStream cancelled its task right after the task started, so the cancel could land before the read reached the core. It then passed on the old whole-stream readAll() too. cancelledReadLeavesNoOpenStream had the same gap, so it did not show that an in-flight read was abandoned. Neither put a timeout on the cancelled task or on the reads that follow. The public readers now hold the core's reader by its generated protocol (ByteStreamReaderProtocol, TextStreamReaderProtocol) instead of the concrete class. That lets a test wrap the real reader in a stand-in that reports when next() or readAll() enters the core; nothing else changes. The cancel tests move to ByteStreamReaderCancellationTests.swift with the writeToFile and waiting-read cancel tests, cancel only after that report, and bound every wait to 5 s. The open-count test now uses the FFI manager's count directly. With readAll() put back to the core's whole-stream read, cancelledReadAllLeavesRestOfStream fails: the rest of the stream reads as 0 bytes. With the current readAll() it reads at least two of the three chunks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…rator Each AsyncChunks iterator counted the bytes it returned and checked them against the declared length when the stream ended. readAll() makes its own iterator, so a valid stream read partly with next() and finished with readAll() failed with .incomplete; the review reproduced it with inline "abcd". A chunk that a cancelled read took was not counted anywhere either. The count now lives on the public reader, in a StateSync, and every iterator and readAll() go through one nextChunk() on it. The count grows inside the operation that abandoningOnCancel runs, so a chunk an abandoned read takes still counts. Every read that reaches the end of a short or long stream reports it, not only the first, so a readAll() after an iterator saw the end still fails on a short stream. New tests: a valid stream, inline or chunked, bytes or text, read with next() and then readAll(), returns everything; a short inline stream read the same way still fails with .incomplete; and a next() cancelled while it waits in the core, then readAll() of the rest, completes. With the per-iterator count, all four split-read cases and the cancelled case fail with .incomplete; the short-stream case passes both ways. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

… file The attachment sends ran in a task group that ended with waitForAll(), which waits for every child even after one has thrown. A file that could not be read therefore failed the call only once every other file had finished, and the review showed sendText hanging while another file's delivery was blocked. Drain the group with `for try await _ in group {}`. The first error leaves the group, which cancels the other sends; they return at once through abandoningOnCancel. The docstrings say so. The attachment tests move to LocalParticipantAttachmentTests.swift. Two new tests use a delegate that holds every file header in the transport: a missing file next to a held one fails the call with .fileInfoUnavailable within 5 s, and cancelling a call whose files are held ends it with CancellationError. Before the fix the first test is still waiting after 5 s; the cancel test passes both ways. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

StreamError pins its NSError codes through CustomNSError but left the user info empty, so Objective-C saw "The operation couldn't be completed" and lost the UTF-8 decoder's message, the file error and the sender's reason that the Swift cases carry. Add errorUserInfo: NSLocalizedDescriptionKey names the case with its values, and NSLocalizedFailureReasonErrorKey holds the detail for .abnormalEnd, .invalidUTF8 and .io. The codes do not change. The new NSError bridge test reads the code, the failure reason and the description of each of those three cases. Before the change the failure reason is nil and the description is the generic one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

writeFailsAndClosesStreamWhenTransportIsGone expected .internalError for a write after the room disconnected. The write races the teardown: it finds either the closed request channel of the outgoing manager the disconnect released (Internal) or the delivery delegate of a room that is gone (SendFailed). A run here got .sendFailed. Accept either; the writer still has to report itself closed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…ng-publish test v2ResponseArrivesDuringPublish injected the ack and response from an unstructured task started by the mocked channel, whose send returned at once. The response could therefore land after the publish completed, and the test did not cover a response during publish. A mocked channel subclass now awaits a handler in send(), and the handler processes the ack and response when the request header goes out, so they are handled before that send, and the publish, return. The after-publish hook records whether the response was already in, and the test expects it was. With the old mock the new assertion held in one run here; the race is real but not reproducible on demand. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…eck large payload framing The room tests all ran between rooms with the default protocol, so the mixed v1/v2 cases were covered by fixtures only. The large-payload test used repetitive text that compresses from 315,000 bytes to about 800, so one-shot sends went inline, and it asserted nothing about framing: it passed with compression turned off. A test-only recorder (DataStreamPacketRecorder.swift) swaps a connected room's outgoing manager for one whose delegate records each packet and then hands it to the room's own delegate, so a test against the server can see the framing the core picked. It can also tell the core that every recipient is a v1 client, which makes the room frame its streams as an SDK that predates v2 does. The text and byte round trips become one test over the peers and the five ways to send. Peers come from RoomTestingOptions.clientProtocol: v2 to v2; v2 to a receiver advertising v1; a sender advertising v1 and framing as v1, to v2; and a v2 broadcast to a v1 and a v2 receiver. Each receiver reads the payload, and the header is inline only for an in-memory one-shot send with no v1 peer; with a v1 peer it is not compressed either. A file is never inline. With the room's registry reporting every recipient as v2, the v1-receiver and mixed cases fail. The v1 sender is this SDK advertising v1 with the recorder's legacy framing. The core picks framing by recipient, so the advertised protocol alone does not make an old sender; the recorder supplies the framing an old SDK would use. The large-payload test now sends 200,000 bytes of seeded text over 16 letters, which still takes several chunks once compressed. It checks the receiver's copy and the framing: no inline content, more than one chunk, DEFLATE with fewer bytes on the wire than the payload for the one-shot text, bytes and file sends, and plain chunks of the full length for the two streams. With compression forced off, the one-shot cases fail. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

The public property added for the two-room payload cap test had no docstring. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…d checks the end The readers count each chunk inside the operation that abandoningOnCancel runs, so a chunk a cancelled read took still counts. But nothing ordered that count against a later read: once the core handed the abandoned read its chunk, the later read could take the next chunk, see the end of the stream, and check the declared length before the abandoned read had added its bytes. A valid stream then failed with .incomplete; the review reproduced it at iteration 62 of a cancel-and-resume probe. Each reader now owns a SerialRunnerActor, and every read runs the core's next(), the count and the end check as one block on it, inside the abandon-on-cancel wrapper. An abandoned read keeps running there, so a later next() or readAll() waits until it has counted its chunk. No new lock or queue. The new test, for both readers, cancels a read waiting in the core, feeds the rest of an 8-byte stream, and reads it with readAll(), 20 times. A stand-in for the core's reader returns the abandoned read's chunk only after 500 yields, which leaves the later read every chance to reach the end first. Before the fix both cases fail with .incomplete in the first round; after it all rounds pass. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

The v1-sender case of the room matrix gave the sender's core a registry that reported every recipient as v1. That forced the legacy framing instead of testing what the room negotiates: a sender that advertises v1 still frames by its recipients, and the review confirmed that the real registry sends inline text to a v2 receiver. The recorder no longer replaces the registry; it only records. The matrix drops its hard-coded legacy flag. Each case now checks that the sender sees each receiver's advertised protocol and each receiver sees the sender's, then derives the expected framing from the participants the sender's registry reports: inline only for an in-memory one-shot send when every recipient reads v2, and no compression unless every recipient also advertises DEFLATE. The v1-to-v2 leg now expects inline text and bytes, which the old override made impossible. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

@objc

A Swift caller that cancelled its task while performRpc waited for the ack or the response kept waiting. performRpc is an @objc async method in an extension of a non-final NSObject subclass. Swift calls it through objc_msgSend and the completion-handler thunk, which runs the body in a new task. The caller's cancellation never reached that task, so the call ended only on the 7 s ack timeout or on the response timeout. 777193a fixed the data stream methods the same way. Mark performRpc final. The Objective-C selector stays the same, and Swift now calls the method directly. The completer wait already ends on cancellation, but it throws LiveKitError(.cancelled). The manager now throws CancellationError when the task is cancelled, like the data stream methods. The existing cleanup removes the pending ack, the pending response, and the ack watchdog. A late ack or response for the cancelled request finds no entry. Its log moves from error to debug, because a late response is not a problem that the app can fix. The tests start a call, wait until the call waits on its completer, and then cancel. They cover a cancel while the call waits for the ack and for the response, a late ack, response packet, and response stream after a cancel, and an uncancelled call that waits and then gets its response. They replace performRpcCleansUpOnCancellation, which passed on the ack timeout. On 9661b42 the three cancel cases time out after 5 s and leave pending state. With final alone they throw LiveKitError 100 instead of CancellationError. With both changes they pass, each in about 1 s including the room connection. The pending maps are private(set) so that the tests can read them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

The cancel tests set an afterPublish hook that captures the room. The room holds the RPC client, which holds the hook, so the room never deinitialized. startWaitingCall now clears the hook once the call waits on its completer. The tests also did not check that the ack watchdog was removed. The watchdog map is now private(set), like the pending maps, and one helper expects all three maps to be empty in the cancel, late response, and uncancelled tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

b06bb75 moved the log for a late response to debug, but a late response stream still went through readAll(). If that read failed, the handler logged an error before it found that no call was waiting. A call that ended while its stream was read, for example on a cancel, had the same problem. The handler now checks that the request is pending before it reads the stream, so a late stream is dropped unread with a debug log. If a read fails, the handler checks the pending call again. It logs an error and fails the call only if the call still waits, and logs at debug level otherwise. The new test cancels a call and then delivers a response stream for it that fails to read. On e108b9e the stream is read once. Now it is not read, and no pending ack, response, or watchdog is left. No test checks the log level, because sharedLogger is fixed at its first use. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…ing its publish A v1 call cancelled while its request was published rethrew the data channel's LiveKitError(.cancelled), but a v2 call threw CancellationError. The publish catch now throws CancellationError when the task is cancelled, after it removes the pending state. The v1 drain still finishes its send first. Only the error changes. A v1 call whose task was already cancelled sent its request and then threw at the wait. A v2 call sent nothing, because the data stream calls check for a cancel first. performRpc now checks before it registers or sends anything, so both transports send nothing. The docstring now says what a cancel does in each phase. The new test cancels its task before it calls performRpc, for a v0 and a v1 remote. It expects CancellationError, no packet sent, and nothing pending. On 501baa3 the v0 case sends 1 packet. The other checks pass there. With this change both cases pass. No test cancels during the publish. The afterPublish hook runs after the publish returns, so it cannot hold a publish open. The afterPublish call is now one line, which keeps performRpc inside the function length limit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

Twelve RPC tests stored an afterPublish hook that captured the room. The room holds the RPC client, which holds the hook, so each of those rooms never deinitialized. Each hook now captures the room weakly and returns if it is gone. The room lives for the whole withRoom block, so the hooks behave as before. The hook that only captures StateSync values and the waiting-call hook, which e108b9e clears, stay as they are. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

…cancel performRpcCancelledDuringPublishThrowsCancellationError gives the room a mocked data channel that never opens, so a v1 request waits on its whenOpen completer. The test waits until that wait registers, cancels the call, and expects CancellationError and no pending ack, response, or watchdog. Without the checkCancellation() in the publish catch, the call throws LiveKitError 100 (Cancelled) instead. performRpcIgnoresReadFailureAfterCancel delivers a response stream whose read waits on a gate. It cancels the call while the read waits, expects CancellationError and nothing pending, and then makes the read fail. The handler returns, the stream is read once, nothing is pending, and the call's completer holds no result. If the catch fails the call from the entry it looked up before the read, instead of rechecking, the completer holds RpcError 1500 and the test fails. WaitingCall now carries the completer so that the test can check it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>

pblazej

I'm kinda self-approving this after many rounds of AI-x-reviews (lack of other fb), follow-ups come in the next PR(s) as this one grows too much.

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