align with spec and add tests

Author: lukasIO

livekit-api/src/signal_client/mod.rs

@@ -42,6 +42,11 @@ mod signal_stream;
 #[cfg(all(test, feature = "signal-client-tokio", feature = "access-token"))]
 mod signal_test;
 
+// Shared mock WsClient/HttpClient for the unit tests below. Gated on the signal client alone,
+// since the tests that use it do not need access-token.
+#[cfg(all(test, feature = "signal-client-tokio"))]
+mod test_transport;
+
 pub use region_url_provider::RegionUrlProvider;
 
 pub type SignalEmitter = mpsc::UnboundedSender<SignalEvent>;
@@ -1055,7 +1060,10 @@ macro_rules! get_async_message {
                     }
                 }
 
-                Err(SignalError::Timeout("connection closed before message received".into()))
+                // The channel only ends when the read task does, i.e. the transport went away
+                // before the server answered. That is a close, not a timeout — nothing waited.
+                // Only the `livekit_runtime::timeout` wrapper below is a genuine timeout.
+                Err(SignalError::Closed)
             };
 
             livekit_runtime::timeout(JOIN_RESPONSE_TIMEOUT, join).await.map_err(|_| {
@@ -1088,7 +1096,10 @@ async fn get_reconnect_response(
             }
         }
 
-        Err(SignalError::Timeout("connection closed before message received".into()))
+        // The channel only ends when the read task does, i.e. the transport went away
+        // before the server answered. That is a close, not a timeout — nothing waited.
+        // Only the `livekit_runtime::timeout` wrapper below is a genuine timeout.
+        Err(SignalError::Closed)
     };
 
     livekit_runtime::timeout(JOIN_RESPONSE_TIMEOUT, join).await.map_err(|_| {
@@ -1134,19 +1145,61 @@ mod tests {
     /// which is fine — these tests only assert which side of the queue each
     /// message lands on.
     fn make_stub_inner() -> Arc<SignalInner> {
+        make_stub_inner_with(proto::JoinResponse::default())
+    }
+
+    /// As `make_stub_inner`, with a join response — `restart` reads the participant sid from it.
+    fn make_stub_inner_with(join_response: proto::JoinResponse) -> Arc<SignalInner> {
         Arc::new(SignalInner {
             stream: AsyncRwLock::new(None),
             token: Mutex::new(String::new()),
             reconnecting: AtomicBool::new(false),
             queue: Default::default(),
             url: "wss://localhost:7880".to_string(),
             options: SignalOptions::default(),
-            join_response: proto::JoinResponse::default(),
+            join_response,
             request_id: AtomicU32::new(1),
             single_pc_mode_active: false,
         })
     }
 
+    fn mute(sid: &str) -> proto::signal_request::Message {
+        proto::signal_request::Message::Mute(proto::MuteTrackRequest {
+            sid: sid.into(),
+            muted: true,
+        })
+    }
+
+    /// The sids of the queued mute requests, in queue order.
+    async fn queued_sids(inner: &Arc<SignalInner>) -> Vec<String> {
+        inner
+            .queue
+            .lock()
+            .await
+            .iter()
+            .filter_map(|signal| match signal {
+                proto::signal_request::Message::Mute(m) => Some(m.sid.clone()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    /// A live stream over the shared mock transport, for the tests that need the
+    /// difference between "held because we are reconnecting" and "held because
+    /// there is nowhere to send".
+    async fn mock_stream() -> SignalStream {
+        use crate::signal_client::test_transport::install_mock_transport;
+        install_mock_transport();
+        SignalStream::connect(
+            url::Url::parse("wss://localhost:7880/rtc").unwrap(),
+            "",
+            Duration::from_secs(1),
+        )
+        .await
+        .expect("the mock transport always connects")
+        .0
+    }
+
     #[cfg(feature = "signal-client-tokio")]
     #[tokio::test]
     async fn send_queues_queueable_signals_during_reconnect() {
@@ -1228,6 +1281,77 @@ mod tests {
         assert!(!inner.reconnecting.load(Ordering::Acquire), "flag must be cleared");
     }
 
+    /// The queue is FIFO, and the release order is the send order. The existing
+    /// `send_queues_queueable_signals_during_reconnect` only counts what landed there.
+    #[cfg(feature = "signal-client-tokio")]
+    #[tokio::test]
+    async fn queued_signals_keep_their_order() {
+        let inner = make_stub_inner();
+        inner.reconnecting.store(true, Ordering::Release);
+
+        inner.send(mute("first")).await;
+        inner.send(mute("second")).await;
+        inner.send(mute("third")).await;
+
+        assert_eq!(queued_sids(&inner).await, vec!["first", "second", "third"]);
+    }
+
+    /// A resume is not complete when the transport comes back — the engine calls
+    /// `set_reconnected` once the media path is back too, which is seconds later. A
+    /// session-scoped send in that window must queue behind what is already waiting rather
+    /// than overtake it.
+    ///
+    /// Distinct from `send_queues_queueable_signals_during_reconnect`: that one runs with no
+    /// stream at all, so its message could have been queued merely because there was nowhere
+    /// to send it. Here the stream is live and only the `reconnecting` flag holds the message.
+    #[cfg(feature = "signal-client-tokio")]
+    #[tokio::test]
+    async fn send_still_queues_after_the_transport_returns() {
+        let inner = make_stub_inner();
+        inner.reconnecting.store(true, Ordering::Release);
+
+        // issued while the resume is in flight
+        inner.send(mute("held-during-resume")).await;
+
+        // the resume has answered and its transport is installed; `restart` deliberately
+        // leaves `reconnecting` set until the engine reports in
+        *inner.stream.write().await = Some(mock_stream().await);
+        assert!(inner.reconnecting.load(Ordering::Acquire), "restart leaves the flag set");
+
+        inner.send(mute("issued-while-catching-up")).await;
+
+        assert_eq!(
+            queued_sids(&inner).await,
+            vec!["held-during-resume", "issued-while-catching-up"],
+            "a live transport must not let a later send overtake a held one"
+        );
+    }
+
+    /// A failed resume has to leave the flag clear, or every later attempt would route its
+    /// sends to a queue that nothing drains.
+    #[cfg(feature = "signal-client-tokio")]
+    #[tokio::test]
+    async fn restart_failure_resets_the_flag_so_a_retry_can_re_enter() {
+        let _ = mock_stream().await; // installs the shared mock transport
+        let inner = make_stub_inner_with(proto::JoinResponse {
+            participant: Some(proto::ParticipantInfo {
+                sid: "PA_test".into(),
+                ..Default::default()
+            }),
+            ..Default::default()
+        });
+
+        // The mock yields one Pong and then ends the stream, so no ReconnectResponse ever
+        // arrives and the resume fails.
+        let err =
+            inner.restart().await.err().expect("restart must fail without a reconnect answer");
+        assert!(matches!(err, SignalError::Closed), "expected a close, got {err:?}");
+        assert!(
+            !inner.reconnecting.load(Ordering::Acquire),
+            "a failed restart must clear the flag so the next attempt can re-enter"
+        );
+    }
+
     #[test]
     fn livekit_url_test() {
         let io = SignalOptions::default();

livekit-api/src/signal_client/signal_test.rs

@@ -371,8 +371,8 @@ async fn close_before_join() {
         .err()
         .expect("connect must fail when the server closes before the join");
     assert!(
-        matches!(err, SignalError::WsError(_)),
-        "expected a WS error for a close before join, got {err:?}"
+        matches!(err, SignalError::Closed),
+        "a server that closes before answering is a close, not a timeout, got {err:?}"
     );
 }
 
@@ -429,8 +429,8 @@ async fn server_unreachable() {
         .err()
         .expect("connecting to a dead port must fail");
     assert!(
-        matches!(err, SignalError::WsError(_)),
-        "expected a transport (WS) error for an unreachable server, got {err:?}"
+        matches!(err, SignalError::Connection(_)),
+        "expected a transport connection error for an unreachable server, got {err:?}"
     );
 }