refactor(signaling): explicit signal lifecycle state machine - #1402
refactor(signaling): explicit signal lifecycle state machine#1402lukasIO wants to merge 6 commits into
Conversation
SignalInner tracked its lifecycle in two fields, a stream slot and a `reconnecting` AtomicBool, that had to be kept in step by hand across restart, set_reconnected, close and send. It now holds one SignalState that owns the transport: Connected, Reconnecting, Offline, Disconnecting, Closed. Every change goes through SignalState::transition, one match that is the whole table and also says which transport each move releases. An input a state does not accept is logged and refused; a resume from a state that cannot accept it fails with SignalError::InvalidState. The resume gate stays a state until the engine confirms, a stale transport cannot report in because SignalClient::restart awaits the old task first, and ReconnectFailed always lands in Offline. The held-signal queue moves to a sync parking_lot Mutex that is never held across an await. The old async queue lock was taken in both orders relative to the stream lock, which with tokio's fair RwLock could deadlock against a pending restart writer. A send that fails with any transport error is now held like a SendError was. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Changeset ✓This PR includes a changeset covering all affected packages:
|
…it/rust-sdks into lukas/signal-state-machine
![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 3 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| let queued = std::mem::take(&mut *self.queue.lock()); | ||
| for signal in queued { | ||
| if let Err(err) = stream.send(signal).await { | ||
| log::error!("failed to send queued signal: {}", err); // Lost message |
There was a problem hiding this comment.
🔴 Queued sends vanish after failure
When a queued send fails, flush_queue discards that request and every remaining drained request. A transient transport failure loses all held session changes.
Prompt for agents
In livekit-signaling/src/lib.rs, make SignalInner::flush_queue preserve unsent messages when SignalStream::send returns an error. The queue is removed wholesale with mem::take, so the failed item and the untouched suffix currently disappear. Restore the failed item and remaining suffix in original order, ahead of messages concurrently added through hold_or_drop. Stop sending once the transport fails. Add a test with multiple queued signals and a stream that fails partway through, then verify a later successful flush sends every unsent signal exactly once and in FIFO order.
Was this helpful? React with 👍 or 👎 to provide feedback.
| let (new_stream, mut events) = | ||
| SignalStream::connect(lk_url, &token, self.options.connect_timeout).await?; | ||
| let reconnect_response = get_reconnect_response(&mut events).await?; | ||
| SignalResult::Ok((new_stream, reconnect_response, events)) |
There was a problem hiding this comment.
🔴 Reconnect failures leak live transports
If get_reconnect_response fails, new_stream drops without closing its transport. The read task keeps the writer and socket alive across retries.
Prompt for agents
In SignalInner::restart in livekit-signaling/src/lib.rs, ensure every successfully created SignalStream becomes owned by lifecycle state immediately or is explicitly closed on every failure and cancellation path. Currently a timeout, closed response channel, parse error, or LeaveRequest from get_reconnect_response drops new_stream without calling SignalStream::close. SignalStream's read task retains an internal_tx clone, so dropping the struct alone does not terminate the write task or transport. Restructure the reconnect attempt with an ownership guard or an intermediate state that guarantees cleanup, and add a test where the server keeps the reconnect WebSocket open but never returns ReconnectResponse.
Was this helpful? React with 👍 or 👎 to provide feedback.
| let mut state = self.state.write().await; | ||
| let old_stream = state.transition(SignalInput::Reconnect).map_err(|_| { | ||
| SignalError::InvalidState(format!("resume the signal session from {state:?}")) | ||
| })?; |
There was a problem hiding this comment.
🟡 Cancelled reconnects block every retry
If restart is cancelled after Reconnect, the state remains Reconnecting(None). Every later retry returns InvalidState, leaving the client unable to reconnect.
Prompt for agents
Make SignalClient::restart and SignalInner::restart cancellation-safe. Once the state transitions to Reconnecting(None), dropping the future must restore a retryable state and clean up any old or newly opened transport. The close phase has the same issue after entering Disconnecting. Consider an RAII transition guard or redesigning the lifecycle operation as an owned task whose cleanup completes independently of the caller future. Add tests that poll restart into the close and connect phases, cancel it, and verify a subsequent restart can proceed.
Was this helpful? React with 👍 or 👎 to provide feedback.
1egoman
left a comment
1egoman
left a comment
There was a problem hiding this comment.
Generally makes sense to me. I focused more on the infrastructure and general approach, and less on the exact states and all the exact transitions between them.
One thing missing which I was expecting to see: some sort of mechanism that a visualizer could subscribe to state changes and use them to build a visualization. Was this expected to be part of a follow up change or did I miss it in here somewhere?
| use SignalInput as In; | ||
| use SignalState::*; | ||
| // `Closed` is a placeholder while the old state is moved out | ||
| let (next_state, released_stream) = match (std::mem::replace(self, Closed), input) { |
There was a problem hiding this comment.
nitpick: I brought this up in our 1:1 but it might be slightly cleaner if SignalState::Closed was the Default value for SignalState and then this could become std::mem::take(self) given the replaced value is bogus anyway.
| let (next_state, released_stream) = match (std::mem::replace(self, Closed), input) { | |
| let (next_state, released_stream) = match (std::mem::take(self), input) { |
| /// | ||
| /// Leaves `reconnecting=true` on success — the engine is expected to call | ||
| /// [`Self::set_reconnected`] once the full resume has succeeded. On failure | ||
| /// resets `reconnecting=false` so subsequent retries can re-enter cleanly. | ||
| /// The stream slot is held under a write lock for the entire close + new | ||
| /// connect, so concurrent senders block on the read side until the new | ||
| /// stream is in place. | ||
| /// The write lock is held for the entire close + connect, so concurrent senders wait on | ||
| /// the read side and land on the new transport instead of in a transport-less gap. |
There was a problem hiding this comment.
thought: This is a pretty clever way to prioritize locks during the close and connect phases. That being said I also think it is potentially somewhat unintuitive. Maybe it makes sense to move this comment to the AsyncRwLock field definition in the struct?
| pub async fn close(&self, notify_close: bool) { | ||
| if let Some(stream) = self.stream.write().await.take() { | ||
| // Already closing or closed: whoever owns that close finishes it. | ||
| let Ok(stream) = self.state.write().await.transition(SignalInput::Close) else { | ||
| return; | ||
| }; | ||
| if let Some(stream) = stream { | ||
| stream.close(notify_close).await; | ||
| } | ||
| let _ = self.state.write().await.transition(SignalInput::CloseComplete); | ||
| } | ||
|
|
There was a problem hiding this comment.
thought: I'm not sure about this idea or not, but I wonder if there would be benefit to driving the state machine side effects by subscribing to a stream of state changes from the state machine, not just running the side effects alongside at every .transition(...) call sizte. What this would look like is some handler consuming the stream of state changes and that handler containing a big match not unlike how .transition(...) works today.
I will say one nice thing about that is it means that places which all issue the same .transition(...) call would be guaranteed to run the same side effects, which could be nice.
| --- | ||
| livekit-signaling: patch | ||
| --- | ||
|
|
||
| # Model the signal connection lifecycle as an explicit state machine | ||
|
|
||
| `SignalInner` tracked its lifecycle in two fields, a stream slot and a `reconnecting` flag, | ||
| that had to be kept in step by hand. It now holds one `SignalState` that owns the transport | ||
| (`Connected`, `Reconnecting`, `Offline`, `Disconnecting`, `Closed`). Every change goes through one |
There was a problem hiding this comment.
nitpick: this file and .changeset/refactor_signaling_explicit_signal_lifecycle_state_machine.md look to be semantically identical?
2 participants
SignalInner tracked its lifecycle in two fields, a stream slot and a
reconnectingAtomicBool, that had to be kept in step by hand across restart, set_reconnected, close and send. It now holds one SignalState that owns the transport: Connected, Reconnecting, Offline, Disconnecting, Closed.Every change goes through SignalState::transition, one match that is the whole table and also says which transport each move releases. An input a state does not accept is logged and refused; a resume from a state that cannot accept it fails with SignalError::InvalidState. The resume gate stays a state until the engine confirms, a stale transport cannot report in because SignalClient::restart awaits the old task first, and ReconnectFailed always lands in Offline.
The held-signal queue moves to a sync parking_lot Mutex that is never held across an await. The old async queue lock was taken in both orders relative to the stream lock, which with tokio's fair RwLock could deadlock against a pending restart writer. A send that fails with any transport error is now held like a SendError was.