Apply signal responses in the order the server sent them - #1091
Open
tarsyang wants to merge 3 commits into
Open
Conversation
notifyDetached spawned a detached task per notification and let those tasks race into SerialRunnerActor, so two notifications issued in sequence could reach the delegate in the opposite order. SignalClient hands almost every signal message to Room through this hop, so any pair of adjacent messages (participant updates, mute, speakers, stream states) could be applied out of order. notifyDetached(inOrder:) only pinned the closures of a single call, and the data-track responses had grown a consumer task of their own that awaited notifyAsync per message to keep that one path in wire order. Every notification now joins one AsyncStream FIFO that a single consumer task drains, so call order is delivery order. The consumer captures the state and the stream only; deinit finishes the stream and the consumer ends once what was queued has been delivered. With that, both workarounds have nothing left to add. notifyDetached(inOrder:) is removed and its two call sites issue two notifyDetached calls. The data-track responses join the same FIFO from onWebSocketMessage, still bypassing the response queue; what the separate consumer also provided, dropping a dead connection's responses instead of applying them to the next, is kept by a connection generation that the message loop carries and each of those notifications checks at delivery. notifyAsync had no other caller and goes with its consumer. The tests assert the new guarantee directly: call order across 2000 calls, a slow notification not overtaken, a released delegate, delivery of what was queued after release.
resume() flipped the state to .resumed before its first await, so a value that came through processIfResumed while the backlog was still draining was processed at once and overtook everything still queued. SignalClient resumes its response queue right after the transports are configured, on every connect and quick reconnect, so the messages queued during that window (offers, participant updates) could be applied after the ones that arrived a moment later. The same loop iterated a snapshot of the queue and then removed everything, which dropped whatever was appended during the drain, and two overlapping resume() calls each processed the whole snapshot. While a drain runs, a value that may be enqueued now joins the tail of the queue; a value that may not (elseEnqueue == false) is processed at once, as it is whenever the state is .resumed, and `condition` still bypasses the queue. resume() drains the live queue one element at a time until it is empty or clear()/suspend() changes the state, and a resume() that finds a drain running returns after marking the state, since that drain reaches every queued element. Tests hold element 0 behind a Gate test helper (a one-shot signal that cannot miss an open) and act mid-drain: arrivals keep arrival order, non-queueable values bypass, a concurrent resume() neither duplicates nor interleaves, and clear() stops the drain.
onWebSocketMessage spawned a Task.detached per response to hand it to the response queue actor. Independent tasks reach the actor in scheduler order, not socket order, and while one response's _process was suspended the next one re-entered the actor, so two adjacent responses could be processed in parallel and out of order. The message loop already awaits onWebSocketMessage per message, so it now awaits processIfResumed in place: responses enter the queue in socket order and are processed one at a time. Nothing in _process waits on the network or on a later message (join resumes a completer and restarts the ping timer, trackPublished resumes a completer, pong cancels a timer, the rest hand off to the delegate FIFO without waiting), so the loop is not held up. With the drain gate in QueueActor and the FIFO in AsyncSerialDelegate, this closes the last hop between the socket and Room's SignalClientDelegate handlers: signal messages are applied in the order the server sent them.
tarsyang
requested review from
hiroshihorie,
pblazej and
xianshijing-lk
as code owners
August 17, 2026 17:02
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Signal responses can be applied out of the order the server sent them. Three hops between the WebSocket and
Roomeach fork a task per message and serialize behind the fork, so an adjacent pair of messages can swap:SignalClient.onWebSocketMessagespawnsTask.detachedper response before the response queue actor; independent tasks reach the actor in scheduler order, and while one response's_processis suspended the next re-enters.QueueActor.resume()flips to.resumedbefore its firstawait, so a response arriving mid-drain runs ahead of the queued backlog; the loop also iterates a snapshot and thenremoveAll()s, dropping what was appended during the drain, and two overlappingresume()calls process the snapshot twice.AsyncSerialDelegate.notifyDetachedspawnsTask.detachedDiscardingper notification and lets them race intoSerialRunnerActor;_processhands almost every message toRoomthis way (20 sites).notifyDetached(inOrder:)only pins the closures of one call, and the data-track responses got a consumer task of their own that awaitsnotifyAsyncper message to keep that one path in wire order.Roomapplies participant updates in delivery order andRemoteParticipant.set(info:)unpublishes anything missing from the update (there is noParticipantInfo.versioncheck), so an inverted pair makes a freshly published track disappear until the next update. Last-writer-wins messages (speakersChanged,connectionQuality,streamStateUpdate,subscriptionPermissionUpdate,mute,roomUpdate,subscribedQualityUpdate,refreshToken) settle on the stale value; so do successivedataTrackSubscriberHandlessnapshots, which the manager applies entry by entry.Changes
AsyncStreamFIFO drained by a single consumer task;notifyDetachedyields, so call order is delivery order. The consumer captures only the state and the stream;deinitfinishes the stream and the consumer ends once what was queued has been delivered. With ordering in the type, both workarounds for its absence go:notifyDetached(inOrder:)is removed (its two call sites issue twonotifyDetachedcalls), and the data-track responses join the same FIFO straight fromonWebSocketMessage. What the separate consumer also provided, dropping a dead connection's responses instead of applying them to the next, is kept by a connection generation: the message loop carries the value of its connection and each of those notifications checks it at delivery;cleanUpadvances it as it finishes, so the cut-off point is the same as the old stream's.notifyAsynchad no other caller and goes with its consumer.resume()drains, a value that may be enqueued joins the tail (non-queueable values andconditionbypass as before); the drain loop reads the live queue until it is empty orclear()/suspend()changes the state; aresume()that finds a drain running returns after marking the state.onWebSocketMessageawaitsprocessIfResumedin place. The message loop already awaits per message, and nothing in_processwaits on the network or on a later message, so the loop is not held up.Behavior kept: join/reconnect/leave still bypass the queue; non-queueable requests are still sent immediately while suspended-or-draining; a dead connection's data-track responses are still dropped.
Tests
AsyncSerialDelegateTests(replaces theinOrdertests): call order across 2000 calls, slow notification not overtaken, released delegate, delivery of the queue after release.QueueActorTests: arrivals during a drain keep arrival order, non-queueable values bypass the drain, concurrentresume()neither duplicates nor interleaves,clear()stops the drain.Gate(test support) holds one step of the code under test.Verification
LiveKitCoreTestsviaxcodebuildas in CI, against a locallivekit-server --dev1.13.5 with the CI config: 344/345 (the one failure isAudioConverterTests.convertFormat, anExtAudioFileOpenURLerror on this machine, identical on unmodifiedmain). The data-track e2e suites (DataTrackLifecycleTestsreconnect scenarios included) pass 32/32.AsyncSerialDelegateTests,QueueActorTests,SerialRunnerActorTests,CompleterTests: 19/19, no reports.