♻️ refactor stream detection and queue sync logic

- Replace `is_continuous_stream_url` with new canonical check using metadata + fallback
- Extract stream duration comparison logic to `queue::stream_duration_*` helpers  
- Improve queue sync concurrency: add worker loop, pending/cancel flags
- Make `stream_duration_*` functions public(crate) for reuse

pmocontrol/src/music_renderer/capabilities.rs

@@ -46,7 +46,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
 
         drop(queue);
 
-        let is_stream = crate::music_renderer::is_continuous_stream_url(&item.uri);
+        let is_stream = crate::music_renderer::is_continuous_stream(item.metadata.as_ref(), &item.uri);
         *self.continuous_stream().lock().unwrap() = is_stream;
 
         self.play_item(&item)

pmocontrol/src/music_renderer/mod.rs

@@ -23,7 +23,7 @@ pub use crate::music_renderer::capabilities::{
 };
 pub use crate::music_renderer::musicrenderer::{MusicRenderer, PlaylistBinding};
 pub use crate::music_renderer::sleep_timer::SleepTimer;
-pub use crate::music_renderer::stream_detection::is_continuous_stream_url;
+pub use crate::music_renderer::stream_detection::{is_continuous_stream, is_continuous_stream_url};
 use crate::{
     errors::ControlPointError, music_renderer::musicrenderer::MusicRendererBackend, RendererInfo,
 };

pmocontrol/src/music_renderer/musicrenderer.rs

