diff --git a/.github/workflows/test-api.yml b/.github/workflows/test-api.yml index 113e5647b..39d9511bd 100644 --- a/.github/workflows/test-api.yml +++ b/.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 signal_client -- --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 -- signal_client signal_async --nocapture services: mock-server: image: livekit/test-server:latest diff --git a/livekit-api/src/signal_client/mod.rs b/livekit-api/src/signal_client/mod.rs index 472642332..b3e5516dc 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -39,8 +39,13 @@ use livekit_net::HttpClientExt; mod region_url_provider; mod signal_stream; -#[cfg(test)] -pub(crate) mod test_transport; +#[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; @@ -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(|_| { @@ -1133,7 +1144,14 @@ 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 { + 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 { Arc::new(SignalInner { stream: AsyncRwLock::new(None), token: Mutex::new(String::new()), @@ -1141,12 +1159,55 @@ mod tests { 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) -> Vec { + 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 +1289,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(); diff --git a/livekit-api/src/signal_client/signal_test.rs b/livekit-api/src/signal_client/signal_test.rs new file mode 100644 index 000000000..67abb2374 --- /dev/null +++ b/livekit-api/src/signal_client/signal_test.rs @@ -0,0 +1,530 @@ +// 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 tests against the shared mock LiveKit server (livekit/livekit +//! cmd/test-server). Point them at a running instance with `LK_TEST_SERVER_URL` +//! (default `http://127.0.0.1:9999`); they no-op when no server is reachable. In +//! CI the server is booted as a Docker container. +//! +//! Unlike the Twirp API tests, signal behavior is not selected by a request +//! header (a WebSocket client cannot set one) but by the `lk.mock` participant +//! attribute embedded in the access token — see cmd/test-server/README.md +//! ("Signal connection (WebSocket) mocking"). Each test mints a token whose +//! `lk.mock` attribute selects a server behavior, then asserts the +//! *client-observable* outcome: how [`SignalClient::connect`] / [`SignalClient::restart`] +//! classify the connection, and what [`SignalEvent`]s reach the caller. +//! +//! These are the Rust counterparts to cmd/test-server/signal_test.go, which +//! exercises the same modes from the server side. + +use std::time::Duration; + +use livekit_protocol as proto; +use tokio::time::timeout; + +use super::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions}; +use crate::access_token::{AccessToken, VideoGrants}; + +/// The mock verifies tokens against this secret by default (matches +/// `livekit-server --dev` and the test-server's `--api-secret` 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-behavior control object from. +const SIGNAL_CONTROL_ATTRIBUTE: &str = "lk.mock"; + +fn base_url() -> String { + std::env::var("LK_TEST_SERVER_URL").unwrap_or_else(|_| "http://127.0.0.1:9999".to_owned()) +} + +/// Reachability probe over a plain TCP connect rather than an HTTP request, so this file +/// needs no HTTP client and can therefore run under a pinned `signal-client-tokio` build +/// (the HTTP client arrives only with `services-tokio`). It also draws the better line: +/// "server offline" (skip, local dev) is told apart from "server up but answering wrongly", +/// which must fail a test rather than silently skip it. `services_async.rs` probes the same +/// way. +fn server_up(base: &str) -> bool { + let authority = base.split("://").nth(1).unwrap_or(base).trim_end_matches('/'); + std::net::TcpStream::connect(authority).is_ok() +} + +/// Mint a token whose `lk.mock` attribute selects `mode` (empty → no attribute, +/// which the mock treats as the happy path). +fn token(mode: &str) -> String { + mint(mode, None) +} + +/// Mint a token whose `lk.mock` control object carries both a `signal` mode and +/// an explicit `leaveAction` (a `LeaveRequest_Action` enum value). +fn token_with_leave_action(mode: &str, leave_action: proto::leave_request::Action) -> String { + mint(mode, Some(leave_action as i32)) +} + +fn mint(mode: &str, leave_action: Option) -> 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 = match leave_action { + Some(action) => format!(r#"{{"signal":"{mode}","leaveAction":{action}}}"#), + None => format!(r#"{{"signal":"{mode}"}}"#), + }; + at = at.with_attributes([(SIGNAL_CONTROL_ATTRIBUTE, control.as_str())]); + } + + at.to_jwt().expect("mint token") +} + +fn options(single_peer_connection: bool) -> SignalOptions { + SignalOptions { single_peer_connection, ..Default::default() } +} + +async fn connect( + base: &str, + token: &str, + single_peer_connection: bool, +) -> super::SignalResult<(SignalClient, proto::JoinResponse, SignalEvents)> { + SignalClient::connect(base, token, options(single_peer_connection), None).await +} + +/// Await the next [`SignalEvent`], failing the test if none arrives within `dur`. +async fn next_event(events: &mut SignalEvents, dur: Duration) -> SignalEvent { + timeout(dur, events.recv()) + .await + .expect("timed out waiting for a signal event") + .expect("signal event stream closed unexpectedly") +} + +macro_rules! skip_if_offline { + ($base:expr) => { + if !server_up(&$base) { + eprintln!("skipping: mock test server not reachable at {}", $base); + return; + } + }; +} + +// -- happy path ------------------------------------------------------------- + +/// `happy` — the WS sends a `JoinResponse` populated with the room from the +/// token and non-zero ping config (so the client arms keepalive). +#[tokio::test] +async fn happy_join() { + let base = base_url(); + skip_if_offline!(base); + + let (client, join, _events) = + connect(&base, &token("happy"), false).await.expect("happy connect should succeed"); + + assert_eq!(join.room.as_ref().expect("join.room").name, TEST_ROOM); + assert!(join.participant.is_some(), "join must carry participant info"); + assert!(join.server_info.is_some(), "join must carry server info"); + assert!( + join.ping_interval > 0 && join.ping_timeout > 0, + "keepalive config must be non-zero: interval={} timeout={}", + join.ping_interval, + join.ping_timeout + ); + + client.close().await; +} + +/// `happy` over the v1 (single-PC) path: `/rtc/v1` behaves identically and still +/// yields a `JoinResponse`. +#[tokio::test] +async fn v1_path_happy() { + let base = base_url(); + skip_if_offline!(base); + + let (client, join, _events) = + connect(&base, &token("happy"), true).await.expect("v1 happy connect should succeed"); + + assert_eq!(join.room.as_ref().expect("join.room").name, TEST_ROOM); + assert!(client.is_single_pc_mode_active(), "v1 path should activate single-PC mode"); + + client.close().await; +} + +/// The client keeps the connection alive: the mock pongs the client's pings, so +/// no `Close` is emitted within a window that exceeds the join's ping timeout. +/// (Mirrors the server-side ping/pong assertions in `TestHappyJoinAndPingPong`.) +#[tokio::test] +async fn happy_stays_connected() { + let base = base_url(); + skip_if_offline!(base); + + let (client, join, mut events) = + connect(&base, &token("happy"), false).await.expect("happy connect should succeed"); + + // Wait comfortably past the ping timeout: if pongs weren't flowing, the + // signal task would emit Close("ping timeout") by then. + let window = Duration::from_secs(join.ping_timeout as u64) + Duration::from_secs(2); + if let Ok(Some(event)) = timeout(window, events.recv()).await { + match event { + SignalEvent::Close(reason) => { + panic!("connection closed while it should stay alive: {reason}") + } + SignalEvent::Message(_) => { /* server-initiated messages are fine */ } + } + } + + client.close().await; +} + +// -- reconnect -------------------------------------------------------------- + +/// A `reconnect=1` connection yields a `ReconnectResponse` rather than a join. +/// The client drives this via [`SignalClient::restart`] after an initial connect. +#[tokio::test] +async fn reconnect_response() { + let base = base_url(); + skip_if_offline!(base); + + let (client, _join, _events) = + connect(&base, &token("happy"), false).await.expect("initial connect should succeed"); + + client.restart().await.expect("restart should yield a ReconnectResponse"); + + client.close().await; +} + +/// `leave_during_reconnect` — on a `reconnect=1` connection the server sends a +/// `LeaveRequest` first, which the client surfaces as `SignalError::LeaveRequest` +/// from `restart()` (the resume cannot complete). A non-reconnect connection is +/// unaffected, so the initial connect still succeeds. +#[tokio::test] +async fn leave_during_reconnect() { + let base = base_url(); + skip_if_offline!(base); + + let (client, _join, _events) = connect(&base, &token("leave_during_reconnect"), false) + .await + .expect("initial (non-reconnect) connect should succeed"); + + let err = client.restart().await.expect_err("restart should surface the server's LeaveRequest"); + match err { + SignalError::LeaveRequest { reason, action } => { + assert_eq!(reason, proto::DisconnectReason::ServerShutdown); + assert_eq!(action, proto::leave_request::Action::Disconnect); + } + other => panic!("expected SignalError::LeaveRequest, got {other:?}"), + } + + client.close().await; +} + +// -- post-join disconnects -------------------------------------------------- + +/// `no_pong` — the server sends the join but never pongs, so the client's +/// keepalive fires and the signal task emits `Close` (ping timeout). +#[tokio::test] +async fn no_pong_times_out() { + let base = base_url(); + skip_if_offline!(base); + + let (client, join, mut events) = connect(&base, &token("no_pong"), false) + .await + .expect("connect should succeed before the ping timeout"); + + // Ping timeout is short (join.ping_timeout ~3s); allow some slack. + let window = Duration::from_secs(join.ping_timeout as u64) + Duration::from_secs(3); + match next_event(&mut events, window).await { + // The reason must identify the ping timeout so the engine can classify + // the disconnect (client-sdk-js asserts the same string). + SignalEvent::Close(reason) => assert!( + reason.contains("ping timeout"), + "expected a ping-timeout close reason, got {reason:?}" + ), + SignalEvent::Message(msg) => panic!("expected Close on ping timeout, got message: {msg:?}"), + } + + client.close().await; +} + +/// `close_when_connected` — the server sends the join, then cleanly closes +/// (code 1011). The client observes the stream ending as a `Close` event. +#[tokio::test] +async fn close_when_connected() { + let base = base_url(); + skip_if_offline!(base); + + let (client, _join, mut events) = connect(&base, &token("close_when_connected"), false) + .await + .expect("connect should succeed before the server closes"); + + match next_event(&mut events, Duration::from_secs(3)).await { + SignalEvent::Close(reason) => { + assert!(!reason.is_empty(), "a transport close should carry a reason") + } + SignalEvent::Message(msg) => { + panic!("expected Close after server close, got message: {msg:?}") + } + } + + client.close().await; +} + +/// `drop_when_connected` — the server sends the join, then abruptly drops the +/// TCP connection (no close handshake → abnormal 1006). The client still +/// surfaces this as a `Close` event. +#[tokio::test] +async fn drop_when_connected() { + let base = base_url(); + skip_if_offline!(base); + + let (client, _join, mut events) = connect(&base, &token("drop_when_connected"), false) + .await + .expect("connect should succeed before the server drops"); + + match next_event(&mut events, Duration::from_secs(3)).await { + SignalEvent::Close(_) => {} + SignalEvent::Message(msg) => { + panic!("expected Close after abrupt drop, got message: {msg:?}") + } + } + + client.close().await; +} + +/// `leave_when_connected` — the server sends the join, then a `LeaveRequest`. +/// The client forwards it to the caller as a `Message(Leave)` (the engine layer +/// decides how to act on it), carrying `SERVER_SHUTDOWN` and the default +/// `DISCONNECT` action. +#[tokio::test] +async fn leave_when_connected() { + let base = base_url(); + skip_if_offline!(base); + + let (client, _join, mut events) = connect(&base, &token("leave_when_connected"), false) + .await + .expect("connect should succeed before the leave"); + + let leave = recv_leave(&mut events).await; + assert_eq!(leave.reason(), proto::DisconnectReason::ServerShutdown); + assert_eq!(leave.action(), proto::leave_request::Action::Disconnect); + + client.close().await; +} + +/// The `leaveAction` control field overrides the action on emitted leaves. +#[tokio::test] +async fn leave_action_override() { + let base = base_url(); + skip_if_offline!(base); + + let tok = + token_with_leave_action("leave_when_connected", proto::leave_request::Action::Reconnect); + let (client, _join, mut events) = + connect(&base, &tok, false).await.expect("connect should succeed before the leave"); + + let leave = recv_leave(&mut events).await; + assert_eq!(leave.action(), proto::leave_request::Action::Reconnect); + + client.close().await; +} + +/// Read events until a `Leave` message arrives (skipping unrelated server +/// messages, e.g. a token refresh), failing if the stream closes first. +async fn recv_leave(events: &mut SignalEvents) -> proto::LeaveRequest { + let deadline = Duration::from_secs(3); + loop { + match next_event(events, deadline).await { + SignalEvent::Message(msg) => { + if let proto::signal_response::Message::Leave(leave) = *msg { + return leave; + } + } + SignalEvent::Close(reason) => panic!("stream closed before a Leave arrived: {reason}"), + } + } +} + +// -- connect-time failures -------------------------------------------------- + +/// `close_before_join` — the WS upgrade succeeds, then the server closes before +/// sending any first message. The client fails the connect (the stream ended +/// while it was waiting for the join). +#[tokio::test] +async fn close_before_join() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("close_before_join"), false) + .await + .err() + .expect("connect must fail when the server closes before the join"); + assert!( + matches!(err, SignalError::Closed), + "a server that closes before answering is a close, not a timeout, got {err:?}" + ); +} + +/// `no_first_message` — the WS is accepted but the server sends nothing. The +/// client times out waiting for the join. +#[tokio::test] +async fn no_first_message_times_out() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("no_first_message"), false) + .await + .err() + .expect("connect must fail when no first message arrives"); + assert!( + matches!(err, SignalError::Timeout(_)), + "expected a Timeout waiting for the join, got {err:?}" + ); +} + +/// `leave_first_message` on an initial (non-reconnect) connection: the server +/// sends a `LeaveRequest` as the very first message instead of a join, so the +/// connect must be rejected. +/// +/// client-sdk-js rejects here by validating the first message and failing fast +/// on the leave. The Rust initial-join path only recognises a `JoinResponse`, +/// so it surfaces the same *outcome* (a rejected connect) but as a +/// `JOIN_RESPONSE` timeout rather than a dedicated leave error — the reconnect +/// path (`get_reconnect_response`) does classify a leave-first as +/// `SignalError::LeaveRequest`; see `leave_during_reconnect`. +#[tokio::test] +async fn leave_first_message_rejects_join() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("leave_first_message"), false) + .await + .err() + .expect("connect must be rejected when a leave arrives as the first message"); + assert!( + matches!(err, SignalError::Timeout(_)), + "expected the initial join to be rejected (as a timeout), got {err:?}" + ); +} + +/// A server that isn't listening: the WS connect is refused and the `validate` +/// probe also fails to connect, so the original transport error is surfaced. +/// (client-sdk-js classifies this as `ServerUnreachable`.) Needs no mock server. +#[tokio::test] +async fn server_unreachable() { + // Nothing is listening on this port, so the connect is refused immediately. + let err = connect("ws://127.0.0.1:59999", &token("happy"), false) + .await + .err() + .expect("connecting to a dead port must fail"); + assert!( + matches!(err, SignalError::Connection(_)), + "expected a transport connection error for an unreachable server, got {err:?}" + ); +} + +// -- validate-endpoint error classification --------------------------------- +// +// When the WS upgrade is refused, the client falls back to the `/rtc/validate` +// fetch to obtain a definitive HTTP status/body, and classifies the result as a +// client (4xx) or server (5xx) error. These assert that classification. + +/// `validate_500` — WS refused with 500; the `validate` fallback returns 500, +/// which Rust surfaces as `SignalError::Server(500)`. +/// +/// DIVERGENCE from client-sdk-js (intentional): there the WS-rejection error +/// shadows the 5xx and the connect is classified as a generic `WebSocket` +/// error — its own comment notes only 401/403/404 override the ws error. Rust's +/// `validate` step exists precisely to recover the real HTTP status (see the +/// issue #1042 fix in `SignalInner::validate`), so preserving the 500 is the +/// more useful behavior; we assert that rather than matching the JS shadowing. +#[tokio::test] +async fn validate_500_is_server_error() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("validate_500"), false) + .await + .err() + .expect("a 500 must fail the connect"); + match err { + SignalError::Server(status, _) => assert_eq!(status.as_u16(), 500), + other => panic!("expected SignalError::Server(500), got {other:?}"), + } +} + +/// `validate_service_not_found` — 404 without the room marker → client error +/// whose body does NOT contain the "requested room does not exist" marker. +#[tokio::test] +async fn validate_service_not_found_is_client_error() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("validate_service_not_found"), false) + .await + .err() + .expect("a 404 must fail the connect"); + match err { + SignalError::Client(status, body) => { + assert_eq!(status.as_u16(), 404); + assert!( + !body.contains("requested room does not exist"), + "service-not-found body must not carry the room marker, got {body:?}" + ); + } + other => panic!("expected SignalError::Client(404), got {other:?}"), + } +} + +/// `room_not_found` — 404 with the "requested room does not exist" marker. +#[tokio::test] +async fn room_not_found_is_client_error_with_marker() { + let base = base_url(); + skip_if_offline!(base); + + let err = connect(&base, &token("room_not_found"), false) + .await + .err() + .expect("a 404 must fail the connect"); + match err { + SignalError::Client(status, body) => { + assert_eq!(status.as_u16(), 404); + assert!( + body.contains("requested room does not exist"), + "room-not-found body must carry the room marker, got {body:?}" + ); + } + other => panic!("expected SignalError::Client(404), got {other:?}"), + } +} + +/// A malformed/unsigned token is rejected: the WS refuses with 401 and the +/// validate fallback confirms it as a client error. +#[tokio::test] +async fn bad_token_is_client_error() { + let base = base_url(); + skip_if_offline!(base); + + let err = + connect(&base, "not-a-jwt", false).await.err().expect("a bad token must fail the connect"); + match err { + SignalError::Client(status, _) => assert_eq!(status.as_u16(), 401), + // A syntactically invalid bearer may be rejected before the validate + // round-trip; either classification is a legitimate rejection. + SignalError::TokenFormat => {} + other => panic!("expected a 401 client error (or TokenFormat), got {other:?}"), + } +} diff --git a/livekit-api/tests/signal_async.rs b/livekit-api/tests/signal_async.rs new file mode 100644 index 000000000..79ab26638 --- /dev/null +++ b/livekit-api/tests/signal_async.rs @@ -0,0 +1,170 @@ +// 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"; + +fn base_url() -> String { + std::env::var("LK_TEST_SERVER_URL").unwrap_or_else(|_| "http://127.0.0.1:9999".to_owned()) +} + +/// Reachability probe kept separate from the assertions, so "server offline" (skip, local +/// dev) is told apart from "server up but the client misbehaved" (a real regression). The +/// tokio suite probes the same way. +fn server_up(base: &str) -> bool { + let authority = base.split("://").nth(1).unwrap_or(base).trim_end_matches('/'); + std::net::TcpStream::connect(authority).is_ok() +} + +/// 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") +} + +macro_rules! skip_if_offline { + ($base:expr) => { + if !server_up(&$base) { + eprintln!("skipping: mock test server not reachable at {}", $base); + return; + } + }; +} + +/// 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 base = base_url(); + skip_if_offline!(base); + + 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 base = base_url(); + skip_if_offline!(base); + + 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 base = base_url(); + skip_if_offline!(base); + + 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() + ); + }); +}