diff --git a/.changeset/report_reconnect_reason_on_resume.md b/.changeset/report_reconnect_reason_on_resume.md new file mode 100644 index 000000000..aeab4089d --- /dev/null +++ b/.changeset/report_reconnect_reason_on_resume.md @@ -0,0 +1,14 @@ +--- +livekit: patch +livekit-api: patch +livekit-ffi: patch +livekit-uniffi: patch +--- + +Report the reconnect reason to the server when resuming. + +Resumes previously sent no reason, so server-side telemetry could not attribute why Rust +clients reconnect — every resume looked like `RR_UNKNOWN`. The engine now records what caused +the episode (signal disconnected, publisher failed, subscriber failed) and reports it on each +resume attempt. The v0 signalling path was also missing the `reconnect_reason` query parameter +entirely, so it would not have been reported even if a reason had been supplied. diff --git a/livekit-api/src/signal_client/mod.rs b/livekit-api/src/signal_client/mod.rs index 29095dc13..749337c29 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -302,10 +302,15 @@ impl SignalClient { /// in the queue. Caller MUST invoke [`Self::set_reconnected`] once the resume /// has fully recovered (PC connected, SyncState sent) to drain the queue and /// re-enable normal sends. - pub async fn restart(&self) -> SignalResult { + /// Reopen the signalling link for a resume, reporting `reason` to the server so it can + /// attribute why the client reconnected. + pub async fn restart( + &self, + reason: proto::ReconnectReason, + ) -> SignalResult { self.close().await; - let (reconnect_response, stream_events) = self.inner.restart().await?; + let (reconnect_response, stream_events) = self.inner.restart(reason).await?; let signal_task = livekit_runtime::spawn(signal_task( self.inner.clone(), self.emitter.clone(), @@ -540,6 +545,7 @@ impl SignalInner { /// stream is in place. pub async fn restart( self: &Arc, + reason: proto::ReconnectReason, ) -> SignalResult<( proto::ReconnectResponse, mpsc::UnboundedReceiver>, @@ -561,7 +567,7 @@ impl SignalInner { &self.options, self.single_pc_mode_active, true, - None, + Some(reason as i32), sid, None, ) @@ -958,6 +964,16 @@ fn get_livekit_url( .query_pairs_mut() .append_pair("reconnect", "1") .append_pair("sid", participant_sid); + + // The server parses this with `strconv.Atoi` (`rtcservice.go`), so it is the + // enum's numeric value stringified, not its name. `reconnect_reason` is already + // an `i32` (the caller casts), so `to_string` yields e.g. "3". On the v1 path the + // equivalent travels inside the JoinRequest protobuf as an integer field. + if let Some(reason_value) = reconnect_reason { + lk_url + .query_pairs_mut() + .append_pair("reconnect_reason", reason_value.to_string().as_str()); + } } } @@ -1105,6 +1121,69 @@ mod tests { proto::JoinRequest::decode(wrapped.join_request.as_slice()).unwrap() } + /// The server attributes client reconnects from this value + /// (`rtcservice.go`: `r.FormValue("reconnect_reason")`), so a resume that omits it is + /// invisible in that telemetry. Both signalling paths must carry it: v0 as a query + /// parameter, v1 inside the JoinRequest protobuf. + #[test] + fn resume_reports_the_reconnect_reason_on_both_signalling_paths() { + let reason = proto::ReconnectReason::RrSubscriberFailed; + + let v0 = get_livekit_url( + "ws://localhost:7880", + &SignalOptions::default(), + /* use_v1_path= */ false, + /* reconnect= */ true, + Some(reason as i32), + "PA_test", + None, + ) + .unwrap(); + let v0_params: Vec<(String, String)> = + v0.query_pairs().map(|(k, v)| (k.into_owned(), v.into_owned())).collect(); + // The server parses this with `strconv.Atoi`, so it must be the enum's numeric value + // stringified ("3"), never its name ("RR_SUBSCRIBER_FAILED"). Pinned as a literal so + // the wire format is asserted rather than restated. + assert_eq!(reason as i32, 3); + assert!( + v0_params.contains(&("reconnect_reason".to_string(), "3".to_string())), + "v0 resume must send reconnect_reason as the stringified number, got {v0_params:?}" + ); + + let v1 = get_livekit_url( + "ws://localhost:7880", + &SignalOptions::default(), + /* use_v1_path= */ true, + /* reconnect= */ true, + Some(reason as i32), + "PA_test", + None, + ) + .unwrap(); + let join_request_param = v1 + .query_pairs() + .find(|(k, _)| k == "join_request") + .map(|(_, v)| v.into_owned()) + .expect("v1 resume must carry a join_request"); + assert_eq!(decode_join_request_param_for_test(&join_request_param).reconnect_reason, 3); + } + + /// An initial connect is not a reconnect, so it must not claim a reason. + #[test] + fn initial_connect_sends_no_reconnect_reason() { + let url = get_livekit_url( + "ws://localhost:7880", + &SignalOptions::default(), + /* use_v1_path= */ false, + /* reconnect= */ false, + None, + "", + None, + ) + .unwrap(); + assert!(url.query_pairs().all(|(k, _)| k != "reconnect_reason")); + } + #[test] fn client_info_sdk_for_name_maps_known_sdks() { assert_eq!(client_info_sdk_for_name("cpp"), proto::client_info::Sdk::Cpp); diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 2895cfab5..66ff571cc 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -246,6 +246,12 @@ struct EngineHandle { // Carried through so that, if reconnection ultimately fails, the engine // closes with the original cause rather than a generic `UnknownReason`. reconnect_reason: DisconnectReason, + + // The cause reported to the server on each resume attempt of this episode, so + // server-side telemetry can attribute why clients reconnect. Set from the + // failure that started the episode and not overwritten by later failures, + // which are consequences of the first. + reported_reconnect_reason: proto::ReconnectReason, engine_task: Option<(JoinHandle<()>, oneshot::Sender<()>)>, } @@ -484,6 +490,7 @@ impl EngineInner { can_reconnect: true, full_reconnect: false, reconnect_reason: DisconnectReason::UnknownReason, + reported_reconnect_reason: proto::ReconnectReason::RrUnknown, engine_task: None, }), options, @@ -580,7 +587,7 @@ impl EngineInner { async fn on_session_event(self: &Arc, event: SessionEvent) -> EngineResult<()> { match event { - SessionEvent::Close { source, reason, action, retry_now } => { + SessionEvent::Close { source, reason, reconnect_reason, action, retry_now } => { match action { proto::leave_request::Action::Resume | proto::leave_request::Action::Reconnect => { @@ -605,6 +612,7 @@ impl EngineInner { retry_now, action == proto::leave_request::Action::Reconnect, reason, + reconnect_reason, ); } proto::leave_request::Action::Disconnect => { @@ -805,6 +813,7 @@ impl EngineInner { retry_now: bool, full_reconnect: bool, reason: DisconnectReason, + reported_reconnect_reason: proto::ReconnectReason, ) { let mut running_handle = self.running_handle.write(); @@ -839,8 +848,9 @@ impl EngineInner { // full reconnect in `try_restart_connection`. running_handle.full_reconnect |= full_reconnect; // Remember the cause so a failed reconnection closes with it rather than - // a generic UnknownReason. + // a generic UnknownReason, and so each attempt can report it to the server. running_handle.reconnect_reason = reason; + running_handle.reported_reconnect_reason = reported_reconnect_reason; livekit_runtime::spawn({ let inner = self.clone(); @@ -1123,7 +1133,12 @@ impl EngineInner { // the next cycle; pre-fix it was dropped and the engine resumed again. if self.fail_transport_during_next_resume.swap(false, Ordering::AcqRel) { log::warn!("test fault injection: simulating concurrent failure during resume"); - self.reconnection_needed(false, false, DisconnectReason::UnknownReason); + self.reconnection_needed( + false, + false, + DisconnectReason::UnknownReason, + proto::ReconnectReason::RrUnknown, + ); } } @@ -1131,7 +1146,8 @@ impl EngineInner { // 1. Reopen the signalling link. The SignalClient stays gated // (`reconnecting=true`) so queueable mutations buffer until step 4. - let reconnect_response = session.restart().await?; + let reported_reason = self.running_handle.read().reported_reconnect_reason; + let reconnect_response = session.restart(reported_reason).await?; // 2. Hand the ReconnectResponse to the room and wait until it has sent // SyncState, which must precede the publisher re-offer. diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index bb6c45f38..e6057e870 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -186,6 +186,10 @@ pub enum SessionEvent { Close { source: String, reason: DisconnectReason, + /// Cause reported to the server on the next resume, so it can attribute + /// client reconnects. Distinct from `reason`, which is the disconnect + /// reason used if recovery ultimately fails. + reconnect_reason: proto::ReconnectReason, action: proto::leave_request::Action, retry_now: bool, }, @@ -819,8 +823,11 @@ impl RtcSession { self.inner.publish_data(data, kind, is_raw_packet).await } - pub async fn restart(&self) -> EngineResult { - self.inner.restart().await + pub async fn restart( + &self, + reason: proto::ReconnectReason, + ) -> EngineResult { + self.inner.restart(reason).await } pub async fn restart_publisher(&self) -> EngineResult<()> { @@ -1119,6 +1126,7 @@ impl SessionInner { self.on_session_disconnected( format!("signal client closed: {:?}", reason).as_str(), DisconnectReason::UnknownReason, + proto::ReconnectReason::RrSignalDisconnected, proto::leave_request::Action::Resume, false, ); @@ -1418,6 +1426,8 @@ impl SessionInner { self.on_session_disconnected( "server request to leave", leave.reason(), + // The server initiated this and already knows why. + proto::ReconnectReason::RrUnknown, leave.action(), true, ); @@ -1589,6 +1599,10 @@ impl SessionInner { self.on_session_disconnected( "pc_state failed", DisconnectReason::UnknownReason, + match target { + SignalTarget::Subscriber => proto::ReconnectReason::RrSubscriberFailed, + SignalTarget::Publisher => proto::ReconnectReason::RrPublisherFailed, + }, proto::leave_request::Action::Resume, false, ); @@ -1961,6 +1975,7 @@ impl SessionInner { &self, source: &str, reason: DisconnectReason, + reconnect_reason: proto::ReconnectReason, action: proto::leave_request::Action, retry_now: bool, ) { @@ -1972,6 +1987,7 @@ impl SessionInner { let _ = self.emitter.send(SessionEvent::Close { source: source.to_owned(), reason, + reconnect_reason, action, retry_now, }); @@ -2188,12 +2204,15 @@ impl SessionInner { /// This reconnection if more seemless compared to the full reconnection implemented in /// ['RTCEngine'] - async fn restart(&self) -> EngineResult { + async fn restart( + &self, + reason: proto::ReconnectReason, + ) -> EngineResult { // Start accumulating before the signal client reconnects: once // `restart` returns, the resumed stream immediately delivers events // (including the post-resume participant snapshot) on a concurrent task. *self.resume_seen_identities.lock() = Some(HashSet::new()); - let reconnect_response = self.signal_client.restart().await?; + let reconnect_response = self.signal_client.restart(reason).await?; log::debug!("received reconnect response: {:?}", reconnect_response); let rtc_config =