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.
Learn more
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.
Learn more
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.
| sid, known := m.subHandles[trackHandle(packet.Handle)] | ||
| track := m.descriptors[sid] | ||
| if !known || track == nil || track.subscription != subscriptionActive { | ||
| m.params.Logger.Debugw("dropping data track packet without subscription", "handle", packet.Handle) |
There was a problem hiding this comment.
The packet will be dropped silently if debugging is not enabled. Should we use info or warn here?
| select { | ||
| case track.packets <- packet: | ||
| default: | ||
| m.params.Logger.Debugw("dropping data track packet, pipeline is behind", "sid", sid) |
| streamList atomic.Pointer[[]*Stream] | ||
|
|
||
| // guarded by manager.mu | ||
| info Info |
There was a problem hiding this comment.
Using the mutex of one structure inside another structure is highly error-prone.
There was a problem hiding this comment.
That makes sense, addressed in 9b0d4c1; please let me know if this is what you had in mind.
There was a problem hiding this comment.
The RemoteManager lock track's mutex directly now.. can we make RemoteManager/RemoteTrack use its own mutex only?
| default: | ||
| } | ||
| select { | ||
| case <-s.frames: |
There was a problem hiding this comment.
It has a chance to drop a frame that can be buffered when the subscriber pop a frame from a full channel. Can we use a queue instead of the channel for the frame buffer?
2fd1c9b to
b7920b9
Compare
b7920b9 to
282a381
Compare
2 participants
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