Playback: retry the start when a skipped/short head runs the player dry

A queue replace starts playback from the first resolved chunk. If that
chunk's track is is_skipped, or is so short it finishes before the next
chunk resolves, play() found nothing playable and stopped the player --
and the later chunks return None (append mode), so playback never
resumed even though playable tracks were arriving right behind it. The
queue sat stopped with tracks in it. This is exactly the short/skipped
leading-track case the read-ahead is meant to cover.

A pending op now carries a wants_start flag: set when a chunk makes a
track current, cleared only once a start is confirmed (play now returns
whether it handed a track to the player). While set, each arriving chunk
retries the start from the current position -- next_playable_urls
advances past skipped/unplayable heads to the first track that has since
resolved. Once playback takes hold the flag clears, so later chunks only
extend the queue and never restart the playing track, and a user stop
after playback started is respected. An op whose whole resolve yields
nothing playable is dropped and the player stays stopped.

play() returns bool; the now-redundant play_if_some helper is removed.
Documented as architecture/progressive-queueing.md D5. Tests cover the
wants_start lifecycle (set on first current-making chunk, held across
later chunks, cleared on mark_started; never set appending behind a
playing queue).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

architecture/progressive-queueing.md

@@ -156,7 +156,21 @@ Each pending op keeps an insertion cursor:
 Playback start reuses the existing `Option<Track>` returns from the
 `QueueManager` mutations — only a chunk that makes a track current (replace,
 or any insert into an empty queue) yields one, so exactly the first relevant
-chunk starts the player and later chunks never restart it.
+chunk starts the player and later chunks never restart a playing track.
+
+That first start can *fail to find anything playable yet*: if the head
+tracks are `is_skipped` or the collection's first track is very short, the
+player can run dry before the next chunk resolves — and then, since later
+chunks return `None`, playback would stay stopped with playable tracks
+arriving right behind it. So the op carries a `wants_start` flag: set when a
+chunk makes a track current, cleared only once a start is *confirmed*
+(`play` returned that it handed a track to the player). While it is set, each
+arriving chunk retries the start from the current position — `next_playable`
+advances past skipped/unplayable heads to the first track that has now
+resolved. The exponential read-ahead (D8) makes that next track arrive
+sooner; the retry makes sure it actually plays when it does. An op whose
+whole resolve finishes with nothing playable is dropped by `finish_resolve`
+and the player simply stays stopped.
 
 Interleaved edits from other clients during a resolve can shift the cursor's
 target (e.g. removing tracks before it). This is accepted as benign:

crabidy-server/src/lib.rs

