Data streams v2 - #1075
Conversation
|
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 |
![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.
|
@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. |
e6f2d11 to
3395925
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
8bdc427 to
530796b
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
@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). |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…d data streams v2
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
…s race `lazy var dataStreams` isn't atomic: two threads racing the first access could each construct a DataStreams (and its FFI managers). Assign it once in `init` after `super.init()` instead, like the eager data-stream managers it replaced.
Drop the local isOpen flag on Byte/TextStreamWriter and query the UniFFI writer's is_open() instead, so a writer reflects the stream actually closing — including when a send fails because the room disconnected — rather than only an explicit local close().
…handling - Add RoomOptions.dataStreamOptions.maxPayloadSize, plumbed into the incoming manager so a receiver bounds the reassembled size of an incoming stream instead of accepting it uncapped. - Key ordered-topic handler serialization by sender identity rather than by topic, so a still-open stream from one sender no longer blocks a concurrent stream from another (e.g. an agent transcript arriving while a user transcript on the same topic is still streaming).
The query-param connect path (buildUrl) sent client_protocol but not capabilities, so peers connected that way (the default, non-single-PC path) never saw CAP_COMPRESSION_DEFLATE_RAW and never compressed. Emit a `capabilities` query param there too (comma-separated enum names, as the server parses), and source both paths from a single advertisedClientCapabilities list.
The incoming manager was built at Room.init with the payload cap read from the initial room options, so a maxPayloadSize supplied at connect time was ignored. Build the incoming manager lazily on the first inbound packet (post-connect), reading the room's current options then. Guarded by StateSync so it's constructed exactly once. reset()/closeStreams(from:) no-op when no packets have arrived.
Rebase adaptation: main moved the protocol layer to nanopb, where messages
are immutable and built with `.with { }` instead of `var msg = T()`.
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>
53ecd83 to
49ba7b2
Compare
|
|
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
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>
| private func handlerCompleted(topic: String, streamID: String) { | ||
| runningHandlers.mutate { $0[topic]?.removeValue(forKey: streamID) } | ||
| finishingHandlers.mutate { $0[topic]?.removeValue(forKey: streamID) } | ||
| streamTopics.mutate { $0[streamID] = nil } | ||
| } |
There was a problem hiding this comment.
🔴 Repeated messages on the same conversation can be delivered out of order
Bookkeeping for in-order text delivery is keyed only by the sender's stream identifier (handlerCompleted at Sources/LiveKit/DataStream/DataStreams.swift:309-313), so when a sender reuses the same identifier for a following message the finished earlier message erases the newer one's tracking and the ordering chain breaks.
Impact: Consecutive transcription/chat messages from the same sender can be surfaced to the app out of order (or their tracking entries left behind), producing scrambled transcripts.
Mechanism: stream-ID-only keying loses the v1 "generation" guard
v1 stored each open stream under a per-descriptor generation UUID precisely so a stale cleanup could not remove a successor that reused the same stream ID (see the deleted Sources/LiveKit/DataStream/Incoming/IncomingStreamManager.swift comments and the deleted reusedStreamIDDeliversEveryStream test). The new coordinator keys runningHandlers, finishingHandlers and streamTopics by stream ID alone:
- Stream
Xopens on an ordered topic →streamTopics[X] = topic,runningHandlers[topic][X] = task1(Sources/LiveKit/DataStream/DataStreams.swift:281-296). Xcloses →handleStreamClosedmovestask1intofinishingHandlers[topic][X](Sources/LiveKit/DataStream/DataStreams.swift:302-306).- The sender reopens a stream with the same ID
Xwhiletask1is still running →runningHandlers[topic][X] = task2. task1finishes →handlerCompleted(topic:streamID: X)removesrunningHandlers[topic][X]— which is nowtask2— and clearsstreamTopics[X].- When the second stream closes,
handleStreamClosedfinds no topic mapping, sotask2never entersfinishingHandlers; a third stream on the topic therefore does not wait for it and can run concurrently/ahead of it.
The same collision occurs when two different senders happen to use the same stream ID, since streamTopics is not keyed by participant.
Prompt for agents
In Sources/LiveKit/DataStream/DataStreams.swift the ordered-text-stream bookkeeping (runningHandlers, finishingHandlers, streamTopics) is keyed by the wire stream ID only. The previous implementation (deleted Sources/LiveKit/DataStream/Incoming/IncomingStreamManager.swift) deliberately keyed by a per-open 'generation' UUID because senders may reuse a stream ID for consecutive streams (transcription does this), and a stale cleanup must not remove a successor's entry. Reintroduce an identity that is unique per stream *open* (e.g. a UUID minted in handleTextStreamOpened, with streamTopics mapping streamID -> (topic, generation) or a per-participant+ID composite) so that handleStreamClosed and handlerCompleted only touch the entry belonging to the open they were created for. Consider also restoring a regression test equivalent to the deleted reusedStreamIDDeliversEveryStream covering reuse on an ordered topic.
Was this helpful? React with 👍 or 👎 to provide feedback.
|
|
||
| extension Room: DataChannelDelegate { | ||
| func dataChannel(_: DataChannelPair, didReceiveDataPacket dataPacket: Livekit_DataPacket) { | ||
| func dataChannel(_: DataChannelPair, didReceiveDataPacket dataPacket: Livekit_DataPacket, encryptionType: EncryptionType) { |
There was a problem hiding this comment.
🟡 Encrypted data messages are reported to apps as unencrypted
The encryption kind handed to the data-message callback is read from the packet after it has already been unscrambled (dataPacket.encryptedPacket.encryptionType at Sources/LiveKit/Core/Room.swift:805), so apps are always told an encrypted message arrived in the clear even though the correct value is now available as a parameter.
Impact: Applications that inspect whether received data was end-to-end encrypted always see "none", so they cannot distinguish encrypted from plaintext senders.
Mechanism: the oneof field carrying the encryption type is overwritten by decryption
The new delegate signature (Sources/LiveKit/Core/DataChannelPair.swift:26-29) exists precisely because EncryptedPacket shares a protobuf oneof with the decrypted payload: decryptedPayload.applyTo(&$0) sets e.g. builder.user, clearing encryptedPacket (Sources/LiveKit/E2EE/Protos+E2EE.swift:68-88). The stream cases were migrated to the new encryptionType parameter, but the .user case still reads the now-cleared field, so engine(_:didReceiveUserPacket:encryptionType:) — and ultimately room(_:participant:didReceiveData:forTopic:encryptionType:) — always receives .none for decrypted packets.
Prompt for agents
In Sources/LiveKit/Core/Room.swift the DataChannelDelegate implementation now receives the as-received `encryptionType` as a parameter (because decryption clears `dataPacket.encryptedPacket`). The `.user` case still derives the value from `dataPacket.encryptedPacket.encryptionType.toLKType()`, which is always `.none` for a packet that went through decryption. Pass the new `encryptionType` parameter through to `engine(_:didReceiveUserPacket:encryptionType:)` instead.
Was this helpful? React with 👍 or 👎 to provide feedback.
|
|
||
| extension Room: DataChannelDelegate { | ||
| func dataChannel(_: DataChannelPair, didReceiveDataPacket dataPacket: Livekit_DataPacket) { | ||
| func dataChannel(_: DataChannelPair, didReceiveDataPacket dataPacket: Livekit_DataPacket, encryptionType: EncryptionType) { |
There was a problem hiding this comment.
🟨 Encryption status of received data messages is misreported as unencrypted
After a packet is decrypted, the encryption kind is read from the protobuf field that decryption overwrites (dataPacket.encryptedPacket.encryptionType at Sources/LiveKit/Core/Room.swift:805), so room(_:participant:didReceiveData:forTopic:encryptionType:) always reports .none. Applications that gate trust on whether inbound data was end-to-end encrypted cannot distinguish an encrypted sender from a plaintext one, which can lead to accepting unauthenticated/plaintext payloads as if they were E2EE-protected. The PR adds the correct as-received value as a parameter and uses it for the data-stream paths, but not for the user-packet path.
Was this helpful? React with 👍 or 👎 to provide feedback.
| id: id, | ||
| mimeType: mimeType, | ||
| name: name, | ||
| totalLength: totalSize.map { UInt64($0) }, |
There was a problem hiding this comment.
🟡 Passing a negative expected size when opening a byte stream crashes the app
A caller-supplied negative expected size is converted straight to an unsigned value (UInt64($0) at Sources/LiveKit/DataStream/StreamOptions.swift:166) when opening a byte stream, so the process traps instead of the call failing gracefully.
Impact: An app that passes an invalid (negative) expected byte-stream size crashes rather than receiving an error.
Mechanism and precedent in this PR
StreamByteOptions.totalSize is a public Int? with no validation in the initializer (Sources/LiveKit/DataStream/StreamOptions.swift:138-156). ffi maps it with totalSize.map { UInt64($0) }; UInt64(_:) on a negative Int traps at runtime.
The same hazard was explicitly handled for the new DataStreamOptions.maxPayloadSize, which normalizes non-positive values to nil at construction (Sources/LiveKit/Types/Options/DataStreamOptions.swift:42-44) with a test noting "used to reach UInt64(_:) at the FFI boundary and trap the process". totalSize reaches the FFI boundary the same way and is left unguarded.
| totalLength: totalSize.map { UInt64($0) }, | |
| totalLength: totalSize.flatMap { $0 >= 0 ? UInt64($0) : nil }, |
Was this helpful? React with 👍 or 👎 to provide feedback.
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>
2 participants
An initial stab at migrating swift's data streams implementation to the
livekit-uniffiprovided data streams v2.This was a very LLM driven migration. I've run some tests to validate everything works in theory but still need to more extensively test this in actual agent scenarios. I've also where possible tried to keep tests which existed previously and adapt them to the new rust core logic.
LLM summary, which may be useful to reviewers
Warning
This pull request was LLM generated and has been reviewed by a human who isn't a swift expert.
A more thorough review of this needs to occur before it should be considered to be in a mergeable state.