From ccab43a133ee1529f578f651b5d1b2366f3f8266 Mon Sep 17 00:00:00 2001 From: Test User Date: Tue, 21 Jul 2026 01:08:34 +0200 Subject: [PATCH] Queue large collections progressively Resolving a nested node used to collect every track before the queue changed: one broadcast at the very end, playback only after the full walk, and the playback loop blocked for the duration. Now provider resolution streams bounded chunks (tidaldy: one per 50-track page), the playback loop applies and broadcasts each chunk as it lands, playback starts with the first chunk, and Replace/Clear cancel in-flight resolves down to the HTTP fetch. Queue.resolving (additive proto field) drives an animated-dots pseudo-item in the TUI queue pane. Also fixes Enter on a non-queueable library item blanking the queue while audio kept playing, and the reversed album order left by the old LIFO walk. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 4 + architecture/progressive-queueing.md | 269 +++++++++++++++++++++++ cbd-tui/src/app/library.rs | 10 +- cbd-tui/src/app/mod.rs | 40 ++++ cbd-tui/src/app/queue.rs | 142 +++++++++++- crabidy-core/Cargo.toml | 5 + crabidy-core/crabidy/v1/crabidy.proto | 4 + crabidy-core/src/lib.rs | 301 ++++++++++++++++++++++++++ crabidy-server/src/lib.rs | 147 +++++++++++++ crabidy-server/src/main.rs | 24 +- crabidy-server/src/playback.rs | 272 +++++++++++++++++------ crabidy-server/src/provider.rs | 79 ++++--- plan/progressive-queueing.md | 94 ++++++++ plan/summary.md | 55 +++++ quality/progressive-queueing.md | 95 ++++++++ tidaldy/Cargo.toml | 1 + tidaldy/src/lib.rs | 119 ++++++++++ 17 files changed, 1542 insertions(+), 119 deletions(-) create mode 100644 architecture/progressive-queueing.md create mode 100644 plan/progressive-queueing.md create mode 100644 quality/progressive-queueing.md diff --git a/Cargo.lock b/Cargo.lock index 8349aa0..748f3e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -723,13 +723,16 @@ dependencies = [ "async-trait", "clap-serde-derive", "dirs", + "flume", "percent-encoding", "prost", "serde", + "tokio", "toml", "tonic", "tonic-prost", "tonic-prost-build", + "tracing", ] [[package]] @@ -3946,6 +3949,7 @@ dependencies = [ "base64", "chrono", "crabidy-core", + "flume", "reqwest", "serde", "serde_json", diff --git a/architecture/progressive-queueing.md b/architecture/progressive-queueing.md new file mode 100644 index 0000000..a16e79f --- /dev/null +++ b/architecture/progressive-queueing.md @@ -0,0 +1,269 @@ +# Progressive queueing of large collections + +## Context and problem statement + +Queueing a nested node (an artist with many albums, a large playlist) today +freezes the UI's mental model: nothing changes for many seconds, then the full +queue appears at once. Three compounding causes, all in the resolve path: + +1. `Playback::resolve_tracks` collects **every** track before the queue is + touched, so the single `Queue` broadcast happens only at the very end and + playback cannot start earlier. +2. The resolve runs **inline in the playback loop**, so every other playback + command — pause, next, volume — is blocked for the duration. +3. `tidaldy` paginates collections to exhaustion (50 tracks per sequential + request) inside one `get_lib_node` call, so even a single large playlist + produces no intermediate result. + +The feature: resolve progressively. Tracks are applied to the queue in chunks +as the provider produces them, each chunk is broadcast, playback starts with +the first chunk, and clients see a loading indicator (animated dots as a +pseudo last queue item) while resolution is still running. + +Run autonomously per standing user instruction; every decision below records +the options considered and the rationale. + +## Assumptions + +- Chunks must arrive **in playback order** — the user explicitly wants "the + first elements first, and later chunks follow". Order is known up front + (collection order), so no reordering step is needed. +- The four resolve-based queue operations (`Replace`, `Queue`, `Append`, + `Insert`) all benefit equally and should share one mechanism. +- Wire changes must stay additive (old clients keep working; they simply see + the queue fill progressively without an indicator). +- Multi-client remains supported: the indicator must be server-derived state, + not client-local guessing. + +## Decisions + +### D1 — End-to-end progressive resolution, not a client-only spinner + +Options considered: + +1. **Client-only indicator**: the TUI shows dots between sending a queue op + and receiving the next `Queue` update. No wire or server change. +2. **Server-side chunked resolution** with a wire-visible "still resolving" + flag; the indicator falls out of the flag. + +**Decision: (2).** Option 1 papers over the latency without fixing it — +playback would still start only after the full resolve, other clients would +see nothing, and the "did it even register?" dead time remains. Option 2 +fixes the actual complaint (start playing early, fill visibly) and gives +every client the indicator for free. The server broadcasts an immediate +`Queue` update (unchanged tracks, `resolving = true`) when the op is +accepted, so feedback appears within one round trip. + +### D2 — `Queue.resolving` field, not a new stream-update variant + +Options considered: + +1. New `GetUpdateStream` oneof variant `resolving(bool)`. +2. New field `bool resolving = 4` on the `Queue` message itself. + +**Decision: (2).** The flag is queue state and must be atomic with the track +snapshot it describes; a separate variant can arrive out of order relative to +`Queue` updates (the broadcast channel is lossy for slow clients). Additive +field, wire-compatible both ways: old clients ignore it, old servers never +set it. + +### D3 — Chunked resolution lives in `ProviderClient`, with a default + +Options considered: + +1. Keep the BFS in `ProviderOrchestrator` (chunk = one node's tracks); no + trait change. A 1000-track playlist is still one 20-request blob. +2. Add a chunked resolve method to the `ProviderClient` trait with a default + implementation (the generic walk, one chunk per node); `tidaldy` + overrides it to stream **page-sized chunks (50)** for the paginated + collections (playlists, albums). + +**Decision: (2).** + +```rust +/// Streams the playable tracks under `path` into `chunk_tx` in playback +/// order. Zero or more chunks, then the sender is dropped: a dropped +/// SENDER means resolution finished. A dropped RECEIVER cancels +/// resolution (the provider stops fetching and returns Ok). +async fn resolve_tracks_into( + &self, + path: &str, + chunk_tx: flume::Sender>, +) -> Result<(), ProviderError>; +``` + +The channel semantics are the contract and are documented on the trait — +this is a local bounded channel used as a stream, not a queue pretending to +be durable. Unreadable nodes are skipped with a warning (today's behavior); +only a completely unresolvable root returns an error. The default +implementation walks the tree **depth-first pre-order** over queueable +descendants, emitting one chunk per node. This replaces the old orchestrator +BFS, and fixes a latent ordering bug while doing so: the old worklist popped +LIFO, so an artist's albums were flattened in *reverse* order. + +`tidaldy` overrides the method: playlist and album paths stream one chunk +per fetched page instead of paginating to exhaustion first (a new +`make_paginated_request` variant hands each page to a sink); everything else +follows the generic walk. Search-term track lists are a single page already. +The playlist arm also stops fetching the playlist metadata (title) — the +resolve needs only tracks. + +`flume` is already a workspace dependency; `crabidy-core` adopts it for the +trait signature. + +### D4 — Neither event loop blocks: spawned resolves, single-writer queue + +Options considered: + +1. Consume chunks inline in the playback loop's `handle_command` (loop still + blocked for the whole resolve; pause/next dead — today's hidden defect). +2. Spawn the resolve; queue mutations travel back to the playback loop as + internal commands, so the loop stays the **single writer** of queue state. + +**Decision: (2), on both loops.** + +- **Provider side**: `ProviderOrchestrator::run` wraps the orchestrator in an + `Arc`; the `ResolveTracks` arm spawns the resolve onto its own task instead + of awaiting it inline. Without this, the first chunk would deadlock the + system: playback applies chunk 1 → `play()` → sends `GetTrackUrls` and + awaits the reply — but the provider loop would still be busy resolving. + Other provider commands keep flowing while (possibly several) resolves run. +- **Playback side**: each queue op registers a *pending op* (id from an + `AtomicU64`, kind, insertion cursor) and spawns a forwarder task that + drives `ProviderCommand::ResolveTracks` per path (sequentially, preserving + multi-path order) and forwards each chunk to the playback channel as + `PlaybackCommand::ApplyResolvedChunk { op_id, tracks }`, followed by + `ResolveFinished { op_id }`. Queue state is only ever mutated inside the + loop, exactly as before; user commands interleave between chunks. + +Backpressure is real at every hop: provider → forwarder over a small bounded +chunk channel, forwarder → playback over the existing `bounded(64)` command +channel. A slow consumer slows the HTTP fetching down instead of buffering +unboundedly. + +### D5 — Chunk application semantics per op kind + +Each pending op keeps an insertion cursor: + +- **Replace**: first chunk `replace_with_tracks` (broadcast resets the + queue, current position 0, playback starts); later chunks append. +- **Append**: every chunk `append_tracks`. +- **Queue** (after current): cursor starts at the current position; each + chunk `insert_tracks(cursor)`, then `cursor += chunk.len()`. +- **Insert**: same, starting at the requested position. + +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. + +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: +`insert_tracks` already clamps, the queue self-heals on the next broadcast, +and simultaneous multi-client edits during a resolve are rare. Under +shuffle, arriving chunks are shuffled behind the current track like any +other insert — chunk order is irrelevant when shuffle is on. Each op counts +its applied tracks; an op that finishes with zero keeps today's +"resolved to no playable tracks" warning. + +### D6 — `Replace` and `Clear` cancel in-flight resolves + +Without cancellation, "replace the queue" or "clear the queue" during a +large resolve would be followed by the old op's remaining chunks trickling +back in — a corrupted queue, and the exact ghost behavior this feature is +meant to kill. Options: let stale chunks land (wrong), or cancel. + +**Decision: cancel.** Each pending op carries an `Arc` shared +with its forwarder. `Replace` and `Clear` mark every pending op cancelled +and drop it from the map. The forwarder checks the flag per chunk and, when +set, drops the chunk receiver — the provider's next `send` fails and the +resolve task stops fetching (the documented receiver-drop semantics from +D3). Chunks already in flight for an unknown op id are ignored by the loop. +`Queue`/`Append`/`Insert` do **not** cancel: concurrent additive ops are +legal; their chunks interleave between ops while each op's internal order is +preserved. + +### D7 — TUI indicator: an animated pseudo-item, outside the list model + +While the latest `Queue` update carries `resolving = true`, the queue pane +renders one extra line after the last track: one to three dots cycling +(~400 ms per step, derived from elapsed time — the render loop already +redraws at least every 100 ms), in `COLOR_SECONDARY`. The pseudo-item is +appended at render time only and never enters `self.list`, so selection, +removal, and `get_size` cannot reach it — no new input states, nothing to +misclick. + +## Flows + +```d2 +shape: sequence_diagram +user: { shape: person } +tui: cbd-tui +rpc: gRPC handler +playback: playback loop +fwd: forwarder task +provider: provider loop +resolve: resolve task +tidal: Tidal API + +user -> tui: queue large artist +tui -> rpc: Append(paths) +rpc -> playback: "PlaybackCommand::Append (fire-and-forget)" +playback -> playback: register pending op +playback -> tui: "Queue update (resolving=true)" +playback -> fwd: spawn +fwd -> provider: "ResolveTracks(path, chunk_tx)" +provider -> resolve: spawn +resolve -> tidal: fetch page 1 +resolve -> fwd: chunk 1 +fwd -> playback: "ApplyResolvedChunk(op, chunk 1)" +playback -> tui: "Queue update (resolving=true)" +playback -> playback: "play() first track" +resolve -> tidal: fetch page 2 +resolve -> fwd: chunk 2 +fwd -> playback: "ApplyResolvedChunk(op, chunk 2)" +playback -> tui: "Queue update (resolving=true)" +fwd -> playback: "ResolveFinished(op)" +playback -> tui: "Queue update (resolving=false)" +``` + +The dots pseudo-item is visible in the TUI exactly while updates carry +`resolving = true`; user commands (pause, next, remove) flow through the +playback loop between chunk applications instead of waiting for the end. + +```d2 +direction: right +core: "ProviderClient::resolve_tracks_into" { + default: "default: pre-order walk,\none chunk per node" +} +tidaldy: "tidaldy override" { + pages: "playlist/album:\none chunk per 50-track page" +} +playback: "playback loop" { + ops: "pending ops:\ncursor + cancel flag" +} +core -> tidaldy: overridden by +tidaldy.pages -> playback.ops: "bounded chunks, in order" +playback.ops -> playback.ops: "apply + broadcast per chunk" +``` + +## Boundaries and risks + +- **Proto**: one additive field (`Queue.resolving = 4`). No RPC shape + changes; the queue ops stay fire-and-forget. +- **Trait**: one new `ProviderClient` method with a default implementation — + existing providers (there is one) compile unchanged if they skip the + override; the override is where the provider-level win lives. +- **Ordering fix is a behavior change**: multi-album artists now queue in + listing order instead of reversed. Strictly a fix, noted here because + someone may have gotten used to the bug. +- **Concurrent additive ops interleave between ops.** Each op's internal + order is kept; the interleaving matches command arrival order at the loop. + Accepted — same semantics a human doing two appends "at once" expects. +- **Old TUI + new server**: queue fills progressively, no indicator — pure + improvement, no breakage. New TUI + old server: `resolving` is always + false, indicator never shows, behavior as today. +- **Not in scope**: pagination of `get_lib_node` for *browsing* (the library + pane still fetches collections to exhaustion before rendering), queue + persistence, a progress percentage (total counts are known per collection + but not aggregated across a nested walk). diff --git a/cbd-tui/src/app/library.rs b/cbd-tui/src/app/library.rs index 8a4718c..b187e7c 100644 --- a/cbd-tui/src/app/library.rs +++ b/cbd-tui/src/app/library.rs @@ -76,10 +76,12 @@ impl Library { .collect(), ); } - if let Some(idx) = self.list_state.selected() { - return Some(vec![self.list[idx].path.to_string()]); - } - None + // Marks are gated on is_queable when set; the bare selection must be + // gated here too, or Enter on a plain folder ships a path the server + // can only resolve to nothing (silently ignored, like % / e / d on + // items without the capability). + let item = self.list.get(self.list_state.selected()?)?; + item.is_queable.then(|| vec![item.path.to_string()]) } pub fn ascend(&mut self) { if let Some(parent) = self.parent.as_ref() { diff --git a/cbd-tui/src/app/mod.rs b/cbd-tui/src/app/mod.rs index 9777726..6979633 100644 --- a/cbd-tui/src/app/mod.rs +++ b/cbd-tui/src/app/mod.rs @@ -570,6 +570,46 @@ mod tests { } } + #[test] + fn queue_ops_ignore_non_queueable_selections() { + use crabidy_core::proto::crabidy::LibraryNodeChild; + let (mut app, rx) = app(); + + // The selected child of the search listing is not queueable: Enter + // (and every other queue op) must send nothing — previously this + // shipped the path anyway and the zero-track replace blanked the + // queue while playback continued. + app.library.update(search_listing(false)); + for action in [ + Action::LibraryQueueReplace, + Action::LibraryQueueAppend, + Action::LibraryQueueNext, + ] { + let _ = app.dispatch(action); + assert!( + rx.try_recv().is_err(), + "non-queueable selection sent a message for {action:?}" + ); + } + + // A queueable selection still goes through. + app.library.update(LibraryNode { + children: vec![LibraryNodeChild::new( + "/tidal/artists/1".to_string(), + "artist".to_string(), + true, + )], + ..creatable_node("/tidal/artists") + }); + let _ = app.dispatch(Action::LibraryQueueReplace); + match rx.try_recv() { + Ok(MessageFromUi::ReplaceQueue(paths)) => { + assert_eq!(paths, vec!["/tidal/artists/1".to_string()]); + } + other => panic!("expected ReplaceQueue, got {:?}", other.is_ok()), + } + } + #[test] fn edit_opens_the_overlay_prefilled_only_on_editable_selections() { let (mut app, _rx) = app(); diff --git a/cbd-tui/src/app/queue.rs b/cbd-tui/src/app/queue.rs index 9fc1c84..3eb1b22 100644 --- a/cbd-tui/src/app/queue.rs +++ b/cbd-tui/src/app/queue.rs @@ -11,12 +11,19 @@ use crabidy_core::proto::crabidy::Queue as QueueData; use super::{ MessageFromUi, StatefulList, UiItem, UiItemKind, COLOR_PRIMARY, COLOR_PRIMARY_DARK, COLOR_RED, + COLOR_SECONDARY, }; pub struct Queue { current_position: usize, list: Vec, list_state: ListState, + /// True while the server is still resolving queued paths (from + /// `Queue.resolving`); the pane then renders an animated-dots + /// pseudo-item after the last track. The pseudo-item exists only at + /// render time — it never enters `list`, so selection and removal + /// cannot reach it. + resolving: bool, tx: Sender, } @@ -26,9 +33,16 @@ impl Queue { current_position: 0, list: Vec::new(), list_state: ListState::default(), + resolving: false, tx, } } + + /// The loading indicator line: one to three dots, cycling with wall + /// time (~400 ms per step). Pure so the animation is testable. + fn loading_dots(elapsed_ms: u128) -> String { + ".".repeat(1 + (elapsed_ms / 400 % 3) as usize) + } pub fn play_next(&self) { let _ = self.tx.send(MessageFromUi::NextTrack); } @@ -54,6 +68,7 @@ impl Queue { } pub fn update_queue(&mut self, queue: QueueData) { self.current_position = queue.current_position as usize; + self.resolving = queue.resolving; self.list = queue .tracks .iter() @@ -73,7 +88,7 @@ impl Queue { } pub fn render(&mut self, f: &mut Frame, area: Rect, focused: bool) { - let queue_items: Vec = self + let mut queue_items: Vec = self .list .iter() .enumerate() @@ -93,6 +108,22 @@ impl Queue { ListItem::new(Span::from(title)).style(style) }) .collect(); + if self.resolving { + // Render-time pseudo-item: more tracks are on their way. It is + // not part of `self.list`, so it can never be selected or + // removed. The render loop redraws at least every 100 ms, + // which keeps the dots moving. + static RENDERED_FIRST_AT: std::sync::OnceLock = + std::sync::OnceLock::new(); + let elapsed = RENDERED_FIRST_AT + .get_or_init(std::time::Instant::now) + .elapsed() + .as_millis(); + queue_items.push( + ListItem::new(Span::from(Self::loading_dots(elapsed))) + .style(Style::default().fg(COLOR_SECONDARY)), + ); + } let queue_list = List::new(queue_items) .block( @@ -129,3 +160,112 @@ impl StatefulList for Queue { self.list_state.selected() } } + +#[cfg(test)] +mod tests { + use super::*; + use crabidy_core::proto::crabidy::Track; + use ratatui::{backend::TestBackend, Terminal}; + + fn queue_data(titles: &[&str], resolving: bool) -> QueueData { + QueueData { + timestamp: 0, + current_position: 0, + tracks: titles + .iter() + .map(|t| Track { + path: format!("/tidal/x/{t}"), + artist: "artist".to_string(), + title: t.to_string(), + duration: None, + album: None, + }) + .collect(), + resolving, + } + } + + fn rendered_rows(queue: &mut Queue) -> Vec { + let backend = TestBackend::new(40, 8); + let mut terminal = Terminal::new(backend).expect("test terminal"); + terminal + .draw(|f| queue.render(f, f.area(), true)) + .expect("draw"); + let buffer = terminal.backend().buffer().clone(); + (0..buffer.area.height) + .map(|y| { + (0..buffer.area.width) + .map(|x| buffer[(x, y)].symbol().to_string()) + .collect::() + }) + .collect() + } + + /// The row content inside the borders, trimmed. + fn inner_rows(queue: &mut Queue) -> Vec { + rendered_rows(queue) + .iter() + .skip(1) + .map(|row| row.trim_matches(['│', ' ']).to_string()) + .collect() + } + + fn dots_row_count(rows: &[String]) -> usize { + rows.iter() + .filter(|row| !row.is_empty() && row.chars().all(|c| c == '.')) + .count() + } + + #[test] + fn resolving_queue_renders_trailing_dots_item() { + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + queue.update_queue(queue_data(&["one", "two"], true)); + let rows = inner_rows(&mut queue); + assert_eq!(dots_row_count(&rows), 1, "rows: {rows:?}"); + // The dots trail the tracks: they come after the last track row. + let last_track = rows.iter().position(|r| r.contains("two")).unwrap(); + let dots = rows + .iter() + .position(|r| !r.is_empty() && r.chars().all(|c| c == '.')) + .unwrap(); + assert!(last_track < dots, "rows: {rows:?}"); + } + + #[test] + fn settled_queue_has_no_dots_item() { + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + queue.update_queue(queue_data(&["one", "two"], false)); + let rows = inner_rows(&mut queue); + assert_eq!(dots_row_count(&rows), 0, "rows: {rows:?}"); + } + + #[test] + fn resolving_flag_clears_with_the_next_update() { + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + queue.update_queue(queue_data(&["one"], true)); + queue.update_queue(queue_data(&["one", "two"], false)); + let rows = inner_rows(&mut queue); + assert_eq!(dots_row_count(&rows), 0, "rows: {rows:?}"); + } + + #[test] + fn dots_item_is_outside_the_selectable_list() { + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + queue.update_queue(queue_data(&["one", "two"], true)); + // Selection, removal and navigation all key off get_size; the + // pseudo-item must not be reachable through any of them. + assert_eq!(queue.get_size(), 2); + } + + #[test] + fn loading_dots_cycle_one_to_three() { + assert_eq!(Queue::loading_dots(0), "."); + assert_eq!(Queue::loading_dots(400), ".."); + assert_eq!(Queue::loading_dots(800), "..."); + assert_eq!(Queue::loading_dots(1200), "."); + } +} diff --git a/crabidy-core/Cargo.toml b/crabidy-core/Cargo.toml index 5ef067b..1b3df49 100644 --- a/crabidy-core/Cargo.toml +++ b/crabidy-core/Cargo.toml @@ -7,13 +7,18 @@ edition.workspace = true async-trait.workspace = true clap-serde-derive.workspace = true dirs.workspace = true +flume.workspace = true percent-encoding.workspace = true prost.workspace = true serde.workspace = true toml.workspace = true tonic.workspace = true +tracing.workspace = true tonic-prost.workspace = true +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt"] } + [build-dependencies] tonic-prost-build.workspace = true diff --git a/crabidy-core/crabidy/v1/crabidy.proto b/crabidy-core/crabidy/v1/crabidy.proto index 9d1386b..9e161d2 100644 --- a/crabidy-core/crabidy/v1/crabidy.proto +++ b/crabidy-core/crabidy/v1/crabidy.proto @@ -205,6 +205,10 @@ message Queue { uint32 current_position = 2; // Without album repeated Track tracks = 3; + // True while the server is still resolving queued paths into tracks: more + // tracks will arrive in subsequent Queue updates. Clients may show a + // loading indicator until an update carries resolving = false. + bool resolving = 4; } message QueueTrack { diff --git a/crabidy-core/src/lib.rs b/crabidy-core/src/lib.rs index 07cf751..a096130 100644 --- a/crabidy-core/src/lib.rs +++ b/crabidy-core/src/lib.rs @@ -60,6 +60,64 @@ pub trait ProviderClient: std::fmt::Debug + Send + Sync { /// refreshed parent node (what a client should display next). Errors: /// [`ProviderError::NotSupported`] when the path is not deletable. async fn delete_lib_node(&self, path: &str) -> Result; + /// Streams the playable tracks under `path` into `chunk_tx`, in playback + /// order. + /// + /// This is a local bounded channel used as a stream, and its delivery + /// semantics are the contract: + /// + /// - Zero or more non-empty chunks are sent, in playback order. + /// - Resolution is finished when the **sender** is dropped (this method + /// returning). There is no end-of-stream marker. + /// - Dropping the **receiver** cancels resolution: the provider stops + /// fetching at the next send and returns `Ok`. + /// - An unreadable node inside the walk is skipped with a warning; only + /// a `path` that cannot be resolved at all is an `Err`. + /// + /// A track path yields exactly one single-track chunk. The default + /// implementation walks the node's queueable descendants depth-first in + /// pre-order and emits one chunk per node holding tracks; providers + /// should override it when they can produce finer-grained chunks (e.g. + /// one per fetched page of a large collection). + async fn resolve_tracks_into( + &self, + path: &str, + chunk_tx: flume::Sender>, + ) -> Result<(), ProviderError> { + if self.is_track_path(path) { + match self.get_metadata_for_track(path).await { + Ok(track) => { + let _ = chunk_tx.send_async(vec![track]).await; + } + Err(err) => tracing::warn!(path, "failed to resolve track: {err}"), + } + return Ok(()); + } + // Depth-first pre-order so tracks arrive in listing order; children + // are pushed reversed because the worklist pops from the back. + let mut nodes_to_go = vec![path.to_string()]; + let mut at_root = true; + while let Some(node_path) = nodes_to_go.pop() { + let node = match self.get_lib_node(&node_path).await { + Ok(node) => node, + Err(err) if at_root => return Err(err), + Err(err) => { + tracing::warn!(node = node_path, "skipping unreadable node: {err}"); + continue; + } + }; + at_root = false; + if !node.is_queable { + continue; + } + if !node.tracks.is_empty() && chunk_tx.send_async(node.tracks).await.is_err() { + // Receiver gone: the consumer cancelled, stop fetching. + return Ok(()); + } + nodes_to_go.extend(node.children.into_iter().rev().map(|c| c.path)); + } + Ok(()) + } } #[derive(Clone, Debug, Hash, PartialEq, Eq)] @@ -271,6 +329,249 @@ mod tests { } } + /// A scripted in-memory provider for exercising the default + /// `resolve_tracks_into` walk. Node lookups are recorded so tests can + /// assert what was (not) fetched. + #[derive(Debug, Default)] + struct FakeProvider { + nodes: std::collections::HashMap>, + track_paths: Vec, + fetched: std::sync::Mutex>, + } + + impl FakeProvider { + fn node(path: &str, tracks: &[&str], children: &[&str], is_queable: bool) -> LibraryNode { + LibraryNode { + path: path.to_string(), + title: path.to_string(), + children: children + .iter() + .map(|c| LibraryNodeChild::new(c.to_string(), c.to_string(), true)) + .collect(), + parent: None, + tracks: tracks + .iter() + .map(|t| Track { + path: t.to_string(), + artist: "artist".to_string(), + title: t.to_string(), + duration: None, + album: None, + }) + .collect(), + is_queable, + is_creatable: false, + } + } + + fn fetched(&self) -> Vec { + self.fetched.lock().map(|f| f.clone()).unwrap_or_default() + } + } + + #[async_trait] + impl ProviderClient for FakeProvider { + async fn init(_: &str) -> Result { + Ok(Self::default()) + } + fn settings(&self) -> String { + String::new() + } + fn is_track_path(&self, path: &str) -> bool { + self.track_paths.iter().any(|p| p == path) + } + async fn get_urls_for_track(&self, _: &str) -> Result, ProviderError> { + Err(ProviderError::NotSupported) + } + async fn get_metadata_for_track(&self, path: &str) -> Result { + Ok(Track { + path: path.to_string(), + artist: "artist".to_string(), + title: path.to_string(), + duration: None, + album: None, + }) + } + fn get_lib_root(&self) -> LibraryNode { + LibraryNode::new() + } + async fn get_lib_node(&self, path: &str) -> Result { + if let Ok(mut fetched) = self.fetched.lock() { + fetched.push(path.to_string()); + } + self.nodes + .get(path) + .cloned() + .unwrap_or(Err(ProviderError::MalformedPath)) + } + async fn create_lib_node(&self, _: &str, _: &str) -> Result { + Err(ProviderError::NotSupported) + } + async fn rename_lib_node(&self, _: &str, _: &str) -> Result { + Err(ProviderError::NotSupported) + } + async fn delete_lib_node(&self, _: &str) -> Result { + Err(ProviderError::NotSupported) + } + } + + /// Runs the default resolve against the fake and collects the chunks it + /// streamed. The channel is bounded but larger than any test tree, so + /// the resolve never blocks on a full buffer here. + async fn resolve_chunks(provider: &FakeProvider, path: &str) -> Vec> { + let (chunk_tx, chunk_rx) = flume::bounded(32); + provider + .resolve_tracks_into(path, chunk_tx) + .await + .expect("resolve failed"); + chunk_rx + .into_iter() + .map(|chunk| chunk.into_iter().map(|t| t.path).collect()) + .collect() + } + + #[tokio::test] + async fn default_resolve_streams_chunks_per_node_in_preorder() { + let mut provider = FakeProvider::default(); + // artist -> [album1, album2], each album carries tracks; the artist + // node itself has none. Pre-order and *listing order*: album1's + // tracks must come before album2's (the old walk popped LIFO and + // reversed siblings). + provider.nodes.insert( + "/p/artist".into(), + Ok(FakeProvider::node( + "/p/artist", + &[], + &["/p/artist/al1", "/p/artist/al2"], + true, + )), + ); + provider.nodes.insert( + "/p/artist/al1".into(), + Ok(FakeProvider::node( + "/p/artist/al1", + &["/p/artist/al1/t1", "/p/artist/al1/t2"], + &[], + true, + )), + ); + provider.nodes.insert( + "/p/artist/al2".into(), + Ok(FakeProvider::node( + "/p/artist/al2", + &["/p/artist/al2/t3"], + &[], + true, + )), + ); + let chunks = resolve_chunks(&provider, "/p/artist").await; + // One chunk per track-bearing node; the trackless artist node adds + // no empty chunk. + assert_eq!( + chunks, + vec![ + vec!["/p/artist/al1/t1".to_string(), "/p/artist/al1/t2".into()], + vec!["/p/artist/al2/t3".to_string()], + ] + ); + } + + #[tokio::test] + async fn default_resolve_yields_one_chunk_for_track_paths() { + let mut provider = FakeProvider::default(); + provider.track_paths.push("/p/al/t9".into()); + let chunks = resolve_chunks(&provider, "/p/al/t9").await; + assert_eq!(chunks, vec![vec!["/p/al/t9".to_string()]]); + assert!( + provider.fetched().is_empty(), + "a track path must not fetch nodes" + ); + } + + #[tokio::test] + async fn default_resolve_skips_unreadable_nodes_and_unqueable_subtrees() { + let mut provider = FakeProvider::default(); + provider.nodes.insert( + "/p/root".into(), + Ok(FakeProvider::node( + "/p/root", + &["/p/root/t0"], + &["/p/root/broken", "/p/root/private", "/p/root/ok"], + true, + )), + ); + provider + .nodes + .insert("/p/root/broken".into(), Err(ProviderError::FetchError)); + provider.nodes.insert( + "/p/root/private".into(), + Ok(FakeProvider::node( + "/p/root/private", + &["/p/root/private/hidden"], + &[], + false, + )), + ); + provider.nodes.insert( + "/p/root/ok".into(), + Ok(FakeProvider::node( + "/p/root/ok", + &["/p/root/ok/t1"], + &[], + true, + )), + ); + let chunks = resolve_chunks(&provider, "/p/root").await; + // The broken sibling is skipped, the non-queueable subtree + // contributes nothing, the rest still resolves in order. + assert_eq!( + chunks, + vec![ + vec!["/p/root/t0".to_string()], + vec!["/p/root/ok/t1".to_string()], + ] + ); + } + + #[tokio::test] + async fn default_resolve_stops_fetching_once_the_receiver_is_gone() { + let mut provider = FakeProvider::default(); + provider.nodes.insert( + "/p/a".into(), + Ok(FakeProvider::node( + "/p/a", + &["/p/a/t1"], + &["/p/a/b", "/p/a/c"], + true, + )), + ); + provider.nodes.insert( + "/p/a/b".into(), + Ok(FakeProvider::node("/p/a/b", &["/p/a/b/t2"], &[], true)), + ); + provider.nodes.insert( + "/p/a/c".into(), + Ok(FakeProvider::node("/p/a/c", &["/p/a/c/t3"], &[], true)), + ); + let (chunk_tx, chunk_rx) = flume::bounded(32); + drop(chunk_rx); + // A dropped receiver is cancellation, not an error ... + provider + .resolve_tracks_into("/p/a", chunk_tx) + .await + .expect("cancellation must not be an error"); + // ... and the walk stops fetching instead of draining the tree. + assert_eq!(provider.fetched(), vec!["/p/a".to_string()]); + } + + #[tokio::test] + async fn default_resolve_errors_only_for_an_unresolvable_root() { + let provider = FakeProvider::default(); + let (chunk_tx, _chunk_rx) = flume::bounded::>(1); + let result = provider.resolve_tracks_into("/p/unknown", chunk_tx).await; + assert!(result.is_err(), "an unreadable root path is an error"); + } + #[test] fn child_new_defaults_all_capability_flags_off() { // Wire contract: plain children are immutable; providers opt into diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index 9193775..058d9f8 100644 --- a/crabidy-server/src/lib.rs +++ b/crabidy-server/src/lib.rs @@ -1,8 +1,88 @@ use crabidy_core::proto::crabidy::{Queue, Track}; use rand::{rng, seq::SliceRandom}; +use std::sync::{atomic::AtomicBool, Arc}; use std::time::SystemTime; use tracing::{debug, error}; +/// How a pending queue operation places its resolved chunks. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ResolveKind { + /// First chunk replaces the whole queue, later chunks append. + Replace, + /// Every chunk appends at the end. + Append, + /// Chunks insert after the given position, each advancing the cursor so + /// the resolved collection stays contiguous and in order. `Queue` + /// (play-after-current) is an `InsertAfter` at the current position. + InsertAfter(u32), +} + +/// The playback loop's bookkeeping for one in-flight resolve operation. +/// +/// Created when a `Replace`/`Queue`/`Append`/`Insert` command arrives, +/// dropped when its forwarder reports completion or a `Replace`/`Clear` +/// cancels it. Chunk application happens exclusively on the playback loop, +/// which keeps the loop the single writer of queue state. +#[derive(Debug)] +pub struct PendingResolve { + kind: ResolveKind, + /// Tracks applied so far; an op finishing at zero is worth a warning. + applied: usize, + /// Shared with the op's forwarder task: set on cancellation so the + /// forwarder drops the chunk receiver, which stops the provider fetch. + cancelled: Arc, +} + +impl PendingResolve { + pub fn new(kind: ResolveKind) -> Self { + Self { + kind, + applied: 0, + cancelled: Arc::new(AtomicBool::new(false)), + } + } + + /// The cancellation flag to hand to this op's forwarder task. + pub fn cancel_flag(&self) -> Arc { + Arc::clone(&self.cancelled) + } + + /// Marks the op cancelled so its forwarder stops feeding chunks. + pub fn cancel(&self) { + self.cancelled + .store(true, std::sync::atomic::Ordering::Relaxed); + } + + /// Total tracks applied by this op so far. + pub fn applied(&self) -> usize { + self.applied + } + + /// 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. + pub fn apply_chunk(&mut self, queue: &mut QueueManager, tracks: &[Track]) -> Option { + self.applied += tracks.len(); + match self.kind { + ResolveKind::Replace => { + // Only the first chunk replaces; the rest of this op + // extends the fresh queue. + self.kind = ResolveKind::Append; + queue.replace_with_tracks(tracks) + } + ResolveKind::Append => queue.append_tracks(tracks), + ResolveKind::InsertAfter(position) => { + // Advance the cursor so this op's next chunk lands right + // behind this one, keeping the collection contiguous. + // `insert_tracks` clamps positions past the end. + self.kind = ResolveKind::InsertAfter(position + tracks.len() as u32); + queue.insert_tracks(position, tracks) + } + } + } +} + #[derive(Clone, Debug)] pub struct QueueManager { created_at: SystemTime, @@ -24,6 +104,10 @@ impl From for Queue { .as_secs(), current_position: queue_manager.current_position() as u32, tracks: queue_manager.tracks, + // The manager cannot know about in-flight resolves; the + // playback loop's broadcast path sets this from its pending-op + // map (see `Playback::broadcast_queue`). + resolving: false, } } } @@ -390,6 +474,69 @@ mod tests { assert_eq!(q.tracks.last().unwrap().title, "track 2"); } + fn titles(q: &QueueManager) -> Vec { + q.tracks.iter().map(|t| t.title.clone()).collect() + } + + #[test] + fn replace_op_replaces_first_then_appends_and_plays_once() { + let mut q = queue_with(2); + let mut op = PendingResolve::new(ResolveKind::Replace); + let first = op.apply_chunk(&mut q, &[track(10), track(11)]); + // The first chunk resets the queue and names the track to start. + assert_eq!(first.unwrap().title, "track 10"); + assert_eq!(titles(&q), vec!["track 10", "track 11"]); + let second = op.apply_chunk(&mut q, &[track(12)]); + // Later chunks extend the same replace without restarting playback. + assert!(second.is_none()); + assert_eq!(titles(&q), vec!["track 10", "track 11", "track 12"]); + assert_eq!(op.applied(), 3); + } + + #[test] + fn insert_after_op_keeps_chunks_contiguous_and_in_order() { + let mut q = queue_with(3); // playing track 0 + let mut op = PendingResolve::new(ResolveKind::InsertAfter(0)); + assert!(op.apply_chunk(&mut q, &[track(10), track(11)]).is_none()); + assert!(op.apply_chunk(&mut q, &[track(12)]).is_none()); + // Both chunks sit as one contiguous run right after the current + // track, in arrival order — not interleaved with the old tail. + assert_eq!( + titles(&q), + vec!["track 0", "track 10", "track 11", "track 12", "track 1", "track 2"] + ); + } + + #[test] + fn insert_after_op_clamps_past_the_end() { + let mut q = queue_with(1); + let mut op = PendingResolve::new(ResolveKind::InsertAfter(99)); + assert!(op.apply_chunk(&mut q, &[track(10)]).is_none()); + assert!(op.apply_chunk(&mut q, &[track(11)]).is_none()); + assert_eq!(titles(&q), vec!["track 0", "track 10", "track 11"]); + } + + #[test] + fn append_op_starts_playback_only_into_an_empty_queue() { + let mut q = QueueManager::new(); + let mut op = PendingResolve::new(ResolveKind::Append); + let first = op.apply_chunk(&mut q, &[track(10)]); + // Landing in an empty queue makes the track current: play it. + assert_eq!(first.unwrap().title, "track 10"); + assert!(op.apply_chunk(&mut q, &[track(11)]).is_none()); + assert_eq!(titles(&q), vec!["track 10", "track 11"]); + assert_eq!(op.applied(), 2); + } + + #[test] + fn cancel_flag_is_shared_with_the_forwarder() { + let op = PendingResolve::new(ResolveKind::Append); + let flag = op.cancel_flag(); + assert!(!flag.load(std::sync::atomic::Ordering::Relaxed)); + op.cancel(); + assert!(flag.load(std::sync::atomic::Ordering::Relaxed)); + } + #[test] fn shuffle_insert_keeps_order_unique() { let mut q = queue_with(5); diff --git a/crabidy-server/src/main.rs b/crabidy-server/src/main.rs index 77dde95..ed7b05b 100644 --- a/crabidy-server/src/main.rs +++ b/crabidy-server/src/main.rs @@ -151,10 +151,15 @@ pub enum ProviderCommand { result_tx: flume::Sender, ProviderError>>, }, /// Resolves a path into playable tracks: a track path yields that single - /// track, a node path yields all tracks reachable below it. + /// track, a node path yields all tracks reachable below it. Streamed: + /// zero or more in-order chunks arrive on `chunk_tx`, the sender is + /// dropped when resolution finishes, and dropping the receiver cancels + /// it (see `ProviderClient::resolve_tracks_into`). The orchestrator + /// handles this command on a spawned task so its loop stays free for + /// other commands (notably `GetTrackUrls` for the first chunk's track). ResolveTracks { path: String, - result_tx: flume::Sender>, + chunk_tx: flume::Sender>, }, /// Creates a child under a creatable node (see /// `ProviderClient::create_lib_node`); replies with the created node. @@ -228,6 +233,19 @@ pub enum PlaybackCommand { position: u32, paths: Vec, }, + /// Internal: a resolved chunk of tracks for the pending queue operation + /// `op_id`, sent by that operation's forwarder task. Chunks for an + /// unknown (finished or cancelled) op are dropped silently. + ApplyResolvedChunk { + op_id: u64, + tracks: Vec, + }, + /// Internal: the forwarder task for `op_id` has seen the provider drop + /// its chunk sender — the operation is complete and the `resolving` + /// flag clears once no pending operations remain. + ResolveFinished { + op_id: u64, + }, Clear { exclude_current: bool, }, @@ -269,6 +287,8 @@ impl PlaybackCommand { Self::Append { .. } => "append", Self::Remove { .. } => "remove", Self::Insert { .. } => "insert", + Self::ApplyResolvedChunk { .. } => "apply_resolved_chunk", + Self::ResolveFinished { .. } => "resolve_finished", Self::Clear { .. } => "clear", Self::SetCurrent { .. } => "set_current", Self::ToggleShuffle => "toggle_shuffle", diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index 1ac2582..f8997f0 100644 --- a/crabidy-server/src/playback.rs +++ b/crabidy-server/src/playback.rs @@ -2,11 +2,13 @@ use crate::{PlaybackCommand, PlaybackMessage, ProviderCommand, ProviderMessage}; use audio_player::Player; use crabidy_core::proto::crabidy::QueueModifiers; use crabidy_core::proto::crabidy::{ - get_update_stream_response::Update as StreamUpdate, InitResponse, PlayState, QueueTrack, Track, - TrackPosition, + get_update_stream_response::Update as StreamUpdate, InitResponse, PlayState, + Queue as ProtoQueue, QueueTrack, Track, TrackPosition, }; use crabidy_core::ProviderError; -use crabidy_server::QueueManager; +use crabidy_server::{PendingResolve, QueueManager, ResolveKind}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Mutex; use tracing::{debug, debug_span, error, instrument, trace, warn, Instrument}; @@ -17,6 +19,11 @@ pub struct Playback { playback_rx: flume::Receiver, queue: Mutex, state: Mutex, + /// In-flight resolve operations by op id. Non-empty means the broadcast + /// `Queue` snapshots carry `resolving = true`. Only the playback loop + /// touches this map (same single-writer discipline as `queue`). + pending: Mutex>, + next_op_id: AtomicU64, pub player: Player, } @@ -36,6 +43,8 @@ impl Playback { playback_rx, queue, state, + pending: Mutex::new(HashMap::new()), + next_op_id: AtomicU64::new(0), player, } } @@ -78,7 +87,9 @@ impl Playback { *play_state }; InitResponse { - queue: Some(queue.clone().into()), + // Snapshot with `resolving`: a client connecting + // mid-resolve must show the indicator right away. + queue: Some(self.queue_snapshot(&queue)), queue_track: Some(queue_track), play_state: play_state as i32, volume: 0.0, @@ -97,48 +108,34 @@ impl Playback { } PlaybackCommand::Replace { paths } => { - let all_tracks = self.resolve_tracks(paths).await; - debug!(count = all_tracks.len(), "replacing queue"); - let current = { - let Ok(mut queue) = self.queue.lock() else { - error!("queue lock poisoned"); - return; - }; - queue.replace_with_tracks(&all_tracks); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); - queue.current_track() - }; - self.play(current).await; + // A replace obsoletes whatever earlier ops are still + // resolving; their late chunks must not land in the new + // queue. + self.cancel_pending_resolves(); + self.start_resolve(ResolveKind::Replace, paths); } PlaybackCommand::Queue { paths } => { - let all_tracks = self.resolve_tracks(paths).await; - debug!(count = all_tracks.len(), "queueing after current"); - let track = { - let Ok(mut queue) = self.queue.lock() else { + let position = { + let Ok(queue) = self.queue.lock() else { error!("queue lock poisoned"); return; }; - let track = queue.queue_tracks(&all_tracks); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); - track + queue.current_position() as u32 }; - self.play_if_some(track).await; + self.start_resolve(ResolveKind::InsertAfter(position), paths); } PlaybackCommand::Append { paths } => { - let all_tracks = self.resolve_tracks(paths).await; - debug!(count = all_tracks.len(), "appending to queue"); - let track = { - let Ok(mut queue) = self.queue.lock() else { - error!("queue lock poisoned"); - return; - }; - let track = queue.append_tracks(&all_tracks); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); - track - }; - self.play_if_some(track).await; + self.start_resolve(ResolveKind::Append, paths); + } + + PlaybackCommand::ApplyResolvedChunk { op_id, tracks } => { + self.apply_resolved_chunk(op_id, tracks).await; + } + + PlaybackCommand::ResolveFinished { op_id } => { + self.finish_resolve(op_id); } PlaybackCommand::Remove { positions } => { @@ -150,7 +147,7 @@ impl Playback { }; let was_last = queue.is_last_track(); let track = queue.remove_tracks(&positions); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); + self.broadcast_queue(&queue); (track, was_last) }; let state = { @@ -172,29 +169,21 @@ impl Playback { } PlaybackCommand::Insert { position, paths } => { - let all_tracks = self.resolve_tracks(paths).await; - debug!(count = all_tracks.len(), position, "inserting into queue"); - let track = { - let Ok(mut queue) = self.queue.lock() else { - error!("queue lock poisoned"); - return; - }; - let track = queue.insert_tracks(position, &all_tracks); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); - track - }; - self.play_if_some(track).await; + self.start_resolve(ResolveKind::InsertAfter(position), paths); } PlaybackCommand::Clear { exclude_current } => { debug!(exclude_current, "clearing queue"); + // Chunks still resolving would repopulate the queue the + // user just emptied. + self.cancel_pending_resolves(); let should_stop = { let Ok(mut queue) = self.queue.lock() else { error!("queue lock poisoned"); return; }; let should_stop = queue.clear(exclude_current); - self.broadcast(StreamUpdate::Queue(queue.clone().into())); + self.broadcast_queue(&queue); should_stop }; if should_stop { @@ -360,32 +349,173 @@ impl Playback { } } - /// Resolves a mixed list of track and node paths into tracks by asking - /// the provider orchestrator. - async fn resolve_tracks(&self, paths: Vec) -> Vec { - let mut all_tracks = Vec::new(); - for path in paths { - let (result_tx, result_rx) = flume::bounded(1); - let message = ProviderMessage::new(ProviderCommand::ResolveTracks { - path: path.clone(), - result_tx, - }); - if let Err(err) = self.provider_tx.send_async(message).await { - error!("provider channel closed: {err}"); - return all_tracks; - } - match result_rx.recv_async().await { - Ok(tracks) => { - if tracks.is_empty() { + /// Registers a pending resolve operation and spawns its forwarder task. + /// + /// The forwarder resolves `paths` one after the other (preserving the + /// request's path order): for each path it sends + /// `ProviderCommand::ResolveTracks` with a fresh bounded chunk channel + /// and forwards every chunk to the playback loop as + /// `PlaybackCommand::ApplyResolvedChunk`; after the last path it sends + /// `ResolveFinished`. When the op's cancellation flag is set, the + /// forwarder drops the chunk receiver instead — the provider's next + /// send fails and the fetch stops. Queue state is never touched here: + /// mutations happen only when the loop processes the forwarded + /// commands. An immediate `Queue` broadcast (unchanged tracks, + /// `resolving = true`) gives clients instant feedback. + fn start_resolve(&self, kind: ResolveKind, paths: Vec) { + let op = PendingResolve::new(kind); + let cancelled = op.cancel_flag(); + let op_id = self.next_op_id.fetch_add(1, Ordering::Relaxed); + { + let Ok(mut pending) = self.pending.lock() else { + error!("pending ops lock poisoned"); + return; + }; + pending.insert(op_id, op); + } + debug!(op_id, ?paths, "starting queue resolve"); + { + let Ok(queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return; + }; + // Instant feedback: clients see resolving = true before the + // first chunk exists. + self.broadcast_queue(&queue); + } + let provider_tx = self.provider_tx.clone(); + let playback_tx = self.playback_tx.clone(); + tokio::spawn( + async move { + for path in &paths { + if cancelled.load(Ordering::Relaxed) { + break; + } + let (chunk_tx, chunk_rx) = flume::bounded(4); + let message = ProviderMessage::new(ProviderCommand::ResolveTracks { + path: path.clone(), + chunk_tx, + }); + if provider_tx.send_async(message).await.is_err() { + error!("provider channel closed"); + break; + } + let mut forwarded = 0usize; + while let Ok(tracks) = chunk_rx.recv_async().await { + // On cancellation this loop exits and drops + // chunk_rx; the provider's next send fails and the + // fetch stops. + if cancelled.load(Ordering::Relaxed) { + break; + } + forwarded += tracks.len(); + let apply = PlaybackCommand::ApplyResolvedChunk { op_id, tracks }; + if playback_tx + .send_async(PlaybackMessage::new(apply)) + .await + .is_err() + { + error!("playback channel closed"); + return; + } + } + if forwarded == 0 && !cancelled.load(Ordering::Relaxed) { warn!(path, "path resolved to no playable tracks"); } - all_tracks.extend(tracks); } - Err(err) => error!(path, "provider dropped resolve_tracks reply: {err}"), + // Always reported — also for cancelled or empty ops — so + // the pending map can never leak a stuck resolving flag. + let finished = PlaybackCommand::ResolveFinished { op_id }; + let _ = playback_tx.send_async(PlaybackMessage::new(finished)).await; } + .in_current_span(), + ); + } + + /// 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. + async fn apply_resolved_chunk(&self, op_id: u64, tracks: Vec) { + let track = { + let Ok(mut queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return; + }; + let track = { + let Ok(mut pending) = self.pending.lock() else { + error!("pending ops lock poisoned"); + return; + }; + let Some(op) = pending.get_mut(&op_id) else { + trace!(op_id, "dropping chunk for a finished or cancelled op"); + return; + }; + op.apply_chunk(&mut queue, &tracks) + }; + self.broadcast_queue(&queue); + track + }; + self.play_if_some(track).await; + } + + /// Removes the finished op and broadcasts the final `Queue` snapshot + /// (clearing `resolving` once no ops remain). An op already removed by + /// cancellation needs no broadcast — the cancelling command mutates + /// the queue and broadcasts itself. + fn finish_resolve(&self, op_id: u64) { + let removed = { + let Ok(mut pending) = self.pending.lock() else { + error!("pending ops lock poisoned"); + return; + }; + pending.remove(&op_id) + }; + let Some(op) = removed else { + trace!(op_id, "resolve finished for a cancelled op"); + return; + }; + debug!(op_id, tracks = op.applied(), "queue resolve finished"); + let Ok(queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return; + }; + self.broadcast_queue(&queue); + } + + /// Cancels every in-flight resolve op (used by `Replace` and `Clear`). + /// The forwarders see the flag, drop their chunk receivers (stopping + /// the fetches) and still report `ResolveFinished`, which is dropped as + /// unknown here. + fn cancel_pending_resolves(&self) { + let Ok(mut pending) = self.pending.lock() else { + error!("pending ops lock poisoned"); + return; + }; + for (op_id, op) in pending.drain() { + debug!(op_id, "cancelling in-flight resolve"); + op.cancel(); } - trace!(count = all_tracks.len(), "resolved tracks"); - all_tracks + } + + /// A wire snapshot of the queue with the `resolving` flag set from the + /// pending-op map. Callers must not hold the `pending` lock (`queue` is + /// fine — the lock order is queue, then pending). + fn queue_snapshot(&self, queue: &QueueManager) -> ProtoQueue { + let resolving = self + .pending + .lock() + .map(|pending| !pending.is_empty()) + .unwrap_or(false); + let mut snapshot: ProtoQueue = queue.clone().into(); + snapshot.resolving = resolving; + snapshot + } + + /// Broadcasts the current queue snapshot. All queue broadcasts go + /// through here so the `resolving` flag can never be forgotten. + fn broadcast_queue(&self, queue: &QueueManager) { + self.broadcast(StreamUpdate::Queue(self.queue_snapshot(queue))); } #[instrument(skip(self))] diff --git a/crabidy-server/src/provider.rs b/crabidy-server/src/provider.rs index c124040..4bde849 100644 --- a/crabidy-server/src/provider.rs +++ b/crabidy-server/src/provider.rs @@ -17,16 +17,22 @@ pub struct ProviderOrchestrator { impl ProviderOrchestrator { pub fn run(self) { tokio::spawn(async move { - while let Ok(ProviderMessage { span, command }) = self.provider_rx.recv_async().await { + // Behind an Arc so long-running resolves can be spawned onto + // their own tasks while the loop keeps serving commands. + let this = Arc::new(self); + while let Ok(ProviderMessage { span, command }) = this.provider_rx.recv_async().await { let handler_span = debug_span!(parent: &span, "provider_command", command = command.name()); - self.handle_command(command).instrument(handler_span).await; + Arc::clone(&this) + .handle_command(command) + .instrument(handler_span) + .await; } warn!("provider message channel closed, loop exiting"); }); } - async fn handle_command(&self, command: ProviderCommand) { + async fn handle_command(self: Arc, command: ProviderCommand) { match command { ProviderCommand::GetLibraryNode { path, result_tx } => { let result = self.get_lib_node(&path).await; @@ -40,11 +46,20 @@ impl ProviderOrchestrator { error!("failed to send get_track_urls result: {err}"); } } - ProviderCommand::ResolveTracks { path, result_tx } => { - let result = self.resolve_tracks(&path).await; - if let Err(err) = result_tx.send_async(result).await { - error!("failed to send resolve_tracks result: {err}"); - } + ProviderCommand::ResolveTracks { path, chunk_tx } => { + // Spawned: a large resolve must not block this loop, or the + // playback side deadlocks waiting for `GetTrackUrls` while + // chunks back up. Dropping `chunk_tx` at the end of the + // task is the completion signal; there is no reply channel. + let this = Arc::clone(&self); + tokio::spawn( + async move { + if let Err(err) = this.resolve_tracks_into(&path, chunk_tx).await { + warn!(path, "resolve produced no tracks: {err}"); + } + } + .in_current_span(), + ); } ProviderCommand::CreateLibraryNode { parent_path, @@ -74,39 +89,6 @@ impl ProviderOrchestrator { } } } - - /// Resolves a path into playable tracks. A track path resolves to that - /// single track; a node path is flattened by walking its queueable - /// descendants. - #[instrument(skip(self))] - async fn resolve_tracks(&self, path: &str) -> Vec { - if self.is_track_path(path) { - return match self.get_metadata_for_track(path).await { - Ok(track) => vec![track], - Err(err) => { - warn!(path, "failed to resolve track: {err}"); - Vec::new() - } - }; - } - let mut tracks = Vec::new(); - let mut nodes_to_go = vec![path.to_string()]; - while let Some(node_path) = nodes_to_go.pop() { - let node = match self.get_lib_node(&node_path).await { - Ok(node) => node, - Err(err) => { - warn!(node = node_path, "skipping unreadable node: {err}"); - continue; - } - }; - if node.is_queable { - tracks.extend(node.tracks); - nodes_to_go.extend(node.children.into_iter().map(|c| c.path)) - } - } - debug!(count = tracks.len(), "resolved path into tracks"); - tracks - } } #[async_trait] @@ -226,6 +208,21 @@ impl ProviderClient for ProviderOrchestrator { Err(ProviderError::NotSupported) } + /// Routes to the provider that owns the path. The synthetic root is not + /// queueable, so only provider-owned paths can resolve. + #[instrument(skip(self, chunk_tx))] + async fn resolve_tracks_into( + &self, + path: &str, + chunk_tx: flume::Sender>, + ) -> Result<(), ProviderError> { + if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") { + return self.tidal_client.resolve_tracks_into(path, chunk_tx).await; + } + warn!(path, "no provider owns this path"); + Err(ProviderError::MalformedPath) + } + /// Routes to the provider that owns the path. The synthetic root's own /// children are fixed and never deletable. #[instrument(skip(self))] diff --git a/plan/progressive-queueing.md b/plan/progressive-queueing.md new file mode 100644 index 0000000..2369231 --- /dev/null +++ b/plan/progressive-queueing.md @@ -0,0 +1,94 @@ +# Plan — progressive queueing + +Ordered tasks for the `implement` stage. Inputs: +`architecture/progressive-queueing.md`, the workspace stubs, gates in +`quality/progressive-queueing.md`. Tests: +`devenv shell -- cargo test --workspace` (13 failing at plan time = the +target: 5 in crabidy-core, 5 in crabidy-server, 2 in cbd-tui, 1 in tidaldy). +Use the session-local `CARGO_TARGET_DIR` if `target/` contains artifacts +owned by the repo owner. + +## 1. Default chunked resolve (crabidy-core) + +- [x] Implement the `resolve_tracks_into` default body: track path → + single chunk from `get_metadata_for_track` (failure → warn, `Ok`, no + chunk — matches today's skip); node path → depth-first **pre-order** + walk (stack of paths, children pushed in reverse so siblings pop in + listing order), per node: skip unreadable with a warning (root + failure → `Err`), skip non-queueable, send `tracks` as one chunk when + non-empty; a failed send (receiver gone) → return `Ok` immediately. + **Verify**: all 5 `default_resolve_*` tests pass. + +## 2. PendingResolve state machine (crabidy-server lib) + +- [x] Implement `new`/`cancel_flag`/`cancel`/`applied`/`apply_chunk`. + `apply_chunk`: `Replace` → `replace_with_tracks` for the first chunk + (returns current track), then mutate kind to `Append`; `Append` → + `append_tracks`; `InsertAfter(pos)` → `insert_tracks(pos)` then + advance the cursor by `tracks.len()` (clamping is `insert_tracks`' + job); count `applied`. **Verify**: the 5 op tests pass + (`replace_op_*`, `insert_after_op_*` ×2, `append_op_*`, + `cancel_flag_*`). + +## 3. Playback loop wiring (crabidy-server) + +- [x] Implement `broadcast_queue` (snapshot → `Queue`, set `resolving` + from the pending map, broadcast) and route the remaining direct + `StreamUpdate::Queue(...)` broadcasts in `handle_command` through it. + **Verify**: gate "resolving is true iff at least one op is pending". +- [x] Implement `start_resolve`: allocate op id, insert `PendingResolve`, + immediate `broadcast_queue`, spawn the forwarder + (`tokio::spawn` + `in_current_span`): per path in order — check the + cancel flag (set → stop), open `flume::bounded(4)` chunk channel, + send `ProviderCommand::ResolveTracks`, forward each chunk as + `ApplyResolvedChunk` (checking the cancel flag between chunks, drop + the receiver on cancel); after all paths (or on any exit path) send + `ResolveFinished`. Track the op's paths for the zero-track warning. + **Verify**: gates "immediate feedback", "cancellation propagates", + "no panics"; `cargo check`. +- [x] Implement `apply_resolved_chunk` (look up op — unknown id: drop + silently; lock queue, `apply_chunk`, `broadcast_queue`, then + `play_if_some` outside the locks), `finish_resolve` (remove op, warn + when `applied() == 0` with the op's paths, `broadcast_queue`), and + `cancel_pending_resolves` (cancel + clear map; no broadcast needed — + the caller mutates and broadcasts next). **Verify**: gates "playback + starts with the first chunk", "late chunks dropped"; existing + playback tests still green. + +## 4. tidaldy page-streamed resolve + +- [x] Implement `make_paginated_request_into` (same loop as + `make_paginated_request`, `sink(page.items)` per page, stop on + `false`). **Verify**: gate "token/refresh path reused"; `cargo check`. +- [x] Implement the `resolve_tracks_into` override: `parse_path` first + (foreign → `MalformedPath` before any I/O); track paths → one + metadata chunk; pre-order worklist walk where `Playlist` and `Album` + nodes stream `tracks/items` pages as chunks (`Track::to_proto` per + page, no playlist-metadata fetch) and all other node kinds fall back + to `get_lib_node` (tracks as one chunk when queueable, children onto + the worklist in listing order). Failed sends → stop, `Ok`. + **Verify**: `resolve_rejects_foreign_paths_before_any_network_call` + passes; gate "order preserved end-to-end". + +## 5. TUI indicator (cbd-tui) + +- [x] Implement `Queue::loading_dots` (1 + (elapsed_ms / 400) % 3 dots) + and render the pseudo-item: when `resolving`, push one extra + `ListItem` (dots, `COLOR_SECONDARY`) after the track rows, computed + from a monotonic clock at render time; `self.list` stays untouched. + **Verify**: `loading_dots_cycle_one_to_three`, + `resolving_queue_renders_trailing_dots_item` pass; the three guard + tests stay green. + +## 6. End-to-end + gates sweep + +- [x] Exercise the full path against the live API if the local tidal + config is available (temporary ignored probe: resolve a multi-album + artist through the orchestrator, assert multiple chunks arrive, in + listing order, first chunk before the walk completes; drop the + receiver mid-stream and confirm fetching stops). Note the outcome in + the summary; if unavailable, rely on the offline tests and say so. +- [x] Sweep: no `todo!()` left (workspace grep), fmt + clippy + tests + green, every `quality/progressive-queueing.md` box checked, docs + updated where behavior shifted. Append the outcome + deviations to + `plan/summary.md`. diff --git a/plan/summary.md b/plan/summary.md index f74cb74..6ba57e9 100644 --- a/plan/summary.md +++ b/plan/summary.md @@ -1,5 +1,60 @@ # Implementation summaries +## progressive-queueing (2026-07-21) + +Built per `plan/progressive-queueing.md`: queueing a large nested collection +now fills the queue progressively instead of freezing until the full +resolve. `ProviderClient` gained `resolve_tracks_into` (chunk-streaming over +a bounded channel; sender-drop = done, receiver-drop = cancel) with a +default pre-order walk; tidaldy overrides it so playlists and albums emit +one chunk per fetched 50-track page. The playback loop registers a pending +op per queue command, spawns a forwarder, applies chunks on the loop +(single-writer preserved), broadcasts after every chunk, and starts playback +with the first chunk that makes a track current. `Replace`/`Clear` cancel +in-flight resolves down to the HTTP fetch. The wire gained +`Queue.resolving = 4` (additive); the TUI renders an animated one-to-three +dots pseudo-item after the last queue row while it is set. All 76 workspace +tests green; every gate in `quality/progressive-queueing.md` checked. +Verified against the live Tidal API with a temporary ignored probe (removed +after passing): a 71-album artist streamed its first 19-track chunk (first +album, listing order) while the walk was still running, and dropping the +receiver mid-stream ended the resolve cleanly in 1.8 s instead of draining +the discography. + +The whole feature ran autonomously per standing instruction; decisions are +recorded in `architecture/progressive-queueing.md` (options + rationale). + +### Deviations from plan / architecture (progressive-queueing) + +- **`make_paginated_request_into` became `stream_track_pages_into`**: the + planned generic `AsyncFnMut` page sink dies on a rustc + "implementation of `Send` is not general enough" limitation inside + `async_trait` methods. The concrete method (fixed `Track` item type, + proto mapping and channel send inlined) sidesteps it with the same + page-loop and cancellation semantics. +- **Zero-track warning lives in the forwarder, not `finish_resolve`**: the + forwarder sees each path and its chunk count, so the existing per-path + "resolved to no playable tracks" message survives verbatim; the planned + op-level warning would have had to smuggle paths into `PendingResolve`. +- **Fixed alongside (user-reported)**: Enter on a non-queueable library + item used to blank the queue while audio kept playing. Two causes, both + fixed: `Library::get_selected` now gates the bare selection on + `is_queable` (marks were already gated), and a replace that resolves to + zero tracks no longer touches the queue at all — structurally, since the + queue is only mutated by arriving chunks. Regression test + `queue_ops_ignore_non_queueable_selections`. +- **`Queue` (play-next) captures the current position when the command + arrives**, not per chunk: chunks of one op stay contiguous after the + track the user was on when they pressed the key, even if playback + advances mid-resolve. +- **Live probe scope**: the first full-discography probe was cut short + (hundreds of album fetches for no extra signal) and replaced by a + receive-two-chunks-then-cancel probe — which also exercises mid-stream + cancellation against the live API, which the drain-everything version + could not. +- **Environment note**: builds/tests again ran with a session-local + `CARGO_TARGET_DIR` (owner-built artifacts in `target/`); no repo change. + ## node-editing (2026-07-20) Built per `plan/node-editing.md`: search-term nodes (created via `%`) are now diff --git a/quality/progressive-queueing.md b/quality/progressive-queueing.md new file mode 100644 index 0000000..1c41895 --- /dev/null +++ b/quality/progressive-queueing.md @@ -0,0 +1,95 @@ +# Quality gates — progressive queueing + +Criteria for `architecture/progressive-queueing.md`. Automatic tests live in +`crabidy-core/src/lib.rs` (default `resolve_tracks_into` walk, fake +provider), `crabidy-server/src/lib.rs` (`PendingResolve` chunk application), +`tidaldy/src/lib.rs` (offline path validation), and +`cbd-tui/src/app/queue.rs` (indicator rendering). They fail while the stubs +are unimplemented — that is the target state for `implement`. + +Run: `devenv shell -- cargo test --workspace` (session-local +`CARGO_TARGET_DIR` when `target/` is owner-built). + +## Channel and concurrency gates (verify by reading) + +- [x] **All new channels are bounded.** The chunk channel is small (≤ 8 + chunks); forwarder → playback reuses the existing `bounded(64)`. No + `unbounded()` anywhere in the feature. +- [x] **No lock is held across an `await`.** The `pending` map and `queue` + mutexes are locked, used, and released inside synchronous blocks only + — same discipline as the existing playback handlers. +- [x] **The playback loop stays the single writer of queue state.** Neither + the forwarder task nor the provider resolve task touches + `QueueManager` or broadcasts; they only send commands. +- [x] **The provider loop never blocks on a resolve.** The `ResolveTracks` + arm spawns; `GetTrackUrls` for the first chunk's track must be + servable while a resolve is still streaming (deadlock check: playback + awaiting `GetTrackUrls` + provider awaiting a chunk send must be + impossible). +- [x] **Channel semantics are documented where they are the contract**: + sender-drop = finished, receiver-drop = cancel, on both + `ProviderClient::resolve_tracks_into` and + `ProviderCommand::ResolveTracks`. + +## Cancellation gates + +- [x] **`Replace` and `Clear` cancel every pending op** before mutating the + queue; `Queue`/`Append`/`Insert` cancel nothing. +- [x] **Cancellation propagates to the network fetch**: the forwarder drops + the chunk receiver, and both resolve implementations (default walk and + tidaldy override) treat a failed chunk send as "stop fetching, return + Ok" — verified by test for the default, by reading for tidaldy's page + loop. +- [x] **Late chunks for finished/cancelled ops are dropped silently** (no + error, no queue mutation). + +## Behavior gates + +- [x] **Immediate feedback**: accepting a resolve op broadcasts a `Queue` + snapshot with `resolving = true` before the first chunk arrives. +- [x] **`resolving` is true iff at least one op is pending**, and every + queue broadcast from the playback loop goes through the one helper + that sets it (grep: no direct `StreamUpdate::Queue(` construction in + the op/chunk paths besides `broadcast_queue`). +- [x] **Playback starts with the first chunk** that makes a track current + (replace, or landing in an empty queue) and never restarts for later + chunks of the same op — covered by the `PendingResolve` tests, plus + reading the `apply_resolved_chunk` → `play_if_some` wiring. +- [x] **Order preserved end-to-end**: pages arrive in collection order, + nodes in listing order (pre-order walk — the old LIFO reversal must + not reappear), multi-path requests resolve sequentially in request + order, and `InsertAfter` keeps the whole op contiguous. +- [x] **An op that finishes with zero tracks logs the existing + "resolved to no playable tracks" warning** with enough context (paths). + +## Error and robustness gates + +- [x] **No panics on provider failure, in any spawned task.** Unreadable + nodes are skipped with a warning; a dead provider channel ends the op + (with `ResolveFinished` still sent, so `resolving` clears) instead of + leaving a stuck indicator. +- [x] **Tracing spans survive the spawns**: forwarder and provider resolve + tasks attribute their events to the originating request's span + (`ProviderMessage::new` capture + `in_current_span`/explicit parent). +- [x] **tidaldy page loop honors the existing token/refresh path** (reuses + `make_request`) and adds no new retry logic. + +## Wire and UI gates + +- [x] **Proto change is additive only**: `Queue.resolving = 4`, no rpc shape + changes; field documented in the proto. +- [x] **The TUI pseudo-item is render-only**: never in the list model, + unreachable by selection/removal/navigation (`get_size` unchanged) — + covered by tests. +- [x] **Indicator style matches the pane** (`COLOR_SECONDARY`, inside the + queue block, after the last track). +- [x] **Old client / new server and new client / old server both degrade + cleanly** (flag ignored / never set — reasoning check against the + generated proto defaults). + +## Documentation gates + +- [x] **`ProviderClient::resolve_tracks_into` docs state the full channel + contract** (order, completion, cancellation, error policy). +- [x] **Deviations from the architecture are recorded** in + `plan/summary.md` under this feature. diff --git a/tidaldy/Cargo.toml b/tidaldy/Cargo.toml index 857bc63..3176421 100644 --- a/tidaldy/Cargo.toml +++ b/tidaldy/Cargo.toml @@ -8,6 +8,7 @@ async-trait.workspace = true base64.workspace = true chrono.workspace = true crabidy-core.workspace = true +flume.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/tidaldy/src/lib.rs b/tidaldy/src/lib.rs index f5462b1..c3e6211 100644 --- a/tidaldy/src/lib.rs +++ b/tidaldy/src/lib.rs @@ -354,6 +354,79 @@ impl crabidy_core::ProviderClient for Client { let parent = crabidy_core::parent_path(path).unwrap_or(PROVIDER_ROOT); self.get_lib_node(parent).await } + + /// Streams tracks chunk-wise (see the trait docs for the channel + /// contract). Overridden so that the paginated collections — playlists + /// and albums — emit one chunk per fetched page (50 tracks) instead of + /// paginating to exhaustion inside a single node fetch; a large playlist + /// starts playing after its first page. All other paths follow the + /// generic pre-order walk, in listing order. + #[instrument(skip(self, chunk_tx))] + async fn resolve_tracks_into( + &self, + path: &str, + chunk_tx: flume::Sender>, + ) -> Result<(), crabidy_core::ProviderError> { + // Validate before any I/O; foreign paths never touch the network. + if let TidalPath::PlaylistTrack { .. } + | TidalPath::AlbumTrack { .. } + | TidalPath::SearchTrack { .. } = parse_path(path)? + { + match self.get_metadata_for_track(path).await { + Ok(track) => { + let _ = chunk_tx.send_async(vec![track]).await; + } + Err(err) => warn!(path, "failed to resolve track: {err}"), + } + return Ok(()); + } + // Depth-first pre-order (children pushed reversed, worklist pops + // from the back) so tracks arrive in listing order. + let mut nodes_to_go = vec![path.to_string()]; + let mut at_root = true; + while let Some(node_path) = nodes_to_go.pop() { + let root = std::mem::take(&mut at_root); + // The track-bearing collections stream one chunk per fetched + // page instead of paginating to exhaustion first. + let paged_uri = match parse_path(&node_path) { + Ok(TidalPath::Playlist(playlist_id)) => { + Some(format!("playlists/{playlist_id}/tracks")) + } + Ok(TidalPath::Album { album, .. }) => Some(format!("albums/{album}/tracks")), + _ => None, + }; + if let Some(uri) = paged_uri { + match self + .stream_track_pages_into(&uri, &node_path, &chunk_tx) + .await + { + Ok(true) => {} + // Receiver gone: the consumer cancelled, stop fetching. + Ok(false) => return Ok(()), + Err(err) if root => return Err(err.into()), + Err(err) => warn!(node = node_path, "skipping unreadable node: {err}"), + } + continue; + } + let node = match self.get_lib_node(&node_path).await { + Ok(node) => node, + Err(err) if root => return Err(err), + Err(err) => { + warn!(node = node_path, "skipping unreadable node: {err}"); + continue; + } + }; + if !node.is_queable { + continue; + } + if !node.tracks.is_empty() && chunk_tx.send_async(node.tracks).await.is_err() { + // Receiver gone: the consumer cancelled, stop fetching. + return Ok(()); + } + nodes_to_go.extend(node.children.into_iter().rev().map(|c| c.path)); + } + Ok(()) + } } /// The root of this provider in the global library tree. @@ -613,6 +686,40 @@ impl Client { Ok(items) } + /// Like [`make_paginated_request`](Self::make_paginated_request) for a + /// paginated track collection, but streams each fetched page into + /// `chunk_tx` (mapped to proto tracks under `parent_path`) as soon as + /// it arrives instead of collecting to exhaustion first. Returns + /// `Ok(false)` when the receiver disappeared before the collection was + /// exhausted — cancellation, not an error — and `Ok(true)` on normal + /// completion. Reuses [`make_request`](Self::make_request), so token + /// refresh and retry behave exactly like every other fetch. + async fn stream_track_pages_into( + &self, + uri: &str, + parent_path: &str, + chunk_tx: &flume::Sender>, + ) -> Result { + let limit: usize = 50; + let mut offset: usize = 0; + loop { + let params: Vec<(&str, String)> = + vec![("limit", limit.to_string()), ("offset", offset.to_string())]; + let page: Page = self.make_request(uri, Some(¶ms)).await?; + let fetched = page.items.len(); + offset += fetched; + let exhausted = fetched == 0 || offset >= page.total_number_of_items; + let chunk: Vec<_> = page.items.iter().map(|t| t.to_proto(parent_path)).collect(); + if !chunk.is_empty() && chunk_tx.send_async(chunk).await.is_err() { + debug!(uri, offset, "paginated fetch cancelled by its consumer"); + return Ok(false); + } + if exhausted { + return Ok(true); + } + } + } + #[instrument(skip(self))] pub async fn make_explorer_request( &self, @@ -1167,6 +1274,18 @@ mod tests { Client::new(config::Settings::default()).expect("offline client") } + #[tokio::test] + async fn resolve_rejects_foreign_paths_before_any_network_call() { + use crabidy_core::ProviderError; + let client = offline_client(); + for path in ["/spotify/x", "/nope", ""] { + let (chunk_tx, chunk_rx) = flume::bounded(1); + let result = client.resolve_tracks_into(path, chunk_tx).await; + assert_eq!(result, Err(ProviderError::MalformedPath), "path {path:?}"); + assert!(chunk_rx.try_recv().is_err(), "no chunks for {path:?}"); + } + } + /// Dumps the raw search payloads to re-verify the models if the API /// drifts (last verified 2026-07-20, shapes matched `Page` + the /// existing Track/Artist/Album models).