From e8ea401af3c4ce4bb52ce9bfac08d1f1141d99ca Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Tue, 2 Jun 2026 16:05:20 -0600 Subject: [PATCH] refactor(qconnect): move loop helpers to qconnect-app (slice 5) Relocate the transport-loop helpers out of the Tauri adapter into QconnectApp methods + qconnect-app free items, so the shared session loop (slice 5.2) and a future Slint adapter use one definition: - QconnectApp::capture_session_state_takeover_input (locks self.sync) - QconnectApp::clear_pending_if_matches (locks self.state) - QconnectApp::send_set_active_renderer -- inline {"renderer_id": i32} payload, byte-identical to the prior QconnectSetActiveRendererRequest serialization - QconnectApp::ask_for_active_renderer_state / QconnectApp::ask_for_queue_state - SessionStateTakeoverInput struct + queue_payload_track_preview (pure free fn) Deleted the service.rs free fns (capture_session_state_takeover_input, send_set_active_renderer_via_app, ask_for_active_renderer_state, ask_for_queue_state, queue_payload_track_preview, clear_pending_if_matches) and the SessionStateTakeoverInput struct; rewired every call site (the loop + deferred_renderer_join + the reconnect/lagged paths) to the methods. The request structs serialize identically to the inline json!, so the wire is unchanged. qconnect-app 38 green, src-tauri qconnect 42 green. cc --- crates/qconnect-app/src/app.rs | 164 +++++++++++++++++++++++- crates/qconnect-app/src/lib.rs | 4 +- src-tauri/src/qconnect/service.rs | 200 +++--------------------------- src-tauri/src/qconnect/session.rs | 2 +- 4 files changed, 183 insertions(+), 187 deletions(-) diff --git a/crates/qconnect-app/src/app.rs b/crates/qconnect-app/src/app.rs index 757e40328..0e39b578c 100644 --- a/crates/qconnect-app/src/app.rs +++ b/crates/qconnect-app/src/app.rs @@ -19,9 +19,10 @@ use tokio::sync::Mutex; use uuid::Uuid; use crate::session::{ - is_peer_renderer_active, normalize_active_renderer_id, refresh_local_renderer_id, - should_arm_renderer_watchdog, LocalIdentity, QconnectRendererInfo, RendererStatus, - PLAYING_STATE_PLAYING, PLAYING_STATE_UNKNOWN, QCONNECT_RENDERER_LOST_TIMEOUT_MS, + 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, }; use crate::{ ensure_session_renderer_state, sync_session_renderer_active_flags, QconnectAppError, @@ -1445,6 +1446,163 @@ where } } +/// Prior-state snapshot captured from the sync accumulator BEFORE a SESSION_STATE +/// frame is applied, plus the server's classified active state derived from the +/// incoming payload. Drives P1-3 takeover arbitration after the apply. Relocated +/// from the Tauri adapter (slice 5) so the shared session loop owns it. +#[derive(Debug, Clone, Copy)] +pub struct SessionStateTakeoverInput { + pub was_active: bool, + pub was_playing: bool, + pub server: ServerActiveState, +} + +/// Preview the first 8 `track_id`s under `payload[key]` for diagnostic emits. +/// Pure; relocated from the Tauri adapter (slice 5). +pub fn queue_payload_track_preview(payload: &Value, key: &str) -> Vec { + payload + .get(key) + .and_then(Value::as_array) + .map(|tracks| { + tracks + .iter() + .filter_map(|track| track.get("track_id").and_then(Value::as_i64)) + .take(8) + .collect::>() + }) + .unwrap_or_default() +} + +impl QconnectApp +where + TTransport: WsTransport + 'static, + TSink: QconnectEventSink + 'static, +{ + /// Capture `was_active`/`was_playing` from the current (pre-apply) sync state + /// and classify the server's active-renderer state from the incoming + /// SESSION_STATE payload, relative to our local renderer id. Pure-ish read; + /// takes no decision. Relocated from the Tauri adapter (slice 5); locks + /// `self.sync` (the same accumulator the apply path mutates). + pub async fn capture_session_state_takeover_input( + &self, + payload: &Value, + ) -> SessionStateTakeoverInput { + let st = self.sync.lock().await; + let local_id = st.session.local_renderer_id; + let was_active = is_local_renderer_active(&st.session); + let was_playing = st.last_renderer_playing_state == Some(PLAYING_STATE_PLAYING); + let incoming_active = payload + .get("active_renderer_id") + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()); + let server = match incoming_active { + None => ServerActiveState::None, + Some(id) if Some(id) == local_id => ServerActiveState::Me, + Some(_) => { + // Another renderer owns the slot. Classify playing vs paused from + // the last observed peer playing_state; default OtherPaused when unknown. + if st.last_renderer_playing_state == Some(PLAYING_STATE_PLAYING) { + ServerActiveState::OtherPlaying + } else { + ServerActiveState::OtherPaused + } + } + }; + SessionStateTakeoverInput { + was_active, + was_playing, + server, + } + } + + /// Clear the pending action slot iff it still matches `action_uuid`. Used for + /// replies that are not reducer-correlated (set-active, ask-for-state). Locks + /// `self.state` (NEVER co-locked with `self.sync`). Relocated from the Tauri + /// adapter (slice 5). + pub async fn clear_pending_if_matches(&self, action_uuid: &str) { + let mut state = self.state.lock().await; + let pending_matches = state + .pending + .current() + .map(|pending| pending.uuid == action_uuid) + .unwrap_or(false); + if pending_matches { + state.pending.clear(); + } + } + + /// Guarded `CtrlSrvrSetActiveRenderer` sender. No-ops when the target equals + /// the current active renderer or the id is invalid. Never resets reconnect + /// counters (#358 latch untouched). Relocated from the Tauri adapter's + /// `send_set_active_renderer_via_app` (slice 5); the request payload + /// `{ "renderer_id": }` is byte-identical to the prior + /// `QconnectSetActiveRendererRequest` serialization. + pub async fn send_set_active_renderer(&self, target_renderer_id: i32) -> Result { + if target_renderer_id < 0 { + return Ok(false); + } + { + let st = self.sync.lock().await; + if st.session.active_renderer_id == Some(target_renderer_id) { + return Ok(false); + } + } + let payload = serde_json::json!({ "renderer_id": target_renderer_id }); + let command = self + .build_queue_command(QueueCommandType::CtrlSrvrSetActiveRenderer, payload) + .await; + let uuid = self + .send_queue_command(command) + .await + .map_err(|err| format!("send set_active_renderer failed: {err}"))?; + self.clear_pending_if_matches(&uuid).await; + Ok(true) + } + + /// Ask the currently-active renderer for its full state (P1-8). No-op when no + /// active renderer is known. The reply is not reducer-correlated, so its + /// pending slot is cleared immediately. Relocated from the Tauri adapter + /// (slice 5); payload byte-identical to the prior + /// `QconnectAskForRendererStateRequest` serialization. + pub async fn ask_for_active_renderer_state(&self) -> Result { + let active_id = { + let st = self.sync.lock().await; + st.session.active_renderer_id + }; + let Some(active_id) = active_id else { + return Ok(false); + }; + let payload = serde_json::json!({ "renderer_id": active_id }); + let cmd = self + .build_queue_command(QueueCommandType::CtrlSrvrAskForRendererState, payload) + .await; + let uuid = self + .send_queue_command(cmd) + .await + .map_err(|err| format!("send ask_for_renderer_state failed: {err}"))?; + self.clear_pending_if_matches(&uuid).await; + Ok(true) + } + + /// Ask the server for the current queue state (P1-8 Lagged-recovery / generic + /// re-sync). The reply is not reducer-correlated, so its pending slot is + /// cleared. Relocated from the Tauri adapter (slice 5). + pub async fn ask_for_queue_state(&self) -> Result<(), String> { + let cmd = self + .build_queue_command( + QueueCommandType::CtrlSrvrAskForQueueState, + serde_json::json!({}), + ) + .await; + let uuid = self + .send_queue_command(cmd) + .await + .map_err(|err| format!("send ask_for_queue_state failed: {err}"))?; + self.clear_pending_if_matches(&uuid).await; + Ok(()) + } +} + #[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 3a1925d8d..19dafc5a7 100644 --- a/crates/qconnect-app/src/lib.rs +++ b/crates/qconnect-app/src/lib.rs @@ -11,7 +11,9 @@ mod state; pub mod startup; mod sync_state; -pub use app::{QconnectApp, SessionApplyOutcome}; +pub use app::{ + queue_payload_track_preview, QconnectApp, SessionApplyOutcome, SessionStateTakeoverInput, +}; pub use startup::{compute_effective_startup, QconnectStartupMode}; pub use error::QconnectAppError; pub use events::{NoOpEventSink, QconnectAppEvent, QconnectEventSink}; diff --git a/src-tauri/src/qconnect/service.rs b/src-tauri/src/qconnect/service.rs index 44f1bf657..73ab43a52 100644 --- a/src-tauri/src/qconnect/service.rs +++ b/src-tauri/src/qconnect/service.rs @@ -51,134 +51,6 @@ use super::{ /// module compile unchanged. pub(super) use qconnect_app::{deferred_join_reason, should_reask_queue_state}; -/// Prior-state snapshot captured from `sync_state` BEFORE a SESSION_STATE frame -/// is applied by `event_sink`, plus the server's classified active state derived -/// from the incoming payload. Drives P1-3 takeover arbitration after the apply. -struct SessionStateTakeoverInput { - was_active: bool, - was_playing: bool, - server: super::session::ServerActiveState, -} - -/// Capture `was_active`/`was_playing` from the current (pre-apply) sync_state and -/// classify the server's active-renderer state from the incoming SESSION_STATE -/// payload, relative to our local renderer id. Pure-ish read; takes no decision. -async fn capture_session_state_takeover_input( - sync_state: &Arc>, - payload: &Value, -) -> SessionStateTakeoverInput { - use super::session::ServerActiveState; - let st = sync_state.lock().await; - let local_id = st.session.local_renderer_id; - let was_active = is_local_renderer_active(&st.session); - let was_playing = st.last_renderer_playing_state == Some(PLAYING_STATE_PLAYING); - let incoming_active = payload - .get("active_renderer_id") - .and_then(Value::as_i64) - .and_then(|value| i32::try_from(value).ok()); - let server = match incoming_active { - None => ServerActiveState::None, - Some(id) if Some(id) == local_id => ServerActiveState::Me, - Some(_) => { - // Another renderer owns the slot. Classify playing vs paused from the - // last observed peer playing_state; default to OtherPaused when unknown. - if st.last_renderer_playing_state == Some(PLAYING_STATE_PLAYING) { - ServerActiveState::OtherPlaying - } else { - ServerActiveState::OtherPaused - } - } - }; - SessionStateTakeoverInput { - was_active, - was_playing, - server, - } -} - -/// Guarded `CtrlSrvrSetActiveRenderer` sender usable from the spawned event loop -/// (mirrors `QconnectServiceState::send_set_active_renderer_if_needed` but takes -/// the already-cloned `app`/`sync_state` handles). No-ops when the target equals -/// the current active renderer or the id is invalid. Never resets reconnect -/// counters (#358 latch untouched). -async fn send_set_active_renderer_via_app( - app: &Arc>, - sync_state: &Arc>, - target_renderer_id: i32, -) -> Result { - if target_renderer_id < 0 { - return Ok(false); - } - { - let st = sync_state.lock().await; - if st.session.active_renderer_id == Some(target_renderer_id) { - return Ok(false); - } - } - let payload = serde_json::to_value(super::QconnectSetActiveRendererRequest { - renderer_id: Some(target_renderer_id), - }) - .map_err(|err| format!("serialize set_active_renderer: {err}"))?; - let command = app - .build_queue_command(QueueCommandType::CtrlSrvrSetActiveRenderer, payload) - .await; - let uuid = app - .send_queue_command(command) - .await - .map_err(|err| format!("send set_active_renderer failed: {err}"))?; - clear_pending_if_matches(app, &uuid).await; - Ok(true) -} - -/// Ask the currently-active renderer for its full state (P1-8). No-op when no -/// active renderer is known. The reply is not reducer-correlated, so its pending -/// slot is cleared immediately. Usable from the spawned event loop and the -/// deferred-join body. -async fn ask_for_active_renderer_state( - app: &Arc>, - sync_state: &Arc>, -) -> Result { - let active_id = { - let st = sync_state.lock().await; - st.session.active_renderer_id - }; - let Some(active_id) = active_id else { - return Ok(false); - }; - let payload = serde_json::to_value(super::QconnectAskForRendererStateRequest { - renderer_id: Some(active_id), - }) - .map_err(|err| format!("serialize ask_for_renderer_state: {err}"))?; - let cmd = app - .build_queue_command(QueueCommandType::CtrlSrvrAskForRendererState, payload) - .await; - let uuid = app - .send_queue_command(cmd) - .await - .map_err(|err| format!("send ask_for_renderer_state failed: {err}"))?; - clear_pending_if_matches(app, &uuid).await; - Ok(true) -} - -/// Ask the server for the current queue state (P1-8 Lagged-recovery / generic -/// re-sync). The reply is not reducer-correlated, so its pending slot is cleared. -async fn ask_for_queue_state( - app: &Arc>, -) -> Result<(), String> { - let cmd = app - .build_queue_command( - QueueCommandType::CtrlSrvrAskForQueueState, - serde_json::json!({}), - ) - .await; - let uuid = app - .send_queue_command(cmd) - .await - .map_err(|err| format!("send ask_for_queue_state failed: {err}"))?; - clear_pending_if_matches(app, &uuid).await; - Ok(()) -} - 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; @@ -199,19 +71,7 @@ struct QconnectServiceInner { /// loop. See `QconnectLifecycleState` (issue #358). lifecycle_state: QconnectLifecycleState, } -fn queue_payload_track_preview(payload: &Value, key: &str) -> Vec { - payload - .get(key) - .and_then(Value::as_array) - .map(|tracks| { - tracks - .iter() - .filter_map(|track| track.get("track_id").and_then(Value::as_i64)) - .take(8) - .collect::>() - }) - .unwrap_or_default() -} + async fn update_lifecycle_state_if_running( inner: &Arc>, sink: &TauriQconnectEventSink, @@ -356,7 +216,8 @@ impl QconnectServiceState { // 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; + let mut takeover_input: Option = + None; // Check for SESSION_STATE to trigger deferred renderer join if let qconnect_transport_ws::TransportEvent::InboundQueueServerEvent( ref evt, @@ -364,11 +225,9 @@ impl QconnectServiceState { { if evt.message_type() == "MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE" { takeover_input = Some( - capture_session_state_takeover_input( - &sync_state_for_loop, - &evt.payload, - ) - .await, + app_for_loop + .capture_session_state_takeover_input(&evt.payload) + .await, ); if !renderer_joined { if let Some(session_uuid) = @@ -423,11 +282,8 @@ impl QconnectServiceState { } // P1-8: resync renderer state too, not just the // queue, when an active renderer is known. - if let Err(err) = ask_for_active_renderer_state( - &app_for_loop, - &sync_state_for_loop, - ) - .await + if let Err(err) = + app_for_loop.ask_for_active_renderer_state().await { log::warn!("[QConnect] AskForRendererState after reconnect failed: {err}"); } @@ -468,8 +324,8 @@ impl QconnectServiceState { "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"), + "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; } @@ -644,12 +500,8 @@ impl QconnectServiceState { st.session.local_renderer_id }; if let Some(local_id) = local_id { - if let Err(err) = send_set_active_renderer_via_app( - &app_for_loop, - &sync_state_for_loop, - local_id, - ) - .await + if let Err(err) = + app_for_loop.send_set_active_renderer(local_id).await { log::warn!( "[QConnect] takeover set_active_renderer failed: {err}" @@ -676,7 +528,7 @@ impl QconnectServiceState { QCONNECT_REASK_QUEUE_STATE_MAX_ATTEMPTS, ) { lagged_reask_attempts = lagged_reask_attempts.saturating_add(1); - if let Err(err) = ask_for_queue_state(&app_for_loop).await { + if let Err(err) = app_for_loop.ask_for_queue_state().await { log::warn!("[QConnect] Lagged-recovery AskForQueueState failed: {err}"); } } @@ -1919,7 +1771,7 @@ async fn bootstrap_remote_presence( // JoinSession typically responds with session/renderer controller events that are not part of // queue reducer correlation. Drop pending slot so queue operations are not blocked for 10s. - clear_pending_if_matches(app, &join_action_uuid).await; + app.clear_pending_if_matches(&join_action_uuid).await; // 2. Ask for current queue state from server let ask_queue_payload = serde_json::json!({}); @@ -1933,7 +1785,7 @@ async fn bootstrap_remote_presence( .send_queue_command(ask_queue_command) .await .map_err(|err| format!("send bootstrap ask_for_queue_state failed: {err}"))?; - clear_pending_if_matches(app, &ask_action_uuid).await; + app.clear_pending_if_matches(&ask_action_uuid).await; // NOTE: Renderer JoinSession requires a session_uuid from the server (type 81 SESSION_STATE). // It is sent as a deferred step from the event loop when SESSION_STATE arrives. @@ -1963,7 +1815,7 @@ async fn deferred_renderer_join( "[QConnect] Deferred join skipped (already joined session_uuid={}); re-asking renderer state", session_uuid ); - if let Err(err) = ask_for_active_renderer_state(app, sync_state).await { + if let Err(err) = app.ask_for_active_renderer_state().await { log::warn!("[QConnect] Idempotent-join AskForRendererState failed: {err}"); } return; @@ -2091,13 +1943,13 @@ async fn deferred_renderer_join( ) .await; if let Ok(action_uuid) = app.send_queue_command(refresh_command).await { - clear_pending_if_matches(app, &action_uuid).await; + app.clear_pending_if_matches(&action_uuid).await; log::info!("[QConnect] Re-requested session state after renderer join"); } // P1-8: also resync the active renderer's full state (not just the queue) // right after the join, so a reconnect rejoin restores renderer state too. - if let Err(err) = ask_for_active_renderer_state(app, sync_state).await { + if let Err(err) = app.ask_for_active_renderer_state().await { log::warn!("[QConnect] Post-join AskForRendererState failed: {err}"); } @@ -2108,19 +1960,3 @@ async fn deferred_renderer_join( st.last_joined_session_uuid = Some(session_uuid.to_string()); } } - -async fn clear_pending_if_matches( - app: &Arc>, - action_uuid: &str, -) { - let state_handle = app.state_handle(); - let mut state = state_handle.lock().await; - let pending_matches = state - .pending - .current() - .map(|pending| pending.uuid == action_uuid) - .unwrap_or(false); - if pending_matches { - state.pending.clear(); - } -} diff --git a/src-tauri/src/qconnect/session.rs b/src-tauri/src/qconnect/session.rs index 858c9fb5d..888609ca6 100644 --- a/src-tauri/src/qconnect/session.rs +++ b/src-tauri/src/qconnect/session.rs @@ -23,7 +23,7 @@ use super::{ /// 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, ServerActiveState}; +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::…` /