@@ -246,6 +246,13 @@ pub struct PendingResolve {
     kind: ResolveKind,
     /// Tracks applied so far; an op finishing at zero is worth a warning.
     applied: usize,
+    /// Set once a chunk of this op makes a track current — the op should be
+    /// driving the initial playback start — and cleared once playback is
+    /// confirmed started ([`Self::mark_started`]). While set, the playback
+    /// loop (re)starts an idle player as later chunks arrive, so an unplayable
+    /// or very short head track that ran the player dry before more tracks
+    /// resolved does not strand the queue stopped.
+    wants_start: bool,
     /// Shared with the op's forwarder task: set on cancellation so the
     /// forwarder drops the chunk receiver, which stops the provider fetch.
     cancelled: Arc<AtomicBool>,
@@ -256,10 +263,23 @@ impl PendingResolve {
         Self {
             kind,
             applied: 0,
+            wants_start: false,
             cancelled: Arc::new(AtomicBool::new(false)),
         }
     }
 
+    /// Whether this op should (re)start playback: a chunk made a track current
+    /// but playback has not been confirmed started yet (see [`Self::apply_chunk`]).
+    pub fn wants_start(&self) -> bool {
+        self.wants_start
+    }
+
+    /// Records that playback started for this op, so later chunks only extend
+    /// the queue and never restart the playing track.
+    pub fn mark_started(&mut self) {
+        self.wants_start = false;
+    }
+
     /// The cancellation flag to hand to this op's forwarder task.
     pub fn cancel_flag(&self) -> Arc<AtomicBool> {
         Arc::clone(&self.cancelled)
@@ -279,10 +299,14 @@ impl PendingResolve {
     /// Applies one resolved chunk to the queue and advances this op's
     /// cursor. Returns the track that should start playing, if this chunk
     /// made one current (first chunk of a replace, or any chunk landing in
-    /// an empty queue) — later chunks of the same op never restart playback.
+    /// an empty queue) — later chunks of the same op never restart a playing
+    /// track. A chunk that makes a track current also sets [`Self::wants_start`],
+    /// so if that track (or the ones after it) turns out unplayable and runs
+    /// the player dry before more tracks resolve, the loop retries the start
+    /// on later chunks instead of leaving the queue stopped.
     pub fn apply_chunk(&mut self, queue: &mut QueueManager, tracks: &[Track]) -> Option<Track> {
         self.applied += tracks.len();
-        match self.kind {
+        let started = match self.kind {
             ResolveKind::Replace => {
                 // Only the first chunk replaces; the rest of this op
                 // extends the fresh queue.
@@ -297,7 +321,11 @@ impl PendingResolve {
                 self.kind = ResolveKind::InsertAfter(position + tracks.len() as u32);
                 queue.insert_tracks(position, tracks)
             }
+        };
+        if started.is_some() {
+            self.wants_start = true;
         }
+        started
     }
 }
 
@@ -768,6 +796,33 @@ mod tests {
         assert!(flag.load(std::sync::atomic::Ordering::Relaxed));
     }
 
+    #[test]
+    fn replace_op_wants_start_until_marked_started() {
+        let mut q = QueueManager::new();
+        let mut op = PendingResolve::new(ResolveKind::Replace);
+        // Nothing applied yet: no start intent.
+        assert!(!op.wants_start());
+        // The first chunk makes a track current, so the op now wants to drive
+        // the initial start — and keeps wanting it across later chunks until
+        // playback is confirmed, so a skipped/short head that runs the player
+        // dry before more resolve is retried rather than stranding the queue.
+        op.apply_chunk(&mut q, &[track(10)]);
+        assert!(op.wants_start());
+        op.apply_chunk(&mut q, &[track(11)]);
+        assert!(op.wants_start(), "still wants start until confirmed");
+        op.mark_started();
+        assert!(!op.wants_start());
+    }
+
+    #[test]
+    fn append_into_playing_queue_never_wants_start() {
+        let mut q = queue_with(2); // already playing track 0
+        let mut op = PendingResolve::new(ResolveKind::Append);
+        op.apply_chunk(&mut q, &[track(10)]);
+        // Appending behind a playing queue must not (re)start playback.
+        assert!(!op.wants_start());
+    }
+
     #[test]
     fn shuffle_insert_keeps_order_unique() {
         let mut q = queue_with(5);

crabidy-server/src/playback.rs

@@ -552,15 +552,17 @@ impl Playback {
 
     /// Applies one chunk to the queue for pending op `op_id`, broadcasts
     /// the grown queue, and starts playback when the chunk made a track
-    /// current. Chunks for an unknown op id (finished or cancelled) are
+    /// current — or when this op still wants to start but its earlier attempt
-    /// dropped silently.
+    /// ran the player dry on skipped/short head tracks before more resolved.
+    /// Chunks for an unknown op id (finished or cancelled) are dropped
+    /// silently.
     async fn apply_resolved_chunk(&self, op_id: u64, tracks: Vec<Track>) {
-        let track = {
+        let attempt = {
             let Ok(mut queue) = self.queue.lock() else {
                 error!("queue lock poisoned");
                 return;
             };
-            let track = {
+            let (designated, wants_start) = {
                 let Ok(mut pending) = self.pending.lock() else {
                     error!("pending ops lock poisoned");
                     return;
@@ -569,12 +571,32 @@ impl Playback {
                     trace!(op_id, "dropping chunk for a finished or cancelled op");
                     return;
                 };
-                op.apply_chunk(&mut queue, &tracks)
+                let designated = op.apply_chunk(&mut queue, &tracks);
+                (designated, op.wants_start())
             };
             self.broadcast_queue(&queue);
-            track
+            // Start on the track this chunk made current; failing that, if the
+            // op still wants to start (a prior attempt found nothing playable
+            // yet), retry from the current position — `next_playable_urls`
+            // advances past skipped/unplayable heads to the first track that
+            // has now resolved.
+            designated.or_else(|| {
+                if wants_start {
+                    queue.current_track()
+                } else {
+                    None
+                }
+            })
         };
-        self.play_if_some(track).await;
+        // Confirming the start clears the op's `wants_start`, so later chunks
+        // only extend the queue and never restart the playing track.
+        if self.play(attempt).await {
+            if let Ok(mut pending) = self.pending.lock() {
+                if let Some(op) = pending.get_mut(&op_id) {
+                    op.mark_started();
+                }
+            }
+        }
     }
 
     /// Removes the finished op and broadcasts the final `Queue` snapshot
@@ -688,13 +710,6 @@ impl Playback {
         }
     }
 
-    /// Plays the given track if there is one; does nothing otherwise.
-    async fn play_if_some(&self, track: Option<Track>) {
-        if track.is_some() {
-            self.play(track).await;
-        }
-    }
-
     /// Finds the stream URLs of the first playable track, starting at
     /// `track` and advancing the queue past unplayable ones. Tracks marked
     /// `is_skipped` (captures recorded their source as uncapturable) are
@@ -747,20 +762,26 @@ impl Playback {
     /// Starts playback of the given track, skipping past unplayable ones
     /// (see [`Self::next_playable_urls`]); stops the player when nothing
     /// in the queue is playable.
+    ///
+    /// Returns `true` when a playable track was found and handed to the
+    /// player. A device error while starting is logged but still counts as
+    /// started — retrying the same track would not help. Returns `false` when
+    /// there was nothing to play or nothing playable remained (the player is
+    /// then stopped), so the caller can retry once more tracks resolve.
     #[instrument(skip(self, track), fields(track = track.as_ref().map(|t| t.path.as_str())))]
-    async fn play(&self, track: Option<Track>) {
+    async fn play(&self, track: Option<Track>) -> bool {
         let Some(track) = track else {
             debug!("nothing to play");
-            return;
+            return false;
         };
         let Some(urls) = self.next_playable_urls(track).await else {
             self.stop_player().await;
-            return;
+            return false;
         };
         {
             let Ok(queue) = self.queue.lock() else {
                 error!("queue lock poisoned");
-                return;
+                return false;
             };
             // Current-track moves (Next/Prev/SetCurrent/skips) change the
             // persisted position without a queue broadcast.
@@ -774,6 +795,7 @@ impl Playback {
         if let Err(err) = self.player.play(&urls[0]).await {
             error!("player failed to start track: {err:?}");
         }
+        true
     }
 }