Data streams

[ladvoc]

This PR adds native data streams support in Rust and a high-level interface for FFI clients.

[ilo-nanpa]

it seems like you haven't added any nanpa changeset files to this PR.

if this pull request includes changes to code, make sure to add a changeset, by writing a file to .nanpa/<unique-name>.kdl:

minor type="added" "Introduce frobnication algorithm"

refer to the manpage for more information.

[bcherry]

lgtm from a protocol level.

[theomonnom]

I missed your previous comment. You mentioned RPC handlers, but I see how the same issue applies to data stream handlers as well. Using data streams as an example, from my current understanding, a race condition can occur when a participant opens a data stream in response to another participant joining, but the newly joined participant hasn't registered a handler for that stream's topic yet:

Yes, it is the same for textstreams/datastreams/rpc

When a topic is preregistered, any incoming stream matching that topic will be queued until the corresponding handler is registered, at which point each queued stream is dequeued and passed to the handler:

Does this mean that every client will have to return all registered handlers after connect? It would be easier if we don't have to.

A key consideration is that when a topic is preregistered, the user must register the corresponding handler within a short window after connecting. Otherwise, incoming streams could accumulate indefinitely without being handled. To prevent this, I propose a timeout interval (e.g., 30 seconds after connect), after which any queued stream readers that don't yet have handlers will be dropped, and an error will be logged. In practice, as long as the handler is registered shortly after connect, the number of queued streams should remain minimal.

If you send directly after connecting (assume the client language already has a registered handler), you could skip this part. We will not be responsible for any buffering

Another solution would be to not register any callback to the room, because it seems like an anti-pattern with our channel based events? In this case everything will be buffered and you "support the register before connect" on all languages for free

[lukasIO]

Another solution would be to not register any callback to the room, because it seems like an anti-pattern with our channel based events? In this case everything will be buffered and you "support the register before connect" on all languages for free

@theomonnom do I understand your suggestion correctly that we could

• buffer all incoming RPC and StreamHeaders (with subsequent chunks) until connected
• when connected, have a short timeout (even 1s should be enough) that allows for patterns like await room.connect(); registerHandlers();
• if there are handlers registered within that timeout, pipe all buffered packets to the handlers
• after the timeout elapsed (and no handlers have been registered) discard all buffered data

[theomonnom]

I think we don't even need to drop them? If we push the data inside the room_events channel. They're buffered but will be consumed as soon as the connection is established. (Like basically a new RoomEvent variant?). No timeout will be needed IMO.

In the code there is no explicit "buffer until we're connected", we just push to our event queue and assume they're going to get consumed by the user event loop. (In this case livekit-ffi)

The issue with the above pattern, is that every language will have to send a registerHandler each time after the connecting. (we have to keep track of all registered handlers before connection)

But if we "always" forward what we receive to the other languages, then no registration is needed.

If we do this "always" forward mode, then the Python/Node doesn't even need to send a RegisterRequest, but they can just read the data they receive from the ffi server and dispatch their own handlers

[theomonnom]

lgtm! very clean impl!