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 <noreply@anthropic.com>
This commit is contained in:
Test User
parent
333c6e040a
commit
ccab43a133
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<Vec<Track>>,
|
||||
) -> 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<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.
|
||||
|
||||
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<AtomicBool>` 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).
|
||||
|
|
@ -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() {
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<UiItem>,
|
||||
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<MessageFromUi>,
|
||||
}
|
||||
|
||||
|
|
@ -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<ListItem> = self
|
||||
let mut queue_items: Vec<ListItem> = 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::time::Instant> =
|
||||
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<String> {
|
||||
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::<String>()
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The row content inside the borders, trimmed.
|
||||
fn inner_rows(queue: &mut Queue) -> Vec<String> {
|
||||
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), ".");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<LibraryNode, ProviderError>;
|
||||
/// 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<Vec<Track>>,
|
||||
) -> 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<String, Result<LibraryNode, ProviderError>>,
|
||||
track_paths: Vec<String>,
|
||||
fetched: std::sync::Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
self.fetched.lock().map(|f| f.clone()).unwrap_or_default()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ProviderClient for FakeProvider {
|
||||
async fn init(_: &str) -> Result<Self, ProviderError> {
|
||||
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<Vec<String>, ProviderError> {
|
||||
Err(ProviderError::NotSupported)
|
||||
}
|
||||
async fn get_metadata_for_track(&self, path: &str) -> Result<Track, ProviderError> {
|
||||
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<LibraryNode, ProviderError> {
|
||||
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<LibraryNode, ProviderError> {
|
||||
Err(ProviderError::NotSupported)
|
||||
}
|
||||
async fn rename_lib_node(&self, _: &str, _: &str) -> Result<LibraryNode, ProviderError> {
|
||||
Err(ProviderError::NotSupported)
|
||||
}
|
||||
async fn delete_lib_node(&self, _: &str) -> Result<LibraryNode, ProviderError> {
|
||||
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<Vec<String>> {
|
||||
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::<Vec<Track>>(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
|
||||
|
|
|
|||
|
|
@ -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<AtomicBool>,
|
||||
}
|
||||
|
||||
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<AtomicBool> {
|
||||
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<Track> {
|
||||
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<QueueManager> 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<String> {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -151,10 +151,15 @@ pub enum ProviderCommand {
|
|||
result_tx: flume::Sender<Result<Vec<String>, 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<Vec<Track>>,
|
||||
chunk_tx: flume::Sender<Vec<Track>>,
|
||||
},
|
||||
/// 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<String>,
|
||||
},
|
||||
/// 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<Track>,
|
||||
},
|
||||
/// 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",
|
||||
|
|
|
|||
|
|
@ -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<PlaybackMessage>,
|
||||
queue: Mutex<QueueManager>,
|
||||
state: Mutex<PlayState>,
|
||||
/// 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<HashMap<u64, PendingResolve>>,
|
||||
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<String>) -> Vec<Track> {
|
||||
let mut all_tracks = Vec::new();
|
||||
for path in paths {
|
||||
let (result_tx, result_rx) = flume::bounded(1);
|
||||
/// 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<String>) {
|
||||
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(),
|
||||
result_tx,
|
||||
chunk_tx,
|
||||
});
|
||||
if let Err(err) = self.provider_tx.send_async(message).await {
|
||||
error!("provider channel closed: {err}");
|
||||
return all_tracks;
|
||||
if provider_tx.send_async(message).await.is_err() {
|
||||
error!("provider channel closed");
|
||||
break;
|
||||
}
|
||||
match result_rx.recv_async().await {
|
||||
Ok(tracks) => {
|
||||
if tracks.is_empty() {
|
||||
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<Track>) {
|
||||
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))]
|
||||
|
|
|
|||
|
|
@ -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<Self>, command: ProviderCommand) {
|
||||
match command {
|
||||
ProviderCommand::GetLibraryNode { path, result_tx } => {
|
||||
let result = self.get_lib_node(&path).await;
|
||||
|
|
@ -40,12 +46,21 @@ 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,
|
||||
title,
|
||||
|
|
@ -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<Track> {
|
||||
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<Vec<Track>>,
|
||||
) -> 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))]
|
||||
|
|
|
|||
|
|
@ -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`.
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<Vec<crabidy_core::proto::crabidy::Track>>,
|
||||
) -> 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<Vec<crabidy_core::proto::crabidy::Track>>,
|
||||
) -> Result<bool, ClientError> {
|
||||
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<Track> = 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<T>` + the
|
||||
/// existing Track/Artist/Album models).
|
||||
|
|
|
|||
Loading…
Reference in New Issue