Add e2e signalling tests

.changeset/add_e2e_signalling_tests.md

@@ -0,0 +1,8 @@
+---
+livekit: patch
+livekit-api: patch
+livekit-ffi: patch
+livekit-uniffi: patch
+---
+
+differentiate signal connection errors correctly from timeouts - #1234 (@lukasIO)

.github/workflows/test-api.yml

@@ -27,9 +27,10 @@ on:
 jobs:
   # Exercise every runtime backend so a regression in one (e.g. the async/isahc
   # server API silently failing to compile) is caught. Each leg pins a single
-  # runtime via --no-default-features; the mock server backs the legs that make
-  # real requests (services-*), and is a harmless no-op for the signal-client
-  # legs, which spin up their own ephemeral listeners.
+  # runtime via --no-default-features, and every leg reaches the mock server: the
+  # services-* legs over HTTP, the signal-client legs over the WebSocket, selecting
+  # server behaviour with the token's `lk.mock` attribute. So the service container is
+  # a prerequisite for all of them, not a convenience for some.
   livekit-api:
     runs-on: ubuntu-latest
     strategy:
@@ -41,9 +42,9 @@ jobs:
           - name: services (async / isahc)
             cmd: cargo test -p livekit-api --no-default-features --features services-async,access-token --test services_async -- --nocapture
           - name: signal-client (tokio)
-            cmd: cargo test -p livekit-api --no-default-features --features signal-client-tokio --lib signal_client -- --nocapture
+            cmd: cargo test -p livekit-api --no-default-features --features signal-client-tokio,access-token --lib --test signal_tokio --test signal_unreachable -- --nocapture
           - name: signal-client (async)
-            cmd: cargo test -p livekit-api --no-default-features --features signal-client-async --lib signal_client -- --nocapture
+            cmd: cargo test -p livekit-api --no-default-features --features signal-client-async,access-token --lib --test signal_async -- --nocapture
     services:
       mock-server:
         image: livekit/test-server:latest

livekit-api/Cargo.toml

@@ -109,6 +109,9 @@ device-info = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "net", "time", "macros", "io-util"] }
+# Already a hard dependency; repeated here so the integration tests under tests/ can build
+# the protobuf messages they assert on.
+livekit-protocol = { workspace = true }
 # Minimal executor to drive the runtime-agnostic `services-async` (isahc) tests.
 futures = "0.3"
 async-trait = "0.1"

livekit-api/src/signal_client/mod.rs

@@ -39,8 +39,10 @@ use livekit_net::HttpClientExt;
 mod region_url_provider;
 mod signal_stream;
 
-#[cfg(test)]
-pub(crate) mod test_transport;
+// 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;
 
@@ -1055,7 +1057,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 +1093,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(|_| {
@@ -1133,20 +1141,70 @@ mod tests {
     /// in `send`. The stream slot is None so any actual write would be dropped,
     /// which is fine — these tests only assert which side of the queue each
     /// message lands on.
+    #[cfg(feature = "signal-client-tokio")]
     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.
+    #[cfg(feature = "signal-client-tokio")]
+    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,
         })
     }
 
+    #[cfg(feature = "signal-client-tokio")]
+    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.
+    #[cfg(feature = "signal-client-tokio")]
+    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".
+    ///
+    /// Gated like its callers: `test_transport` only exists for the tokio flavour, so an
+    /// ungated helper would break the `signal-client-async` build.
+    #[cfg(feature = "signal-client-tokio")]
+    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 +1286,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/tests/signal_async.rs

