diff --git a/crates/qbz-app/src/listen_log/logger.rs b/crates/qbz-app/src/listen_log/logger.rs index 6bf3fc036..230d50444 100644 --- a/crates/qbz-app/src/listen_log/logger.rs +++ b/crates/qbz-app/src/listen_log/logger.rs @@ -1,10 +1,11 @@ //! The async facade both hosts drive: a [`ListenTracker`] plus a //! [`ListenStore`], every store call inside `spawn_blocking`. //! -//! Hosts call the five verbs (`track_started`, `tick`, `ended_naturally`, -//! `stopped`, `shutdown`) and never touch SQLite or the state machine -//! themselves. "Listening history OFF" (`paused`) is honoured HERE, in one -//! place: a paused logger opens no row, so nothing downstream has to ask. +//! Hosts call explicit lifecycle verbs (`track_started`, `tick`, +//! `ended_naturally`, `stopped`, `handoff`, `shutdown`) and never touch SQLite +//! or the state machine themselves. "Listening history OFF" (`paused`) is +//! honoured HERE, in one place: a paused logger opens no row, so nothing +//! downstream has to ask. use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -170,6 +171,14 @@ impl ListenLogger { self.close(closed).await; } + /// Playback authority moved away from this process. Close the owner row + /// as `handoff`; delegated tracks are intentionally never opened here. + /// Idempotent so repeated lifecycle observations cannot rewrite the row. + pub async fn handoff(&self) { + let closed = self.tracker.lock().unwrap().handoff(now_unix()); + self.close(closed).await; + } + /// Orderly exit: close the row in flight as `shutdown`. SYNCHRONOUS on /// purpose — the hosts call it after their event loop is gone. pub fn shutdown_blocking(&self) { @@ -382,4 +391,24 @@ mod tests { assert_eq!(rows[0].end_reason, Some(EndReason::Natural)); assert_eq!(rows[1].end_reason, Some(EndReason::Stop)); } + + #[tokio::test] + async fn handoff_closes_the_owner_row_once() { + let dir = tempfile::tempdir().unwrap(); + let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install) + .await + .unwrap(); + l.track_started(meta("1", 100_000), false).await; + l.tick(0, true).await; + l.tick(1_000, true).await; + + l.handoff().await; + l.handoff().await; + + assert!(!l.has_open_row()); + let rows = l.rows().await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].end_reason, Some(EndReason::Handoff)); + assert_eq!(rows[0].played_ms, 1_000); + } } diff --git a/crates/qbz-app/src/listen_log/tracker.rs b/crates/qbz-app/src/listen_log/tracker.rs index ea2b3a66c..466798a60 100644 --- a/crates/qbz-app/src/listen_log/tracker.rs +++ b/crates/qbz-app/src/listen_log/tracker.rs @@ -167,6 +167,13 @@ impl ListenTracker { self.close_with(EndReason::Shutdown, now) } + /// Playback authority moved to another renderer. The owner row is closed + /// explicitly and no delegated playback is folded into it. Calling this + /// more than once is harmless: after the first close there is no open row. + pub fn handoff(&mut self, now: i64) -> Option { + self.close_with(EndReason::Handoff, now) + } + /// For hosts that only see "a different track started": was the previous /// one within 2 s of its end? Mirrors the desktop's /// `seen_position + 2 >= duration` (playback_driver.rs:226-264). @@ -353,7 +360,7 @@ mod tests { } #[test] - fn stop_shutdown_error_reasons() { + fn stop_shutdown_error_handoff_reasons() { let mut t = ListenTracker::new(); started(&mut t, 1); assert_eq!(t.stopped(2, false).unwrap().reason, EndReason::Stop); @@ -361,7 +368,10 @@ mod tests { assert_eq!(t.shutdown(3).unwrap().reason, EndReason::Shutdown); started(&mut t, 3); assert_eq!(t.errored(4).unwrap().reason, EndReason::Error); - assert_eq!(t.shutdown(5), None); + started(&mut t, 4); + assert_eq!(t.handoff(5).unwrap().reason, EndReason::Handoff); + assert_eq!(t.handoff(6), None, "handoff is idempotent"); + assert_eq!(t.shutdown(7), None); } #[test] diff --git a/crates/qbz-app/src/playback_driver.rs b/crates/qbz-app/src/playback_driver.rs index 40ec070c6..540278e0e 100644 --- a/crates/qbz-app/src/playback_driver.rs +++ b/crates/qbz-app/src/playback_driver.rs @@ -19,6 +19,8 @@ // * `advance_and_play` — the full advance ritual (skip-walk → play → prefetch // → persist), reused verbatim by the CLI next/prev routes +use std::any::Any; +use std::fmt; use std::sync::Arc; use std::time::Duration; @@ -85,6 +87,18 @@ pub enum DriverAction { QueueFinished, } +/// Whether an action may run under the current authority. Playback/queue +/// mutation, owner-side resolution and session persistence are fenced while a +/// delegated renderer owns playback; diagnostics and renderer reports remain +/// live so the guest session stays observable. +fn action_allowed(action: &DriverAction, owner_actions_allowed: bool) -> bool { + owner_actions_allowed + || matches!( + action, + DriverAction::LatchError(_) | DriverAction::ReportEdge + ) +} + /// The previous tick's snapshot: the desktop loop's `last_track_id` / /// `seen_position` / `was_playing` (plus duration for [`DriverState::after`]). #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -422,6 +436,26 @@ pub fn advance_state( } } +/// Keep delegated observations useful for reports without ever promoting them +/// into the owner's carried playback state. The delegated state is discarded +/// on the first owner-admitted tick; restore then plans only from the last +/// owner snapshot captured before handoff. +fn advance_authority_state( + owner_state: &mut DriverState, + delegated_state: &mut Option, + ev: &PlaybackEvent, + actions: &[DriverAction], + owner_actions_allowed: bool, +) { + if owner_actions_allowed { + *owner_state = advance_state(owner_state, ev, actions); + *delegated_state = None; + } else { + let previous = delegated_state.take().unwrap_or_default(); + *delegated_state = Some(advance_state(&previous, ev, actions)); + } +} + /// Map the desktop `ui_prefs.streaming_quality` key to a request-layer /// [`Quality`]. Byte-identical contract to `crates/qbz/src/ui_prefs.rs:823` /// (`streaming_quality_for_key`), replicated here because the desktop crate is @@ -445,6 +479,19 @@ pub fn quality_from_key(key: &str) -> Quality { pub struct DriverDeps { /// Resolve the streaming quality at play time (qbzd passes the daemon prefs). pub quality: Arc Quality + Send + Sync>, + /// True only while the daemon's owner authority may mutate/resolve/persist. + /// Delegated playback keeps the driver alive for diagnostics and reports, + /// but fences every owner-side action. + pub owner_actions_allowed: Arc bool + Send + Sync>, + /// Atomically classify authority before reading player/queue state. An + /// owner result carries the exact generation and a drain-visible permit; + /// delegated ticks remain report-only, while fenced ticks retry unread. + pub observe_authority: Arc DriverAuthorityObservation + Send + Sync>, + /// Re-admit a spawned continuation only under the exact owner generation + /// that produced it. A guest round-trip must never promote stale work into + /// a later owner merely because owner actions are allowed again. + pub readmit_owner_action: + Arc Option + Send + Sync>, /// Report-edge signal (T10 wires the QConnect renderer report). pub on_edge: Arc, /// Latch a drained error under a category ("stream" | "transport" | "auth"). @@ -453,6 +500,71 @@ pub struct DriverDeps { pub on_tick: Arc, } +/// Type-erased RAII permit supplied by the host's authority coordinator. +/// `qbz-app` needs only the drop boundary; it deliberately knows nothing about +/// the daemon's concrete authority stamps. +pub struct DriverActionPermit { + _inner: Box, +} + +/// Type-erased exact owner token supplied by the host authority coordinator. +/// The concrete token remains opaque to `qbz-app`; only the host callback can +/// downcast and validate it. +#[derive(Clone)] +pub struct DriverOwnerToken(Arc); + +impl DriverOwnerToken { + pub fn new(token: T) -> Self { + Self(Arc::new(token)) + } + + pub fn downcast_ref(&self) -> Option<&T> { + self.0.downcast_ref::() + } +} + +impl fmt::Debug for DriverOwnerToken { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("DriverOwnerToken(..)") + } +} + +/// One exact owner observation plus the live permit covering the tick input. +pub struct DriverOwnerAdmission { + token: DriverOwnerToken, + _permit: DriverActionPermit, +} + +impl DriverOwnerAdmission { + pub fn new(token: DriverOwnerToken, permit: DriverActionPermit) -> Self { + Self { + token, + _permit: permit, + } + } + + fn token(&self) -> DriverOwnerToken { + self.token.clone() + } +} + +/// Host classification captured before a driver tick reads player or queue. +/// A lifecycle fence is distinct from delegated playback: fenced ticks retry +/// without reading or altering either authority's carried state. +pub enum DriverAuthorityObservation { + Owner(DriverOwnerAdmission), + Delegated, + Fenced, +} + +impl DriverActionPermit { + pub fn new(permit: T) -> Self { + Self { + _inner: Box::new(permit), + } + } +} + /// Result of a gapless byte fetch that ran outside the 450 ms control loop. /// Keeping the predecessor alongside the successor lets the loop reject a /// late download after a manual track/queue change instead of appending it to @@ -539,6 +651,7 @@ pub async fn run_driver( mut shutdown: tokio::sync::watch::Receiver, ) { let mut state = DriverState::default(); + let mut delegated_state: Option = None; let mut ticker = tokio::time::interval(Duration::from_millis(TICK_MS)); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // Gapless downloads may outlive the fixed ten-second arm window, but they @@ -561,7 +674,13 @@ pub async fn run_driver( } completed = gapless_fetches.join_next(), if !gapless_fetches.is_empty() => { match completed { - Some(Ok(result)) => finish_gapless_fetch(&runtime, result).await, + Some(Ok((owner_token, result))) => { + if let Some(_permit) = (deps.readmit_owner_action)(&owner_token) { + finish_gapless_fetch(&runtime, result).await; + } else { + log::debug!("[qbzd] driver: discarded gapless result from a stale owner authority"); + } + } Some(Err(error)) if error.is_cancelled() => { log::debug!("[qbzd] driver: stale gapless fetch cancelled"); } @@ -583,6 +702,25 @@ pub async fn run_driver( _ = ticker.tick() => {} } + // Capture owner admission BEFORE the first player/queue read. If this + // tick begins under guest authority it remains guest-only even when an + // owner restore completes during `queue_snapshot().await`. + let owner_admission = match (deps.observe_authority)() { + DriverAuthorityObservation::Owner(admission) => Some(admission), + DriverAuthorityObservation::Delegated => None, + DriverAuthorityObservation::Fenced => { + // The transition fence is waiting for every exact permit held + // by these network jobs. Abort them on the first fenced tick, + // not only after delegated authority is already installed, or + // a slow prefetch could stretch handoff to the stream timeout. + gapless_fetches.abort_all(); + warm_tasks.abort_all(); + continue; + } + }; + let owner_token = owner_admission + .as_ref() + .map(DriverOwnerAdmission::token); let core = runtime.core(); let player = core.player(); let ev = player.get_playback_event(); @@ -590,7 +728,30 @@ pub async fn run_driver( let stream_error = player.state.take_stream_error_message(); let queue = queue_snapshot(core).await; - let actions = plan_tick(&state, &ev, &queue, stream_error.as_deref()); + let owner_actions_allowed = owner_admission.is_some(); + if owner_actions_allowed && delegated_state.is_some() { + // Rebase on the first fully owner-admitted observation after a + // delegated span. Durable player generations may have advanced + // while the guest was active; comparing those guest edges with the + // pre-handoff owner snapshot could synthesize an owner auto-advance. + state = DriverState::after(&ev); + delegated_state = None; + } + let delegated_baseline = DriverState::default(); + let planning_state = if owner_actions_allowed { + &state + } else { + delegated_state.as_ref().unwrap_or(&delegated_baseline) + }; + let actions = plan_tick(planning_state, &ev, &queue, stream_error.as_deref()); + + // Authority can change while an owner prefetch is in flight. Cancel + // those best-effort jobs at the first observed delegated tick; their + // completion paths are fenced independently above. + if !owner_actions_allowed { + gapless_fetches.abort_all(); + warm_tasks.abort_all(); + } // Once the engine/queue edge has moved, an outstanding fetch belongs // to the old predecessor. Drop its network future immediately so it @@ -607,7 +768,12 @@ pub async fn run_driver( gapless_fetches.abort_all(); } + let mut executed_actions = Vec::with_capacity(actions.len()); for action in &actions { + if !action_allowed(action, owner_actions_allowed) { + continue; + } + executed_actions.push(action.clone()); match action { DriverAction::SyncCursorTo(id) => { core.sync_current_to_id(*id).await; @@ -617,11 +783,28 @@ pub async fn run_driver( let predecessor_id = ev.track_id; let successor_id = *id; let task_runtime = Arc::clone(&runtime); + let task_owner_actions_allowed = Arc::clone(&deps.owner_actions_allowed); + let task_readmit_owner_action = Arc::clone(&deps.readmit_owner_action); + let task_owner_token = owner_token + .as_ref() + .expect("owner action admitted without an owner token") + .clone(); log::info!( "[qbzd] driver: fetching gapless track {successor_id} after {predecessor_id}" ); let cached = player.is_track_cached(successor_id); gapless_fetches.spawn(async move { + let Some(_permit) = task_readmit_owner_action(&task_owner_token) else { + return ( + task_owner_token, + GaplessFetchResult { + predecessor_id, + successor_id, + bytes: None, + streamed: false, + }, + ); + }; let core = task_runtime.core(); // COLD successor: append its initial CMAF buffer as an // incremental source (the desktop's streaming @@ -633,34 +816,57 @@ pub async fn run_driver( if !cached { match core.queue_gapless_streaming(successor_id, quality).await { Ok(()) => { - return GaplessFetchResult { - predecessor_id, - successor_id, - bytes: None, - streamed: true, - } + return ( + task_owner_token, + GaplessFetchResult { + predecessor_id, + successor_id, + bytes: None, + streamed: true, + }, + ) } Err(error) => log::warn!( "[qbzd] driver: gapless stream setup for {successor_id} failed: {error}; trying byte fallback" ), } } - let bytes = core - .fetch_for_gapless_resolved(successor_id, quality, None, None) - .await; - GaplessFetchResult { - predecessor_id, - successor_id, - bytes, - streamed: false, - } + let bytes = if task_owner_actions_allowed() { + core.fetch_for_gapless_resolved(successor_id, quality, None, None) + .await + } else { + None + }; + ( + task_owner_token, + GaplessFetchResult { + predecessor_id, + successor_id, + bytes, + streamed: false, + }, + ) }); } DriverAction::WarmSuccessor => { let quality = (deps.quality)(); let task_runtime = Arc::clone(&runtime); + let task_owner_actions_allowed = Arc::clone(&deps.owner_actions_allowed); + let task_readmit_owner_action = Arc::clone(&deps.readmit_owner_action); + let task_owner_token = owner_token + .as_ref() + .expect("owner action admitted without an owner token") + .clone(); warm_tasks.spawn(async move { - prefetch_successors(task_runtime.as_ref(), quality).await; + let Some(_permit) = task_readmit_owner_action(&task_owner_token) else { + return; + }; + prefetch_successors( + task_runtime.as_ref(), + quality, + task_owner_actions_allowed.as_ref(), + ) + .await; }); } DriverAction::PauseStopAfter => { @@ -708,7 +914,13 @@ pub async fn run_driver( } } - state = advance_state(&state, &ev, &actions); + advance_authority_state( + &mut state, + &mut delegated_state, + &ev, + &executed_actions, + owner_actions_allowed, + ); (deps.on_tick)(); } warm_tasks.abort_all(); @@ -803,7 +1015,11 @@ pub async fn advance_and_play( async fn prefetch_successors( runtime: &AppRuntime, quality: Quality, + owner_actions_allowed: &(dyn Fn() -> bool + Send + Sync), ) { + if !owner_actions_allowed() { + return; + } // Memory-pressure watchdog: critical pressure latched a halt — do not // start a new download (issue #660). if crate::memory_watchdog::prefetch_halted() { @@ -830,6 +1046,9 @@ async fn prefetch_successors( if player.is_track_cached(next.id) { return; } + if !owner_actions_allowed() { + return; + } let client_lock = core.client(); let guard = client_lock.read().await; let Some(client) = guard.as_ref() else { @@ -842,6 +1061,9 @@ async fn prefetch_successors( let allow_hires = profile.allow_hires_prefetch && !crate::memory_watchdog::hires_prefetch_paused(); let prefetch_quality = cap_prefetch_quality(quality, allow_hires); + if !owner_actions_allowed() { + return; + } if let Err(e) = player .prefetch_into_cache(client, next.id, prefetch_quality) .await @@ -891,9 +1113,7 @@ async fn queue_snapshot( /// `None`). Mirrors `crates/qbz/src/session_persist.rs::capture_and_save`, minus /// the desktop-only `persist_session` gate (the daemon's store IS its queue /// persistence, so it always saves). -pub async fn save_session_now( - runtime: &AppRuntime, -) { +pub async fn save_session_now(runtime: &AppRuntime) { let core = runtime.core(); let (tracks, current_index) = core.get_all_queue_tracks().await; let full = core.get_queue_state_full().await; @@ -1073,10 +1293,103 @@ mod tests { } } + #[test] + fn delegated_authority_fences_every_owner_action() { + let owner_actions = [ + DriverAction::SyncCursorTo(2), + DriverAction::ArmGapless(2), + DriverAction::WarmSuccessor, + DriverAction::AdvanceAndPlay, + DriverAction::PauseStopAfter, + DriverAction::SavePosition(17), + DriverAction::QueueFinished, + ]; + + for action in &owner_actions { + assert!(action_allowed(action, true), "owner rejected {action:?}"); + assert!( + !action_allowed(action, false), + "delegated authority accepted {action:?}" + ); + } + + for action in [ + DriverAction::LatchError("stream failed".to_string()), + DriverAction::ReportEdge, + ] { + assert!(action_allowed(&action, true)); + assert!(action_allowed(&action, false)); + } + } + + #[test] + fn delegated_filter_keeps_diagnostics_in_planned_order() { + let state = DriverState::after(&ev(1, true, 580, 581)); + let planned = plan_tick( + &state, + &ev(1, false, 581, 581), + &q(1, &[(2, true)], "off", None), + Some("stream failed"), + ); + assert!(planned.contains(&DriverAction::AdvanceAndPlay)); + + let executable: Vec<_> = planned + .into_iter() + .filter(|action| action_allowed(action, false)) + .collect(); + assert_eq!( + executable, + vec![ + DriverAction::LatchError("stream failed".to_string()), + DriverAction::ReportEdge, + ] + ); + } + + #[test] + fn delegated_snapshots_never_replace_owner_driver_state() { + let owner_event = ev(11, true, 42, 300); + let mut owner_state = DriverState::after(&owner_event); + let original_owner_state = owner_state.clone(); + let mut delegated_state = None; + let guest_event = ev(99, true, 120, 240); + + advance_authority_state( + &mut owner_state, + &mut delegated_state, + &guest_event, + &[DriverAction::ReportEdge], + false, + ); + + assert_eq!(owner_state, original_owner_state); + assert_eq!( + delegated_state.as_ref().map(|state| state.last.track_id), + Some(99) + ); + + let restored_event = ev(11, true, 43, 300); + advance_authority_state( + &mut owner_state, + &mut delegated_state, + &restored_event, + &[], + true, + ); + assert_eq!(owner_state.last.track_id, 11); + assert_eq!(owner_state.last.position, 43); + assert!(delegated_state.is_none()); + } + #[test] fn end_edge_advances() { let s = DriverState::after(&ev(1, true, 580, 581)); - let a = plan_tick(&s, &ev(1, false, 581, 581), &q(1, &[(2, true)], "off", None), None); + let a = plan_tick( + &s, + &ev(1, false, 581, 581), + &q(1, &[(2, true)], "off", None), + None, + ); assert!(a.contains(&DriverAction::AdvanceAndPlay)); assert!(a.contains(&DriverAction::ReportEdge)); // play-state edge } @@ -1091,12 +1404,7 @@ mod tests { current.engine_empty_generation = 1; current.engine_empty_track_id = 1; - let actions = plan_tick( - &s, - ¤t, - &q(1, &[(2, true)], "off", None), - None, - ); + let actions = plan_tick(&s, ¤t, &q(1, &[(2, true)], "off", None), None); assert!(actions.contains(&DriverAction::AdvanceAndPlay)); } @@ -1165,7 +1473,12 @@ mod tests { #[test] fn mid_track_pause_does_not_advance() { let s = DriverState::after(&ev(1, true, 100, 581)); - let a = plan_tick(&s, &ev(1, false, 100, 581), &q(1, &[(2, true)], "off", None), None); + let a = plan_tick( + &s, + &ev(1, false, 100, 581), + &q(1, &[(2, true)], "off", None), + None, + ); assert!(!a.contains(&DriverAction::AdvanceAndPlay)); assert!(a.contains(&DriverAction::ReportEdge)); } @@ -1179,12 +1492,7 @@ mod tests { let mut underrun = ready.clone(); underrun.buffer_state = PlaybackBufferState::Underrun; - let actions = plan_tick( - &state, - &underrun, - &q(41, &[(42, true)], "off", None), - None, - ); + let actions = plan_tick(&state, &underrun, &q(41, &[(42, true)], "off", None), None); assert!(actions.contains(&DriverAction::ReportEdge)); } @@ -1257,12 +1565,7 @@ mod tests { let mut current = ev(1, false, 581, 581); current.gapless_ready = true; let previous = DriverState::after(&ev(1, true, 580, 581)); - let actions = plan_tick( - &previous, - ¤t, - &q(1, &[(2, true)], "off", None), - None, - ); + let actions = plan_tick(&previous, ¤t, &q(1, &[(2, true)], "off", None), None); assert!(!actions .iter() .any(|action| matches!(action, DriverAction::ArmGapless(_)))); @@ -1342,13 +1645,21 @@ mod tests { #[test] fn queue_finished_when_nothing_playable() { let s = DriverState::after(&ev(1, true, 580, 581)); - let a = plan_tick(&s, &ev(1, false, 581, 581), &q(1, &[(2, false)], "off", None), None); + let a = plan_tick( + &s, + &ev(1, false, 581, 581), + &q(1, &[(2, false)], "off", None), + None, + ); assert!(a.contains(&DriverAction::QueueFinished)); } #[test] fn skip_walk_bounds() { - assert_eq!(next_playable(&[(2, false), (3, false), (4, true)], 50), Some((2, 4))); + assert_eq!( + next_playable(&[(2, false), (3, false), (4, true)], 50), + Some((2, 4)) + ); let all_bad: Vec<(u64, bool)> = (0..60).map(|i| (i, false)).collect(); assert_eq!(next_playable(&all_bad, 50), None); // bounded — never walks forever } @@ -1371,7 +1682,12 @@ mod tests { #[test] fn seamless_gapless_transition_syncs_cursor() { let s = DriverState::after(&ev(1, true, 580, 581)); - let a = plan_tick(&s, &ev(2, true, 0, 547), &q(1, &[(2, true)], "off", None), None); + let a = plan_tick( + &s, + &ev(2, true, 0, 547), + &q(1, &[(2, true)], "off", None), + None, + ); assert!(a.contains(&DriverAction::SyncCursorTo(2))); } @@ -1390,7 +1706,12 @@ mod tests { #[test] fn duration_zero_never_advances() { let s = DriverState::after(&ev(1, true, 580, 581)); - let a = plan_tick(&s, &ev(1, false, 580, 0), &q(1, &[(2, true)], "off", None), None); + let a = plan_tick( + &s, + &ev(1, false, 580, 0), + &q(1, &[(2, true)], "off", None), + None, + ); assert!(!a.contains(&DriverAction::AdvanceAndPlay)); assert!(a.contains(&DriverAction::ReportEdge)); // play-state edge } diff --git a/crates/qbz-core/src/core.rs b/crates/qbz-core/src/core.rs index f7b2337c7..0e4235ddc 100644 --- a/crates/qbz-core/src/core.rs +++ b/crates/qbz-core/src/core.rs @@ -6,6 +6,15 @@ use std::sync::Arc; use tokio::sync::RwLock; +use qbz_integrations::musicbrainz::cache::{MusicBrainzCache, QobuzArtistMatch}; +use qbz_integrations::musicbrainz::genre::{extract_affinity_seeds, genre_summary, is_broad_genre}; +use qbz_integrations::musicbrainz::location::compute_affinity_score; +use qbz_integrations::musicbrainz::{ + location, AffinitySeeds, AlbumAppearance, ArtistMetadata, ArtistRelationships, + DiscoveryArtist, DiscoveryResponse, LocationCandidate, LocationDiscoveryResponse, + MusicBrainzClient, MusicianAppearances, MusicianConfidence, Period, RelatedArtist, + ResolvedArtist, ResolvedMusician, Tag, +}; use qbz_models::{ ArtistStoryResponse, AssetOrigin, ExternalStreamAsset, StreamQualityInfo, @@ -17,16 +26,9 @@ use qbz_models::{ RepeatMode, SearchAllResults, SearchResultsPage, StreamUrl, Track, TrackToAnalyse, TracksContainer, UserSession, }; -use qbz_integrations::musicbrainz::cache::{MusicBrainzCache, QobuzArtistMatch}; -use qbz_integrations::musicbrainz::genre::{extract_affinity_seeds, genre_summary, is_broad_genre}; -use qbz_integrations::musicbrainz::location::compute_affinity_score; -use qbz_integrations::musicbrainz::{ - location, AffinitySeeds, AlbumAppearance, ArtistMetadata, ArtistRelationships, - DiscoveryArtist, DiscoveryResponse, LocationCandidate, LocationDiscoveryResponse, - MusicBrainzClient, MusicianAppearances, MusicianConfidence, Period, RelatedArtist, - ResolvedArtist, ResolvedMusician, Tag, +use qbz_player::{ + PlaybackState, Player, QueueAuthoritySnapshot as PlayerQueueAuthoritySnapshot, QueueManager, }; -use qbz_player::{PlaybackState, Player, QueueManager}; use qbz_qobuz::QobuzClient; use crate::error::CoreError; @@ -264,13 +266,26 @@ pub struct QbzCore { initialized: Arc>, /// D8 guard: true when the current queue was built from an OFFLINE-ONLY /// local playlist — such a queue must never be pushed to the Qobuz - /// Connect cloud. Cleared by every queue REPLACEMENT (`set_queue` / - /// `set_queue_with_order` / `clear_queue`); append-style ops preserve it. - /// Set explicitly by the frontend's local-playlist play path right after - /// its `set_queue`. + /// Connect cloud. Ordinary queue replacement (`set_queue` / + /// `set_queue_with_order` / `clear_queue`) clears it; append-style ops + /// preserve it. An offline local-playlist replacement sets it atomically + /// through `set_queue_with_offline_only` before publishing QueueUpdated. queue_offline_only: Arc, } +/// Exact in-process playback authority owned by [`QbzCore`]. +/// +/// The player snapshot preserves the queue mechanics while `offline_only` +/// preserves the cloud-publication policy coupled to that queue. Keeping both +/// opaque prevents a delegation handoff from restoring an offline playlist as +/// if it were eligible for QConnect synchronization. +#[must_use = "an authority snapshot must be retained until it is restored or deliberately discarded"] +#[derive(Clone)] +pub struct QueueAuthoritySnapshot { + queue: PlayerQueueAuthoritySnapshot, + offline_only: bool, +} + impl QbzCore { /// Create a new QbzCore instance with the given frontend adapter and player /// @@ -291,17 +306,19 @@ impl QbzCore { } /// Mark (or unmark) the current queue as built from an OFFLINE-ONLY local - /// playlist (D8). Call right after the `set_queue` that loaded it. - pub fn set_queue_offline_only(&self, on: bool) { + /// playlist (D8). Queue replacement code should prefer + /// [`Self::set_queue_with_offline_only`] so the flag is visible before the + /// `QueueUpdated` event can trigger cloud synchronization. + fn set_queue_offline_only(&self, on: bool) { self.queue_offline_only - .store(on, std::sync::atomic::Ordering::Relaxed); + .store(on, std::sync::atomic::Ordering::Release); } /// True when the current queue originates from an offline-only local /// playlist — QConnect must skip its cloud queue push. pub fn queue_is_offline_only(&self) -> bool { self.queue_offline_only - .load(std::sync::atomic::Ordering::Relaxed) + .load(std::sync::atomic::Ordering::Acquire) } /// Install a MusicBrainz cache. The frontend owns the data path @@ -484,6 +501,59 @@ impl QbzCore { queue.get_all_tracks() } + /// Capture the complete queue authority state atomically for a temporary + /// renderer handoff. The returned payload is opaque and in-process only. + pub async fn capture_authority_snapshot(&self) -> QueueAuthoritySnapshot { + let queue = self.queue.read().await; + QueueAuthoritySnapshot { + queue: queue.capture_authority_snapshot(), + offline_only: self.queue_is_offline_only(), + } + } + + /// Restore a previously captured queue authority state in one atomic, + /// side-effect-free replacement. The authority coordinator publishes the + /// restored projection only after sealing its origin and generation. + pub async fn restore_authority_snapshot(&self, snapshot: QueueAuthoritySnapshot) { + let queue = self.queue.write().await; + self.set_queue_offline_only(snapshot.offline_only); + queue.restore_authority_snapshot(snapshot.queue); + } + + /// Commit an authority stamp and restore its exact queue as one synchronous + /// visibility boundary. The outer Tokio lock is acquired with `try_write` + /// because authority coordinators call this from their non-async commit + /// gate. If either the queue is busy or `commit` rejects, ownership of the + /// opaque snapshot is returned unchanged and no queue state is mutated. + pub fn try_commit_authority_snapshot( + &self, + snapshot: QueueAuthoritySnapshot, + commit: F, + ) -> Result + where + F: FnOnce() -> bool, + { + let QueueAuthoritySnapshot { + queue: queue_snapshot, + offline_only, + } = snapshot; + let Ok(queue) = self.queue.try_write() else { + return Err(QueueAuthoritySnapshot { + queue: queue_snapshot, + offline_only, + }); + }; + if !commit() { + return Err(QueueAuthoritySnapshot { + queue: queue_snapshot, + offline_only, + }); + } + self.set_queue_offline_only(offline_only); + queue.restore_authority_snapshot(queue_snapshot); + Ok(queue.get_state()) + } + /// Capture queue rows, cursor, and oldest-first playback history under one /// queue lock for session persistence. pub async fn get_persistable_queue_state( @@ -562,8 +632,8 @@ impl QbzCore { /// slot — use when nothing is actively playing and the user wants a full /// reset. pub async fn clear_queue(&self, keep_current: bool) { - self.set_queue_offline_only(false); let queue = self.queue.write().await; + self.set_queue_offline_only(false); queue.clear(keep_current); self.emit(CoreEvent::QueueUpdated { state: queue.get_state(), @@ -615,10 +685,21 @@ impl QbzCore { /// Set the entire queue (replaces existing) pub async fn set_queue(&self, tracks: Vec, start_index: Option) { - // Any queue replacement drops the offline-only-playlist stamp; the - // local-playlist play path re-sets it right after when it applies. - self.set_queue_offline_only(false); + self.set_queue_with_offline_only(tracks, start_index, false) + .await; + } + + /// Replace the queue and publish its offline-only origin as one ordered + /// operation. The flag is set before `QueueUpdated`, closing the window in + /// which a QConnect subscriber could upload an offline-only playlist. + pub async fn set_queue_with_offline_only( + &self, + tracks: Vec, + start_index: Option, + offline_only: bool, + ) { let queue = self.queue.write().await; + self.set_queue_offline_only(offline_only); queue.set_queue(tracks, start_index); self.emit(CoreEvent::QueueUpdated { state: queue.get_state(), @@ -634,8 +715,8 @@ impl QbzCore { shuffle_enabled: bool, shuffle_order: Option>, ) { - self.set_queue_offline_only(false); let queue = self.queue.write().await; + self.set_queue_offline_only(false); queue.set_queue_with_order(tracks, start_index, shuffle_enabled, shuffle_order); self.emit(CoreEvent::QueueUpdated { state: queue.get_state(), diff --git a/crates/qbz-core/src/lib.rs b/crates/qbz-core/src/lib.rs index 728a957eb..70aeb3ad8 100644 --- a/crates/qbz-core/src/lib.rs +++ b/crates/qbz-core/src/lib.rs @@ -61,7 +61,7 @@ pub mod system_capabilities; pub use qbz_models::{CoreEvent, FrontendAdapter, LoggingAdapter, NoOpAdapter}; // Re-exports from this crate -pub use core::{normalize_artist_name, QbzCore}; +pub use core::{normalize_artist_name, QbzCore, QueueAuthoritySnapshot}; // Scene discovery ("artists from the same place"). A frontend driving it needs // all four: the token to cancel a run it has navigated away from, the phase for // the progress label, and the typed error to tell "offline" and "the upstreams diff --git a/crates/qbz-player/src/lib.rs b/crates/qbz-player/src/lib.rs index 6b5919648..c90268838 100644 --- a/crates/qbz-player/src/lib.rs +++ b/crates/qbz-player/src/lib.rs @@ -41,4 +41,4 @@ pub use player::{ InMemorySource, PlaybackBufferState, PlaybackEvent, PlaybackState, Player, SharedState, StreamingConfig, }; -pub use queue::QueueManager; +pub use queue::{QueueAuthoritySnapshot, QueueManager}; diff --git a/crates/qbz-player/src/player/mod.rs b/crates/qbz-player/src/player/mod.rs index 1d8620ae0..2d84c3a7f 100644 --- a/crates/qbz-player/src/player/mod.rs +++ b/crates/qbz-player/src/player/mod.rs @@ -5343,9 +5343,9 @@ impl Player { let stream_url = client .get_stream_url_with_fallback(track_id, quality) .await - .map_err(|e| { - log::error!("Player: Failed to get stream URL: {}", e); - format!("Failed to get stream URL: {}", e) + .map_err(|_| { + log::error!("Player: Failed to get stream URL (details omitted)"); + "Failed to get stream URL".to_string() })?; if !self.is_current_play(gen) { @@ -5355,11 +5355,7 @@ impl Player { return Ok(()); } - log::info!( - "Player: Got stream URL: {} (format: {})", - stream_url.url, - stream_url.mime_type - ); + log::info!("Player: Got stream URL (format: {})", stream_url.mime_type); // Download the audio data log::info!("Player: Starting audio caching..."); @@ -6695,7 +6691,7 @@ impl Player { .timeout(Duration::from_secs(60)) .connect_timeout(Duration::from_secs(10)) .build() - .map_err(|e| format!("Failed to create HTTP client: {}", e))?; + .map_err(|_| "Failed to create HTTP client".to_string())?; log::info!("Caching audio from URL..."); @@ -6704,7 +6700,10 @@ impl Player { .header("User-Agent", "Mozilla/5.0") .send() .await - .map_err(|e| format!("Failed to fetch audio: {}", e))?; + .map_err(|e| format!( + "Failed to fetch audio: {}", + crate::remote_stream::describe_reqwest_error(&e) + ))?; if !response.status().is_success() { return Err(format!("HTTP error: {}", response.status())); @@ -6715,7 +6714,10 @@ impl Player { let bytes = response .bytes() .await - .map_err(|e| format!("Failed to read audio bytes: {}", e))?; + .map_err(|e| format!( + "Failed to read audio bytes: {}", + crate::remote_stream::describe_reqwest_error(&e) + ))?; log::info!("Cached {} bytes", bytes.len()); Ok(bytes.to_vec()) diff --git a/crates/qbz-player/src/queue.rs b/crates/qbz-player/src/queue.rs index 214ae1e4a..dc3f974e4 100644 --- a/crates/qbz-player/src/queue.rs +++ b/crates/qbz-player/src/queue.rs @@ -21,6 +21,7 @@ enum QueueMoveDirection { const MAX_HISTORY_LEN: usize = 50; /// Internal queue state - all in one struct to avoid deadlocks +#[derive(Clone)] struct InternalState { /// All tracks in the queue (original order) tracks: Vec, @@ -45,6 +46,16 @@ struct InternalState { manual_next_count: usize, } +/// An exact, in-process snapshot of queue playback authority. +/// +/// The payload is deliberately opaque: callers can preserve and later restore +/// the queue, but cannot manufacture a partial or internally inconsistent +/// state. It is not a persistence format and is intended for transactional +/// authority handoffs such as QConnect delegation. +#[must_use = "an authority snapshot must be retained until it is restored or deliberately discarded"] +#[derive(Clone)] +pub struct QueueAuthoritySnapshot(InternalState); + /// Queue manager for handling playback queue pub struct QueueManager { state: Mutex, @@ -370,7 +381,10 @@ impl QueueManager { // History stores indices into `tracks`. Remap by track id // so entries for removed rows drop and any entry that still // refers to the kept track points at index 0. - Self::remap_history_by_track_id_internal(&mut state, std::slice::from_ref(&kept)); + Self::remap_history_by_track_id_internal( + &mut state, + std::slice::from_ref(&kept), + ); state.tracks = vec![kept]; state.current_index = Some(0); } else { @@ -1058,9 +1072,7 @@ impl QueueManager { // shrinks the block; exhausting or wrapping the queue dissolves it. match next_idx { Some(0) | None => state.manual_next_count = 0, - Some(_) => { - state.manual_next_count = state.manual_next_count.saturating_sub(1) - } + Some(_) => state.manual_next_count = state.manual_next_count.saturating_sub(1), } next_idx.and_then(|idx| state.tracks.get(idx).cloned()) } @@ -1487,6 +1499,27 @@ impl QueueManager { (state.tracks.clone(), state.current_index) } + /// Capture the complete queue authority state under one lock. + /// + /// Unlike the frontend and persistence projections, this preserves every + /// playback-order detail without caps or normalization: canonical tracks, + /// current cursor, shuffle order and cursor, repeat mode, chronological + /// history, stop-after marker, and the manual-next block size. + pub fn capture_authority_snapshot(&self) -> QueueAuthoritySnapshot { + let state = self.state.lock().unwrap(); + QueueAuthoritySnapshot(state.clone()) + } + + /// Atomically replace the complete queue authority state with `snapshot`. + /// + /// Consuming the opaque snapshot prevents callers from restoring only a + /// subset of the coupled queue fields or mutating them between capture and + /// restore. + pub fn restore_authority_snapshot(&self, snapshot: QueueAuthoritySnapshot) { + let mut state = self.state.lock().unwrap(); + *state = snapshot.0; + } + /// Capture the queue rows, cursor, and chronological play-history indices /// under one lock for durable session persistence. /// @@ -1777,6 +1810,89 @@ mod tests { } } + #[test] + fn authority_snapshot_round_trips_complete_shuffled_state() { + let queue = QueueManager::new(); + let mut tracks: Vec<_> = (101..=105).map(create_test_track).collect(); + tracks[2].version = Some("Authority snapshot edition".to_string()); + tracks[2].bit_depth = Some(24); + tracks[2].sample_rate = Some(192.0); + tracks[2].isrc = Some("TEST00000103".to_string()); + + let shuffle_order = vec![4, 2, 0, 3, 1]; + queue.set_queue_with_order(tracks, Some(2), true, Some(shuffle_order.clone())); + queue.play_index(3); + queue.set_repeat(RepeatMode::All); + queue.set_stop_after(105); + + let expected_tracks = { + let state = queue.state.lock().unwrap(); + format!("{:?}", state.tracks) + }; + let snapshot = queue.capture_authority_snapshot(); + + queue.clear(false); + queue.add_track(create_test_track(999)); + queue.play_index(0); + queue.set_repeat(RepeatMode::One); + + queue.restore_authority_snapshot(snapshot); + + let state = queue.state.lock().unwrap(); + assert_eq!(format!("{:?}", state.tracks), expected_tracks); + assert_eq!(state.current_index, Some(3)); + assert!(state.shuffle); + assert_eq!(state.shuffle_order, shuffle_order); + assert_eq!(state.shuffle_position, 3); + assert_eq!(state.repeat, RepeatMode::All); + assert_eq!(state.history, VecDeque::from([2])); + assert_eq!(state.stop_after_track_id, Some(105)); + assert_eq!(state.manual_next_count, 0); + } + + #[test] + fn authority_snapshot_round_trips_manual_next_block() { + let queue = QueueManager::new(); + queue.set_queue( + vec![ + create_test_track(1), + create_test_track(2), + create_test_track(3), + ], + Some(0), + ); + queue.add_track_next(create_test_track(10)); + queue.add_track_later(create_test_track(11)); + queue.set_repeat(RepeatMode::One); + queue.set_stop_after(3); + + let (expected_shuffle_order, expected_shuffle_position) = { + let state = queue.state.lock().unwrap(); + (state.shuffle_order.clone(), state.shuffle_position) + }; + let snapshot = queue.capture_authority_snapshot(); + queue.clear(false); + queue.restore_authority_snapshot(snapshot); + + let state = queue.state.lock().unwrap(); + assert_eq!( + state + .tracks + .iter() + .map(|track| track.id) + .collect::>(), + vec![1, 10, 11, 2, 3] + ); + assert_eq!(state.current_index, Some(0)); + assert!(!state.shuffle); + assert_eq!(state.shuffle_order, expected_shuffle_order); + assert_eq!(state.shuffle_position, expected_shuffle_position); + assert_eq!(state.repeat, RepeatMode::One); + assert!(state.history.is_empty()); + assert_eq!(state.stop_after_track_id, Some(3)); + assert_eq!(state.manual_next_count, 2); + } + #[test] fn repeat_all_wraps_a_local_album_for_renderer_advance() { let queue = QueueManager::new(); @@ -3140,10 +3256,7 @@ mod tests { fn test_remove_upcoming_after_linear() { let queue = QueueManager::new(); // 101 playing, upcoming = [102, 103, 104, 105]. - queue.set_queue( - (101..=105).map(create_test_track).collect(), - Some(0), - ); + queue.set_queue((101..=105).map(create_test_track).collect(), Some(0)); // Keep upcoming positions 0..=1 (102, 103); drop 2, 3 (104, 105). let removed = queue.remove_upcoming_after(1); diff --git a/crates/qbz-player/src/remote_stream.rs b/crates/qbz-player/src/remote_stream.rs index 9d2e98ce9..ab3be2edc 100644 --- a/crates/qbz-player/src/remote_stream.rs +++ b/crates/qbz-player/src/remote_stream.rs @@ -459,20 +459,53 @@ fn parsed_request_headers( Ok(parsed) } -/// reqwest's `Display` hides the source chain — which is exactly where the -/// diagnosis lives (Akamai's >100-header small-object flood surfaces as hyper's -/// "message head is too large" two levels down). Walk `source()` and join the -/// chain so logs AND signature matching see the real cause. +/// Return a bounded, URL-free diagnostic for a reqwest failure. +/// +/// The raw error and its source chain can contain a signed CDN URL. Inspect the +/// chain only to preserve header-limit classification; never copy an arbitrary +/// cause into logs or a returned error. pub fn describe_reqwest_error(err: &reqwest::Error) -> String { + if error_chain_has_header_limit(err) { + return safe_transport_diagnostic("message head is too large").to_string(); + } + + if err.is_timeout() { + "HTTP transport timed out".to_string() + } else if err.is_connect() { + "HTTP transport connection failed".to_string() + } else if err.is_body() { + "HTTP response body failed".to_string() + } else if err.is_decode() { + "HTTP response decode failed".to_string() + } else if err.is_status() { + "HTTP status rejected".to_string() + } else { + "HTTP transport request failed".to_string() + } +} + +fn error_chain_has_header_limit(err: &reqwest::Error) -> bool { use std::error::Error as _; - let mut out = err.to_string(); + + if is_header_flood_error(&err.to_string()) { + return true; + } let mut source = err.source(); while let Some(cause) = source { - out.push_str(": "); - out.push_str(&cause.to_string()); + if is_header_flood_error(&cause.to_string()) { + return true; + } source = cause.source(); } - out + false +} + +fn safe_transport_diagnostic(message: &str) -> &'static str { + if is_header_flood_error(message) { + "HTTP response header limit exceeded (message head is too large)" + } else { + "HTTP transport request failed" + } } /// True when an error message (already chain-expanded by @@ -487,7 +520,19 @@ pub fn is_header_flood_error(message: &str) -> bool { #[cfg(test)] mod tests { - use super::{content_range_total, parsed_request_headers, probe_content_length}; + use super::{ + content_range_total, parsed_request_headers, probe_content_length, + safe_transport_diagnostic, + }; + + #[test] + fn transport_diagnostic_never_echoes_signed_url() { + let marker = "https://cdn.example/audio.flac?jwt=secret&request_sig=signed"; + let diagnostic = safe_transport_diagnostic(marker); + assert_eq!(diagnostic, "HTTP transport request failed"); + assert!(!diagnostic.contains(marker)); + assert!(!diagnostic.contains("secret")); + } #[test] fn source_request_headers_are_parsed_for_the_http_request() { diff --git a/crates/qbz-qobuz/src/client.rs b/crates/qbz-qobuz/src/client.rs index cad70aaea..d7b543102 100644 --- a/crates/qbz-qobuz/src/client.rs +++ b/crates/qbz-qobuz/src/client.rs @@ -23,25 +23,29 @@ use zeroize::Zeroizing; const USER_AGENT: &str = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0"; -/// Read a short, log-safe preview of a response body — for diagnosing an -/// unexpected non-2xx (e.g. distinguishing an edge/WAF HTML 403 from the API's -/// JSON error envelope, issue #637). Bounded so a large/HTML body can't bloat -/// the log; prefixed with " : " so it reads well appended to an error message. +/// Return a bounded, content-free diagnostic for an unexpected response body. +/// +/// Remote bodies are untrusted and can contain credentials or echo signed +/// request parameters. They must never enter logs or a displayable `ApiError`. +/// We retain only whether a body was present. async fn body_preview(response: reqwest::Response) -> String { match response.text().await { Ok(body) => { - let trimmed = body.trim(); - if trimmed.is_empty() { - " : ".to_string() - } else { - let preview: String = trimmed.chars().take(200).collect(); - format!(" : {preview}") - } + let body = Zeroizing::new(body); + redacted_body_diagnostic(body.as_str()).to_string() } Err(_) => String::new(), } } +fn redacted_body_diagnostic(body: &str) -> &'static str { + if body.trim().is_empty() { + " : " + } else { + " : " + } +} + /// CMAF session state (session/start + infos for key derivation) struct CmafSession { session_id: String, @@ -3264,6 +3268,16 @@ impl Default for QobuzClient { mod tests { use super::*; + #[test] + fn remote_body_diagnostic_never_echoes_payload() { + let marker = "jwt=secret&request_sig=signed-value"; + let diagnostic = redacted_body_diagnostic(marker); + assert_eq!(diagnostic, " : "); + assert!(!diagnostic.contains(marker)); + assert!(!diagnostic.contains("secret")); + assert_eq!(redacted_body_diagnostic(" \n"), " : "); + } + /// Install the process-level rustls `CryptoProvider`, once. /// /// UNRELATED DRIVE-BY FIX, 2026-08-16: the two gate tests below were failing diff --git a/crates/qbz-qobuz/src/error.rs b/crates/qbz-qobuz/src/error.rs index 5dd1ecd41..ec3a1c9cb 100644 --- a/crates/qbz-qobuz/src/error.rs +++ b/crates/qbz-qobuz/src/error.rs @@ -34,8 +34,9 @@ pub enum ApiError { /// Qobuz answered 403 Forbidden on an authenticated request. The account is /// authenticated but not currently allowed to perform the action (entitlement /// not restored after an outage, geo/concurrency limit, or an edge/WAF block). - /// The string is a short body preview for diagnostics. Terminal, never a - /// per-quality restriction — abort the fallback loop instead of retrying. + /// The string contains only a redacted body-presence diagnostic. Terminal, + /// never a per-quality restriction — abort the fallback loop instead of + /// retrying. #[error("Access forbidden by Qobuz (HTTP 403){0}")] Forbidden(String), diff --git a/crates/qconnect-app/src/app.rs b/crates/qconnect-app/src/app.rs index f910a5206..0993c8bea 100644 --- a/crates/qconnect-app/src/app.rs +++ b/crates/qconnect-app/src/app.rs @@ -1,11 +1,13 @@ use std::{ - sync::Arc, + collections::VecDeque, + sync::{Arc, Mutex as StdMutex}, time::{Duration, SystemTime, UNIX_EPOCH}, }; use qconnect_core::{ apply_event, apply_renderer_command, telemetry, PendingCorrelation, PendingQueueAction, - QConnectQueueState, QConnectRendererState, QueueEvent, QueueItem, QueueVersion, RendererCommand, + QConnectQueueState, QConnectRendererState, QueueEvent, QueueItem, QueueVersion, + RendererCommand, }; use qconnect_protocol::{ build_qconnect_outbound_envelope, build_qconnect_renderer_outbound_envelope, @@ -52,6 +54,17 @@ where /// (slice 6) both lock THIS one, exactly as they share it today via the Tauri /// adapter. See `MASTER-qconnect-to-slint-plan.md` §0 (one Mutex, not two). sync: Arc>, + /// Timers belong to this app instance, not to the process runtime. Keeping + /// their handles here lets disconnect/retirement abort them synchronously + /// before an old sink (and its renderer engine/stream feeder) can outlive + /// the authority that created it. + background_tasks: Arc>, +} + +#[derive(Default)] +struct BackgroundTasks { + accepting: bool, + handles: Vec>, } impl Clone for QconnectApp @@ -65,6 +78,7 @@ where sink: Arc::clone(&self.sink), state: Arc::clone(&self.state), sync: Arc::clone(&self.sync), + background_tasks: Arc::clone(&self.background_tasks), } } } @@ -90,6 +104,10 @@ where sink, state: Arc::new(Mutex::new(QconnectRuntimeState::default())), sync, + background_tasks: Arc::new(StdMutex::new(BackgroundTasks { + accepting: true, + handles: Vec::new(), + })), } } @@ -110,6 +128,7 @@ where pub async fn connect(&self, config: WsTransportConfig) -> Result<(), QconnectAppError> { self.transport.connect(config).await?; + self.set_background_tasks_accepting(true); { let mut state = self.state.lock().await; state.transport_connected = true; @@ -121,7 +140,8 @@ where } pub async fn disconnect(&self) -> Result<(), QconnectAppError> { - self.transport.disconnect().await?; + self.abort_background_tasks().await; + let transport_result = self.transport.disconnect().await; { let mut state = self.state.lock().await; state.transport_connected = false; @@ -130,7 +150,7 @@ where self.sink .on_event(QconnectAppEvent::TransportDisconnected) .await; - Ok(()) + transport_result.map_err(Into::into) } pub async fn queue_state_snapshot(&self) -> QConnectQueueState { @@ -573,9 +593,10 @@ where fn spawn_pending_timeout_watch(&self, action_uuid: String) { let app = self.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { app.watch_pending_action_timeout(action_uuid).await; }); + self.own_background_task(handle); } async fn watch_pending_action_timeout(&self, action_uuid: String) { @@ -1481,9 +1502,48 @@ where /// `watch_pending_action_timeout`; relocated from the Tauri sink (slice 2+4). pub fn arm_renderer_watchdog(&self, renderer_id: i32, generation: u64) { let app = self.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { app.run_renderer_watchdog(renderer_id, generation).await; }); + self.own_background_task(handle); + } + + fn own_background_task(&self, handle: tokio::task::JoinHandle<()>) { + let mut tasks = self + .background_tasks + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if !tasks.accepting { + handle.abort(); + return; + } + tasks.handles.retain(|task| !task.is_finished()); + tasks.handles.push(handle); + } + + fn set_background_tasks_accepting(&self, accepting: bool) { + let mut tasks = self + .background_tasks + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + tasks.accepting = accepting; + } + + async fn abort_background_tasks(&self) { + let handles = { + let mut tasks = self + .background_tasks + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + tasks.accepting = false; + std::mem::take(&mut tasks.handles) + }; + for handle in &handles { + handle.abort(); + } + for handle in handles { + let _ = handle.await; + } } async fn run_renderer_watchdog(&self, renderer_id: i32, generation: u64) { @@ -1585,9 +1645,8 @@ where // raw parse would classify -1 as `Some(-1)` and fall into the `Some(_)` // peer-active arm — suppressing the idle auto-take. Filtering `>= 0` maps // -1 (and any negative sentinel) to None, the genuine "session idle" state. - let incoming_active = normalize_active_renderer_id( - payload.get("active_renderer_id").and_then(Value::as_i64), - ); + let incoming_active = + normalize_active_renderer_id(payload.get("active_renderer_id").and_then(Value::as_i64)); let server = match incoming_active { None => ServerActiveState::None, Some(id) if Some(id) == local_id => ServerActiveState::Me, @@ -1743,10 +1802,33 @@ where /// 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, + transport_rx: tokio::sync::broadcast::Receiver, + idle_retry_active: bool, + ) { + self.run_session_loop_with_prefetched( + host, + transport_rx, + idle_retry_active, + VecDeque::new(), + ) + .await; + } + + /// Run the shared transport loop after draining events that this exact + /// receiver already consumed during a transactional preflight. + /// + /// `prefetched` is processed once, in FIFO order, before the receiver is + /// polled again. Callers must move (not copy) consumed events into this + /// queue; the receiver then resumes at its existing cursor, so the handoff + /// has neither a loss window nor a replay duplicate. + pub async fn run_session_loop_with_prefetched( &self, host: Arc, mut transport_rx: tokio::sync::broadcast::Receiver, idle_retry_active: bool, + mut prefetched: VecDeque, ) { // NOTE: `transport_rx` is subscribed by the adapter SYNCHRONOUSLY before // this task is spawned (and before any further await), so the receiver is @@ -1772,7 +1854,11 @@ where // until the session_uuid is confirmed. Reset on disconnect. let mut lagged_reask_attempts: u32 = 0; loop { - match transport_rx.recv().await { + let next_event = match prefetched.pop_front() { + Some(event) => Ok(event), + None => transport_rx.recv().await, + }; + match next_event { Ok(event) => { // P1-3: on every SESSION_STATE, capture our prior active/playing // state + classify the server's active renderer BEFORE the sink @@ -1781,8 +1867,10 @@ where 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); + 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()) @@ -1794,7 +1882,9 @@ where ) .await; } else { - log::warn!("[QConnect] SESSION_STATE received but no session_uuid in payload: {}", evt.payload); + log::warn!( + "[QConnect] SESSION_STATE received without session_uuid" + ); } } } @@ -1914,16 +2004,10 @@ where message ); } - TransportEvent::CloudError { - msg_id, - code, - descr, - } => { + TransportEvent::CloudError { msg_id, code } => { log::warn!( - "[QConnect/Transport] Cloud rejected session: msg_id={} code={} descr={:?} (issue #358)", - msg_id, - code, - descr + "[QConnect/Transport] Cloud rejected session: msg_id={} code={} (issue #358)", + msg_id, code ); self.sink .on_event(QconnectAppEvent::Diagnostic { @@ -1932,7 +2016,7 @@ where payload: serde_json::json!({ "msg_id": msg_id, "code": code, - "descr": descr, + "category": "cloud-rejected", }), }) .await; @@ -2105,20 +2189,23 @@ where #[cfg(test)] mod tests { + use std::collections::VecDeque; use std::sync::Arc; use async_trait::async_trait; use qconnect_core::QueueVersion; use qconnect_transport_ws::InMemoryWsTransport; - use crate::session::{LocalIdentity, QconnectRendererInfo, ServerActiveState}; + use crate::session::{ + LocalIdentity, QconnectLifecycleState, QconnectRendererInfo, ServerActiveState, + }; use crate::QconnectRemoteSyncState; use serde_json::json; use tokio::sync::Mutex; use crate::{QconnectAppEvent, QconnectEventSink}; - use super::{queue_hashes_diverge, QconnectApp}; + use super::{queue_hashes_diverge, QconnectApp, SessionLoopHost}; use qconnect_core::QConnectQueueState; use qconnect_protocol::{ QueueCommandType, QueueEventType, QueueServerEvent, RendererCommandType, @@ -2144,6 +2231,33 @@ mod tests { } } + #[derive(Default)] + struct TestSessionLoopHost { + lifecycles: Mutex>, + } + + #[async_trait] + impl SessionLoopHost for TestSessionLoopHost { + async fn update_lifecycle(&self, state: QconnectLifecycleState) { + self.lifecycles.lock().await.push(state); + } + + async fn bootstrap_after_reconnect(&self) {} + + async fn deferred_renderer_join(&self, _session_uuid: String, _reason: i32) {} + + async fn on_reconnect_exhausted( + &self, + _attempts: u32, + _last_reason: String, + _idle_retry_active: bool, + ) -> bool { + true + } + + async fn on_loop_error(&self, _message: String) {} + } + fn test_config() -> WsTransportConfig { let mut config = WsTransportConfig::default(); config.endpoint_url = "wss://example.invalid/ws".to_string(); @@ -2169,6 +2283,66 @@ mod tests { (app, sink, transport, events_rx) } + #[tokio::test] + async fn session_loop_drains_prefetched_events_once_before_live_receiver() { + // Build without `app.connect()`: that helper intentionally emits its own + // TransportConnected sink event, which is outside the prefetched/live + // receiver sequence this test measures. + let transport = Arc::new(InMemoryWsTransport::new()); + let sink = TestSink::default(); + let app = QconnectApp::new( + transport, + Arc::new(sink.clone()), + Arc::new(Mutex::new(QconnectRemoteSyncState::default())), + ); + let (events_tx, mut events_rx) = tokio::sync::broadcast::channel(8); + events_tx + .send(qconnect_transport_ws::TransportEvent::Connected) + .expect("preflight receiver exists"); + events_tx + .send(qconnect_transport_ws::TransportEvent::SessionEstablished) + .expect("preflight receiver exists"); + let prefetched = VecDeque::from([ + events_rx.recv().await.expect("prefetched connected event"), + events_rx + .recv() + .await + .expect("prefetched established event"), + ]); + events_tx + .send(qconnect_transport_ws::TransportEvent::Disconnected) + .expect("live receiver exists"); + drop(events_tx); + + let host = Arc::new(TestSessionLoopHost::default()); + let loop_host: Arc = host.clone(); + app.run_session_loop_with_prefetched(loop_host, events_rx, false, prefetched) + .await; + + assert_eq!( + *host.lifecycles.lock().await, + vec![ + QconnectLifecycleState::Connected, + QconnectLifecycleState::Reconnecting, + ] + ); + let sink_events = sink.snapshot().await; + assert_eq!( + sink_events + .iter() + .filter(|event| matches!(event, QconnectAppEvent::TransportConnected)) + .count(), + 1 + ); + assert_eq!( + sink_events + .iter() + .filter(|event| matches!(event, QconnectAppEvent::TransportDisconnected)) + .count(), + 1 + ); + } + fn renderer_info(renderer_id: i32, device_uuid: &str) -> QconnectRendererInfo { QconnectRendererInfo { renderer_id, @@ -2194,7 +2368,10 @@ mod tests { let mut state = handle.lock().await; state.session.local_renderer_id = Some(1); state.session.active_renderer_id = Some(2); - state.session.renderers = vec![renderer_info(1, "local-uuid"), renderer_info(2, "peer-uuid")]; + state.session.renderers = vec![ + renderer_info(1, "local-uuid"), + renderer_info(2, "peer-uuid"), + ]; } /// SESSION_STATE writes session topology under the sync lock and reports the @@ -2356,6 +2533,35 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn disconnect_aborts_runtime_owned_watchdog() { + let (app, sink, _transport, _rx) = build_connected_app().await; + seed_local_and_active_peer(&app).await; + let armed = app + .apply_session_management_event( + "MESSAGE_TYPE_SRVR_CTRL_RENDERER_STATE_UPDATED", + &json!({ "renderer_id": 2, "player_state": { "playing_state": 2 } }), + &local_identity("local-uuid"), + ) + .await; + let (renderer_id, generation) = armed.watchdog_arm.expect("arm"); + app.arm_renderer_watchdog(renderer_id, generation); + + app.disconnect().await.expect("disconnect"); + tokio::task::yield_now().await; + tokio::time::advance(std::time::Duration::from_millis(12_010)).await; + tokio::task::yield_now().await; + + assert!( + !sink + .snapshot() + .await + .iter() + .any(|event| matches!(event, QconnectAppEvent::RendererUnreachable { .. })), + "a retired app watchdog must not retain or publish through its sink" + ); + } + /// ACTIVE_DISCONNECTED(status==2) for the active peer reports the /// disconnected renderer; freezing emits RendererDisconnected + UNKNOWN(0). #[tokio::test] @@ -2427,6 +2633,42 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn disconnect_aborts_pending_timeout_and_resync() { + let (app, sink, transport, _events_rx) = build_connected_app().await; + let command = app + .build_queue_command( + QueueCommandType::CtrlSrvrQueueAddTracks, + json!({"track_ids":[101]}), + ) + .await; + app.send_queue_command(command) + .await + .expect("send queue command"); + let sent_before_disconnect = transport.sent_messages().await.len(); + + app.disconnect().await.expect("disconnect"); + tokio::task::yield_now().await; + tokio::time::advance(std::time::Duration::from_millis( + QconnectApp::::PENDING_ACTION_TIMEOUT_MS + 10, + )) + .await; + tokio::task::yield_now().await; + + let events = sink.snapshot().await; + assert!(!events + .iter() + .any(|event| matches!(event, QconnectAppEvent::PendingActionTimedOut { .. }))); + assert!(!events + .iter() + .any(|event| matches!(event, QconnectAppEvent::QueueResyncTriggered))); + assert_eq!( + transport.sent_messages().await.len(), + sent_before_disconnect, + "retired timeout must not send ask-for-state" + ); + } + #[tokio::test] async fn concurrent_remote_event_cancels_pending_and_requests_resync() { let (app, sink, transport, _events_rx) = build_connected_app().await; diff --git a/crates/qconnect-app/src/authority.rs b/crates/qconnect-app/src/authority.rs new file mode 100644 index 000000000..cc76a1509 --- /dev/null +++ b/crates/qconnect-app/src/authority.rs @@ -0,0 +1,1234 @@ +//! Exact authority stamps for QConnect runtimes. +//! +//! A stamp is reserved while a replacement runtime is being prepared, but it +//! does not become authoritative until [`AuthorityCell::install`] succeeds. +//! Long-lived tasks capture that exact stamp and re-check it immediately before +//! publishing or mutating shared state. This prevents a retired owner or guest +//! runtime from acting on a replacement runtime after an asynchronous callback +//! completes late. + +use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use tokio::sync::Notify; + +/// One process-local enable epoch for the global QConnect lifecycle. +/// +/// A token is intentionally opaque: adapters may carry it across network +/// awaits, but every authority/runtime/LAN publication must revalidate it at a +/// short synchronous commit boundary. `disable()` advances the epoch before it +/// starts any asynchronous teardown, so work that finishes late cannot publish +/// a replacement runtime or resurrect LAN discovery. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct QconnectEnableToken(u64); + +/// Exact disabled intent produced by one teardown boundary. Automatic restore +/// may re-enable only while this remains the latest intent; a later manual +/// enable or repeated disable invalidates the token. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct QconnectDisabledToken(u64); + +#[derive(Debug, Clone, Copy)] +struct QconnectEnableState { + enabled: bool, + epoch: u64, +} + +/// Synchronous latest-intent gate for enable/disable versus long-running +/// connect work. +/// +/// The mutex is never held across an await. Callers may use +/// [`commit_if_current`](Self::commit_if_current) only for a bounded in-memory +/// install (authority stamp, runtime slot or LAN slot). This lets disable win +/// immediately without waiting behind DNS, HTTP, QWS or API I/O. +#[derive(Debug)] +pub struct QconnectEnableIntent { + state: Mutex, + changed: Notify, +} + +impl QconnectEnableIntent { + pub fn new(enabled: bool) -> Self { + Self { + state: Mutex::new(QconnectEnableState { enabled, epoch: 0 }), + changed: Notify::new(), + } + } + + /// Express enabled intent and return its stable epoch. Repeated enable + /// requests are idempotent and therefore do not invalidate an in-flight + /// connect for the same user intent. + pub fn enable(&self) -> QconnectEnableToken { + let mut state = self.lock_enable_state(); + if !state.enabled { + state.epoch = next_enable_epoch(state.epoch); + state.enabled = true; + } + QconnectEnableToken(state.epoch) + } + + /// Express a distinct enabled intent even when QConnect was already + /// logically enabled. Interactive connect entrypoints use this so a fresh + /// user click supersedes an automatic restore that has not installed a + /// runtime yet; retry loops should retain the token returned by `enable()` + /// or this method instead of minting one per attempt. + pub fn enable_new_intent(&self) -> QconnectEnableToken { + let mut state = self.lock_enable_state(); + state.epoch = next_enable_epoch(state.epoch); + state.enabled = true; + self.changed.notify_waiters(); + QconnectEnableToken(state.epoch) + } + + /// Invalidate every token issued before this disable boundary. The epoch is + /// advanced even for an idempotent repeated disable, so a later-invoked + /// shutdown always wins over older work. + pub fn disable(&self) -> QconnectDisabledToken { + let mut state = self.lock_enable_state(); + state.epoch = next_enable_epoch(state.epoch); + state.enabled = false; + // Notify while the state mutex is still held. `cancelled()` registers + // its waiter before inspecting this same state, which closes both + // possible missed-wake windows around the epoch check. + self.changed.notify_waiters(); + QconnectDisabledToken(state.epoch) + } + + /// Atomically disable one exact enabled intent. This is the inverse of + /// [`enable_if_disabled`](Self::enable_if_disabled): a renderer handoff + /// may replace a consumed restore token without disabling a newer manual + /// enable that won the race. + pub fn disable_if_current(&self, token: QconnectEnableToken) -> Option { + let mut state = self.lock_enable_state(); + if !state.enabled || state.epoch != token.0 { + return None; + } + state.epoch = next_enable_epoch(state.epoch); + state.enabled = false; + self.changed.notify_waiters(); + Some(QconnectDisabledToken(state.epoch)) + } + + /// Atomically enable only from one exact disabled boundary. On success the + /// epoch advances, transferring restore ownership to the returned enabled + /// token. A stale automatic restore can never overwrite newer user intent. + pub fn enable_if_disabled(&self, token: QconnectDisabledToken) -> Option { + let mut state = self.lock_enable_state(); + if state.enabled || state.epoch != token.0 { + return None; + } + state.epoch = next_enable_epoch(state.epoch); + state.enabled = true; + Some(QconnectEnableToken(state.epoch)) + } + + pub fn is_disabled_current(&self, token: QconnectDisabledToken) -> bool { + let state = self.lock_enable_state(); + !state.enabled && state.epoch == token.0 + } + + pub fn current_token(&self) -> Option { + let state = self.lock_enable_state(); + state.enabled.then_some(QconnectEnableToken(state.epoch)) + } + + pub fn is_current(&self, token: QconnectEnableToken) -> bool { + let state = self.lock_enable_state(); + state.enabled && state.epoch == token.0 + } + + /// Wait until `token` no longer represents enabled intent. + /// + /// Registration is deliberately enabled before the state check. If a + /// disable wins first, the state check returns immediately; if it wins + /// after registration, the notification is retained for the await. + pub async fn cancelled(&self, token: QconnectEnableToken) { + loop { + let notified = self.changed.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if !self.is_current(token) { + return; + } + notified.await; + } + } + + /// Run one short, non-async publication only if `token` still represents + /// the latest enabled intent. Disable is serialized against the closure, so + /// either the publication wins first and teardown observes it, or disable + /// wins first and the publication is rejected. + pub fn commit_if_current( + &self, + token: QconnectEnableToken, + commit: impl FnOnce() -> R, + ) -> Option { + let state = self.lock_enable_state(); + if !state.enabled || state.epoch != token.0 { + return None; + } + Some(commit()) + } + + /// Publish a short disabled-state side effect only while no newer enable + /// has won. This is the inverse commit boundary used by status mirrors: + /// either the disabled publication completes first and a later enable + /// overwrites it, or the newer enable rejects the stale publication. + pub fn commit_if_disabled(&self, commit: impl FnOnce() -> R) -> Option { + let state = self.lock_enable_state(); + if state.enabled { + return None; + } + Some(commit()) + } + + fn lock_enable_state(&self) -> MutexGuard<'_, QconnectEnableState> { + self.state.lock().unwrap_or_else(|poisoned| { + log::error!("[QConnect] enable-intent gate recovered from a poisoned lock"); + let guard = poisoned.into_inner(); + self.state.clear_poison(); + guard + }) + } +} + +fn next_enable_epoch(epoch: u64) -> u64 { + epoch + .checked_add(1) + .expect("QConnect enable-intent epoch space exhausted") +} + +/// Credential authority represented by a runtime. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum AuthorityOrigin { + /// The daemon's authenticated account. + Owner, + /// A controller-delegated identity, scoped to one coordinator generation. + Delegated { generation: u64 }, +} + +impl AuthorityOrigin { + pub const fn delegated_generation(self) -> Option { + match self { + Self::Owner => None, + Self::Delegated { generation } => Some(generation), + } + } +} + +/// Unforgeable-in-module identity for one prepared/installed runtime. +/// +/// The epoch distinguishes repeated owner runtimes as well as delegated +/// generations. Checking only `Owner` would let an old owner's reconnect task +/// mutate a freshly restored owner runtime. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct AuthorityStamp { + epoch: u64, + origin: AuthorityOrigin, +} + +impl AuthorityStamp { + pub const fn epoch(self) -> u64 { + self.epoch + } + + pub const fn origin(self) -> AuthorityOrigin { + self.origin + } +} + +#[derive(Clone, Copy, Default, PartialEq, Eq)] +struct AuthorityState { + current: Option, + /// Monotonic fence for installs. A clear advances this even while no runtime + /// is active, invalidating candidates reserved before the clear. + installed_through_epoch: u64, +} + +/// Exact owner-side observation of one authority generation. +/// +/// The fields stay private deliberately: callers may carry this token across +/// queues/awaits, but only [`AuthorityCell`] can mint or validate it. Comparing +/// only `current` is insufficient because an off-state can travel through +/// `None -> delegated -> None` and look identical again. The captured +/// high-water mark makes every successful install/clear part of the identity. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct OwnerAuthorityToken { + current: Option, + installed_through_epoch: u64, +} + +impl fmt::Debug for OwnerAuthorityToken { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("OwnerAuthorityToken(..)") + } +} + +/// Synchronous, short-held authority cell shared by callback/task boundaries. +/// +/// No guard is exposed, so callers cannot accidentally hold this mutex across +/// an `.await`. Poisoning is recovered deliberately: a panic in diagnostics or +/// teardown must not disable the stale-callback safety boundary. +pub struct AuthorityCell { + next_epoch: AtomicU64, + state: Mutex, + owner_action_fences: AtomicU64, + active_actions: AtomicU64, + actions_drained: Notify, + owner_actions_resumed: Notify, +} + +impl Default for AuthorityCell { + fn default() -> Self { + Self { + // Epoch zero is reserved as the initial high-water mark. + next_epoch: AtomicU64::new(1), + state: Mutex::new(AuthorityState::default()), + owner_action_fences: AtomicU64::new(0), + active_actions: AtomicU64::new(0), + actions_drained: Notify::new(), + owner_actions_resumed: Notify::new(), + } + } +} + +/// RAII admission for one authority-sensitive action. +/// +/// Handoffs first close admission and then wait for every live permit to drop +/// before capturing or restoring the queue. This turns the authority checks +/// from a racy check-before-await into a real transaction boundary. +pub struct AuthorityActionPermit { + authority: Arc, +} + +impl fmt::Debug for AuthorityActionPermit { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("AuthorityActionPermit(..)") + } +} + +impl Drop for AuthorityActionPermit { + fn drop(&mut self) { + let previous = self.authority.active_actions.fetch_sub(1, Ordering::AcqRel); + debug_assert!(previous > 0, "authority action permit count underflow"); + if previous == 1 { + self.authority.actions_drained.notify_waiters(); + } + } +} + +/// Atomic classification of one owner-side observation attempt. +/// +/// `Delegated` is an actual guest authority snapshot. `Fenced` means a +/// lifecycle transition is in progress (or raced the observation) and callers +/// must retry without treating it as a guest handoff. +#[derive(Debug)] +pub enum OwnerAuthorityObservation { + Owner { + token: OwnerAuthorityToken, + permit: AuthorityActionPermit, + }, + Delegated, + Fenced, +} + +/// Exact re-admission result for work already stamped by an owner observation. +#[derive(Debug)] +pub enum ExactOwnerAuthorityObservation { + Admitted(AuthorityActionPermit), + Stale, + Fenced, +} + +impl fmt::Debug for AuthorityCell { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let state = self.lock_state(); + f.debug_struct("AuthorityCell") + .field("current", &state.current) + .field("installed_through_epoch", &state.installed_through_epoch) + .finish_non_exhaustive() + } +} + +impl AuthorityCell { + pub fn new() -> Self { + Self::default() + } + + /// Reserve a unique epoch without making it active. + pub fn reserve(&self, origin: AuthorityOrigin) -> AuthorityStamp { + AuthorityStamp { + epoch: self.reserve_epoch(), + origin, + } + } + + /// Install `stamp` when it is newer than every prior install/clear fence. + /// + /// Returning `false` is the expected outcome for a late candidate whose + /// reservation lost a latest-wins race. The current authority is untouched. + pub fn install(&self, stamp: AuthorityStamp) -> bool { + let mut state = self.lock_state(); + if stamp.epoch <= state.installed_through_epoch { + return false; + } + state.installed_through_epoch = stamp.epoch; + state.current = Some(stamp); + true + } + + /// Clear every current authority and fence out all stamps reserved before + /// this call. Intended for global disable/logout/shutdown. + pub fn clear(&self) -> Option { + let fence = self.reserve_epoch(); + let mut state = self.lock_state(); + state.installed_through_epoch = state.installed_through_epoch.max(fence); + state.current.take() + } + + /// Clear only if `expected` is still current, while fencing candidates that + /// were already reserved. A retired runtime therefore cannot clear its + /// replacement during late teardown. + pub fn clear_if_current(&self, expected: AuthorityStamp) -> bool { + let fence = self.reserve_epoch(); + let mut state = self.lock_state(); + if state.current != Some(expected) { + return false; + } + state.installed_through_epoch = state.installed_through_epoch.max(fence); + state.current = None; + true + } + + pub fn current(&self) -> Option { + self.lock_state().current + } + + pub fn is_current(&self, stamp: AuthorityStamp) -> bool { + self.current() == Some(stamp) + } + + /// Owner-only daemon side effects remain enabled while QConnect is off or + /// an owner runtime is installed, and are fenced for every guest epoch. + pub fn owner_actions_allowed(&self) -> bool { + self.owner_action_fences.load(Ordering::Acquire) == 0 + && !matches!( + self.current().map(AuthorityStamp::origin), + Some(AuthorityOrigin::Delegated { .. }) + ) + } + + /// Admit a daemon-local owner action and keep it visible to an authority + /// handoff until the returned permit is dropped. Local actions are valid + /// while QConnect is off or an owner runtime is current, never for a guest. + pub fn try_owner_action_permit(self: &Arc) -> Option { + self.try_owner_action_permit_observed() + .map(|(_, permit)| permit) + } + + /// Admit owner work and capture the exact authority generation under which + /// its input will be read. The permit must remain live until that read and + /// every directly governed mutation complete. The token may then stamp a + /// queued continuation for exact re-admission at its eventual consumer. + pub fn try_owner_action_permit_observed( + self: &Arc, + ) -> Option<(OwnerAuthorityToken, AuthorityActionPermit)> { + match self.observe_owner_authority() { + OwnerAuthorityObservation::Owner { token, permit } => Some((token, permit)), + OwnerAuthorityObservation::Delegated | OwnerAuthorityObservation::Fenced => None, + } + } + + /// Atomically distinguish a real delegated observation from a transient + /// lifecycle fence. Consumers that publish authority edges must skip a + /// `Fenced` observation: treating every failed owner admission as delegated + /// would fabricate a handoff while a guest candidate is merely activating + /// and may still fail. + pub fn observe_owner_authority(self: &Arc) -> OwnerAuthorityObservation { + if self.owner_action_fences.load(Ordering::Acquire) != 0 { + return OwnerAuthorityObservation::Fenced; + } + let observed_state = *self.lock_state(); + let Some(observed) = owner_token_for(observed_state) else { + // This was a real delegated snapshot. A fence that starts after the + // read cannot make the guest observation unsafe: callers will stamp + // it delegated and never perform an owner-only effect. + return OwnerAuthorityObservation::Delegated; + }; + + self.active_actions + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_add(1) + }) + .expect("QConnect authority action permit count exhausted"); + + // Close both races: fence acquisition and an authority install/clear + // between the observation and the active-action increment. + let fenced = self.owner_action_fences.load(Ordering::Acquire) != 0; + let current_state = *self.lock_state(); + if !fenced && current_state == observed_state { + OwnerAuthorityObservation::Owner { + token: observed, + permit: AuthorityActionPermit { + authority: Arc::clone(self), + }, + } + } else { + self.release_rejected_action(); + if fenced { + OwnerAuthorityObservation::Fenced + } else if owner_token_for(current_state).is_none() { + OwnerAuthorityObservation::Delegated + } else { + // An owner/clear generation changed without leaving a stable + // delegated snapshot at this commit boundary. Retry instead of + // publishing either authority from a torn observation. + OwnerAuthorityObservation::Fenced + } + } + } + + /// Re-admit a queued owner continuation only if the complete authority + /// observation that produced it is still current. This rejects stale work + /// even when the visible origin cycles back to owner or `None`. + pub fn try_owner_action_permit_exact( + self: &Arc, + expected: OwnerAuthorityToken, + ) -> Option { + match self.observe_exact_owner_authority(expected) { + ExactOwnerAuthorityObservation::Admitted(permit) => Some(permit), + ExactOwnerAuthorityObservation::Stale | ExactOwnerAuthorityObservation::Fenced => None, + } + } + + /// Re-admit an exact owner token while preserving the distinction between + /// a permanently stale generation and a transient lifecycle fence. + pub fn observe_exact_owner_authority( + self: &Arc, + expected: OwnerAuthorityToken, + ) -> ExactOwnerAuthorityObservation { + if self.owner_action_fences.load(Ordering::Acquire) != 0 { + return ExactOwnerAuthorityObservation::Fenced; + } + if owner_token_for(*self.lock_state()) != Some(expected) { + return ExactOwnerAuthorityObservation::Stale; + } + + self.active_actions + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_add(1) + }) + .expect("QConnect authority action permit count exhausted"); + + let fenced = self.owner_action_fences.load(Ordering::Acquire) != 0; + let current_matches = owner_token_for(*self.lock_state()) == Some(expected); + if !fenced && current_matches { + ExactOwnerAuthorityObservation::Admitted(AuthorityActionPermit { + authority: Arc::clone(self), + }) + } else { + self.release_rejected_action(); + if fenced { + ExactOwnerAuthorityObservation::Fenced + } else { + ExactOwnerAuthorityObservation::Stale + } + } + } + + /// Preserve one queued owner event across a transient lifecycle fence, then + /// either re-admit it under the same exact token or reject it as stale once + /// the transition outcome is known. + pub async fn wait_for_exact_owner_action_permit( + self: &Arc, + expected: OwnerAuthorityToken, + ) -> Option { + loop { + match self.observe_exact_owner_authority(expected) { + ExactOwnerAuthorityObservation::Admitted(permit) => return Some(permit), + ExactOwnerAuthorityObservation::Stale => return None, + ExactOwnerAuthorityObservation::Fenced => { + self.wait_for_owner_actions_resumed().await; + } + } + } + } + + /// Admit an origin-agnostic transport action such as pause, seek or volume. + /// + /// Those controls are valid for both the signed-in owner and the currently + /// delegated renderer, but they must still participate in the handoff + /// fence. Otherwise a local seek admitted just before guest -> owner restore + /// could land on the restored owner's stream after the authority swap. + pub fn try_transport_action_permit(self: &Arc) -> Option { + self.try_action_permit_matching_state(|_| true) + } + + /// Admit work for one exact renderer runtime. A reserved, retired, or + /// replaced stamp cannot obtain a permit, and an in-progress handoff closes + /// admission before waiting for already admitted work to drain. + pub fn try_runtime_action_permit( + self: &Arc, + expected: AuthorityStamp, + ) -> Option { + self.try_action_permit_matching_state(|state| state.current == Some(expected)) + } + + fn try_action_permit_matching_state( + self: &Arc, + matches_authority: F, + ) -> Option + where + F: Fn(&AuthorityState) -> bool, + { + if self.owner_action_fences.load(Ordering::Acquire) != 0 { + return None; + } + if !matches_authority(&self.lock_state()) { + return None; + } + + self.active_actions + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_add(1) + }) + .expect("QConnect authority action permit count exhausted"); + + // Close the acquisition-vs-fence race. If a transition began after the + // first check, this increment is either observed by its drain or rolled + // back here before any caller-visible action can start. + if self.owner_action_fences.load(Ordering::Acquire) == 0 + && matches_authority(&self.lock_state()) + { + Some(AuthorityActionPermit { + authority: Arc::clone(self), + }) + } else { + self.release_rejected_action(); + None + } + } + + fn release_rejected_action(&self) { + let previous = self.active_actions.fetch_sub(1, Ordering::AcqRel); + debug_assert!(previous > 0, "authority action permit count underflow"); + if previous == 1 { + self.actions_drained.notify_waiters(); + } + } + + /// Fence daemon owner-only background work across a guest -> owner restore + /// or global teardown, while the queue snapshot is still being replaced. + pub fn suspend_owner_actions(&self) { + self.owner_action_fences + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_add(1) + }) + .expect("QConnect owner-action fence count exhausted"); + } + + pub fn resume_owner_actions(&self) { + match self + .owner_action_fences + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_sub(1) + }) { + Ok(1) => self.owner_actions_resumed.notify_waiters(), + Ok(_) => {} + Err(_) => log::error!("[QConnect] unbalanced owner-action fence release"), + } + } + + /// Wait until the outermost lifecycle fence has been released. The notify + /// future is registered before the atomic check so a release cannot be + /// missed between observation and suspension. + pub async fn wait_for_owner_actions_resumed(&self) { + loop { + let notified = self.owner_actions_resumed.notified(); + if self.owner_action_fences.load(Ordering::Acquire) == 0 { + return; + } + notified.await; + } + } + + /// Wait until every action admitted before the fence has completed. + /// Callers must hold at least one owner-action fence for the whole wait. + pub async fn wait_for_actions_drained(&self) { + loop { + let notified = self.actions_drained.notified(); + if self.active_actions.load(Ordering::Acquire) == 0 { + return; + } + notified.await; + } + } + + fn reserve_epoch(&self) -> u64 { + self.next_epoch + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |epoch| { + epoch.checked_add(1) + }) + .expect("QConnect authority epoch space exhausted") + } + + fn lock_state(&self) -> MutexGuard<'_, AuthorityState> { + self.state.lock().unwrap_or_else(|poisoned| { + log::error!("[QConnect] authority cell recovered from a poisoned lock"); + let guard = poisoned.into_inner(); + self.state.clear_poison(); + guard + }) + } +} + +fn owner_token_for(state: AuthorityState) -> Option { + if matches!( + state.current.map(AuthorityStamp::origin), + Some(AuthorityOrigin::Delegated { .. }) + ) { + return None; + } + Some(OwnerAuthorityToken { + current: state.current, + installed_through_epoch: state.installed_through_epoch, + }) +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; + use std::sync::Arc; + use std::time::Duration; + + use super::*; + + #[test] + fn disable_invalidates_inflight_enable_token() { + let intent = QconnectEnableIntent::new(false); + let first = intent.enable(); + assert_eq!(intent.enable(), first, "repeated enable must be idempotent"); + assert!(intent.is_current(first)); + + intent.disable(); + assert!(!intent.is_current(first)); + assert!(intent + .commit_if_current(first, || panic!("stale connect publication ran")) + .is_none()); + + let replacement = intent.enable(); + assert_ne!(replacement, first); + assert!(intent.is_current(replacement)); + assert!(!intent.is_current(first)); + } + + #[tokio::test] + async fn cancellation_observes_disable_that_precedes_the_wait() { + let intent = QconnectEnableIntent::new(true); + let token = intent.current_token().expect("initial enabled token"); + intent.disable(); + + tokio::time::timeout(Duration::from_secs(1), intent.cancelled(token)) + .await + .expect("pre-existing cancellation was missed"); + } + + #[tokio::test] + async fn cancellation_wakes_a_registered_waiter() { + let intent = Arc::new(QconnectEnableIntent::new(true)); + let token = intent.current_token().expect("initial enabled token"); + let waiter_intent = Arc::clone(&intent); + let waiter = tokio::spawn(async move { waiter_intent.cancelled(token).await }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + intent.disable(); + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("registered cancellation waiter timed out") + .expect("registered cancellation waiter panicked"); + } + + #[tokio::test] + async fn exact_disable_wakes_its_enabled_waiter() { + let intent = Arc::new(QconnectEnableIntent::new(true)); + let token = intent.current_token().expect("initial enabled token"); + let waiter_intent = Arc::clone(&intent); + let waiter = tokio::spawn(async move { waiter_intent.cancelled(token).await }); + tokio::task::yield_now().await; + + let disabled = intent + .disable_if_current(token) + .expect("exact enabled intent should disable"); + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("exact-disable cancellation waiter timed out") + .expect("exact-disable cancellation waiter panicked"); + assert!(intent.is_disabled_current(disabled)); + } + + #[test] + fn disabled_publication_is_rejected_after_a_new_enable() { + let intent = QconnectEnableIntent::new(true); + intent.disable(); + let replacement = intent.enable(); + + assert!(intent + .commit_if_disabled(|| panic!("stale disabled publication ran")) + .is_none()); + assert!(intent.is_current(replacement)); + } + + #[test] + fn later_manual_disable_invalidates_automatic_restore() { + let intent = QconnectEnableIntent::new(true); + let cast_suspend = intent.disable(); + let manual_disable = intent.disable(); + + assert!(!intent.is_disabled_current(cast_suspend)); + assert!(intent.enable_if_disabled(cast_suspend).is_none()); + assert!(intent.is_disabled_current(manual_disable)); + } + + #[test] + fn exact_disabled_intent_restores_once() { + let intent = QconnectEnableIntent::new(true); + let disabled = intent.disable(); + let restored = intent + .enable_if_disabled(disabled) + .expect("exact disabled intent should restore"); + + assert!(intent.is_current(restored)); + assert!(intent.enable_if_disabled(disabled).is_none()); + } + + #[test] + fn stale_enabled_intent_cannot_disable_its_replacement() { + let intent = QconnectEnableIntent::new(true); + let stale = intent.current_token().expect("initial enabled intent"); + let disabled = intent.disable(); + let replacement = intent + .enable_if_disabled(disabled) + .expect("replacement enabled intent"); + + assert!(intent.disable_if_current(stale).is_none()); + assert!(intent.is_current(replacement)); + } + + #[test] + fn manual_enable_supersedes_cast_disable_of_consumed_restore() { + let intent = QconnectEnableIntent::new(true); + let suspended = intent.disable(); + let automatic_restore = intent + .enable_if_disabled(suspended) + .expect("automatic restore should consume its exact token"); + let manual_connect = intent.enable_new_intent(); + + assert_ne!(manual_connect, automatic_restore); + assert!(intent.disable_if_current(automatic_restore).is_none()); + assert!(intent.is_current(manual_connect)); + } + + #[tokio::test] + async fn new_enabled_intent_wakes_the_superseded_waiter() { + let intent = Arc::new(QconnectEnableIntent::new(true)); + let stale = intent.current_token().expect("initial enabled intent"); + let waiter_intent = Arc::clone(&intent); + let waiter = tokio::spawn(async move { waiter_intent.cancelled(stale).await }); + tokio::task::yield_now().await; + + let replacement = intent.enable_new_intent(); + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("superseded enabled waiter timed out") + .expect("superseded enabled waiter panicked"); + assert!(intent.is_current(replacement)); + } + + #[test] + fn consumed_restore_can_be_replaced_by_exact_cast_disable() { + let intent = QconnectEnableIntent::new(true); + let first_disabled = intent.disable(); + let restored = intent + .enable_if_disabled(first_disabled) + .expect("exact restore should enable"); + let replacement_disabled = intent + .disable_if_current(restored) + .expect("exact restored intent should disable"); + + assert_ne!(replacement_disabled, first_disabled); + assert!(intent.is_disabled_current(replacement_disabled)); + assert!(intent.enable_if_disabled(first_disabled).is_none()); + assert!(intent.enable_if_disabled(replacement_disabled).is_some()); + } + + #[test] + fn manual_disable_after_cast_replacement_invalidates_restore() { + let intent = QconnectEnableIntent::new(true); + let first_disabled = intent.disable(); + let restored = intent + .enable_if_disabled(first_disabled) + .expect("exact restore should enable"); + let cast_disabled = intent + .disable_if_current(restored) + .expect("Cast should replace the consumed restore token"); + let manual_disabled = intent.disable(); + + assert!(!intent.is_disabled_current(cast_disabled)); + assert!(intent.enable_if_disabled(cast_disabled).is_none()); + assert!(intent.is_disabled_current(manual_disabled)); + } + + #[test] + fn disable_serializes_after_an_already_committing_publication() { + let intent = Arc::new(QconnectEnableIntent::new(true)); + let token = intent.current_token().expect("initial enabled token"); + let entered = Arc::new(std::sync::Barrier::new(2)); + let release = Arc::new(std::sync::Barrier::new(2)); + let published = Arc::new(AtomicBool::new(false)); + + let worker_intent = Arc::clone(&intent); + let worker_entered = Arc::clone(&entered); + let worker_release = Arc::clone(&release); + let worker_published = Arc::clone(&published); + let worker = std::thread::spawn(move || { + worker_intent + .commit_if_current(token, || { + worker_entered.wait(); + worker_release.wait(); + worker_published.store(true, AtomicOrdering::Release); + }) + .expect("current publication was rejected"); + }); + + entered.wait(); + let disable_intent = Arc::clone(&intent); + let disable = std::thread::spawn(move || disable_intent.disable()); + assert!( + !disable.is_finished(), + "disable bypassed the commit boundary" + ); + release.wait(); + worker.join().expect("publication thread panicked"); + disable.join().expect("disable thread panicked"); + + assert!(published.load(AtomicOrdering::Acquire)); + assert!(intent.current_token().is_none()); + assert!(!intent.is_current(token)); + } + + #[test] + fn reserve_does_not_activate_a_runtime() { + let cell = AuthorityCell::new(); + let stamp = cell.reserve(AuthorityOrigin::Owner); + + assert_eq!(stamp.epoch(), 1); + assert_eq!(stamp.origin(), AuthorityOrigin::Owner); + assert_eq!(cell.current(), None); + assert!(!cell.is_current(stamp)); + } + + #[test] + fn install_is_exact_and_monotonic() { + let cell = AuthorityCell::new(); + let old = cell.reserve(AuthorityOrigin::Owner); + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 7 }); + + assert!(cell.install(guest)); + assert!(cell.is_current(guest)); + assert_eq!(guest.origin().delegated_generation(), Some(7)); + assert!(!cell.install(old)); + assert!(cell.is_current(guest)); + } + + #[test] + fn clear_fences_previously_reserved_candidates() { + let cell = AuthorityCell::new(); + let owner = cell.reserve(AuthorityOrigin::Owner); + let pending = cell.reserve(AuthorityOrigin::Delegated { generation: 1 }); + assert!(cell.install(owner)); + + assert_eq!(cell.clear(), Some(owner)); + assert!(!cell.install(pending)); + assert_eq!(cell.current(), None); + + let replacement = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(replacement)); + assert!(cell.is_current(replacement)); + } + + #[test] + fn retired_runtime_cannot_clear_its_replacement() { + let cell = AuthorityCell::new(); + let old = cell.reserve(AuthorityOrigin::Owner); + let replacement = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(old)); + assert!(cell.install(replacement)); + + assert!(!cell.clear_if_current(old)); + assert!(cell.is_current(replacement)); + } + + #[test] + fn owner_actions_are_fenced_only_for_delegated_authority() { + let cell = AuthorityCell::new(); + assert!(cell.owner_actions_allowed()); + let owner = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(owner)); + assert!(cell.owner_actions_allowed()); + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 3 }); + assert!(cell.install(guest)); + assert!(!cell.owner_actions_allowed()); + cell.suspend_owner_actions(); + assert!(!cell.owner_actions_allowed()); + cell.clear(); + assert!(!cell.owner_actions_allowed()); + cell.suspend_owner_actions(); + cell.resume_owner_actions(); + assert!(!cell.owner_actions_allowed()); + cell.resume_owner_actions(); + assert!(cell.owner_actions_allowed()); + } + + #[test] + fn permits_are_exact_to_the_installed_authority() { + let cell = Arc::new(AuthorityCell::new()); + assert!(cell.try_owner_action_permit().is_some()); + + let owner = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(owner)); + assert!(cell.try_owner_action_permit().is_some()); + assert!(cell.try_runtime_action_permit(owner).is_some()); + + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 9 }); + assert!(cell.install(guest)); + assert!(cell.try_owner_action_permit().is_none()); + assert!(cell.try_transport_action_permit().is_some()); + assert!(cell.try_runtime_action_permit(owner).is_none()); + assert!(cell.try_runtime_action_permit(guest).is_some()); + } + + #[test] + fn owner_observation_distinguishes_transition_fence_from_delegated() { + let cell = Arc::new(AuthorityCell::new()); + match cell.observe_owner_authority() { + OwnerAuthorityObservation::Owner { permit, .. } => drop(permit), + other => panic!("initial owner observation was {other:?}"), + } + + cell.suspend_owner_actions(); + assert!(matches!( + cell.observe_owner_authority(), + OwnerAuthorityObservation::Fenced + )); + cell.resume_owner_actions(); + + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 12 }); + assert!(cell.install(guest)); + assert!(matches!( + cell.observe_owner_authority(), + OwnerAuthorityObservation::Delegated + )); + } + + #[test] + fn exact_observation_distinguishes_transition_fence_from_stale_token() { + let cell = Arc::new(AuthorityCell::new()); + let (token, permit) = cell + .try_owner_action_permit_observed() + .expect("initial owner observation"); + drop(permit); + + cell.suspend_owner_actions(); + assert!(matches!( + cell.observe_exact_owner_authority(token), + ExactOwnerAuthorityObservation::Fenced + )); + cell.resume_owner_actions(); + match cell.observe_exact_owner_authority(token) { + ExactOwnerAuthorityObservation::Admitted(permit) => drop(permit), + other => panic!("same owner token was not re-admitted: {other:?}"), + } + + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 13 }); + assert!(cell.install(guest)); + assert!(matches!( + cell.observe_exact_owner_authority(token), + ExactOwnerAuthorityObservation::Stale + )); + } + + #[tokio::test] + async fn queued_exact_admission_waits_for_fence_outcome() { + let cell = Arc::new(AuthorityCell::new()); + let (token, permit) = cell + .try_owner_action_permit_observed() + .expect("initial owner observation"); + drop(permit); + cell.suspend_owner_actions(); + let waiter_cell = Arc::clone(&cell); + let waiter = + tokio::spawn( + async move { waiter_cell.wait_for_exact_owner_action_permit(token).await }, + ); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + cell.resume_owner_actions(); + let permit = tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("exact admission waiter timed out") + .expect("exact admission waiter panicked") + .expect("same token became stale across a fence-only transition"); + drop(permit); + } + + #[test] + fn exact_owner_token_rejects_none_guest_none_cycle() { + let cell = Arc::new(AuthorityCell::new()); + let (initial, permit) = cell + .try_owner_action_permit_observed() + .expect("initial off-state owner admission"); + drop(permit); + + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 17 }); + assert!(cell.install(guest)); + assert_eq!(cell.clear(), Some(guest)); + assert_eq!(cell.current(), None); + + assert!(cell.try_owner_action_permit_exact(initial).is_none()); + let (replacement, _) = cell + .try_owner_action_permit_observed() + .expect("fresh off-state owner admission"); + assert_ne!(replacement, initial); + assert!(cell.try_owner_action_permit_exact(replacement).is_some()); + } + + #[test] + fn exact_owner_token_rejects_clear_while_current_stays_none() { + let cell = Arc::new(AuthorityCell::new()); + let (before_clear, permit) = cell + .try_owner_action_permit_observed() + .expect("initial off-state owner admission"); + drop(permit); + + assert_eq!(cell.clear(), None); + assert_eq!(cell.current(), None); + assert!(cell.try_owner_action_permit_exact(before_clear).is_none()); + + let (after_clear, _) = cell + .try_owner_action_permit_observed() + .expect("post-clear off-state owner admission"); + assert_ne!(after_clear, before_clear); + } + + #[test] + fn exact_owner_token_rejects_owner_guest_owner_cycle() { + let cell = Arc::new(AuthorityCell::new()); + let owner = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(owner)); + let (old_owner, permit) = cell + .try_owner_action_permit_observed() + .expect("installed owner admission"); + drop(permit); + + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 18 }); + assert!(cell.install(guest)); + let restored = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(restored)); + + assert!(cell.try_owner_action_permit_exact(old_owner).is_none()); + let (new_owner, _) = cell + .try_owner_action_permit_observed() + .expect("restored owner admission"); + assert_ne!(new_owner, old_owner); + assert!(cell.try_owner_action_permit_exact(new_owner).is_some()); + } + + #[test] + fn transport_actions_allow_guest_but_not_a_transition_fence() { + let cell = Arc::new(AuthorityCell::new()); + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 2 }); + assert!(cell.install(guest)); + assert!(cell.try_transport_action_permit().is_some()); + + cell.suspend_owner_actions(); + assert!(cell.try_transport_action_permit().is_none()); + cell.resume_owner_actions(); + assert!(cell.try_transport_action_permit().is_some()); + } + + #[tokio::test] + async fn fence_closes_admission_and_waits_for_live_permits() { + let cell = Arc::new(AuthorityCell::new()); + let owner = cell.reserve(AuthorityOrigin::Owner); + assert!(cell.install(owner)); + let permit = cell + .try_owner_action_permit() + .expect("owner action admitted before fence"); + + cell.suspend_owner_actions(); + assert!(cell.try_owner_action_permit().is_none()); + assert!(cell.try_runtime_action_permit(owner).is_none()); + + let waiter_cell = Arc::clone(&cell); + let waiter = tokio::spawn(async move { + waiter_cell.wait_for_actions_drained().await; + }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + drop(permit); + tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("permit drain timed out") + .expect("permit drain task panicked"); + cell.resume_owner_actions(); + assert!(cell.try_owner_action_permit().is_some()); + } + + #[tokio::test] + async fn fence_drains_active_delegated_runtime_permits() { + let cell = Arc::new(AuthorityCell::new()); + let guest = cell.reserve(AuthorityOrigin::Delegated { generation: 4 }); + assert!(cell.install(guest)); + let permit = cell + .try_runtime_action_permit(guest) + .expect("delegated runtime action admitted before fence"); + + cell.suspend_owner_actions(); + assert!(cell.try_runtime_action_permit(guest).is_none()); + + let waiter_cell = Arc::clone(&cell); + let waiter = tokio::spawn(async move { + waiter_cell.wait_for_actions_drained().await; + }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + drop(permit); + tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("delegated permit drain timed out") + .expect("delegated permit drain task panicked"); + cell.resume_owner_actions(); + assert!(cell.try_runtime_action_permit(guest).is_some()); + } + + #[test] + fn poison_is_recovered() { + let cell = Arc::new(AuthorityCell::new()); + let stamp = cell.reserve(AuthorityOrigin::Owner); + let panic_cell = Arc::clone(&cell); + let _ = std::thread::spawn(move || { + let _guard = panic_cell.state.lock().expect("initial lock"); + panic!("poison authority cell for recovery test"); + }) + .join(); + + assert!(cell.install(stamp)); + assert!(!cell.state.is_poisoned()); + assert!(cell.is_current(stamp)); + } +} diff --git a/crates/qconnect-app/src/delegation.rs b/crates/qconnect-app/src/delegation.rs index 54228a3db..1b120cb80 100644 --- a/crates/qconnect-app/src/delegation.rs +++ b/crates/qconnect-app/src/delegation.rs @@ -12,7 +12,7 @@ use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use async_trait::async_trait; -use futures_util::FutureExt; +use futures_util::{stream::FuturesUnordered, FutureExt, StreamExt}; use serde::{Deserialize, Serialize}; use tokio::sync::{watch, Mutex}; use tokio::task::JoinHandle; @@ -426,9 +426,18 @@ impl DelegationCoordinator { /// Declare that the already-prepared owner authority is ready. Intended for /// initial global-QConnect startup after owner authentication. pub async fn declare_owner_ready(&self) -> bool { + self.declare_owner_ready_if(|| true).await + } + + /// Declare owner readiness only while an external lifecycle intent remains + /// current. The predicate runs inside the coordinator commit gate and must + /// be synchronous and bounded. This closes disable-vs-connect races without + /// making teardown wait behind owner network preparation. + pub async fn declare_owner_ready_if(&self, is_current: impl FnOnce() -> bool) -> bool { let _gate = self.inner.commit_gate.lock().await; let mut state = self.inner.state.lock().await; - if state.lifecycle != CoordinatorLifecycle::Disabled + if !is_current() + || state.lifecycle != CoordinatorLifecycle::Disabled || state.pending.is_some() || !matches!(state.active, ActiveControl::Owner) { @@ -523,9 +532,7 @@ impl DelegationCoordinator { let mut tasks = recover_std_lock(&self.inner.tasks); tasks.drain().map(|(_, task)| task).collect::>() }; - for task in tasks { - await_or_abort(task, self.inner.config.cleanup_timeout).await; - } + await_all_or_abort(tasks, self.inner.config.cleanup_timeout).await; let shutdown_timed_out = tokio::time::timeout( self.inner.config.cleanup_timeout, self.inner.host.shutdown_authority(), @@ -1108,11 +1115,30 @@ fn recover_std_lock(mutex: &std::sync::Mutex) -> std::sync::MutexGuard<'_, .unwrap_or_else(|poisoned| poisoned.into_inner()) } -async fn await_or_abort(mut task: JoinHandle<()>, timeout: Duration) { - if tokio::time::timeout(timeout, &mut task).await.is_err() { - task.abort(); - let _ = task.await; +/// Give every candidate/restore transaction one shared graceful-cleanup +/// budget, then abort the remainder together. A hostile burst must not turn a +/// nominal two-second shutdown budget into two seconds per generation. +async fn await_all_or_abort(tasks: Vec>, timeout: Duration) { + let mut tasks = tasks.into_iter().collect::>(); + let timed_out = tokio::time::timeout(timeout, async { + while tasks.next().await.is_some() {} + }) + .await + .is_err(); + if !timed_out { + return; } + + for task in tasks.iter() { + task.abort(); + } + // Aborted Tokio tasks normally join on the next poll. Keep even this tail + // bounded: cancellation-safe host cleanup and authority stamps own the + // late-resource safety if a task refuses to yield promptly. + let _ = tokio::time::timeout(Duration::from_millis(250), async { + while tasks.next().await.is_some() {} + }) + .await; } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -1155,6 +1181,25 @@ mod tests { use super::*; + #[tokio::test(start_paused = true)] + async fn candidate_cleanup_uses_one_shared_shutdown_budget() { + let tasks = (0..3) + .map(|_| tokio::spawn(std::future::pending::<()>())) + .collect(); + let cleanup = tokio::spawn(await_all_or_abort(tasks, Duration::from_secs(2))); + + tokio::task::yield_now().await; + assert!(!cleanup.is_finished()); + tokio::time::advance(Duration::from_secs(2)).await; + tokio::task::yield_now().await; + + assert!( + cleanup.is_finished(), + "cleanup multiplied its timeout by the number of candidate tasks" + ); + cleanup.await.expect("cleanup task panicked"); + } + #[derive(Clone, Copy)] struct FakeCandidate { id: u64, @@ -1683,6 +1728,21 @@ mod tests { coordinator.shutdown().await; } + #[tokio::test] + async fn declare_owner_ready_revalidates_external_intent_inside_commit_gate() { + let host = Arc::new(FakeHost::default()); + let coordinator = DelegationCoordinator::disabled( + Arc::clone(&host), + DelegationCoordinatorConfig::default(), + ); + + assert!(!coordinator.declare_owner_ready_if(|| false).await); + assert_eq!(coordinator.snapshot().phase, DelegationPhase::Disabled); + assert!(coordinator.declare_owner_ready_if(|| true).await); + assert_eq!(coordinator.snapshot().phase, DelegationPhase::OwnerReady); + coordinator.shutdown().await; + } + #[tokio::test] async fn published_snapshot_revisions_are_strictly_monotonic() { let host = Arc::new(FakeHost::default()); diff --git a/crates/qconnect-app/src/error.rs b/crates/qconnect-app/src/error.rs index 1cff9fd2c..ff68e4128 100644 --- a/crates/qconnect-app/src/error.rs +++ b/crates/qconnect-app/src/error.rs @@ -3,6 +3,46 @@ use qconnect_protocol::ProtocolError; use qconnect_transport_ws::WsTransportError; use thiserror::Error; +/// Stable, payload-free diagnostics for owner-authority failures that cross +/// the QConnect renderer boundary. +/// +/// Owner API and playback errors can contain signed URLs, credential-bearing routing +/// material, or server-controlled response text. Frontend adapters must first +/// classify those errors into this enum and may only format this value. +#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)] +pub enum QconnectOwnerFailure { + #[error("owner service is unavailable")] + Unavailable, + #[error("owner network request failed")] + Network, + #[error("owner authentication failed")] + Authentication, + #[error("owner authorization failed")] + Authorization, + #[error("owner service is offline")] + Offline, + #[error("owner request was rate limited")] + RateLimited, + #[error("owner server request failed")] + Server, + #[error("owner service returned an invalid response")] + InvalidResponse, + #[error("owner track is unavailable")] + TrackUnavailable, + #[error("owner playback failed")] + Playback, + #[error("owner internal operation failed")] + Internal, +} + +impl QconnectOwnerFailure { + /// Collapse a legacy playback `String` without inspecting or retaining it. + /// The player layer predates typed errors and may have embedded a signed URL. + pub const fn from_opaque_playback_error(_error: &str) -> Self { + Self::Playback + } +} + #[derive(Debug, Error)] pub enum QconnectAppError { #[error(transparent)] diff --git a/crates/qconnect-app/src/lib.rs b/crates/qconnect-app/src/lib.rs index 4a6cc1626..a842d5102 100644 --- a/crates/qconnect-app/src/lib.rs +++ b/crates/qconnect-app/src/lib.rs @@ -3,6 +3,7 @@ //! Application adapter that composes qconnect core + protocol + transport. mod app; +mod authority; mod delegation; mod error; mod events; @@ -12,35 +13,26 @@ pub mod renderer; mod renderer_engine; mod reporting; pub mod session; -mod state; pub mod startup; +mod state; mod sync_state; pub use app::{ queue_payload_track_preview, QconnectApp, SessionApplyOutcome, SessionLoopHost, SessionStateTakeoverInput, }; -pub use startup::{compute_effective_startup, QconnectStartupMode}; +pub use authority::{ + AuthorityActionPermit, AuthorityCell, AuthorityOrigin, AuthorityStamp, + ExactOwnerAuthorityObservation, OwnerAuthorityObservation, OwnerAuthorityToken, + QconnectDisabledToken, QconnectEnableIntent, QconnectEnableToken, +}; pub use delegation::{ CommitRejected, CredentialOrigin, DelegationCancellation, DelegationCandidate, DelegationCoordinator, DelegationCoordinatorConfig, DelegationCoordinatorError, DelegationErrorCode, DelegationHost, DelegationPhase, DelegationSnapshot, RestoreReason, }; -pub use error::QconnectAppError; +pub use error::{QconnectAppError, QconnectOwnerFailure}; pub use events::{NoOpEventSink, QconnectAppEvent, QconnectEventSink}; -pub use session::{ - build_effective_renderer_snapshot, build_session_renderer_snapshot, compute_connection_state, - deferred_join_reason, - find_unique_renderer_id, is_local_renderer_active, is_peer_renderer_active, - normalize_active_renderer_id, quality_from_max_audio_quality, queue_item_snapshot_for_cursor, - qconnect_millis_from_secs, - refresh_local_renderer_id, renderer_allows_remote_volume, should_arm_renderer_watchdog, - should_reask_queue_state, ConnectionDecision, LocalIdentity, QconnectFileAudioQualitySnapshot, - QconnectLifecycleState, QconnectRendererInfo, QconnectSessionRendererState, - QconnectSessionState, RendererStatus, ServerActiveState, - JOIN_SESSION_REASON_CONTROLLER_REQUEST, JOIN_SESSION_REASON_RECONNECTION, - QCONNECT_RENDERER_LOST_TIMEOUT_MS, -}; pub use feature_flags::{ QBZ_QCONNECT_PANEL_SWITCH, QBZ_QCONNECT_QUEUE_MODEL, QBZ_QCONNECT_STRICT_DOMAIN_ISOLATION, QBZ_QCONNECT_TRANSPORT, @@ -58,6 +50,18 @@ pub use reporting::{ build_renderer_playback_report, qconnect_report_track_id, renderer_buffer_state, RendererPlaybackSnapshot, }; +pub use session::{ + build_effective_renderer_snapshot, build_session_renderer_snapshot, compute_connection_state, + deferred_join_reason, find_unique_renderer_id, is_local_renderer_active, + is_peer_renderer_active, normalize_active_renderer_id, qconnect_millis_from_secs, + quality_from_max_audio_quality, queue_item_snapshot_for_cursor, refresh_local_renderer_id, + renderer_allows_remote_volume, should_arm_renderer_watchdog, should_reask_queue_state, + ConnectionDecision, LocalIdentity, QconnectFileAudioQualitySnapshot, QconnectLifecycleState, + QconnectRendererInfo, QconnectSessionRendererState, QconnectSessionState, RendererStatus, + ServerActiveState, JOIN_SESSION_REASON_CONTROLLER_REQUEST, JOIN_SESSION_REASON_RECONNECTION, + QCONNECT_RENDERER_LOST_TIMEOUT_MS, +}; +pub use startup::{compute_effective_startup, QconnectStartupMode}; pub use state::QconnectRuntimeState; pub use sync_state::{ cache_renderer_snapshot, ensure_session_renderer_state, sync_session_renderer_active_flags, diff --git a/crates/qconnect-lan/src/mdns.rs b/crates/qconnect-lan/src/mdns.rs index 9060a3dd5..bbf0b644c 100644 --- a/crates/qconnect-lan/src/mdns.rs +++ b/crates/qconnect-lan/src/mdns.rs @@ -1,98 +1,516 @@ use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; use std::time::{Duration, Instant}; -use mdns_sd::{DaemonEvent, IfKind, ServiceDaemon, ServiceInfo, UnregisterStatus}; +use mdns_sd::{ + DaemonEvent, IfKind, Receiver, RecvTimeoutError, ServiceDaemon, ServiceInfo, UnregisterStatus, +}; use crate::server::{LanError, SERVICE_TYPE}; const ANNOUNCE_TIMEOUT: Duration = Duration::from_secs(2); const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); +const RECONCILE_INTERVAL: Duration = Duration::from_secs(30); +const MONITOR_WAKE_INTERVAL: Duration = Duration::from_millis(500); const GOODBYE_RESEND_DELAY: Duration = Duration::from_millis(150); const DAEMON_SHUTDOWN_RESERVE: Duration = Duration::from_millis(250); pub(crate) struct MdnsRegistration { + control: Arc, + monitor: Option>, + monitor_thread: Option>, + initial_addresses: Option>, +} + +#[derive(Clone)] +pub(crate) struct MdnsShutdownHandle { + control: Arc, +} + +#[derive(Debug, Clone)] +enum AddressSource { + Dynamic, + Fixed(Vec), +} + +struct MdnsControl { daemon: ServiceDaemon, fullname: String, + device_uuid: String, + port: u16, + sdk_version: String, + address_source: AddressSource, + state: Mutex, +} + +#[derive(Debug, Default)] +struct RegistrationState { + stopped: bool, + published_addresses: Option>, + pending_interface_changes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum ReconcileAction { + Unchanged, + Withdraw, + Publish(Vec), + Replace(Vec), +} + +#[derive(Debug, Clone, Copy)] +enum InterfaceChange { + Added(IpAddr), + Removed(IpAddr), } impl MdnsRegistration { - pub(crate) fn register( + /// Prepare discovery without publishing it. The server installs the + /// shutdown handle before its HTTP threads start, so a fatal thread exit + /// can always fence discovery even while the first announcement is in + /// flight. + pub(crate) fn prepare( device_uuid: &str, port: u16, sdk_version: &str, addresses: Option>, ) -> Result { - let mut addresses = match addresses { - Some(addresses) => addresses, - None => lan_ipv4_addresses()?, + let (address_source, initial_addresses) = match addresses { + Some(addresses) => { + let addresses = normalize_addresses(addresses); + (AddressSource::Fixed(addresses.clone()), addresses) + } + None => ( + AddressSource::Dynamic, + normalize_addresses(lan_ipv4_addresses()?), + ), }; - addresses.retain(|ip| ip.is_ipv4() && !ip.is_loopback() && !ip.is_unspecified()); - addresses.sort_unstable(); - addresses.dedup(); - if addresses.is_empty() { + if initial_addresses.is_empty() { return Err(LanError::NoLanAddresses); } - let info = service_info(device_uuid, port, sdk_version, &addresses)?; + let info = service_info(device_uuid, port, sdk_version, &initial_addresses)?; let fullname = info.get_fullname().to_string(); let daemon = ServiceDaemon::new()?; let monitor = daemon.monitor()?; - if daemon.register(info).is_err() { - shutdown_daemon(&daemon, Some(&fullname), SHUTDOWN_TIMEOUT); - return Err(LanError::Mdns); - } + let control = Arc::new(MdnsControl { + daemon, + fullname, + device_uuid: device_uuid.to_string(), + port, + sdk_version: sdk_version.to_string(), + address_source, + state: Mutex::new(RegistrationState::default()), + }); - let deadline = Instant::now() + ANNOUNCE_TIMEOUT; - let announced = loop { - let Some(remaining) = deadline.checked_duration_since(Instant::now()) else { - break false; - }; - match monitor.recv_timeout(remaining) { - Ok(DaemonEvent::Announce(announced, _)) - if announced.eq_ignore_ascii_case(&fullname) => - { - break true; - } - Ok(DaemonEvent::Error(_) | DaemonEvent::NameChange(_)) | Err(_) => break false, - Ok(_) => {} - } - }; - if !announced { - shutdown_daemon(&daemon, Some(&fullname), SHUTDOWN_TIMEOUT); - return Err(LanError::Mdns); - } - - Ok(Self { daemon, fullname }) + Ok(Self { + control, + monitor: Some(monitor), + monitor_thread: None, + initial_addresses: Some(initial_addresses), + }) } - pub(crate) fn shutdown(self) { - shutdown_daemon(&self.daemon, Some(&self.fullname), SHUTDOWN_TIMEOUT); + pub(crate) fn shutdown_handle(&self) -> MdnsShutdownHandle { + MdnsShutdownHandle { + control: Arc::clone(&self.control), + } + } + + /// Publish only after the HTTP acceptor and workers exist. The monitor + /// supervises daemon health and reconciles route-selected IPv4 changes. + pub(crate) fn publish(&mut self, on_fatal: F) -> Result<(), LanError> + where + F: Fn() + Send + Sync + 'static, + { + let monitor = self.monitor.take().ok_or(LanError::Mdns)?; + let initial_addresses = self.initial_addresses.take().ok_or(LanError::Mdns)?; + self.control + .reconcile(initial_addresses, &monitor, Some(ANNOUNCE_TIMEOUT))?; + if matches!(&self.control.address_source, AddressSource::Dynamic) { + // The initial acknowledgement may have shared the monitor queue + // with an interface event. Re-read route selection once before the + // long-lived monitor takes over; runtime publishes do not consume + // monitor events while waiting for an acknowledgement. + self.control + .reconcile(self.control.observed_addresses(), &monitor, None)?; + } + + let control = Arc::clone(&self.control); + let monitor_control = Arc::clone(&control); + let on_fatal = Arc::new(on_fatal); + let monitor_fatal = Arc::clone(&on_fatal); + let thread = std::thread::Builder::new() + .name("qconnect-lan-mdns".to_string()) + .spawn(move || { + let mut exit_guard = + MonitorExitGuard::new(Arc::clone(&monitor_control), Arc::clone(&monitor_fatal)); + if monitor_loop(&monitor_control, &monitor) { + return; + } + exit_guard.disarm(); + }) + .map_err(|_| LanError::Mdns); + + match thread { + Ok(thread) => { + self.monitor_thread = Some(thread); + Ok(()) + } + Err(error) => { + control.stop(); + Err(error) + } + } + } + + pub(crate) fn shutdown(&mut self) { + self.control.stop(); + if let Some(thread) = self.monitor_thread.take() { + let _ = thread.join(); + } } } -fn shutdown_daemon(daemon: &ServiceDaemon, fullname: Option<&str>, timeout: Duration) { - let deadline = Instant::now() + timeout; - if let Some(fullname) = fullname { - if let Ok(receiver) = daemon.unregister(fullname) { - if let Some(remaining) = deadline.checked_duration_since(Instant::now()) { - if matches!(receiver.recv_timeout(remaining), Ok(UnregisterStatus::OK)) { - // mdns-sd schedules a second goodbye 120 ms after the - // unregister acknowledgement. Keep the daemon alive long - // enough to send it, while reserving time for shutdown. - if let Some(remaining) = deadline.checked_duration_since(Instant::now()) { - let goodbye_budget = remaining.saturating_sub(DAEMON_SHUTDOWN_RESERVE); - std::thread::sleep(GOODBYE_RESEND_DELAY.min(goodbye_budget)); - } - } +impl Drop for MdnsRegistration { + fn drop(&mut self) { + self.shutdown(); + } +} + +impl MdnsShutdownHandle { + pub(crate) fn shutdown(&self) { + self.control.stop(); + } +} + +impl MdnsControl { + fn reconcile( + &self, + observed_addresses: Vec, + monitor: &Receiver, + announce_timeout: Option, + ) -> Result<(), LanError> { + let observed_addresses = normalize_addresses(observed_addresses); + let mut state = recover_lock(&self.state); + let action = reconcile_action( + state.stopped, + state.published_addresses.as_deref(), + observed_addresses, + ); + match action { + ReconcileAction::Unchanged => Ok(()), + ReconcileAction::Withdraw => { + self.withdraw_locked(&mut state, Instant::now() + SHUTDOWN_TIMEOUT) + } + ReconcileAction::Publish(addresses) => { + self.publish_locked(&mut state, addresses, monitor, announce_timeout) + } + ReconcileAction::Replace(addresses) => { + self.withdraw_locked(&mut state, Instant::now() + SHUTDOWN_TIMEOUT)?; + self.publish_locked(&mut state, addresses, monitor, announce_timeout) } } } - if let Ok(receiver) = daemon.shutdown() { - if let Some(remaining) = deadline.checked_duration_since(Instant::now()) { - let _ = receiver.recv_timeout(remaining); + + fn publish_locked( + &self, + state: &mut RegistrationState, + addresses: Vec, + monitor: &Receiver, + announce_timeout: Option, + ) -> Result<(), LanError> { + if state.stopped { + return Ok(()); + } + let info = service_info(&self.device_uuid, self.port, &self.sdk_version, &addresses)?; + self.daemon.register(info).map_err(|_| LanError::Mdns)?; + // Record the command immediately. If acknowledgement fails, stop() + // still knows that it must unregister before shutting the daemon down. + state.published_addresses = Some(addresses); + match announce_timeout { + Some(timeout) => { + state.pending_interface_changes.extend(wait_for_announce( + monitor, + &self.fullname, + timeout, + )?); + Ok(()) + } + None => Ok(()), } } + + fn withdraw_locked( + &self, + state: &mut RegistrationState, + deadline: Instant, + ) -> Result<(), LanError> { + if state.published_addresses.is_none() { + return Ok(()); + } + let receiver = self + .daemon + .unregister(&self.fullname) + .map_err(|_| LanError::Mdns)?; + let remaining = deadline + .checked_duration_since(Instant::now()) + .ok_or(LanError::Mdns)?; + match receiver.recv_timeout(remaining) { + Ok(UnregisterStatus::OK) | Ok(UnregisterStatus::NotFound) => { + state.published_addresses = None; + } + Err(_) => return Err(LanError::Mdns), + } + + // mdns-sd schedules a second goodbye after its acknowledgement. Keep + // the daemon alive for it, but preserve a bounded shutdown reserve. + if let Some(remaining) = deadline.checked_duration_since(Instant::now()) { + let goodbye_budget = remaining.saturating_sub(DAEMON_SHUTDOWN_RESERVE); + std::thread::sleep(GOODBYE_RESEND_DELAY.min(goodbye_budget)); + } + Ok(()) + } + + fn stop(&self) { + let deadline = Instant::now() + SHUTDOWN_TIMEOUT; + let mut state = recover_lock(&self.state); + if state.stopped { + return; + } + // This is the linearization point. Reconciliation uses the same lock + // and checks this flag before every register command, so publication + // cannot restart after shutdown has won the lock. + state.stopped = true; + let _ = self.withdraw_locked(&mut state, deadline); + if let Ok(receiver) = self.daemon.shutdown() { + if let Some(remaining) = deadline.checked_duration_since(Instant::now()) { + let _ = receiver.recv_timeout(remaining); + } + } + } + + fn is_stopped(&self) -> bool { + recover_lock(&self.state).stopped + } + + fn observed_addresses(&self) -> Vec { + match &self.address_source { + AddressSource::Dynamic => lan_ipv4_addresses().unwrap_or_default(), + AddressSource::Fixed(_) => recover_lock(&self.state) + .published_addresses + .clone() + .unwrap_or_default(), + } + } + + fn addresses_after_event(&self, added: bool, address: IpAddr) -> Option> { + match &self.address_source { + AddressSource::Dynamic => Some(self.observed_addresses()), + AddressSource::Fixed(configured) => { + let state = recover_lock(&self.state); + fixed_addresses_after_event( + configured, + state.published_addresses.as_deref(), + added, + address, + ) + } + } + } + + fn take_pending_interface_change(&self) -> Option { + let mut state = recover_lock(&self.state); + (!state.pending_interface_changes.is_empty()) + .then(|| state.pending_interface_changes.remove(0)) + } +} + +struct MonitorExitGuard +where + F: Fn() + Send + Sync + 'static, +{ + control: Arc, + on_fatal: Arc, + armed: bool, +} + +impl MonitorExitGuard +where + F: Fn() + Send + Sync + 'static, +{ + fn new(control: Arc, on_fatal: Arc) -> Self { + Self { + control, + on_fatal, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for MonitorExitGuard +where + F: Fn() + Send + Sync + 'static, +{ + fn drop(&mut self) { + if self.armed { + self.control.stop(); + (self.on_fatal)(); + } + } +} + +/// Returns true for an unexpected exit. The receiver blocks between actual +/// interface events. Its bounded wake only observes shutdown; route discovery +/// remains a 30-second missed-event repair pass, not a fast polling loop. +fn monitor_loop(control: &MdnsControl, monitor: &Receiver) -> bool { + let mut next_reconcile = Instant::now() + RECONCILE_INTERVAL; + loop { + if control.is_stopped() { + return false; + } + if let Some(change) = control.take_pending_interface_change() { + let result = match change { + InterfaceChange::Added(address) => { + reconcile_interface_event(control, monitor, true, address) + } + InterfaceChange::Removed(address) => { + reconcile_interface_event(control, monitor, false, address) + } + }; + if result.is_err() { + return true; + } + next_reconcile = Instant::now() + RECONCILE_INTERVAL; + continue; + } + let wait = next_reconcile + .saturating_duration_since(Instant::now()) + .min(MONITOR_WAKE_INTERVAL); + match monitor.recv_timeout(wait) { + Ok(DaemonEvent::IpAdd(address)) => { + if reconcile_interface_event(control, monitor, true, address).is_err() { + return true; + } + next_reconcile = Instant::now() + RECONCILE_INTERVAL; + } + Ok(DaemonEvent::IpDel(address)) => { + if reconcile_interface_event(control, monitor, false, address).is_err() { + return true; + } + next_reconcile = Instant::now() + RECONCILE_INTERVAL; + } + Err(RecvTimeoutError::Timeout) => { + if Instant::now() >= next_reconcile { + if matches!(&control.address_source, AddressSource::Dynamic) + && control + .reconcile(control.observed_addresses(), monitor, None) + .is_err() + { + return true; + } + next_reconcile = Instant::now() + RECONCILE_INTERVAL; + } + } + Ok(DaemonEvent::Error(_) | DaemonEvent::NameChange(_)) => return true, + Err(RecvTimeoutError::Disconnected) => return !control.is_stopped(), + Ok(_) => {} + } + } +} + +fn reconcile_interface_event( + control: &MdnsControl, + monitor: &Receiver, + added: bool, + address: IpAddr, +) -> Result<(), LanError> { + match control.addresses_after_event(added, address) { + Some(addresses) => control.reconcile(addresses, monitor, None), + None => Ok(()), + } +} + +fn wait_for_announce( + monitor: &Receiver, + fullname: &str, + timeout: Duration, +) -> Result, LanError> { + let deadline = Instant::now() + timeout; + let mut interface_changes = Vec::new(); + loop { + let remaining = deadline + .checked_duration_since(Instant::now()) + .ok_or(LanError::Mdns)?; + match monitor.recv_timeout(remaining) { + Ok(DaemonEvent::Announce(announced, _)) if announced.eq_ignore_ascii_case(fullname) => { + return Ok(interface_changes) + } + Ok(DaemonEvent::IpAdd(address)) => { + interface_changes.push(InterfaceChange::Added(address)) + } + Ok(DaemonEvent::IpDel(address)) => { + interface_changes.push(InterfaceChange::Removed(address)) + } + Ok(DaemonEvent::Error(_) | DaemonEvent::NameChange(_)) | Err(_) => { + return Err(LanError::Mdns) + } + Ok(_) => {} + } + } +} + +fn reconcile_action( + stopped: bool, + published_addresses: Option<&[IpAddr]>, + observed_addresses: Vec, +) -> ReconcileAction { + if stopped { + return ReconcileAction::Unchanged; + } + match (published_addresses, observed_addresses.is_empty()) { + (None, true) => ReconcileAction::Unchanged, + (Some(_), true) => ReconcileAction::Withdraw, + (None, false) => ReconcileAction::Publish(observed_addresses), + (Some(published), false) if published == observed_addresses.as_slice() => { + ReconcileAction::Unchanged + } + (Some(_), false) => ReconcileAction::Replace(observed_addresses), + } +} + +fn fixed_addresses_after_event( + configured: &[IpAddr], + published: Option<&[IpAddr]>, + added: bool, + address: IpAddr, +) -> Option> { + if !configured.contains(&address) { + return None; + } + let mut observed = published.unwrap_or_default().to_vec(); + if added { + observed.push(address); + } else { + observed.retain(|candidate| *candidate != address); + } + Some(normalize_addresses(observed)) +} + +fn normalize_addresses(addresses: Vec) -> Vec { + let mut addresses = addresses + .into_iter() + .filter(|ip| ip.is_ipv4() && !ip.is_loopback() && !ip.is_unspecified()) + .collect::>(); + addresses.sort_unstable(); + addresses.dedup(); + addresses } fn service_info( @@ -155,10 +573,20 @@ fn short_id(device_uuid: &str) -> String { } } +fn recover_lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + #[cfg(test)] mod tests { use super::*; + fn ip(value: &str) -> IpAddr { + value.parse().unwrap() + } + #[test] fn instance_short_id_is_dns_safe_and_stable() { assert_eq!(short_id("550e8400-e29b-41d4-a716-446655440000"), "550e8400"); @@ -168,7 +596,7 @@ mod tests { #[test] fn service_info_has_official_type_and_required_txt_union() { let uuid = "550e8400-e29b-41d4-a716-446655440000"; - let info = service_info(uuid, 43210, "0.9.5", &["192.0.2.10".parse().unwrap()]).unwrap(); + let info = service_info(uuid, 43210, "0.9.5", &[ip("192.0.2.10")]).unwrap(); assert_eq!(info.get_type(), SERVICE_TYPE); assert_eq!(info.get_port(), 43210); @@ -178,9 +606,75 @@ mod tests { assert!(!info.is_addr_auto()); assert_eq!( info.get_addresses(), - &["192.0.2.10".parse::().unwrap()] - .into_iter() - .collect() + &[ip("192.0.2.10")].into_iter().collect() + ); + } + + #[test] + fn address_normalization_is_ipv4_only_sorted_and_deduplicated() { + assert_eq!( + normalize_addresses(vec![ + ip("192.0.2.20"), + ip("::1"), + ip("127.0.0.1"), + ip("0.0.0.0"), + ip("192.0.2.10"), + ip("192.0.2.20"), + ]), + vec![ip("192.0.2.10"), ip("192.0.2.20")] + ); + } + + #[test] + fn address_reconciliation_withdraws_recovers_and_replaces_in_order() { + let first = vec![ip("192.0.2.10")]; + let second = vec![ip("192.0.2.20")]; + + assert_eq!( + reconcile_action(false, Some(&first), vec![]), + ReconcileAction::Withdraw + ); + assert_eq!( + reconcile_action(false, None, first.clone()), + ReconcileAction::Publish(first.clone()) + ); + assert_eq!( + reconcile_action(false, Some(&first), second.clone()), + ReconcileAction::Replace(second) + ); + assert_eq!( + reconcile_action(false, Some(&first), first.clone()), + ReconcileAction::Unchanged + ); + } + + #[test] + fn stopped_reconciliation_never_republishes() { + assert_eq!( + reconcile_action(true, None, vec![ip("192.0.2.10")]), + ReconcileAction::Unchanged + ); + assert_eq!( + reconcile_action(true, Some(&[ip("192.0.2.10")]), vec![ip("192.0.2.20")]), + ReconcileAction::Unchanged + ); + } + + #[test] + fn fixed_address_events_withdraw_and_restore_only_configured_ipv4() { + let configured = vec![ip("192.0.2.10")]; + + assert_eq!( + fixed_addresses_after_event(&configured, Some(&configured), false, ip("192.0.2.10")), + Some(vec![]) + ); + assert_eq!( + fixed_addresses_after_event(&configured, None, true, ip("192.0.2.10")), + Some(configured.clone()) + ); + assert_eq!( + fixed_addresses_after_event(&configured, Some(&configured), false, ip("192.0.2.99")), + None ); } } diff --git a/crates/qconnect-lan/src/model.rs b/crates/qconnect-lan/src/model.rs index 922bd3de8..c0fcda4f8 100644 --- a/crates/qconnect-lan/src/model.rs +++ b/crates/qconnect-lan/src/model.rs @@ -93,6 +93,10 @@ impl LanJwtToken { Self { endpoint, exp, jwt } } + fn empty() -> Self { + Self::new(String::new(), 0, String::new()) + } + pub fn endpoint(&self) -> &str { &self.endpoint } @@ -104,6 +108,16 @@ impl LanJwtToken { pub fn jwt(&self) -> &str { &self.jwt } + + /// Consume the token without cloning either sensitive allocation. The + /// returned endpoint and JWT become the caller's zeroization responsibility. + pub fn into_parts(mut self) -> (String, i64, String) { + ( + std::mem::take(&mut self.endpoint), + self.exp, + std::mem::take(&mut self.jwt), + ) + } } impl Drop for LanJwtToken { @@ -152,6 +166,16 @@ impl HandoffCandidate { pub const fn become_active(&self) -> bool { self.become_active } + + /// Consume the complete handoff without duplicating credential buffers. + pub fn into_parts(mut self) -> (String, LanJwtToken, LanJwtToken, bool) { + ( + std::mem::take(&mut self.session_id), + std::mem::replace(&mut self.jwt_api, LanJwtToken::empty()), + std::mem::replace(&mut self.jwt_qconnect, LanJwtToken::empty()), + self.become_active, + ) + } } impl Drop for HandoffCandidate { diff --git a/crates/qconnect-lan/src/server.rs b/crates/qconnect-lan/src/server.rs index 06e91576f..92bf9fea7 100644 --- a/crates/qconnect-lan/src/server.rs +++ b/crates/qconnect-lan/src/server.rs @@ -11,7 +11,7 @@ use socket2::{Domain, Protocol, Socket, Type}; use zeroize::Zeroizing; use crate::admission::{admission_channel, AdmissionInbox, AdmissionSender, SubmitError}; -use crate::mdns::MdnsRegistration; +use crate::mdns::{MdnsRegistration, MdnsShutdownHandle}; use crate::validation::{parse_and_validate, EndpointPolicy, ValidationError, MAX_BODY_BYTES}; use crate::LanProjection; @@ -115,10 +115,7 @@ impl LanServiceConfig { pub struct LanService { port: u16, - admission: AdmissionSender, - accepting: Arc, - accept_wake_addr: SocketAddr, - active_connections: Arc, + lifecycle: Arc, accept_thread: Option>, worker_threads: Vec>, mdns: Option, @@ -171,6 +168,82 @@ impl ActiveConnections { } } +struct ServiceLifecycle { + running: AtomicBool, + admission: AdmissionSender, + accept_wake_addr: SocketAddr, + active_connections: Arc, + discovery: Mutex>, +} + +impl ServiceLifecycle { + fn new( + admission: AdmissionSender, + accept_wake_addr: SocketAddr, + active_connections: Arc, + ) -> Self { + Self { + running: AtomicBool::new(true), + admission, + accept_wake_addr, + active_connections, + discovery: Mutex::new(None), + } + } + + fn is_running(&self) -> bool { + self.running.load(Ordering::Acquire) + } + + fn install_discovery(&self, discovery: MdnsShutdownHandle) { + let stop_immediately = { + let mut slot = recover_lock(&self.discovery); + if self.is_running() { + *slot = Some(discovery.clone()); + false + } else { + true + } + }; + if stop_immediately { + discovery.shutdown(); + } + } + + /// One fail-stop edge owns teardown. Every later caller observes the + /// fenced state and cannot revive admission or discovery. + fn stop(&self) { + if !self.running.swap(false, Ordering::AcqRel) { + return; + } + self.admission.close(); + // Withdraw discovery before interrupting the endpoint. This handle is + // also called from unexpected thread-exit guards, not only Drop. + let discovery = recover_lock(&self.discovery).clone(); + if let Some(discovery) = discovery { + discovery.shutdown(); + } + self.active_connections.shutdown_all(); + wake_acceptor(self.accept_wake_addr); + } +} + +struct UnexpectedThreadExit { + lifecycle: Arc, +} + +impl UnexpectedThreadExit { + fn new(lifecycle: Arc) -> Self { + Self { lifecycle } + } +} + +impl Drop for UnexpectedThreadExit { + fn drop(&mut self) { + self.lifecycle.stop(); + } +} + struct ConnectionJob { stream: TcpStream, active_connections: Arc, @@ -195,145 +268,172 @@ impl LanService { let accept_wake_addr = accept_wake_addr(listener_addr); let (admission, inbox) = admission_channel(); - let accepting = Arc::new(AtomicBool::new(true)); let rate_limiter = Arc::new(Mutex::new(PostRateLimiter::default())); let (request_tx, request_rx) = mpsc::sync_channel::(HTTP_WORKERS); let request_rx = Arc::new(Mutex::new(request_rx)); let in_flight = Arc::new(AtomicUsize::new(0)); let active_connections = Arc::new(ActiveConnections::default()); + let lifecycle = Arc::new(ServiceLifecycle::new( + admission.clone(), + accept_wake_addr, + Arc::clone(&active_connections), + )); + + // Discovery is prepared and its fail-stop handle installed before any + // supervised HTTP thread starts. Publication itself remains last, once + // the endpoint can actually serve official-client requests. + let mut mdns = if config.advertise_mdns { + let advertised_addresses = config.advertised_addresses.clone().or_else(|| { + (!config.bind_addr.ip().is_unspecified()).then(|| vec![config.bind_addr.ip()]) + }); + let registration = MdnsRegistration::prepare( + &config.device_uuid, + port, + &config.sdk_version, + advertised_addresses, + )?; + lifecycle.install_discovery(registration.shutdown_handle()); + Some(registration) + } else { + None + }; let mut worker_threads = Vec::with_capacity(HTTP_WORKERS); for _ in 0..HTTP_WORKERS { let worker_rx = Arc::clone(&request_rx); - let worker_accepting = Arc::clone(&accepting); + let worker_lifecycle = Arc::clone(&lifecycle); let worker_admission = admission.clone(); let worker_projection = config.projection.clone(); let worker_policy = config.endpoint_policy.clone(); let worker_limiter = Arc::clone(&rate_limiter); let worker_observer = config.request_observer.clone(); let read_timeout = config.read_timeout; - worker_threads.push(std::thread::spawn(move || loop { - let job = { - let receiver = recover_lock(&worker_rx); - receiver.recv() - }; - let mut job = match job { - Ok(job) => job, - Err(_) => break, - }; - if worker_accepting.load(Ordering::Acquire) { - handle_connection( - &mut job.stream, - read_timeout, - &worker_projection, - &worker_policy, - &worker_admission, - &worker_limiter, - worker_observer.as_ref(), - ); - } else { - let _ = write_response(&mut job.stream, &error_response(503, "unavailable")); + worker_threads.push(std::thread::spawn(move || { + let _exit_guard = UnexpectedThreadExit::new(Arc::clone(&worker_lifecycle)); + loop { + let job = { + let receiver = recover_lock(&worker_rx); + receiver.recv() + }; + let mut job = match job { + Ok(job) => job, + Err(_) => break, + }; + if worker_lifecycle.is_running() { + handle_connection( + &mut job.stream, + read_timeout, + &worker_projection, + &worker_policy, + &worker_admission, + &worker_limiter, + worker_observer.as_ref(), + ); + } else { + let _ = + write_response(&mut job.stream, &error_response(503, "unavailable")); + } } })); } - let thread_accepting = Arc::clone(&accepting); + let accept_lifecycle = Arc::clone(&lifecycle); let accept_in_flight = Arc::clone(&in_flight); let accept_active_connections = Arc::clone(&active_connections); - let accept_admission = admission.clone(); let accept_observer = config.request_observer.clone(); // The listener deliberately has no SO_RCVTIMEO: that option also // times out accept() on Linux. Shutdown wakes this blocking accept - // with one bounded local connection after clearing `accepting`. - let accept_thread = std::thread::spawn(move || loop { - match listener.accept() { - Ok((mut stream, _)) => { - if !thread_accepting.load(Ordering::Acquire) { - break; - } - let Some(permit) = InFlightPermit::try_acquire(Arc::clone(&accept_in_flight)) - else { - observe_status(accept_observer.as_ref(), 429); - reject_connection(&mut stream, &error_response(429, "busy")); - continue; - }; - let shutdown_stream = match stream.try_clone() { - Ok(stream) => stream, - Err(_) => { - drop(permit); - observe_status(accept_observer.as_ref(), 503); - reject_connection(&mut stream, &error_response(503, "unavailable")); - continue; - } - }; - let active_id = accept_active_connections.insert(shutdown_stream); - let job = ConnectionJob { - stream, - active_connections: Arc::clone(&accept_active_connections), - active_id, - _permit: permit, - }; - match request_tx.try_send(job) { - Ok(()) => {} - Err(TrySendError::Full(mut job)) => { - observe_status(accept_observer.as_ref(), 429); - reject_connection(&mut job.stream, &error_response(429, "busy")); - } - Err(TrySendError::Disconnected(mut job)) => { - observe_status(accept_observer.as_ref(), 503); - reject_connection(&mut job.stream, &error_response(503, "unavailable")); - thread_accepting.store(false, Ordering::Release); - accept_admission.close(); - accept_active_connections.shutdown_all(); + // with one bounded local connection after fencing the lifecycle. + let accept_thread = std::thread::spawn(move || { + let _exit_guard = UnexpectedThreadExit::new(Arc::clone(&accept_lifecycle)); + loop { + match listener.accept() { + Ok((mut stream, _)) => { + if !accept_lifecycle.is_running() { break; } + let Some(permit) = + InFlightPermit::try_acquire(Arc::clone(&accept_in_flight)) + else { + observe_status(accept_observer.as_ref(), 429); + reject_connection(&mut stream, &error_response(429, "busy")); + continue; + }; + let shutdown_stream = match stream.try_clone() { + Ok(stream) => stream, + Err(_) => { + drop(permit); + observe_status(accept_observer.as_ref(), 503); + reject_connection(&mut stream, &error_response(503, "unavailable")); + continue; + } + }; + let active_id = accept_active_connections.insert(shutdown_stream); + let job = ConnectionJob { + stream, + active_connections: Arc::clone(&accept_active_connections), + active_id, + _permit: permit, + }; + match request_tx.try_send(job) { + Ok(()) => {} + Err(TrySendError::Full(mut job)) => { + observe_status(accept_observer.as_ref(), 429); + reject_connection(&mut job.stream, &error_response(429, "busy")); + } + Err(TrySendError::Disconnected(mut job)) => { + observe_status(accept_observer.as_ref(), 503); + reject_connection( + &mut job.stream, + &error_response(503, "unavailable"), + ); + accept_lifecycle.stop(); + break; + } + } + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(_) if !accept_lifecycle.is_running() => break, + Err(_) => { + accept_lifecycle.stop(); + break; } - } - Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, - Err(_) if !thread_accepting.load(Ordering::Acquire) => break, - Err(_) => { - thread_accepting.store(false, Ordering::Release); - accept_admission.close(); - accept_active_connections.shutdown_all(); - break; } } }); - let mdns = if config.advertise_mdns { - let advertised_addresses = config.advertised_addresses.or_else(|| { - (!config.bind_addr.ip().is_unspecified()).then(|| vec![config.bind_addr.ip()]) - }); - match MdnsRegistration::register( - &config.device_uuid, - port, - &config.sdk_version, - advertised_addresses, - ) { - Ok(mdns) => Some(mdns), - Err(error) => { - admission.close(); - accepting.store(false, Ordering::Release); - active_connections.shutdown_all(); - wake_acceptor(accept_wake_addr); - let _ = accept_thread.join(); - for worker in worker_threads { - let _ = worker.join(); - } - return Err(error); + if let Some(registration) = mdns.as_mut() { + let weak_lifecycle = Arc::downgrade(&lifecycle); + if let Err(error) = registration.publish(move || { + if let Some(lifecycle) = weak_lifecycle.upgrade() { + lifecycle.stop(); } + }) { + lifecycle.stop(); + registration.shutdown(); + let _ = accept_thread.join(); + for worker in worker_threads { + let _ = worker.join(); + } + return Err(error); } - } else { - None - }; + } + if !lifecycle.is_running() { + lifecycle.stop(); + if let Some(registration) = mdns.as_mut() { + registration.shutdown(); + } + let _ = accept_thread.join(); + for worker in worker_threads { + let _ = worker.join(); + } + return Err(LanError::Mdns); + } Ok(( Self { port, - admission, - accepting, - accept_wake_addr, - active_connections, + lifecycle, accept_thread: Some(accept_thread), worker_threads, mdns, @@ -346,19 +446,17 @@ impl LanService { self.port } + /// False after explicit shutdown or any supervised HTTP/mDNS fatal exit. + pub fn is_running(&self) -> bool { + self.lifecycle.is_running() + } + pub fn shutdown(&mut self) { - self.admission.close(); - self.accepting.store(false, Ordering::Release); - // Stop advertising before making the HTTP endpoint unreachable. A - // strict official client may cache a discovered renderer for minutes, - // so publishing a dead port even briefly is worse than keeping the - // already-closed admission socket alive during the bounded goodbye. - if let Some(mdns) = self.mdns.take() { + self.lifecycle.stop(); + if let Some(mut mdns) = self.mdns.take() { mdns.shutdown(); } - self.active_connections.shutdown_all(); if let Some(thread) = self.accept_thread.take() { - wake_acceptor(self.accept_wake_addr); let _ = thread.join(); } for thread in self.worker_threads.drain(..) { @@ -932,4 +1030,23 @@ mod tests { *recover_lock(&value) = 1; assert_eq!(*recover_lock(&value), 1); } + + #[test] + fn unexpected_thread_exit_fail_stops_admission_once() { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let wake_addr = listener.local_addr().unwrap(); + let (admission, inbox) = admission_channel(); + let lifecycle = Arc::new(ServiceLifecycle::new( + admission, + wake_addr, + Arc::new(ActiveConnections::default()), + )); + + drop(UnexpectedThreadExit::new(Arc::clone(&lifecycle))); + lifecycle.stop(); + + assert!(!lifecycle.is_running()); + assert!(inbox.is_closed()); + drop(listener); + } } diff --git a/crates/qconnect-lan/src/validation.rs b/crates/qconnect-lan/src/validation.rs index a334d2103..cd8afda51 100644 --- a/crates/qconnect-lan/src/validation.rs +++ b/crates/qconnect-lan/src/validation.rs @@ -109,6 +109,7 @@ fn validate_token( let url = Url::parse(&raw.endpoint).map_err(|_| ValidationError::EndpointInvalid)?; if url.scheme() != required_scheme + || url.port_or_known_default() != Some(443) || !url.username().is_empty() || url.password().is_some() || url.fragment().is_some() @@ -139,6 +140,7 @@ fn validate_token( fn endpoint_host(endpoint: &str, required_scheme: &str) -> Result { let url = Url::parse(endpoint).map_err(|_| ValidationError::EndpointInvalid)?; if url.scheme() != required_scheme + || url.port_or_known_default() != Some(443) || !url.username().is_empty() || url.password().is_some() || url.fragment().is_some() @@ -195,6 +197,11 @@ mod tests { assert_eq!(candidate.session_id(), "controller-session"); assert_eq!(candidate.api_token().jwt(), "api-secret"); assert_eq!(candidate.qconnect_token().jwt(), "qws-secret"); + let (session_id, api, qconnect, become_active) = candidate.into_parts(); + assert_eq!(session_id, "controller-session"); + assert_eq!(api.into_parts().2, "api-secret"); + assert_eq!(qconnect.into_parts().2, "qws-secret"); + assert!(become_active); } #[test] @@ -233,6 +240,7 @@ mod tests { fn rejects_wrong_schemes_userinfo_and_inactive_handoff() { for api in [ "http://api.qobuz.test", + "https://api.qobuz.test:8443", "https://user@api.qobuz.test", "https://api.qobuz.test/#fragment", ] { @@ -241,6 +249,15 @@ mod tests { ValidationError::EndpointInvalid ); } + assert_eq!( + validation_error(&body( + "https://api.qobuz.test", + "wss://qws.qobuz.test:8443", + NOW + 600, + true, + )), + ValidationError::EndpointInvalid + ); assert_eq!( validation_error(&body( "https://api.qobuz.test", diff --git a/crates/qconnect-lan/tests/official_wire.rs b/crates/qconnect-lan/tests/official_wire.rs index 597140c0e..296f312e5 100644 --- a/crates/qconnect-lan/tests/official_wire.rs +++ b/crates/qconnect-lan/tests/official_wire.rs @@ -186,6 +186,7 @@ fn handoff(session_id: &str, become_active: bool) -> serde_json::Value { async fn serves_official_get_union_and_admits_exact_post_shape() { let (mut service, inbox, base, _) = start_service(); let client = reqwest::Client::new(); + assert!(service.is_running()); let display = client .get(format!("{base}/get-display-info")) @@ -246,6 +247,7 @@ async fn serves_official_get_union_and_admits_exact_post_shape() { service.shutdown(); service.shutdown(); + assert!(!service.is_running()); assert!(inbox.is_closed()); } diff --git a/crates/qconnect-transport-ws/src/transport.rs b/crates/qconnect-transport-ws/src/transport.rs index 81d9af5c9..aca41b576 100644 --- a/crates/qconnect-transport-ws/src/transport.rs +++ b/crates/qconnect-transport-ws/src/transport.rs @@ -32,6 +32,7 @@ const MSG_TYPE_SUBSCRIBE: u8 = 2; const MSG_TYPE_PAYLOAD: u8 = 6; const MSG_TYPE_ERROR: u8 = 9; const MSG_TYPE_DISCONNECT: u8 = 10; +const SHUTDOWN_CLOSE_TIMEOUT: Duration = Duration::from_millis(500); #[derive(Debug, Clone, Serialize, Deserialize)] pub enum TransportEvent { @@ -78,14 +79,10 @@ pub enum TransportEvent { /// Decoded `MSG_TYPE_ERROR` (cloud_type=9) frame from Qobuz cloud, per the /// `qws.proto` `ErrorMessage` definition. Emitted whenever the qws frontend /// rejects a session (e.g. zombie session, conflicting device, expired - /// JWT). Carries the cloud-side `code` and human-readable `descr` so - /// downstream consumers can reason about *why* the cloud is rejecting us - /// instead of only seeing a payload byte count (issue #358). - CloudError { - msg_id: u32, - code: u32, - descr: String, - }, + /// JWT). Only numeric protocol metadata crosses this boundary: `descr` is + /// server-controlled and may reflect credentials, endpoints, account data, + /// or session identifiers, so it is zeroized at decode time. + CloudError { msg_id: u32, code: u32 }, InboundQueueServerEvent(QueueServerEvent), InboundRendererServerCommand(RendererServerCommand), InboundReceived(InboundEnvelope), @@ -362,12 +359,8 @@ async fn run_native_transport_loop( let (mut ws, _) = match connect_result { Ok(Ok((ws, response))) => (ws, response), - Ok(Err(err)) => { - let reason = redact_sensitive( - format!("connect_error:{err}"), - config.jwt_qws.as_deref(), - Some(&config.endpoint_url), - ); + Ok(Err(_)) => { + let reason = "connect_error".to_string(); match handle_reconnect_delay( &events_tx, &mut shutdown_rx, @@ -443,16 +436,15 @@ async fn run_native_transport_loop( emit(&events_tx, TransportEvent::Connected); if let Some(jwt_qws) = config.jwt_qws.as_ref() { - if let Err(err) = send_authenticate(&mut ws, &mut msg_id, jwt_qws).await { + if send_authenticate(&mut ws, &mut msg_id, jwt_qws) + .await + .is_err() + { emit( &events_tx, TransportEvent::TransportError { stage: "authenticate".to_string(), - message: redact_sensitive( - err.to_string(), - config.jwt_qws.as_deref(), - Some(&config.endpoint_url), - ), + message: "authenticate_failed".to_string(), }, ); let _ = ws.close(None).await; @@ -540,23 +532,20 @@ async fn run_native_transport_loop( } if config.auto_subscribe { - if let Err(err) = send_subscribe( + if send_subscribe( &mut ws, &mut msg_id, config.qcloud_proto, &config.subscribe_channels, ) .await + .is_err() { emit( &events_tx, TransportEvent::TransportError { stage: "subscribe".to_string(), - message: redact_sensitive( - err.to_string(), - config.jwt_qws.as_deref(), - Some(&config.endpoint_url), - ), + message: "subscribe_failed".to_string(), }, ); let _ = ws.close(None).await; @@ -618,20 +607,24 @@ async fn run_native_transport_loop( tokio::select! { _ = shutdown_rx.changed() => { if *shutdown_rx.borrow() { - let _ = ws.close(None).await; + let _ = tokio::time::timeout( + SHUTDOWN_CLOSE_TIMEOUT, + ws.close(None), + ) + .await; break "shutdown".to_string(); } } maybe_envelope = outbound_rx.recv() => { match maybe_envelope { Some(envelope) => { - if let Err(err) = send_outbound_payload( + if send_outbound_payload( &mut ws, &mut msg_id, config.qcloud_proto, &envelope, - ).await { - break format!("send_error:{err}"); + ).await.is_err() { + break "send_error".to_string(); } emit( &events_tx, @@ -658,8 +651,8 @@ async fn run_native_transport_loop( ) { break "keepalive_timeout".to_string(); } - if let Err(err) = ws.send(WsMessage::Ping(Vec::new().into())).await { - break format!("keepalive_ping_error:{err}"); + if ws.send(WsMessage::Ping(Vec::new().into())).await.is_err() { + break "keepalive_ping_error".to_string(); } outstanding_pings = outstanding_pings.saturating_add(1); emit(&events_tx, TransportEvent::KeepalivePingSent); @@ -708,8 +701,8 @@ async fn run_native_transport_loop( break "remote_close".to_string(); } Some(Ok(_)) => {} - Some(Err(err)) => { - break format!("ws_read_error:{err}"); + Some(Err(_)) => { + break "ws_read_error".to_string(); } None => { break "ws_stream_end".to_string(); @@ -979,25 +972,13 @@ fn handle_incoming_binary( Ok(error) => { let msg_id = error.msg_id.unwrap_or(0); let code = error.code.unwrap_or(0); - let descr = redact_sensitive( - error.descr.unwrap_or_default(), - jwt_qws, - sensitive_endpoint, - ); + let mut descr = error.descr.unwrap_or_default(); + descr.zeroize(); log::warn!( - "[QConnect/Transport] Cloud error frame: msg_id={} code={} descr={:?}", - msg_id, - code, - descr - ); - emit( - events_tx, - TransportEvent::CloudError { - msg_id, - code, - descr, - }, + "[QConnect/Transport] Cloud error frame: msg_id={} code={}", + msg_id, code ); + emit(events_tx, TransportEvent::CloudError { msg_id, code }); } Err(err) => { log::warn!( @@ -1320,7 +1301,7 @@ mod tests { } /// Issue #358: a qws-level `MSG_TYPE_ERROR` frame must surface as a - /// `CloudError` event carrying the decoded code + descr, NOT as an opaque + /// `CloudError` event carrying the decoded numeric code, NOT as an opaque /// `bytes=N` `TransportError`. Without this, the upper layer can't /// distinguish a zombie session from any other transport hiccup and the /// reconnect loop spins blindly. @@ -1347,21 +1328,16 @@ mod tests { assert!(matches!(first, TransportEvent::InboundFrameDecoded { .. })); match events_rx.recv().await.expect("second event") { - TransportEvent::CloudError { - msg_id, - code, - descr, - } => { + TransportEvent::CloudError { msg_id, code } => { assert_eq!(msg_id, 42); assert_eq!(code, 403); - assert_eq!(descr, "zombie_session"); } other => panic!("expected CloudError, got {other:?}"), } } #[tokio::test] - async fn cloud_error_redacts_reflected_jwt() { + async fn cloud_error_discards_server_description() { const JWT: &str = "header.payload.secret-signature"; let (events_tx, mut events_rx) = broadcast::channel::(8); @@ -1380,12 +1356,13 @@ mod tests { let event = events_rx.recv().await.expect("cloud error event"); let debug = format!("{event:?}"); assert!(!debug.contains(JWT)); - match event { - TransportEvent::CloudError { descr, .. } => { - assert_eq!(descr, "rejected credential [REDACTED]"); + assert!(matches!( + event, + TransportEvent::CloudError { + msg_id: 7, + code: 403 } - other => panic!("expected CloudError, got {other:?}"), - } + )); } #[test]