From a07bb3632728eab550828b8d1899c8295b69ce11 Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Tue, 2 Jun 2026 16:23:46 -0600 Subject: [PATCH] feat(qconnect): factor transport loop into QconnectApp::run_session_loop (slice 5) Move the spawned transport event-loop body out of the Tauri adapter into the shared QconnectApp::run_session_loop, so both frontends inherit takeover arbitration / deferred renderer-join / reconnect resync / Lagged re-ask. The adapter-coupled seams go through a new SessionLoopHost trait: - update_lifecycle -- Tauri gates on a live runtime + dedups, emits via sink - bootstrap_after_reconnect - deferred_renderer_join -- R2: the CoreBridge track-duration read stays adapter-side - on_reconnect_exhausted -- R3: Exhausted + last_error + runtime=None teardown + the raw qconnect:status_changed emit; returns whether the loop should break - on_loop_error -- the raw qconnect:error channel TauriSessionLoopHost implements it over the existing service handles; connect() is reduced to: build host -> spawn(app.run_session_loop(host, idle_retry)). The diagnostic/lifecycle/resync emits still route through self.sink, so the Svelte wire surface is byte-identical. hex_preview + queue_payload_track_preview + the re-ask budget const relocate with the loop; the dead Tauri copies are removed. R3 (QconnectRuntime/JoinHandle ownership, #358 latch, disconnect teardown) stays adapter-side. qconnect-app 38 green, src-tauri qconnect 42 green (the byte-identity fence). cc --- crates/qconnect-app/src/app.rs | 353 ++++++++++++++++++++- crates/qconnect-app/src/lib.rs | 3 +- src-tauri/src/qconnect/service.rs | 458 ++++++---------------------- src-tauri/src/qconnect/session.rs | 7 - src-tauri/src/qconnect/transport.rs | 13 - 5 files changed, 446 insertions(+), 388 deletions(-) diff --git a/crates/qconnect-app/src/app.rs b/crates/qconnect-app/src/app.rs index 0e39b578c..b848c93bd 100644 --- a/crates/qconnect-app/src/app.rs +++ b/crates/qconnect-app/src/app.rs @@ -18,11 +18,14 @@ use serde_json::Value; use tokio::sync::Mutex; use uuid::Uuid; +use async_trait::async_trait; + use crate::session::{ - is_local_renderer_active, is_peer_renderer_active, normalize_active_renderer_id, - refresh_local_renderer_id, should_arm_renderer_watchdog, LocalIdentity, QconnectRendererInfo, - RendererStatus, ServerActiveState, PLAYING_STATE_PLAYING, PLAYING_STATE_UNKNOWN, - QCONNECT_RENDERER_LOST_TIMEOUT_MS, + compute_connection_state, deferred_join_reason, is_local_renderer_active, + is_peer_renderer_active, normalize_active_renderer_id, refresh_local_renderer_id, + should_arm_renderer_watchdog, should_reask_queue_state, LocalIdentity, QconnectLifecycleState, + QconnectRendererInfo, RendererStatus, ServerActiveState, PLAYING_STATE_PLAYING, + PLAYING_STATE_UNKNOWN, QCONNECT_RENDERER_LOST_TIMEOUT_MS, }; use crate::{ ensure_session_renderer_state, sync_session_renderer_active_flags, QconnectAppError, @@ -1603,6 +1606,348 @@ where } } +/// Attempt budget for the Lagged-recovery re-AskForQueueState loop (P1-8). +const QCONNECT_REASK_QUEUE_STATE_MAX_ATTEMPTS: u32 = 5; + +/// Adapter-provided seams for the shared session loop. `run_session_loop` is +/// frontend-agnostic; it calls these for the bits that must stay adapter-side: +/// the lifecycle gating (depends on the adapter's runtime ownership), the +/// reconnect-exhausted teardown (R3), controller bootstrap, the renderer-join +/// reports (which read the engine's track duration — R2), and the raw error +/// surface. Both the Tauri adapter and a future Slint adapter implement this. +#[async_trait] +pub trait SessionLoopHost: Send + Sync { + /// Surface a lifecycle transition. The Tauri adapter gates + dedups this + /// (emits `qconnect:status_changed` via the sink only while a runtime is + /// still alive); a Slint adapter does its own. + async fn update_lifecycle(&self, state: QconnectLifecycleState); + /// Re-bootstrap controller presence after a reconnect (JoinSession + ask + /// queue state). Logs its own errors. + async fn bootstrap_after_reconnect(&self); + /// Renderer JoinSession + initial reports for `session_uuid` (R2: reads the + /// current track duration from the adapter's engine). Idempotent per uuid. + async fn deferred_renderer_join(&self, session_uuid: String, reason: i32); + /// Handle MaxReconnectAttemptsExceeded (R3): set Exhausted + last_error, drop + /// the runtime unless idle-retry is active, surface the terminal status. + /// Returns true if the loop should break (terminate), false to keep idling. + async fn on_reconnect_exhausted( + &self, + attempts: u32, + last_reason: String, + idle_retry_active: bool, + ) -> bool; + /// Surface a transport-handling error (the raw `qconnect:error` channel on Tauri). + async fn on_loop_error(&self, message: String); +} + +/// Hex preview of up to `max_bytes` bytes for diagnostic logging. Mirrors the +/// Tauri adapter's `hex_preview` (relocated with the loop, slice 5). +fn hex_preview(data: &[u8], max_bytes: usize) -> String { + let take = data.len().min(max_bytes); + let hex: String = data[..take] + .iter() + .map(|b| format!("{b:02x}")) + .collect::>() + .join(""); + if data.len() > max_bytes { + format!("{hex}...({}B total)", data.len()) + } else { + hex + } +} + +impl QconnectApp +where + TTransport: WsTransport + 'static, + TSink: QconnectEventSink + 'static, +{ + /// The shared transport event loop. Relocated from the Tauri adapter (slice + /// 5): consumes transport events and drives takeover arbitration / deferred + /// renderer join / reconnect resync / Lagged re-ask, routing diagnostics + + /// lifecycle through `self.sink` + the injected `host`. Byte-identical to the + /// prior inline Tauri loop; the adapter-coupled seams (lifecycle gating, + /// teardown, bootstrap, renderer-join, raw error) go through `host` so both + /// frontends share this control flow. + pub async fn run_session_loop(&self, host: Arc, idle_retry_active: bool) { + let mut transport_rx = self.subscribe_transport_events(); + log::info!("[QConnect/EventLoop] Started listening for transport events"); + let mut renderer_joined = false; + let mut has_disconnected = false; + // P1-8: budget for re-AskForQueueState after Lagged broadcast drops, + // until the session_uuid is confirmed. Reset on disconnect. + let mut lagged_reask_attempts: u32 = 0; + loop { + match transport_rx.recv().await { + Ok(event) => { + // P1-3: on every SESSION_STATE, capture our prior active/playing + // state + classify the server's active renderer BEFORE the sink + // applies the new state, so takeover arbitration runs on the + // prior snapshot once the apply lands below. + let mut takeover_input: Option = None; + if let TransportEvent::InboundQueueServerEvent(ref evt) = event { + if evt.message_type() == "MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE" { + takeover_input = + Some(self.capture_session_state_takeover_input(&evt.payload).await); + if !renderer_joined { + if let Some(session_uuid) = + evt.payload.get("session_uuid").and_then(|v| v.as_str()) + { + renderer_joined = true; + host.deferred_renderer_join( + session_uuid.to_string(), + deferred_join_reason(has_disconnected), + ) + .await; + } else { + log::warn!("[QConnect] SESSION_STATE received but no session_uuid in payload: {}", evt.payload); + } + } + } + } + match &event { + TransportEvent::Connected => { + log::info!("[QConnect/Transport] WebSocket connected"); + } + TransportEvent::Disconnected => { + log::warn!("[QConnect/Transport] WebSocket disconnected — resetting renderer_joined flag"); + renderer_joined = false; + has_disconnected = true; + lagged_reask_attempts = 0; + // Surface "Reconnecting" to the UI, but only if we're + // not in a teardown path (Off/Exhausted). The host gates + // this on a live runtime. + host.update_lifecycle(QconnectLifecycleState::Reconnecting) + .await; + } + TransportEvent::Authenticated => { + log::info!("[QConnect/Transport] Authenticated with JWT"); + } + TransportEvent::Subscribed => { + log::info!("[QConnect/Transport] Subscribed to channels"); + // Re-bootstrap only after a reconnection (not on initial + // connect, where connect() already bootstraps). + if has_disconnected { + log::info!("[QConnect] Re-bootstrapping after reconnect..."); + host.bootstrap_after_reconnect().await; + // P1-8: resync renderer state too, not just the queue. + if let Err(err) = self.ask_for_active_renderer_state().await { + log::warn!("[QConnect] AskForRendererState after reconnect failed: {err}"); + } + // P1-11: signal the frontend that the post-reconnect + // resync has been issued. + self.sink.on_event(QconnectAppEvent::ResyncComplete).await; + } + } + TransportEvent::KeepalivePingSent => { + log::debug!("[QConnect/Transport] Keepalive ping sent"); + } + TransportEvent::KeepalivePongReceived => { + log::debug!("[QConnect/Transport] Keepalive pong received"); + } + TransportEvent::ReconnectScheduled { + attempt, + backoff_ms, + reason, + } => { + log::warn!("[QConnect/Transport] Reconnect scheduled: attempt={} backoff={}ms reason={}", attempt, backoff_ms, reason); + } + TransportEvent::InboundQueueServerEvent(evt) => { + log::info!( + "[QConnect] <-- Inbound queue event: {} payload={}", + evt.message_type(), + evt.payload + ); + self.sink + .on_event(QconnectAppEvent::Diagnostic { + channel: "qconnect:inbound_queue_event".to_string(), + level: "info".to_string(), + payload: serde_json::json!({ + "message_type": evt.message_type(), + "action_uuid": evt.action_uuid.clone(), + "queue_version": evt.queue_version, + "track_count": evt.payload.get("tracks").and_then(|value| value.as_array()).map(|tracks| tracks.len()).unwrap_or(0), + "autoplay_track_count": evt.payload.get("autoplay_tracks").and_then(|value| value.as_array()).map(|tracks| tracks.len()).unwrap_or(0), + "preview_track_ids": queue_payload_track_preview(&evt.payload, "tracks"), + "preview_autoplay_track_ids": queue_payload_track_preview(&evt.payload, "autoplay_tracks"), + }), + }) + .await; + } + TransportEvent::InboundRendererServerCommand(cmd) => { + log::info!( + "[QConnect] <-- Inbound renderer command: {} payload={}", + cmd.message_type(), + cmd.payload + ); + } + TransportEvent::InboundFrameDecoded { + cloud_message_type, + payload_size, + } => { + log::info!( + "[QConnect/Transport] <-- Frame decoded: cloud_type={} size={}", + cloud_message_type, + payload_size + ); + } + TransportEvent::InboundPayloadBytes { + cloud_message_type, + payload, + } => { + log::info!("[QConnect/Transport] <-- Payload bytes: cloud_type={} len={} hex={}", cloud_message_type, payload.len(), hex_preview(payload, 64)); + } + TransportEvent::OutboundSent { + message_type, + action_uuid, + } => { + log::info!( + "[QConnect/Transport] --> Outbound sent: {} uuid={}", + message_type, + action_uuid + ); + } + TransportEvent::TransportError { stage, message } => { + log::error!( + "[QConnect/Transport] Error: stage={} message={}", + stage, + message + ); + } + TransportEvent::CloudError { + msg_id, + code, + descr, + } => { + log::warn!( + "[QConnect/Transport] Cloud rejected session: msg_id={} code={} descr={:?} (issue #358)", + msg_id, + code, + descr + ); + self.sink + .on_event(QconnectAppEvent::Diagnostic { + channel: "qconnect:cloud_error".to_string(), + level: "warning".to_string(), + payload: serde_json::json!({ + "msg_id": msg_id, + "code": code, + "descr": descr, + }), + }) + .await; + } + TransportEvent::InboundReceived(_envelope) => { + log::info!("[QConnect/Transport] <-- InboundReceived (JSON envelope)"); + } + TransportEvent::SessionEstablished => { + log::info!( + "[QConnect/Transport] Session established — backoff counters reset" + ); + host.update_lifecycle(QconnectLifecycleState::Connected).await; + } + TransportEvent::MaxReconnectAttemptsExceeded { + attempts, + last_reason, + } => { + log::error!( + "[QConnect/Transport] Max reconnect attempts exceeded: attempts={} last_reason={}", + attempts, + last_reason + ); + self.sink + .on_event(QconnectAppEvent::Diagnostic { + channel: "qconnect:max_reconnect_attempts_exceeded".to_string(), + level: "error".to_string(), + payload: serde_json::json!({ + "attempts": attempts, + "last_reason": last_reason, + }), + }) + .await; + // R3: the adapter owns the runtime/teardown. It sets + // Exhausted + last_error, drops the runtime unless idle- + // retry is active, surfaces the terminal status, and tells + // us whether to break (terminate) or keep idling. + let should_break = host + .on_reconnect_exhausted( + *attempts, + last_reason.clone(), + idle_retry_active, + ) + .await; + if should_break { + break; + } + // gap #7 (idle_retry_active): keep the loop alive for the + // re-armed transport; skip the reducer for this terminal + // diagnostic and wait for the next event. + continue; + } + } + if let Err(err) = self.handle_transport_event(event).await { + let message = format!("qconnect app transport handling error: {err}"); + log::error!("{message}"); + host.on_loop_error(message).await; + } + // P1-3: now that the sink has applied the SESSION_STATE, run + // takeover arbitration on the prior snapshot captured above. + // Only the claim-active path is wired here; never resets + // reconnect counters (#358 latch untouched). + if let Some(input) = takeover_input { + let decision = compute_connection_state( + input.was_active, + input.was_playing, + input.server, + // queue_equal is not yet derivable here without a + // controller/renderer queue diff; default false so the + // reduced matrix never spuriously suppresses a push. + false, + ); + if decision.should_set_active_renderer { + let local_id = { + let st = self.sync.lock().await; + st.session.local_renderer_id + }; + if let Some(local_id) = local_id { + if let Err(err) = self.send_set_active_renderer(local_id).await { + log::warn!( + "[QConnect] takeover set_active_renderer failed: {err}" + ); + } + } + } + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + log::warn!("[QConnect] Transport event lagged by {skipped} messages"); + // P1-8: a Lagged drop may have eaten the SESSION_STATE carrying + // the session_uuid. Re-ask for queue state until the uuid is + // confirmed or the attempt budget is spent. + let session_uuid_known = { + let st = self.sync.lock().await; + st.session.session_uuid.is_some() + }; + if should_reask_queue_state( + session_uuid_known, + lagged_reask_attempts, + QCONNECT_REASK_QUEUE_STATE_MAX_ATTEMPTS, + ) { + lagged_reask_attempts = lagged_reask_attempts.saturating_add(1); + if let Err(err) = self.ask_for_queue_state().await { + log::warn!("[QConnect] Lagged-recovery AskForQueueState failed: {err}"); + } + } + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + log::warn!("[QConnect/EventLoop] Transport channel closed, stopping"); + break; + } + } + } + log::info!("[QConnect/EventLoop] Stopped"); + } +} + #[cfg(test)] mod tests { use std::sync::Arc; diff --git a/crates/qconnect-app/src/lib.rs b/crates/qconnect-app/src/lib.rs index 19dafc5a7..6daf94df9 100644 --- a/crates/qconnect-app/src/lib.rs +++ b/crates/qconnect-app/src/lib.rs @@ -12,7 +12,8 @@ pub mod startup; mod sync_state; pub use app::{ - queue_payload_track_preview, QconnectApp, SessionApplyOutcome, SessionStateTakeoverInput, + queue_payload_track_preview, QconnectApp, SessionApplyOutcome, SessionLoopHost, + SessionStateTakeoverInput, }; pub use startup::{compute_effective_startup, QconnectStartupMode}; pub use error::QconnectAppError; diff --git a/src-tauri/src/qconnect/service.rs b/src-tauri/src/qconnect/service.rs index 73ab43a52..493a65fdc 100644 --- a/src-tauri/src/qconnect/service.rs +++ b/src-tauri/src/qconnect/service.rs @@ -28,13 +28,12 @@ use super::queue_resolution::{ QconnectRemoteSkipDirection, }; use super::session::{ - build_effective_renderer_snapshot, build_visible_queue_projection, compute_connection_state, + build_effective_renderer_snapshot, build_visible_queue_projection, ensure_session_renderer_state, is_local_renderer_active, is_peer_renderer_active, renderer_allows_remote_volume, QconnectFileAudioQualitySnapshot, }; use super::transport::{ - default_qconnect_device_info, default_qconnect_device_info_with_name, hex_preview, - load_persisted_device_name, + default_qconnect_device_info, default_qconnect_device_info_with_name, load_persisted_device_name, }; use super::{ qconnect_now_ms, QconnectConnectionStatus, QconnectJoinSessionRequest, @@ -46,14 +45,11 @@ use super::{ }; /// `deferred_join_reason` and `should_reask_queue_state` now live in the -/// frontend-agnostic `qconnect_app::session` module. Re-exported here so the -/// existing `deferred_join_reason`/`should_reask_queue_state` call sites in this -/// module compile unchanged. -pub(super) use qconnect_app::{deferred_join_reason, should_reask_queue_state}; +/// frontend-agnostic `qconnect_app::session` module; the session loop that +/// consumed them now lives in `qconnect_app::QconnectApp::run_session_loop` +/// (slice 5), so they are no longer referenced on the Tauri side. const QCONNECT_PLAY_TRACK_HANDOFF_WAIT_MS: u64 = 1_500; -/// Attempt budget for the Lagged-recovery re-AskForQueueState loop (P1-8). -const QCONNECT_REASK_QUEUE_STATE_MAX_ATTEMPTS: u32 = 5; const QCONNECT_PLAY_TRACK_HANDOFF_POLL_MS: u64 = 50; struct QconnectRuntime { @@ -115,6 +111,82 @@ pub(super) fn emit_qconnect_diagnostic( log::warn!("[QConnect] Failed to emit diagnostic {channel}: {err}"); } } +/// Tauri-side implementation of the shared session-loop seams (slice 5). Holds +/// the adapter handles the loop reaches back into: the service runtime +/// (lifecycle gating + reconnect-exhausted teardown), the sync accumulator + +/// app_handle (renderer join + its CoreBridge duration read), and the sink +/// (lifecycle emit). `QconnectApp::run_session_loop` owns the control flow. +struct TauriSessionLoopHost { + app: Arc>, + sync_state: Arc>, + inner: Arc>, + sink: Arc, + app_handle: AppHandle, +} + +#[async_trait::async_trait] +impl qconnect_app::SessionLoopHost for TauriSessionLoopHost { + async fn update_lifecycle(&self, state: QconnectLifecycleState) { + update_lifecycle_state_if_running(&self.inner, &self.sink, state).await; + } + + async fn bootstrap_after_reconnect(&self) { + if let Err(err) = bootstrap_remote_presence(&self.app, None).await { + log::error!("[QConnect] Re-bootstrap after reconnect failed: {err}"); + } + } + + async fn deferred_renderer_join(&self, session_uuid: String, reason: i32) { + deferred_renderer_join( + &self.app, + &self.sync_state, + &self.app_handle, + &session_uuid, + reason, + ) + .await; + } + + async fn on_reconnect_exhausted( + &self, + attempts: u32, + last_reason: String, + idle_retry_active: bool, + ) -> bool { + // Surface the Exhausted lifecycle in both modes. The difference is what + // happens to the runtime + event loop afterwards. + { + let mut guard = self.inner.lock().await; + guard.lifecycle_state = QconnectLifecycleState::Exhausted; + guard.last_error = Some(format!( + "Reconnect attempts exhausted ({attempts}): {last_reason}" + )); + if !idle_retry_active { + // Legacy terminate path: take the runtime out so a fresh + // user-initiated connect() succeeds. Dropping it detaches this + // task's own JoinHandle (fine — the loop breaks right after) (#358). + guard.runtime = None; + } + // gap #7 (idle_retry_active): leave the runtime in place; the + // transport is idling and will re-arm, and the loop keeps consuming. + } + let _ = self.app_handle.emit( + "qconnect:status_changed", + json!({ + "state": "exhausted", + "reason": "max_reconnect_attempts_exceeded", + "attempts": attempts, + "last_reason": last_reason, + }), + ); + !idle_retry_active + } + + async fn on_loop_error(&self, message: String) { + let _ = self.app_handle.emit("qconnect:error", &message); + } +} + pub struct QconnectServiceState { inner: Arc>, pub(super) custom_device_name: Arc>>, @@ -184,362 +256,22 @@ impl QconnectServiceState { return Err(msg); } - let mut transport_rx = app.subscribe_transport_events(); - let app_for_loop = Arc::clone(&app); - let app_for_errors = app_handle.clone(); - let inner_for_loop = Arc::clone(&self.inner); - let sync_state_for_loop = Arc::clone(&sync_state); - let app_handle_for_status = app_handle.clone(); - // The session/liveness/diagnostic emits route THROUGH the event sink - // (Slice 3): the sink maps each new QconnectAppEvent variant back to the - // exact same `qconnect:*` channel + payload as the prior raw emits, so - // the Svelte wire surface is byte-identical. - let sink_for_loop = Arc::clone(&sink); - // gap #7: when the transport is configured to idle-retry after - // Exhausted, the transport loop does NOT terminate — it idles then - // re-arms and emits fresh ReconnectScheduled / SessionEstablished - // events. The event loop must therefore stay alive and keep the runtime, - // otherwise the re-armed transport has no listener. let idle_retry_active = config.reconnect_idle_retry_ms > 0; - + // The session/liveness/diagnostic emits route THROUGH the event sink + // (Slice 3); the adapter-coupled seams (lifecycle gating, reconnect- + // exhausted teardown, controller bootstrap, renderer join, raw error) + // go through the SessionLoopHost. The Svelte wire surface stays + // byte-identical. The shared loop lives in qconnect-app (slice 5). + let host: Arc = Arc::new(TauriSessionLoopHost { + app: Arc::clone(&app), + sync_state: Arc::clone(&sync_state), + inner: Arc::clone(&self.inner), + sink: Arc::clone(&sink), + app_handle: app_handle.clone(), + }); + let app_for_loop = Arc::clone(&app); let event_loop = tauri::async_runtime::spawn(async move { - log::info!("[QConnect/EventLoop] Started listening for transport events"); - let mut renderer_joined = false; - let mut has_disconnected = false; - // P1-8: budget for re-AskForQueueState after Lagged broadcast drops, - // until the session_uuid is confirmed. Reset on disconnect. - let mut lagged_reask_attempts: u32 = 0; - loop { - match transport_rx.recv().await { - Ok(event) => { - // P1-3: on every SESSION_STATE, capture our prior - // active/playing state + classify the server's active - // renderer BEFORE event_sink applies the new state, so we - // can run takeover arbitration once the apply lands below. - let mut takeover_input: Option = - None; - // Check for SESSION_STATE to trigger deferred renderer join - if let qconnect_transport_ws::TransportEvent::InboundQueueServerEvent( - ref evt, - ) = event - { - if evt.message_type() == "MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE" { - takeover_input = Some( - app_for_loop - .capture_session_state_takeover_input(&evt.payload) - .await, - ); - if !renderer_joined { - if let Some(session_uuid) = - evt.payload.get("session_uuid").and_then(|v| v.as_str()) - { - renderer_joined = true; - deferred_renderer_join( - &app_for_loop, - &sync_state_for_loop, - &app_handle_for_status, - session_uuid, - deferred_join_reason(has_disconnected), - ) - .await; - } else { - log::warn!("[QConnect] SESSION_STATE received but no session_uuid in payload: {}", evt.payload); - } - } - } - } - match &event { - qconnect_transport_ws::TransportEvent::Connected => { - log::info!("[QConnect/Transport] WebSocket connected"); - } - qconnect_transport_ws::TransportEvent::Disconnected => { - log::warn!("[QConnect/Transport] WebSocket disconnected — resetting renderer_joined flag"); - renderer_joined = false; - has_disconnected = true; - lagged_reask_attempts = 0; - // Surface "Reconnecting" to the UI, but only if - // we're not in a teardown path (Off/Exhausted). - // The reconnect loop will keep retrying until - // SessionEstablished or MaxReconnectAttemptsExceeded. - update_lifecycle_state_if_running( - &inner_for_loop, - &sink_for_loop, - QconnectLifecycleState::Reconnecting, - ) - .await; - } - qconnect_transport_ws::TransportEvent::Authenticated => { - log::info!("[QConnect/Transport] Authenticated with JWT"); - } - qconnect_transport_ws::TransportEvent::Subscribed => { - log::info!("[QConnect/Transport] Subscribed to channels"); - // Re-bootstrap only after a reconnection (not on initial connect, - // where connect() already calls bootstrap_remote_presence). - if has_disconnected { - log::info!("[QConnect] Re-bootstrapping after reconnect..."); - if let Err(err) = bootstrap_remote_presence(&app_for_loop, None).await { - log::error!("[QConnect] Re-bootstrap after reconnect failed: {err}"); - } - // P1-8: resync renderer state too, not just the - // queue, when an active renderer is known. - if let Err(err) = - app_for_loop.ask_for_active_renderer_state().await - { - log::warn!("[QConnect] AskForRendererState after reconnect failed: {err}"); - } - // P1-11: signal the frontend that the post-reconnect - // resync has been issued, so it can evaluate whether - // to replay or advise on the last transport intent. - // Slice 3: routed through the sink; the `ResyncComplete` - // arm emits the identical `qconnect:diagnostic` - // (channel=qconnect:resync_complete, level=info, {}). - sink_for_loop.on_event(QconnectAppEvent::ResyncComplete).await; - } - } - qconnect_transport_ws::TransportEvent::KeepalivePingSent => { - log::debug!("[QConnect/Transport] Keepalive ping sent"); - } - qconnect_transport_ws::TransportEvent::KeepalivePongReceived => { - log::debug!("[QConnect/Transport] Keepalive pong received"); - } - qconnect_transport_ws::TransportEvent::ReconnectScheduled { - attempt, - backoff_ms, - reason, - } => { - log::warn!("[QConnect/Transport] Reconnect scheduled: attempt={} backoff={}ms reason={}", attempt, backoff_ms, reason); - } - qconnect_transport_ws::TransportEvent::InboundQueueServerEvent(evt) => { - log::info!( - "[QConnect] <-- Inbound queue event: {} payload={}", - evt.message_type(), - evt.payload - ); - sink_for_loop.on_event(QconnectAppEvent::Diagnostic { - channel: "qconnect:inbound_queue_event".to_string(), - level: "info".to_string(), - payload: json!({ - "message_type": evt.message_type(), - "action_uuid": evt.action_uuid.clone(), - "queue_version": evt.queue_version, - "track_count": evt.payload.get("tracks").and_then(|value| value.as_array()).map(|tracks| tracks.len()).unwrap_or(0), - "autoplay_track_count": evt.payload.get("autoplay_tracks").and_then(|value| value.as_array()).map(|tracks| tracks.len()).unwrap_or(0), - "preview_track_ids": qconnect_app::queue_payload_track_preview(&evt.payload, "tracks"), - "preview_autoplay_track_ids": qconnect_app::queue_payload_track_preview(&evt.payload, "autoplay_tracks"), - }), - }).await; - } - qconnect_transport_ws::TransportEvent::InboundRendererServerCommand( - cmd, - ) => { - log::info!( - "[QConnect] <-- Inbound renderer command: {} payload={}", - cmd.message_type(), - cmd.payload - ); - } - qconnect_transport_ws::TransportEvent::InboundFrameDecoded { - cloud_message_type, - payload_size, - } => { - log::info!( - "[QConnect/Transport] <-- Frame decoded: cloud_type={} size={}", - cloud_message_type, - payload_size - ); - } - qconnect_transport_ws::TransportEvent::InboundPayloadBytes { - cloud_message_type, - payload, - } => { - log::info!("[QConnect/Transport] <-- Payload bytes: cloud_type={} len={} hex={}", cloud_message_type, payload.len(), hex_preview(payload, 64)); - } - qconnect_transport_ws::TransportEvent::OutboundSent { - message_type, - action_uuid, - } => { - log::info!( - "[QConnect/Transport] --> Outbound sent: {} uuid={}", - message_type, - action_uuid - ); - } - qconnect_transport_ws::TransportEvent::TransportError { - stage, - message, - } => { - log::error!( - "[QConnect/Transport] Error: stage={} message={}", - stage, - message - ); - } - qconnect_transport_ws::TransportEvent::CloudError { - msg_id, - code, - descr, - } => { - log::warn!( - "[QConnect/Transport] Cloud rejected session: msg_id={} code={} descr={:?} (issue #358)", - msg_id, - code, - descr - ); - sink_for_loop.on_event(QconnectAppEvent::Diagnostic { - channel: "qconnect:cloud_error".to_string(), - level: "warning".to_string(), - payload: json!({ - "msg_id": msg_id, - "code": code, - "descr": descr, - }), - }).await; - } - qconnect_transport_ws::TransportEvent::InboundReceived(_envelope) => { - log::info!( - "[QConnect/Transport] <-- InboundReceived (JSON envelope)" - ); - } - qconnect_transport_ws::TransportEvent::SessionEstablished => { - log::info!( - "[QConnect/Transport] Session established — backoff counters reset" - ); - update_lifecycle_state_if_running( - &inner_for_loop, - &sink_for_loop, - QconnectLifecycleState::Connected, - ) - .await; - } - qconnect_transport_ws::TransportEvent::MaxReconnectAttemptsExceeded { - attempts, - last_reason, - } => { - log::error!( - "[QConnect/Transport] Max reconnect attempts exceeded: attempts={} last_reason={}", - attempts, - last_reason - ); - sink_for_loop.on_event(QconnectAppEvent::Diagnostic { - channel: "qconnect:max_reconnect_attempts_exceeded".to_string(), - level: "error".to_string(), - payload: json!({ - "attempts": attempts, - "last_reason": last_reason, - }), - }).await; - // Surface the Exhausted lifecycle to the UI in - // both modes. The difference is what we do with - // the runtime + event loop afterwards: - { - let mut guard = inner_for_loop.lock().await; - guard.lifecycle_state = QconnectLifecycleState::Exhausted; - guard.last_error = Some(format!( - "Reconnect attempts exhausted ({attempts}): {last_reason}" - )); - if !idle_retry_active { - // Legacy terminate path: the transport - // loop already broke, so no more events - // will fire. Take the runtime out so a - // fresh user-initiated `connect()` - // succeeds. Dropping the runtime detaches - // this task's own JoinHandle (fine — we - // break right after) (issue #358). - guard.runtime = None; - } - // gap #7 (idle_retry_active): leave the - // runtime in place. The transport is idling - // and will re-arm; this loop must keep - // consuming its subsequent events. - } - let _ = app_handle_for_status.emit( - "qconnect:status_changed", - json!({ - "state": "exhausted", - "reason": "max_reconnect_attempts_exceeded", - "attempts": attempts, - "last_reason": last_reason, - }), - ); - if idle_retry_active { - // Keep the event loop alive for the re-armed - // transport. Skip handle_transport_event for - // this terminal-diagnostic event and wait for - // the next one. - continue; - } - break; - } - } - if let Err(err) = app_for_loop.handle_transport_event(event).await { - let message = format!("qconnect app transport handling error: {err}"); - log::error!("{message}"); - let _ = app_for_errors.emit("qconnect:error", &message); - } - // P1-3: now that event_sink has applied the SESSION_STATE, - // run takeover arbitration on the prior snapshot we - // captured above. Only the claim-active path is wired here; - // queue push/ask flows reuse the existing reducer + the - // P1-8 resync. Never resets reconnect counters (#358 latch - // untouched). - if let Some(input) = takeover_input { - let decision = compute_connection_state( - input.was_active, - input.was_playing, - input.server, - // queue_equal is not yet derivable here without a - // controller/renderer queue diff; default false so - // the reduced matrix never spuriously suppresses a - // push. Refining this is part of the deferred - // bundle cross-check. - false, - ); - if decision.should_set_active_renderer { - let local_id = { - let st = sync_state_for_loop.lock().await; - st.session.local_renderer_id - }; - if let Some(local_id) = local_id { - if let Err(err) = - app_for_loop.send_set_active_renderer(local_id).await - { - log::warn!( - "[QConnect] takeover set_active_renderer failed: {err}" - ); - } - } - } - } - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { - log::warn!("[QConnect] Transport event lagged by {skipped} messages"); - // P1-8: a Lagged drop may have eaten the SESSION_STATE that - // carries the session_uuid / active_renderer_id. Re-ask for - // queue state until the session_uuid is confirmed or the - // attempt budget is spent. Never resets reconnect counters - // (#358 latch untouched). - let session_uuid_known = { - let st = sync_state_for_loop.lock().await; - st.session.session_uuid.is_some() - }; - if should_reask_queue_state( - session_uuid_known, - lagged_reask_attempts, - QCONNECT_REASK_QUEUE_STATE_MAX_ATTEMPTS, - ) { - lagged_reask_attempts = lagged_reask_attempts.saturating_add(1); - if let Err(err) = app_for_loop.ask_for_queue_state().await { - log::warn!("[QConnect] Lagged-recovery AskForQueueState failed: {err}"); - } - } - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - log::warn!("[QConnect/EventLoop] Transport channel closed, stopping"); - break; - } - } - } - log::info!("[QConnect/EventLoop] Stopped"); + app_for_loop.run_session_loop(host, idle_retry_active).await; }); let runtime = QconnectRuntime { diff --git a/src-tauri/src/qconnect/session.rs b/src-tauri/src/qconnect/session.rs index 888609ca6..7448a070c 100644 --- a/src-tauri/src/qconnect/session.rs +++ b/src-tauri/src/qconnect/session.rs @@ -18,13 +18,6 @@ use super::{ QconnectQueueVersionPayload, QconnectRemoteSyncState, QconnectVisibleQueueProjection, }; -/// `ServerActiveState`, `ConnectionDecision`, and `compute_connection_state` -/// now live in the frontend-agnostic `qconnect_app::session` module. Re-exported -/// here so existing `super::session::…` references compile unchanged. -/// (`ConnectionDecision` is the return type — consumed via field access, so it -/// is not named directly on the Tauri side; it lives in `qconnect_app`.) -pub(super) use qconnect_app::compute_connection_state; - /// Session topology types now live in the frontend-agnostic `qconnect_app::session` /// module (slice 2+4). Re-exported here so existing `super::session::…` / /// `super::…` references inside this module compile unchanged, and so the Tauri diff --git a/src-tauri/src/qconnect/transport.rs b/src-tauri/src/qconnect/transport.rs index c68cae661..f8bb52476 100644 --- a/src-tauri/src/qconnect/transport.rs +++ b/src-tauri/src/qconnect/transport.rs @@ -321,19 +321,6 @@ async fn fetch_qconnect_transport_credentials( Ok((endpoint_url, jwt_qws)) } -pub(super) fn hex_preview(data: &[u8], max_bytes: usize) -> String { - let take = data.len().min(max_bytes); - let hex: String = data[..take] - .iter() - .map(|b| format!("{b:02x}")) - .collect::>() - .join(""); - if data.len() > max_bytes { - format!("{hex}...({}B total)", data.len()) - } else { - hex - } -} fn normalize_opt_string(value: Option) -> Option { value.and_then(|value| {