@@ -0,0 +1,164 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Signal-connection coverage for the non-tokio runtime flavour.
+//!
+//! The in-crate `signal_test` suite is tokio-only — every case there is a
+//! `#[tokio::test]` gated on `signal-client-tokio` — so under `signal-client-async` the
+//! signal client is type-checked and never exercised. That gap matters because the parts
+//! of the client that differ per runtime are exactly the timing ones:
+//! `livekit_runtime::timeout` around the first-message wait, and the keepalive
+//! `interval` in `signal_task`. A timeout that never fires under this flavour would be
+//! invisible today.
+//!
+//! Same shape as `services_async.rs`: an integration binary pinned to the non-tokio
+//! flavour, driven by `futures::executor::block_on`, run against the shared mock server
+//! (`LK_TEST_SERVER_URL`, default `http://127.0.0.1:9999`). It no-ops when the server is
+//! unreachable.
+//!
+//! Only the public surface is visible from here — `SignalInner`, its queue and the mock
+//! transport are all private to the crate — so this covers client-observable behaviour,
+//! which is the right level for the timing paths anyway.
+#![cfg(all(
+    feature = "signal-client-async",
+    feature = "access-token",
+    not(feature = "signal-client-tokio")
+))]
+
+use std::time::{Duration, Instant};
+
+use livekit_api::access_token::{AccessToken, VideoGrants};
+use livekit_api::signal_client::{SignalClient, SignalError, SignalOptions};
+
+/// The mock verifies tokens against this secret by default.
+const TEST_SECRET: &str = "secret";
+const TEST_API_KEY: &str = "APItest";
+const TEST_ROOM: &str = "test-room";
+const TEST_IDENTITY: &str = "tester";
+
+/// The attribute key the mock reads its signal-behaviour control object from.
+const SIGNAL_CONTROL_ATTRIBUTE: &str = "lk.mock";
+
+/// The mock server's base URL, if it is up.
+///
+/// `None` means "no server here" — a local checkout without one, where the tests it gates
+/// return early instead of failing. Probing the port only, never a response, is deliberate:
+/// a server that is up but answering wrongly is a real regression and must fail a test
+/// rather than be waved through as "offline". Plain TCP also keeps this file free of an
+/// HTTP client, which `access-token` alone does not provide.
+fn online_lk_test_server() -> Option<String> {
+    let base =
+        std::env::var("LK_TEST_SERVER_URL").unwrap_or_else(|_| "http://127.0.0.1:9999".to_owned());
+    let authority = base.split("://").nth(1).unwrap_or(&base).trim_end_matches('/');
+    if std::net::TcpStream::connect(authority).is_ok() {
+        return Some(base);
+    }
+    eprintln!("skipping: mock test server not reachable at {base}");
+    None
+}
+
+/// Mint a token whose `lk.mock` control object selects a server behaviour.
+fn token(mode: &str) -> String {
+    let mut at = AccessToken::with_api_key(TEST_API_KEY, TEST_SECRET)
+        .with_ttl(Duration::from_secs(60 * 60))
+        .with_identity(TEST_IDENTITY)
+        .with_grants(VideoGrants {
+            room_join: true,
+            room: TEST_ROOM.to_owned(),
+            ..Default::default()
+        });
+    if !mode.is_empty() {
+        let control = format!(r#"{{"signal":"{mode}"}}"#);
+        at = at.with_attributes([(SIGNAL_CONTROL_ATTRIBUTE, control.as_str())]);
+    }
+    at.to_jwt().expect("mint token")
+}
+
+/// The connect path end to end on this flavour: WS upgrade, join response, keepalive
+/// config. Proves the transport seam and `livekit_runtime::spawn` work here at all.
+#[test]
+fn signal_async_happy_join() {
+    let Some(base) = online_lk_test_server() else { return };
+
+    futures::executor::block_on(async {
+        let (client, join, _events) =
+            SignalClient::connect(&base, &token(""), SignalOptions::default(), None)
+                .await
+                .expect("connect must succeed against the mock");
+
+        assert_eq!(join.room.expect("room").name, TEST_ROOM);
+        assert!(join.ping_interval > 0, "the mock supplies keepalive config");
+        assert!(join.ping_timeout > 0, "the mock supplies keepalive config");
+        client.close().await;
+    });
+}
+
+/// A server that closes before answering is a close, not a timeout. Same assertion as the
+/// tokio suite's `close_before_join`, which cannot run on this flavour — and the
+/// classification lives in `get_async_message!`, one of the two runtime-sensitive spots.
+#[test]
+fn signal_async_close_before_join_is_a_close() {
+    let Some(base) = online_lk_test_server() else { return };
+
+    futures::executor::block_on(async {
+        let err = SignalClient::connect(
+            &base,
+            &token("close_before_join"),
+            SignalOptions::default(),
+            None,
+        )
+        .await
+        .err()
+        .expect("connect must fail when the server closes before the join");
+
+        assert!(
+            matches!(err, SignalError::Closed),
+            "a close before the answer is a close, not a timeout, got {err:?}"
+        );
+    });
+}
+
+/// The one that actually tests the runtime: the mock accepts the socket and stays silent,
+/// so nothing but `livekit_runtime::timeout` can end the wait. If timers do not drive
+/// under this flavour's executor, this hangs rather than fails — which is itself the
+/// finding.
+#[test]
+fn signal_async_no_first_message_times_out() {
+    let Some(base) = online_lk_test_server() else { return };
+
+    futures::executor::block_on(async {
+        let started = Instant::now();
+        let err = SignalClient::connect(
+            &base,
+            &token("no_first_message"),
+            SignalOptions::default(),
+            None,
+        )
+        .await
+        .err()
+        .expect("connect must fail when the server never answers");
+
+        assert!(
+            matches!(err, SignalError::Timeout(_)),
+            "a silent server is a timeout, got {err:?}"
+        );
+        // The wait is the client's own deadline, so it must have actually elapsed —
+        // otherwise something else failed the connect and the timer was never proven.
+        assert!(
+            started.elapsed() >= Duration::from_secs(1),
+            "the timeout fired after {:?}, too fast to be the first-message deadline",
+            started.elapsed()
+        );
+    });
+}