refactor(signaling): explicit signal lifecycle state machine - #1402
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}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
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?
| 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.
There was a problem hiding this comment.
Yeah, agree, I would like to move into that direction eventually.
I think it should be rather easy to transition (pun intended) to that design later on once we also represent the PC connection as more of a state machine pattern?
There was a problem hiding this comment.
Makes sense, cool, let's revisit it later then.
yeah, that's intended as follow up work |
The machine now starts in Connecting, which is also its Default, so the transition table can use mem::take instead of a placeholder. The constructor drives the first transition before there is a client to hold the state: ConnectComplete(stream) lands in Connected, and any failure, transport or missing JoinResponse, lands in Closed and is dropped with the error. The v1/v0 transport selection moves out of connect into open_transport so the constructor reads as: open, join, transition, build. Behaviour is unchanged. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…it/rust-sdks into lukas/signal-state-machine
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
flush_queue took the whole queue out under a sync lock and then sent each signal with the lock released. The state read lock is shared, so a second Connected sender could see an empty queue and put a newer signal on the wire between two held ones: Mute(true), Mute(false) held, then a concurrent Mute(true) landed before the held Mute(false), and the server kept the older unmute. The queue is an async mutex again, always taken under the state lock so the order is fixed, and held for the whole drain so a concurrent send waits behind the batch. This restores the gate main had before the state machine change dropped it. SignalStream::spawn splits the task setup out of connect so a test can drive a connection that stays alive. The new test spawns two senders against a held batch and asserts the batch reaches the wire first; on the old drain it fails with ["held-1", "b", "held-2", "a"]. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
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: Connecting, 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 lock is now always taken under the state lock, which fixes the lock order: the old code took the queue and stream locks in both orders, and with tokio's fair RwLock that could deadlock against a pending restart writer. The lock stays held while a batch drains, so a concurrent Connected send waits behind older held signals instead of interleaving with them; a test drives two senders against a held batch and asserts the batch reaches the wire first. A send that fails with any transport error is now held like a SendError was.