diff --git a/architecture/progressive-queueing.md b/architecture/progressive-queueing.md index 4408408..a42fc14 100644 --- a/architecture/progressive-queueing.md +++ b/architecture/progressive-queueing.md @@ -156,7 +156,21 @@ Each pending op keeps an insertion cursor: Playback start reuses the existing `Option` 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: diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index 87955b0..bd6d70f 100644 --- a/crabidy-server/src/lib.rs +++ b/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, @@ -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 { 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 { 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); diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index 2844c82..3391d93 100644 --- a/crabidy-server/src/playback.rs +++ b/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 - /// dropped silently. + /// current — or when this op still wants to start but its earlier attempt + /// 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) { - 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) { - 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) { + async fn play(&self, track: Option) -> 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 } }