feat(rtc_engine): report the reconnect reason to the server on resume

.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.

livekit-api/src/signal_client/mod.rs

@@ -302,10 +302,15 @@
     /// 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<proto::ReconnectResponse> {
+    /// 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<proto::ReconnectResponse> {
         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 @@
     /// stream is in place.
     pub async fn restart(
         self: &Arc<Self>,
+        reason: proto::ReconnectReason,
     ) -> SignalResult<(
         proto::ReconnectResponse,
         mpsc::UnboundedReceiver<Box<proto::signal_response::Message>>,
@@ -561,7 +567,7 @@
             &self.options,
             self.single_pc_mode_active,
             true,
-            None,
+            Some(reason as i32),
             sid,
             None,
         )
@@ -958,6 +964,16 @@
                 .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 @@
         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);
@@ -1133,7 +1212,7 @@
             single_pc_mode_active: false,
         })
     }
 
     #[cfg(feature = "signal-client-tokio")]
     #[tokio::test]
     async fn send_queues_queueable_signals_during_reconnect() {

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<Self>, 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,15 +1133,21 @@ 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,
+                );
             }
         }
 
         let session = self.running_handle.read().session.clone();
 
         // 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.

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<proto::ReconnectResponse> {
-        self.inner.restart().await
+    pub async fn restart(
+        &self,
+        reason: proto::ReconnectReason,
+    ) -> EngineResult<proto::ReconnectResponse> {
+        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<proto::ReconnectResponse> {
+    async fn restart(
+        &self,
+        reason: proto::ReconnectReason,
+    ) -> EngineResult<proto::ReconnectResponse> {
         // 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 =