@@ -492,39 +492,19 @@ impl MusicRenderer {
                     if let Some(ref new_duration) = position.track_duration {
                         let mut state = self.state.lock().unwrap();
 
-                        // Parse durations to compare (HH:MM:SS format)
-                        let parse_duration = |dur_str: &str| -> Option<u32> {
-                            let parts: Vec<&str> = dur_str.split(':').collect();
-                            if parts.len() == 3 {
-                                let h: u32 = parts[0].parse().ok()?;
-                                let m: u32 = parts[1].parse().ok()?;
-                                let s: u32 = parts[2].parse().ok()?;
-                                Some(h * 3600 + m * 60 + s)
-                            } else {
-                                None
-                            }
-                        };
-
                         match &state.current_track_duration {
                             Some(stored_duration) => {
-                                // Compare new duration with stored one
+                                if crate::queue::stream_duration_increased(stored_duration, new_duration) {
-                                if let (Some(stored_secs), Some(new_secs)) = (
+                                    tracing::debug!(
-                                    parse_duration(stored_duration),
+                                        "MusicRenderer [{}]: Stream duration increased: {} -> {}",
-                                    parse_duration(new_duration),
+                                        self.info.friendly_name(),
-                                ) {
+                                        stored_duration,
-                                    if new_secs > stored_secs {
+                                        new_duration
-                                        // Duration increased: update stored value and use new one
+                                    );
-                                        tracing::debug!(
+                                    state.current_track_duration = Some(new_duration.clone());
-                                            "MusicRenderer [{}]: Stream duration increased: {} -> {}",
+                                } else if crate::queue::stream_duration_decreased(stored_duration, new_duration) {
-                                            self.info.friendly_name(),
+                                    // Duration decreased: keep stored value
-                                            stored_duration,
+                                    position.track_duration = Some(stored_duration.clone());
-                                            new_duration
-                                        );
-                                        state.current_track_duration = Some(new_duration.clone());
-                                    } else {
-                                        // Duration decreased or equal: keep stored value
-                                        position.track_duration = Some(stored_duration.clone());
-                                    }
                                 }
                             }
                             None => {

pmocontrol/src/music_renderer/stream_detection.rs

@@ -195,6 +195,17 @@ fn check_stream_headers(url: &str) -> Result<bool, String> {
     Ok(is_stream)
 }
 
+/// Canonical check: returns `true` if this item should be treated as a continuous stream.
+///
+/// Checks `metadata.is_continuous_stream` first (already computed at ingest time),
+/// then falls back to the URL-based HTTP detection.
+///
+/// Use this function everywhere transport-layer code needs to decide whether playback is
+/// a continuous stream (radio) vs bounded media (file/album track).
+pub fn is_continuous_stream(metadata: Option<&crate::model::TrackMetadata>, uri: &str) -> bool {
+    metadata.map(|m| m.is_continuous_stream).unwrap_or(false) || is_continuous_stream_url(uri)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

pmocontrol/src/queue/mod.rs

@@ -19,7 +19,7 @@ use crate::{errors::ControlPointError, RendererInfo};
 
 /// Returns true if `new_dur` < `old_dur` (both parseable as HH:MM:SS/MM:SS/SS).
 /// Used to protect stream durations from decreasing for the same track.
-pub(super) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool {
+pub(crate) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool {
     match (
         parse_time_flexible(old_dur).ok(),
         parse_time_flexible(new_dur).ok(),
@@ -30,7 +30,7 @@ pub(super) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool {
 }
 
 /// Returns true if `new_dur` > `old_dur` (both parseable as HH:MM:SS/MM:SS/SS).
-pub(super) fn stream_duration_increased(old_dur: &str, new_dur: &str) -> bool {
+pub(crate) fn stream_duration_increased(old_dur: &str, new_dur: &str) -> bool {
     match (
         parse_time_flexible(old_dur).ok(),
         parse_time_flexible(new_dur).ok(),

pmocontrol/src/queue/music_queue.rs

@@ -129,6 +129,14 @@ impl MusicQueue {
         thread::Builder::new()
             .name(thread_name)
             .spawn(move || {
+                // Protocol for the three AtomicBools:
+                //   sync_in_progress : set to true before spawn, cleared on Drop via Guard.
+                //   sync_pending     : set to true by a concurrent caller that arrives while
+                //                      a sync is already running.  The worker re-fetches items
+                //                      and loops when it detects this flag on exit.
+                //   sync_cancel_token: set to true when a new sync request interrupts an
+                //                      in-progress one.  Passed into QueueBackend::sync_queue
+                //                      so it can abort early.
                 struct Guard(Arc<AtomicBool>);
                 impl Drop for Guard {
                     fn drop(&mut self) {
@@ -136,113 +144,133 @@ impl MusicQueue {
                     }
                 }
                 let _guard = Guard(Arc::clone(&sync_in_progress));
-
-                let mut current_items = items;
-                let mut current_on_ready = Some(on_ready);
-                let mut on_complete = Some(on_complete);
-
                 tracing::debug!(thread = %std::thread::current().name().unwrap_or("?"), "queue-sync thread started");
-
+                Self::sync_worker_loop(
-                loop {
+                    queue_arc,
-                    sync_pending.store(false, SeqCst);
+                    items,
-                    sync_cancel_token.store(false, SeqCst);
+                    pending_items_fn,
-
+                    on_ready,
-                    // Extract the real on_ready BEFORE locking the queue.
+                    on_complete,
-                    // on_ready may call play_from_queue() which re-locks the queue,
+                    sync_pending,
-                    // so we must NOT call it while holding queue_arc.
+                    sync_cancel_token,
-                    let real_on_ready = current_on_ready.take().flatten();
+                );
-                    let on_ready_triggered = Arc::new(AtomicBool::new(false));
-                    let proxy_on_ready: Option<Box<dyn FnOnce() + Send + 'static>> =
-                        real_on_ready.as_ref().map(|_| {
-                            let flag = Arc::clone(&on_ready_triggered);
-                            Box::new(move || {
-                                flag.store(true, SeqCst);
-                            }) as Box<dyn FnOnce() + Send + 'static>
-                        });
-
-                    tracing::debug!(
-                        thread = %std::thread::current().name().unwrap_or("?"),
-                        items = current_items.len(),
-                        has_on_ready = real_on_ready.is_some(),
-                        "queue-sync: calling sync_queue"
-                    );
-
-                    let result = {
-                        let mut q = queue_arc.lock().unwrap();
-                        <MusicQueue as QueueBackend>::sync_queue(
-                            &mut q,
-                            current_items,
-                            &sync_cancel_token,
-                            proxy_on_ready,
-                        )
-                    };
-                    // Queue lock is released here.
-                    // Now safe to call on_ready (which may re-lock the queue).
-                    // If on_ready was triggered by the proxy, consume and call it.
-                    // If not (cancelled before first insert), keep it to pass to retry.
-                    let carry_on_ready = if on_ready_triggered.load(SeqCst) {
-                        tracing::debug!(
-                            thread = %std::thread::current().name().unwrap_or("?"),
-                            "queue-sync: on_ready triggered, calling callback"
-                        );
-                        if let Some(f) = real_on_ready {
-                            f();
-                        }
-                        None
-                    } else {
-                        if real_on_ready.is_some() {
-                            tracing::debug!(
-                                thread = %std::thread::current().name().unwrap_or("?"),
-                                "queue-sync: on_ready not triggered, carrying to next attempt"
-                            );
-                        }
-                        real_on_ready
-                    };
-
-                    match result {
-                        Err(ControlPointError::SyncCancelled) => {
-                            tracing::debug!(
-                                thread = %std::thread::current().name().unwrap_or("?"),
-                                "queue-sync: cancelled"
-                            );
-                        }
-                        Err(e) => {
-                            tracing::warn!("queue-sync error: {}", e);
-                        }
-                        Ok(()) => {
-                            tracing::debug!(
-                                thread = %std::thread::current().name().unwrap_or("?"),
-                                "queue-sync: completed successfully"
-                            );
-                            if let Some(cb) = on_complete.take() {
-                                let queue_len = queue_arc.lock().unwrap().len().unwrap_or(0);
-                                cb(queue_len);
-                            }
-                        }
-                    }
-
-                    if !sync_pending.load(SeqCst) {
-                        break;
-                    }
-
-                    match pending_items_fn() {
-                        Ok(new_items) => {
-                            current_items = new_items;
-                            current_on_ready = Some(carry_on_ready);
-                        }
-                        Err(e) => {
-                            tracing::warn!("queue-sync pending re-fetch error: {}", e);
-                            break;
-                        }
-                    }
-                }
-
                 tracing::debug!(thread = %std::thread::current().name().unwrap_or("?"), "queue-sync thread done");
             })
             .expect("Failed to spawn queue-sync thread");
 
         SyncScheduleOutcome::Scheduled
     }
+
+    /// Inner loop executed by the sync worker thread.
+    ///
+    /// Runs at least once with `initial_items`.  If a new sync request arrives while the
+    /// loop is running (`sync_pending` becomes true), it re-fetches items via
+    /// `pending_items_fn` and iterates again, allowing the latest playlist state to win.
+    fn sync_worker_loop(
+        queue_arc: Arc<Mutex<MusicQueue>>,
+        initial_items: Vec<PlaybackItem>,
+        pending_items_fn: Box<dyn Fn() -> Result<Vec<PlaybackItem>, ControlPointError> + Send>,
+        initial_on_ready: Option<Box<dyn FnOnce() + Send>>,
+        on_complete: Box<dyn Fn(usize) + Send>,
+        sync_pending: Arc<AtomicBool>,
+        sync_cancel_token: Arc<AtomicBool>,
+    ) {
+        let mut current_items = initial_items;
+        let mut current_on_ready = Some(initial_on_ready);
+        let mut on_complete = Some(on_complete);
+
+        loop {
+            sync_pending.store(false, SeqCst);
+            sync_cancel_token.store(false, SeqCst);
+
+            // Extract the real on_ready BEFORE locking the queue.
+            // on_ready may call play_from_queue() which re-locks the queue,
+            // so we must NOT call it while holding queue_arc.
+            let real_on_ready = current_on_ready.take().flatten();
+            let on_ready_triggered = Arc::new(AtomicBool::new(false));
+            let proxy_on_ready: Option<Box<dyn FnOnce() + Send + 'static>> =
+                real_on_ready.as_ref().map(|_| {
+                    let flag = Arc::clone(&on_ready_triggered);
+                    Box::new(move || {
+                        flag.store(true, SeqCst);
+                    }) as Box<dyn FnOnce() + Send + 'static>
+                });
+
+            tracing::debug!(
+                thread = %std::thread::current().name().unwrap_or("?"),
+                items = current_items.len(),
+                has_on_ready = real_on_ready.is_some(),
+                "queue-sync: calling sync_queue"
+            );
+
+            let result = {
+                let mut q = queue_arc.lock().unwrap();
+                <MusicQueue as QueueBackend>::sync_queue(
+                    &mut q,
+                    current_items,
+                    &sync_cancel_token,
+                    proxy_on_ready,
+                )
+            };
+            // Queue lock is released here.
+            // Now safe to call on_ready (which may re-lock the queue).
+            let carry_on_ready = if on_ready_triggered.load(SeqCst) {
+                tracing::debug!(
+                    thread = %std::thread::current().name().unwrap_or("?"),
+                    "queue-sync: on_ready triggered, calling callback"
+                );
+                if let Some(f) = real_on_ready {
+                    f();
+                }
+                None
+            } else {
+                if real_on_ready.is_some() {
+                    tracing::debug!(
+                        thread = %std::thread::current().name().unwrap_or("?"),
+                        "queue-sync: on_ready not triggered, carrying to next attempt"
+                    );
+                }
+                real_on_ready
+            };
+
+            match result {
+                Err(ControlPointError::SyncCancelled) => {
+                    tracing::debug!(
+                        thread = %std::thread::current().name().unwrap_or("?"),
+                        "queue-sync: cancelled"
+                    );
+                }
+                Err(e) => {
+                    tracing::warn!("queue-sync error: {}", e);
+                }
+                Ok(()) => {
+                    tracing::debug!(
+                        thread = %std::thread::current().name().unwrap_or("?"),
+                        "queue-sync: completed successfully"
+                    );
+                    if let Some(cb) = on_complete.take() {
+                        let queue_len = queue_arc.lock().unwrap().len().unwrap_or(0);
+                        cb(queue_len);
+                    }
+                }
+            }
+
+            if !sync_pending.load(SeqCst) {
+                break;
+            }
+
+            match pending_items_fn() {
+                Ok(new_items) => {
+                    current_items = new_items;
+                    current_on_ready = Some(carry_on_ready);
+                }
+                Err(e) => {
+                    tracing::warn!("queue-sync pending re-fetch error: {}", e);
+                    break;
+                }
+            }
+        }
+    }
 }
 
 impl QueueBackend for MusicQueue {