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.
| room.log("Failed to decode outgoing data stream packet", .warning) | ||
| continue | ||
| } | ||
| try? await room.send(dataPacket: packet) |
There was a problem hiding this comment.
🟡 Failures to transmit stream data are discarded without any log
A failure to actually transmit an outgoing piece of stream data is thrown away silently (try? await room.send(...) at Sources/LiveKit/DataStream/DataStreams.swift:307), so data that never left the device leaves no trace at all.
Impact: When a send fails (for example while the connection is down) the transfer silently loses content with no error and no log for diagnosis.
Details
In the previous implementation the packet handler's error propagated back to write(_:)/close(...) so callers learned that the stream failed. With the FFI split, the send result cannot be returned to the writer, but the current code additionally discards it with try? without logging. The sibling failure path two lines above does log (room.log("Failed to decode outgoing data stream packet", .warning)), so the omission looks unintentional. At minimum the error should be logged at .warning/.error, and ideally the stream should be aborted so the reader on the other side does not hang waiting for chunks that will never arrive.
| try? await room.send(dataPacket: packet) | |
| do { | |
| try await room.send(dataPacket: packet) | |
| } catch { | |
| room.log("Failed to send outgoing data stream packet: \(error)", .warning) | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
(Not sure if this is problematic or not, will defer to @pblazej)
| public var maxPayloadSizeNumber: NSNumber? { | ||
| maxPayloadSize.map { NSNumber(value: $0) } | ||
| } | ||
|
|
||
| public init(maxPayloadSize: Int?) { | ||
| self.maxPayloadSize = maxPayloadSize | ||
| } | ||
|
|
||
| override public init() { | ||
| maxPayloadSize = nil | ||
| } | ||
|
|
||
| public convenience init(maxPayloadSizeNumber: NSNumber?) { | ||
| self.init(maxPayloadSize: maxPayloadSizeNumber?.intValue) |
There was a problem hiding this comment.
🟡 New public data stream option members ship without documentation comments
The newly added public members of the data stream options type are published without documentation (maxPayloadSizeNumber and the initializers at Sources/LiveKit/Types/Options/DataStreamOptions.swift:28-41), which the repository's contributor rules require for every public API.
Impact: Generated API documentation for the new option type is incomplete for SDK users.
Rule reference
AGENTS.md (Coding Style): "/// Docstrings for every public API using Swift markdown". maxPayloadSizeNumber, init(maxPayloadSize:), init() and init(maxPayloadSizeNumber:) are all public and undocumented. (The existing ObjC-compatibility initializer on StreamByteOptions shows the expected pattern of documenting the NSNumber bridge.)
Was this helpful? React with 👍 or 👎 to provide feedback.
|
@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). |
| // a user transcript on the same topic is still streaming). A single sender's streams are | ||
| // sequential in practice, so per-sender serialization preserves ordering without stalling peers. | ||
| private let orderedTopics = StateSync<Set<String>>([]) | ||
| private let orderedTails = StateSync<[String: [String: Task<Void, Never>]]>([:]) |
There was a problem hiding this comment.
🟡 Per-sender bookkeeping for ordered stream topics grows without bound
Finished handler entries are never removed from the per-sender bookkeeping map (orderedTails at Sources/LiveKit/DataStream/DataStreams.swift:64), so the map keeps one entry for every sender that has ever opened a stream on an in-order topic for the lifetime of the room.
Impact: In long-running rooms with many participants, memory use slowly creeps up and never comes back down.
Missing completion cleanup compared to the removed implementation
orderedTails is keyed [topic: [identity: Task]] and written in handleTextStreamOpened (Sources/LiveKit/DataStream/DataStreams.swift:243-253). Entries are only cleared when the whole topic is unregistered (Sources/LiveKit/DataStream/DataStreams.swift:140). The previous IncomingStreamManager removed each handler entry on completion via handlerCompleted(topic:generation:). A cheap fix is to clear the tail entry when the chained task finishes if it is still the current tail for that (topic, identity).
Was this helpful? React with 👍 or 👎 to provide feedback.
| let maxPayloadSize = room?._state.roomOptions.dataStreamOptions.maxPayloadSize | ||
| let manager = LiveKitUniFFI.IncomingDataStreamManager( | ||
| delegate: delegate, | ||
| maxPayloadByteLength: maxPayloadSize.map { UInt64($0) }, | ||
| ) | ||
| existing = manager |
There was a problem hiding this comment.
🟡 A negative maximum payload size crashes the app instead of being rejected
The app-supplied maximum incoming payload size is converted to an unsigned value without validation (UInt64($0) at Sources/LiveKit/DataStream/DataStreams.swift:87), so a negative value terminates the process on the first received data-stream packet.
Impact: An app that passes a negative limit crashes as soon as any data stream arrives, rather than getting an error or a sane default.
Unvalidated Int → UInt64 conversion on a public option
DataStreamOptions.maxPayloadSize is a public Int? (Sources/LiveKit/Types/Options/DataStreamOptions.swift:26) with no validation in its initializers. It is read lazily on the first inbound packet in DataStreams.incomingManager() and converted with UInt64($0), which traps for negative values. Per AGENTS.md, crashing consumer code is not allowed and unsafe conversions should be wrapped. Clamping (max(0, $0)) or ignoring non-positive values would be sufficient.
Was this helpful? React with 👍 or 👎 to provide feedback.
…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
|
|
| let (stream, continuation) = AsyncStream.makeStream(of: [Data].self) | ||
| self.continuation = continuation | ||
| pump = Task.detached { [weak room] in |
There was a problem hiding this comment.
🔴 Sending a large file or stream can buffer the whole payload in memory
Outgoing stream packets are queued into an unlimited in-memory queue (AsyncStream.makeStream(of:) at Sources/LiveKit/DataStream/DataStreams.swift:323) that the sender never has to wait on, so a writer can produce data far faster than the network drains it and the backlog grows without bound.
Impact: Sending a large file or a fast stream can balloon the app's memory use and, on a slow or stalled connection, keep growing until the app is killed.
Loss of the backpressure the previous implementation provided
The old OutgoingStreamManager awaited Room.send(dataPacket:) for each chunk inside write, and DataChannelPair.send waits on the data channel's buffered-amount-low signal, so a writer was naturally flow-controlled by the transport.
Now onPacketsAvailable (Sources/LiveKit/DataStream/DataStreams.swift:343-345) yields into an AsyncStream created with the default .unbounded buffering policy and returns immediately; the drain task at Sources/LiveKit/DataStream/DataStreams.swift:325-336 is the only consumer. Because the FFI callback is synchronous and returns Void, there is no way for the transport to slow the producer down. sendFile in particular hands the whole file to the FFI (Sources/LiveKit/DataStream/DataStreams.swift:175), which chunks it eagerly, so every chunk of a multi-hundred-megabyte file can end up resident in the queue at once.
A bounded buffering policy alone would silently drop packets (fatal for stream reassembly), so the fix likely needs a bounded queue plus a way to make the FFI producer wait — or an explicit drop-with-stream-failure policy.
Prompt for agents
The outgoing data stream pump in Sources/LiveKit/DataStream/DataStreams.swift (OutgoingDelegate) yields FFI-produced packets into an AsyncStream created with the default unbounded buffering policy, and onPacketsAvailable is a synchronous Void-returning FFI callback. This removes the backpressure the previous OutgoingStreamManager had, where each chunk send was awaited through Room.send(dataPacket:) and therefore flow-controlled by DataChannelPair's buffered-amount-low gating. As a result, a large sendFile or a fast streamBytes producer can enqueue the entire payload in memory before the drain task sends any of it. Investigate whether the UniFFI outgoing manager offers a way to signal backpressure (e.g. an async or blocking delegate callback, or a pull-based packet API). If it does, drive the pump from that instead of an unbounded AsyncStream. If it does not, consider a bounded queue combined with an explicit stream failure when the bound is exceeded — note that simply dropping packets is not acceptable because the receiver fails a stream on a non-consecutive chunk index.
Was this helpful? React with 👍 or 👎 to provide feedback.
| private func incomingManager() -> LiveKitUniFFI.IncomingDataStreamManager { | ||
| // Fast path: after the first packet of a session this is a plain read, keeping the exclusive | ||
| // lock off the per-packet inbound path. `mutate` re-checks, so the race is still safe. | ||
| if let existing = _incoming.copy() { return existing } | ||
| return _incoming.mutate { existing in | ||
| if let existing { return existing } | ||
| let delegate = IncomingDelegate() | ||
| delegate.coordinator = self | ||
| // `nil` → the core's default cap. Read now (first packet, i.e. post-connect) so a | ||
| // `maxPayloadSize` supplied via `connect(roomOptions:)` is honored. | ||
| let maxPayloadSize = room?._state.roomOptions.dataStreamOptions.maxPayloadSize | ||
| let manager = LiveKitUniFFI.IncomingDataStreamManager( | ||
| delegate: delegate, | ||
| maxPayloadByteLength: maxPayloadSize.map { UInt64($0) }, | ||
| ) | ||
| existing = manager | ||
| return manager | ||
| } | ||
| } |
There was a problem hiding this comment.
🟨 Incoming data-stream payload size cap is only applied on the first packet of the first session
The incoming UniFFI manager is constructed lazily on the first inbound stream packet, reading maxPayloadSize from the room options at that moment (Sources/LiveKit/DataStream/DataStreams.swift:78-96). If a remote participant sends a data-stream packet before the app's connect(roomOptions:) finalizes options, the manager is built with the SDK's default cap (documented as 5 GB in Sources/LiveKit/Types/Options/DataStreamOptions.swift:25) and that cap is fixed for the whole session, silently ignoring a smaller limit the app configured. Combined with the removal of the Swift-side lengthExceeded accounting that the deleted IncomingStreamManager performed per chunk, a malicious or buggy peer can make the receiver buffer far more data than the app intended.
Was this helpful? React with 👍 or 👎 to provide feedback.
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.