Data track subscribing - #992
Conversation
ce38d1d to
a1d1643
Compare
a1d1643 to
bee8224
Compare
bee8224 to
2fd1c9b
Compare
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
There was a problem hiding this comment.
Devin Review found 5 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) | ||
| r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) | ||
| r.LocalParticipant.dataTracks = r.localDataTracks | ||
| r.remoteDataTracks = datatrack.NewRemoteManager(datatrack.RemoteManagerParams{Transport: remoteDataTrackTransport{room: r}, Logger: r.log}) |
There was a problem hiding this comment.
🔴 Encrypted tracks return ciphertext
With WithDataEncryption, remoteDataTracks still receives no Decryptor. processPacket then returns ciphertext as a valid frame. Subscribers cannot consume encrypted publications.
Prompt for agents
Wire remote data-track decryption into Room session encryption. The manager created in room.go receives no Decryptor even when WithDataEncryption configures engine.dataCryptor. Add an adapter from the session DataCryptor/key provider to datatrack.Decryptor, resolve it for the current session, and ensure encrypted tracks fail clearly rather than yielding ciphertext when encryption is unavailable. Cover encrypted subscriptions through the Room API, including reconnects and key rotation.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if withdraw != nil { | ||
| m.sendSubscriptionUpdate(*withdraw) |
There was a problem hiding this comment.
🟡 Concurrent resubscribe gets cancelled
When the last stream closes during Subscribe, the new subscribe can send before this deferred unsubscribe. The stale unsubscribe cancels the new request, leaving Subscribe pending until timeout.
Prompt for agents
Serialize subscription state transitions with their outgoing UpdateDataSubscription messages. In datatrack/remote.go, removeStream marks the track unsubscribed under manager.mu but sends the unsubscribe after unlocking. A concurrent Subscribe can therefore send a newer subscribe before the older unsubscribe. Introduce ordered outbound updates or generation-aware reconciliation so stale withdrawals cannot overtake newer subscriptions. Apply the same ordering protection to timeout withdrawals and reconnect resubscriptions, then add a deterministic concurrent Close/Subscribe test.
Was this helpful? React with 👍 or 👎 to provide feedback.
| r.runParticipantDefers(newSid, rp) | ||
| } | ||
| } | ||
| r.remoteDataTracks.HandleParticipantUpdate(participants, r.LocalParticipant.Identity()) |
There was a problem hiding this comment.
🟡 Disconnect loses participant callback
A disconnected publisher is removed before HandleParticipantUpdate emits data-track removals. OnTrackUnpublished then skips its participant callback and passes nil to the room callback.
Prompt for agents
Preserve the RemoteParticipant through data-track unpublication on participant disconnect. Room.OnParticipantUpdate currently calls OnParticipantDisconnect, which removes the participant, before RemoteManager.HandleParticipantUpdate emits removals. Reorder or carry the participant reference so RemoteParticipant.Callback.OnDataTrackUnpublished runs and RoomCallback receives the documented participant. Keep participant and room callback ordering consistent with media-track unpublication.
Was this helpful? React with 👍 or 👎 to provide feedback.
| // Subscribe, without stalling signal handling. | ||
| func (t remoteDataTrackTransport) OnTrackPublished(track *datatrack.RemoteTrack) { | ||
| rp := t.room.GetParticipantByIdentity(track.PublisherIdentity()) | ||
| go func() { |
There was a problem hiding this comment.
🟡 Publication callbacks arrive reversed
Each event starts an independent goroutine, so rapid publication and removal can run OnDataTrackUnpublished first. Applications observe removal before discovery or subscribe after removal.
Prompt for agents
Dispatch data-track lifecycle callbacks asynchronously but in publication order. The two remoteDataTrackTransport methods currently launch independent goroutines, allowing publish and unpublish events for the same track to overtake each other. Use a serialized room-level or per-track callback queue that does not block signal handling, while retaining participant-before-room callback ordering.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for _, stream := range *t.streamList.Load() { | ||
| stream.push(frame) | ||
| } |
There was a problem hiding this comment.
🟡 Resubscribe leaks old frames
A closed pipeline can drain buffered packets after a new subscription replaces streamList. Its old worker then pushes pre-subscription frames into the new streams.
Prompt for agents
Prevent workers from delivering packets across subscription generations. runPipeline reads the track-wide current streamList, while deactivateLocked only closes the old packet channel and does not wait for its worker. After resubscription, that worker can drain old buffered packets into newly installed streams. Tie each worker to its subscription generation and stream snapshot, or synchronously stop it before exposing new streams. Add a test that pauses an old worker, closes the final stream, resubscribes, then releases the worker.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Adds support for subscribing to data tracks.
Architecture and public API closely match Rust and JS clients:
OnDataTrackPublishedevent to receive the track object which they can then use to subscribe if desired.RemoteManageris owned by the room to manage internal state and transitions for all remote data tracks.Areas to review:
RemoteManager) are currently public. Would like to understand if there is a better way to organize packages to avoid this.RemoteManageris currently owned byLocalParticipant, but I am not sure this is the right place for it to live.Closes BOT-541