diff --git a/architecture/captures.md b/architecture/captures.md index 6e27357..74fba2a 100644 --- a/architecture/captures.md +++ b/architecture/captures.md @@ -1,5 +1,11 @@ # Captures (downloaded subtrees) +> **Partially superseded** by `incremental-captures.md`: download captures +> are now incremental (no tmp-and-swap, re-capturing a name resumes it), +> uncapturable tracks are recorded as *skipped* tomls instead of being +> omitted, and the capture RPC streams progress. Bookmarks keep the +> tmp-and-swap described here. + ## Context and problem statement Bookmarks (`w`) mirror a library subtree as **link** files — replaying a diff --git a/architecture/incremental-captures.md b/architecture/incremental-captures.md new file mode 100644 index 0000000..05db3ff --- /dev/null +++ b/architecture/incremental-captures.md @@ -0,0 +1,193 @@ +# Incremental captures, skipped tracks, and capture progress + +## Context and problem statement + +Download captures (`W`, architecture/captures.md) are all-or-nothing: the +whole subtree is built in a hidden temp folder and swapped into place; any +failure destroys everything downloaded so far. For a large node that means +hours of downloading can evaporate on one bad track, and re-running restarts +from zero. Mixed-provider sources (queues, bookmarks) silently *omit* +uncapturable tracks, so the capture's track list quietly diverges from the +source. And while a capture runs, the user sees nothing — worse, the TUI's +poll loop awaits the capture RPC, so the client is effectively frozen until +the capture finishes. + +This design makes download captures **incremental and resumable**, records +uncapturable tracks as a first-class **skipped** playable, streams **capture +progress** to clients, and warns about long captures up front. A small, +unrelated fix rides along: colored library items (editable/creatable/marked) +are unreadable under the focused selection bar (D7). + +## Assumptions + +- "Capture" here means the *download* capture (`W`). Bookmark captures (`w`) + stay atomic tmp-and-swap: they are cheap, and "overwrite = refresh" is the + right semantic for links. They do adopt the skipped playable for skipped + source tracks (D1) and report progress (D5). +- Resuming keys on the **name**: capturing into an existing capture name + merges into that folder. Entry identity is the deterministic toml file name + (`NNNN .cbd-track.toml`), so resuming assumes the source keeps its + order — appending to a queue is fine, reordering it re-captures under new + names and leaves stale files behind (the user can delete the capture and + start over). Accepted. +- One capture per name at a time is the user's responsibility (same as the + old racing-tmp behavior); concurrent same-name captures interleave per + file, last writer wins. Accepted. + +## D1 — A `skipped` playable + +Track files get a fourth playable: `[playable] skipped = true`, validated +with the same exactly-one cardinality as `file`/`url`/`link` +(`skipped = false` counts as unset and is rejected). Semantics: *this +position in the tree is a real track whose audio could not be captured*. + +- `fsdy::Playable::Skipped`; `TrackFile::from_track_skipped(track)` builds + one from a wire track. +- Wire: `Track.is_skipped` (proto field 6). `TrackFile::to_track` sets it; + the track's `path` stays the lib path (like `file`), there is nothing to + route to. +- `TrackFile::from_track` (queue persistence, bookmarks) writes a skipped + playable when the source track `is_skipped` — skipped-ness survives queue + persistence and bookmark round trips instead of degrading into a dead + link. +- `get_urls_for_track` on a skipped file returns `ProviderError::FetchError` + (playback never asks, see D4; direct callers get a normal typed error). + +Alternative considered: model skipped-ness as *absence* (keep omitting the +track) plus a client-side diff against the source. Rejected — the source may +be gone tomorrow; the capture itself must record the gap. + +## D2 — Incremental download captures + +`capture_into` splits into two phases: + +1. **Enumerate**: the existing iterative pre-order walk collects every + directory and track (with its listing index) first, enforcing + `max_dirs`/`max_tracks`. This makes the total known before the first + download — progress can be a real ratio — and costs only metadata calls. +2. **Fetch**: process the collected tracks in order, feeding progress after + each one. + +The sink decides the write mode: + +- `Sink::Link` (bookmarks): unchanged tmp-and-swap into `.tmp-<name>`, + all-or-nothing. +- `Sink::Download` (captures): writes **directly** into `dir/<name>/`, + creating directories as needed, never deleting existing content. Per + track, in order: + - The target toml exists, parses, and its playable is *not* skipped, and + (for a `file` playable) the referenced audio exists → **reuse** (counts + as done, no download). A `url`/`link` playable also counts as satisfied + — only this store writes here, but hand-edited files should not be + clobbered. + - Otherwise the track is (re)captured: a source that cannot be captured — + the track is itself skipped, its stream fails to resolve, resolves to + nothing, or to a non-http(s) target — writes a **skipped toml** + (`from_track_skipped`) and counts as skipped. This replaces the old + silent omission. + - A real download failure (HTTP status, transport, timeout, byte budget) + **aborts the run but keeps everything written so far** — re-running the + same name resumes exactly where it stopped, re-attempting skipped and + missing entries only. + +Audio is still written before its toml, so a crash mid-download leaves a +toml-less audio file that the resume simply re-downloads (truncating on +create). The byte budget counts only bytes downloaded *this run*, so resuming +a large capture is never starved by what is already on disk. + +```d2 +direction: right +walk: capture_into { + enumerate: "phase 1: enumerate\n(dirs + tracks, caps)" + fetch: "phase 2: fetch\n(per track, in order)" + enumerate -> fetch: "total known" +} +walk.fetch -> reuse: "toml ok + audio present" +walk.fetch -> skipped: "source uncapturable\n→ skipped = true toml" +walk.fetch -> download: "download + toml" +walk.fetch -> abort: "download failure\n(keeps progress)" +``` + +## D3 — Skipped tracks in the queue + +Skipped tracks queue like any other (the user sees the gap instead of a +silently shorter queue). The TUI renders them red; playback skips them. + +`Playback::play` already loops past tracks whose URLs fail to resolve. It now +additionally: + +- skips `is_skipped` tracks **without a provider round trip**, and +- bounds the whole skip loop by the queue length at entry — an all-skipped + queue with repeat on used to be an infinite provider-hammering spin; now + it stops the player with a warning after one full pass. + +## D4 — Capture progress on the update stream + +New stream update (proto): + +```proto +message CaptureProgress { + string name = 1; // capture / bookmark name + bool download = 2; // W capture vs w bookmark + uint32 tracks_done = 3; // settled: reused + downloaded + linked + + // skipped — reaches tracks_total on success + uint32 tracks_total = 4; // known after enumeration (0 until then) + uint32 tracks_skipped = 5; // of those, skipped tomls written this run + bool finished = 6; + string error = 7; // set iff finished with a failure +} +``` + +`CaptureLibraryNode` now returns once the capture is **accepted**: the +provider validates the name, the store, and the source's download blessing, +replies, and runs the walk on its spawned task, streaming `CaptureProgress` +through a bounded channel that the RPC layer forwards into the existing +update broadcast. Completion and failure arrive as the final progress event +(`finished`, `error`), not as the RPC result. + +Rationale: the TUI's orchestration loop `select!`s over one RPC at a time — +a capture RPC that lasts an hour freezes every other interaction. Validation +errors still come back synchronously with the old status mapping; walk +errors move to the stream (and the server log, as before). + +## D5 — TUI: progress, red skipped tracks, warnings + +- **Skipped tracks are red** (and not bold) in both the queue and library + listings, driven by `Track.is_skipped`. The playing-track red keeps + precedence in the queue. +- **Progress lines** render at the bottom of the library pane, one per + active capture: `capturing <name> 12/34 (2 skipped)` (bookmarks: + `bookmarking`). A finished capture lingers ~5 s as + `captured <name>: 34 tracks (2 skipped)`; a failed one shows the error in + red for ~10 s. State lives in the `App`, fed by the update stream; the + 100 ms render tick handles expiry. +- **Warnings**: the `W` help-table description and the capture input + overlay's label both say a download capture can take a long time (and that + re-capturing the same name resumes it). + +## D6 — Out of scope + +- Cancelling a running capture from the TUI. +- Retrying real download failures within a run (rerun-to-resume covers it). +- Garbage-collecting stale entries when the source shrank or reordered. +- Multi-hop link resolution for skipped detection (a link whose target is a + skipped file plays as a normal link failure). + +## D7 — Focused-selection contrast fix + +Library items styled with a foreground color (creatable/editable/deletable → +secondary, marked → green; queue: skipped/current → red) are hard to read +when the focused selection bar (`bg = COLOR_PRIMARY`, a light blue) sits on +them. Fix: when an item is the selected row of a *focused* pane, its +foreground switches to the dark `COLOR_PRIMARY_DARK` so it reads against the +light bar. The unfocused bar is dark and keeps the colored foregrounds. + +## Risks + +- Enumerate-then-fetch holds the full entry list in memory: bounded by + `max_tracks` (500 download / 20 000 bookmark) — fine. +- A source whose listing order changes between runs duplicates content under + new prefixes (assumption above). Accepted; documented in the help text via + the "resumes by name" phrasing. +- The progress channel is bounded (64); a slow broadcast consumer only slows + the walk, never blocks it permanently (the forwarder drains continuously). diff --git a/cbd-tui/src/app/bindings.rs b/cbd-tui/src/app/bindings.rs index 321a828..796eabd 100644 --- a/cbd-tui/src/app/bindings.rs +++ b/cbd-tui/src/app/bindings.rs @@ -275,7 +275,7 @@ pub const BINDINGS: &[Binding] = &[ mods: KeyModifiers::SHIFT, code: KeyCode::Char('W'), action: Action::LibraryDownloadNode, - description: "Download selection as capture", + description: "Download selection as capture (can take long; same name resumes)", }, Binding { scope: Scope::Library, diff --git a/cbd-tui/src/app/library.rs b/cbd-tui/src/app/library.rs index 6b1eaee..3b35525 100644 --- a/cbd-tui/src/app/library.rs +++ b/cbd-tui/src/app/library.rs @@ -13,7 +13,7 @@ use crabidy_core::proto::crabidy::LibraryNode; use super::{ MessageFromUi, StatefulList, UiItem, UiItemKind, COLOR_GREEN, COLOR_PRIMARY, - COLOR_PRIMARY_DARK, COLOR_SECONDARY, + COLOR_PRIMARY_DARK, COLOR_RED, COLOR_SECONDARY, }; pub struct Library { @@ -198,6 +198,7 @@ impl Library { // Tracks carry no wire flag: they inherit their node's // blessing (architecture/captures.md D4). is_downloadable: node.is_downloadable, + is_skipped: t.is_skipped, }) .chain(node.children.iter().map(|c| UiItem { path: c.path.clone(), @@ -209,6 +210,7 @@ impl Library { is_editable: c.is_editable, is_deletable: c.is_deletable, is_downloadable: c.is_downloadable, + is_skipped: false, })) .collect(); @@ -216,10 +218,12 @@ impl Library { } pub fn render(&mut self, f: &mut Frame, area: Rect, focused: bool) { + let selected = self.list_state.selected(); let library_items: Vec<ListItem> = self .list .iter() - .map(|i| { + .enumerate() + .map(|(idx, i)| { let mut text = if i.marked { format!("* {}", i.title) } else { @@ -239,15 +243,24 @@ impl Library { } text.push(']'); } - let style = if i.marked { + let mut style = if i.marked { Style::default() .fg(COLOR_GREEN) .add_modifier(Modifier::BOLD) + } else if i.is_skipped { + // Skipped tracks have no playable audio; playback + // skips them (architecture/incremental-captures.md). + Style::default().fg(COLOR_RED) } else if i.is_creatable || i.is_editable || i.is_deletable { Style::default().fg(COLOR_SECONDARY) } else { Style::default() }; + // A colored foreground is unreadable on the light focused + // selection bar — switch it to the dark tone there (D7). + if focused && selected == Some(idx) && style.fg.is_some() { + style = style.fg(COLOR_PRIMARY_DARK); + } ListItem::new(Span::from(text)).style(style) }) .collect(); diff --git a/cbd-tui/src/app/mod.rs b/cbd-tui/src/app/mod.rs index 8e48bf5..9c6911c 100644 --- a/cbd-tui/src/app/mod.rs +++ b/cbd-tui/src/app/mod.rs @@ -14,7 +14,8 @@ use ratatui::{ }; use crabidy_core::proto::crabidy::{ - get_update_stream_response::Update as StreamUpdate, InitResponse as InitialData, LibraryNode, + get_update_stream_response::Update as StreamUpdate, CaptureProgress, + InitResponse as InitialData, LibraryNode, }; pub use list::StatefulList; @@ -51,6 +52,9 @@ struct UiItem { /// This item allows download captures (`W`). Tracks inherit their /// containing node's flag; child nodes carry their own. is_downloadable: bool, + /// The track has no playable audio (`Track.is_skipped`) — rendered + /// red; playback skips it. Always false for nodes. + is_skipped: bool, } pub const COLOR_PRIMARY: Color = Color::Rgb(129, 161, 193); @@ -147,6 +151,87 @@ pub enum InputPurpose { Capture { path: String, download: bool }, } +/// How long a finished capture's line lingers before it disappears. +const CAPTURE_DONE_LINGER: std::time::Duration = std::time::Duration::from_secs(5); +/// Failures linger longer — the user should get to read them. +const CAPTURE_ERROR_LINGER: std::time::Duration = std::time::Duration::from_secs(10); + +/// One capture's latest progress plus, once finished, when it finished — +/// the render loop expires finished entries after their linger time. +struct CaptureEntry { + progress: CaptureProgress, + finished_at: Option<std::time::Instant>, +} + +/// Live capture progress lines (architecture/incremental-captures.md D5): +/// fed from `CaptureProgress` stream updates, keyed by capture name, +/// expired at render time via [`Self::lines`]. +#[derive(Default)] +pub struct CaptureBoard { + entries: Vec<CaptureEntry>, +} + +impl CaptureBoard { + /// Applies one stream update: replaces the entry of the same name (a + /// re-run supersedes the lingering result of the previous run) and + /// stamps the finish time on terminal events. + pub fn apply(&mut self, progress: CaptureProgress) { + let finished_at = progress.finished.then(std::time::Instant::now); + let entry = CaptureEntry { + progress, + finished_at, + }; + match self + .entries + .iter_mut() + .find(|e| e.progress.name == entry.progress.name) + { + Some(existing) => *existing = entry, + None => self.entries.push(entry), + } + } + + /// The lines to render right now, oldest first, paired with an + /// is-error flag. Drops finished entries past their linger time. + fn lines(&mut self) -> Vec<(String, bool)> { + self.entries.retain(|e| match e.finished_at { + None => true, + Some(at) if e.progress.error.is_empty() => at.elapsed() < CAPTURE_DONE_LINGER, + Some(at) => at.elapsed() < CAPTURE_ERROR_LINGER, + }); + self.entries + .iter() + .map(|e| (Self::line(&e.progress), !e.progress.error.is_empty())) + .collect() + } + + /// One entry's display line. Pure so it is testable without clocks. + fn line(p: &CaptureProgress) -> String { + let verb = if p.download { + ("capturing", "captured", "capture") + } else { + ("bookmarking", "bookmarked", "bookmark") + }; + let skipped = if p.tracks_skipped > 0 { + format!(" ({} skipped)", p.tracks_skipped) + } else { + String::new() + }; + if !p.finished { + let total = if p.tracks_total > 0 { + p.tracks_total.to_string() + } else { + "?".to_string() + }; + format!("{} {} {}/{total}{skipped}", verb.0, p.name, p.tracks_done) + } else if p.error.is_empty() { + format!("{} {}: {} tracks{skipped}", verb.1, p.name, p.tracks_done) + } else { + format!("{} {} failed: {}", verb.2, p.name, p.error) + } + } +} + /// State of the one-line text input overlay (node creation and rename). /// /// While this is `Some` on [`App`], key events bypass `bindings::lookup` @@ -165,6 +250,9 @@ pub struct App { /// `Some` while the input overlay is open; takes precedence over the /// bindings table (checked first in the event loop). pub input: Option<InputState>, + /// Progress of running (and recently finished) captures, rendered as + /// status lines at the bottom of the library pane. + pub captures: CaptureBoard, pub library: Library, pub now_playing: NowPlaying, pub queue: Queue, @@ -180,6 +268,7 @@ impl App { focus: UiFocus::Library, show_help: false, input: None, + captures: CaptureBoard::default(), library, now_playing, queue, @@ -417,7 +506,10 @@ impl App { InputPurpose::Capture { download: false, .. } => "bookmark", - InputPurpose::Capture { download: true, .. } => "capture", + // Downloading a whole subtree can take a long time; + // re-capturing the same name resumes it (D5 warning — + // kept short so the line fits narrow panes). + InputPurpose::Capture { download: true, .. } => "capture (slow, resumable)", }; let line = Rect::new(area.x + 1, area.y + area.height - 2, area.width - 2, 1); f.render_widget(Clear, line); @@ -429,6 +521,38 @@ impl App { } } + // Capture progress lines: stacked up from the bottom of the library + // pane, above the input overlay when that is open. Failures render + // red. At most three — more concurrent captures than that keep + // running fine, only their lines wait for a free row. + let capture_lines = self.captures.lines(); + if !capture_lines.is_empty() { + let area = main[0]; + let bottom_offset = if self.input.is_some() { 3 } else { 2 }; + for (i, (text, is_error)) in capture_lines.iter().take(3).enumerate() { + let offset = bottom_offset + i as u16; + if area.height <= offset + 1 || area.width < 4 { + break; + } + let line = Rect::new( + area.x + 1, + area.y + area.height - 1 - offset, + area.width - 2, + 1, + ); + let color = if *is_error { + COLOR_RED + } else { + COLOR_SECONDARY + }; + f.render_widget(Clear, line); + f.render_widget( + Paragraph::new(text.as_str()).style(Style::default().fg(color)), + line, + ); + } + } + // The help modal renders last so it overlays every pane. if self.show_help { help::render(f); @@ -904,6 +1028,7 @@ mod tests { title: "one".to_string(), duration: None, album: None, + is_skipped: false, }; let listing = |downloadable| LibraryNode { tracks: vec![track.clone()], @@ -961,6 +1086,7 @@ mod tests { title: "track".to_string(), duration: None, album: None, + is_skipped: false, }], resolving: false, } @@ -1001,6 +1127,131 @@ mod tests { assert!(rx.try_recv().is_err(), "Esc must not save"); } + fn render_text(app: &mut App) -> String { + let backend = ratatui::backend::TestBackend::new(80, 24); + let mut terminal = ratatui::Terminal::new(backend).expect("test terminal"); + terminal.draw(|f| app.render(f)).expect("draw app"); + let buf = terminal.backend().buffer(); + let mut text = String::new(); + for y in 0..buf.area.height { + for x in 0..buf.area.width { + text.push_str(buf[(x, y)].symbol()); + } + text.push('\n'); + } + text + } + + #[test] + fn capture_board_lines_cover_all_states() { + let mut p = CaptureProgress { + name: "faves".to_string(), + download: true, + tracks_done: 3, + tracks_total: 0, + tracks_skipped: 0, + finished: false, + error: String::new(), + }; + // Before the enumeration finishes the total is unknown. + assert_eq!(CaptureBoard::line(&p), "capturing faves 3/?"); + p.tracks_total = 12; + p.tracks_skipped = 2; + assert_eq!(CaptureBoard::line(&p), "capturing faves 3/12 (2 skipped)"); + p.finished = true; + p.tracks_done = 12; + assert_eq!( + CaptureBoard::line(&p), + "captured faves: 12 tracks (2 skipped)" + ); + p.error = "boom".to_string(); + assert_eq!(CaptureBoard::line(&p), "capture faves failed: boom"); + // Bookmarks use their own verbs. + let b = CaptureProgress { + name: "b".to_string(), + tracks_done: 1, + tracks_total: 2, + ..Default::default() + }; + assert_eq!(CaptureBoard::line(&b), "bookmarking b 1/2"); + } + + #[test] + fn capture_progress_renders_and_reruns_replace_finished_lines() { + let (mut app, _rx) = app(); + app.captures.apply(CaptureProgress { + name: "faves".to_string(), + download: true, + tracks_done: 3, + tracks_total: 12, + tracks_skipped: 1, + finished: false, + error: String::new(), + }); + let text = render_text(&mut app); + assert!( + text.contains("capturing faves 3/12 (1 skipped)"), + "progress line: {text}" + ); + // A later event for the same name replaces the line. + app.captures.apply(CaptureProgress { + name: "faves".to_string(), + download: true, + tracks_done: 12, + tracks_total: 12, + tracks_skipped: 1, + finished: true, + error: String::new(), + }); + let text = render_text(&mut app); + assert!(!text.contains("capturing faves"), "{text}"); + assert!( + text.contains("captured faves: 12 tracks (1 skipped)"), + "{text}" + ); + } + + #[test] + fn download_capture_overlay_warns_about_long_captures() { + let (mut app, _rx) = app(); + app.library.update(downloadable_listing()); + let _ = app.dispatch(Action::LibraryDownloadNode); + let text = render_text(&mut app); + assert!( + text.contains("capture (slow, resumable): artist"), + "warning label: {text}" + ); + } + + #[test] + fn focused_selection_darkens_colored_library_items() { + // The editable [ed] item is rendered in the secondary color; under + // the focused selection bar that was unreadable — the foreground + // must switch to the dark tone (D7). + let (mut app, _rx) = app(); + app.library.update(search_listing(true)); + let backend = ratatui::backend::TestBackend::new(80, 24); + let mut terminal = ratatui::Terminal::new(backend).expect("test terminal"); + terminal.draw(|f| app.render(f)).expect("draw app"); + let buf = terminal.backend().buffer().clone(); + let mut found = None; + for y in 0..buf.area.height { + let row: String = (0..buf.area.width) + .map(|x| buf[(x, y)].symbol().to_string()) + .collect(); + if let Some(col) = row.find("abba [ed]") { + found = Some((col as u16, y)); + break; + } + } + let (x, y) = found.expect("editable row rendered"); + assert_eq!( + buf[(x, y)].style().fg, + Some(COLOR_PRIMARY_DARK), + "focused selected editable item must be readable" + ); + } + #[test] fn queue_clear_actions_carry_the_keep_current_flag() { let (mut app, rx) = app(); diff --git a/cbd-tui/src/app/now_playing.rs b/cbd-tui/src/app/now_playing.rs index 760089f..0277024 100644 --- a/cbd-tui/src/app/now_playing.rs +++ b/cbd-tui/src/app/now_playing.rs @@ -210,6 +210,7 @@ mod tests { title: "title".to_string(), duration: None, album: None, + is_skipped: false, }), } } diff --git a/cbd-tui/src/app/queue.rs b/cbd-tui/src/app/queue.rs index ec1377d..bca0fc0 100644 --- a/cbd-tui/src/app/queue.rs +++ b/cbd-tui/src/app/queue.rs @@ -82,6 +82,7 @@ impl Queue { is_editable: false, is_deletable: false, is_downloadable: false, + is_skipped: t.is_skipped, }) .collect(); @@ -89,6 +90,7 @@ impl Queue { } pub fn render(&mut self, f: &mut Frame, area: Rect, focused: bool) { + let selected = self.list_state.selected(); let mut queue_items: Vec<ListItem> = self .list .iter() @@ -101,11 +103,21 @@ impl Queue { } else { item.title.to_string() }; - let style = if active { + let mut style = if active { Style::default().fg(COLOR_RED).add_modifier(Modifier::BOLD) + } else if item.is_skipped { + // No playable audio: rendered red (not bold — the + // playing marker keeps precedence), skipped by + // playback (architecture/incremental-captures.md). + Style::default().fg(COLOR_RED) } else { Style::default() }; + // A colored foreground is unreadable on the light focused + // selection bar — switch it to the dark tone there (D7). + if focused && selected == Some(idx) && style.fg.is_some() { + style = style.fg(COLOR_PRIMARY_DARK); + } ListItem::new(Span::from(title)).style(style) }) .collect(); @@ -180,6 +192,7 @@ mod tests { title: t.to_string(), duration: None, album: None, + is_skipped: false, }) .collect(), resolving, @@ -269,4 +282,63 @@ mod tests { assert_eq!(Queue::loading_dots(800), "..."); assert_eq!(Queue::loading_dots(1200), "."); } + + /// Renders and returns the buffer plus the y of the row containing + /// `needle` and the x of its first character. + fn render_and_find(queue: &mut Queue, needle: &str) -> (ratatui::buffer::Buffer, u16, u16) { + 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(); + for y in 0..buffer.area.height { + let row: String = (0..buffer.area.width) + .map(|x| buffer[(x, y)].symbol().to_string()) + .collect(); + if let Some(col) = row.find(needle) { + return (buffer, col as u16, y); + } + } + panic!("row containing {needle:?} not found"); + } + + #[test] + fn skipped_tracks_render_red() { + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + let mut data = queue_data(&["one", "two"], false); + data.tracks[1].is_skipped = true; + queue.update_queue(data); + // Selection sits on row 0; the unselected skipped row is red. + let (buffer, x, y) = render_and_find(&mut queue, "artist - two"); + assert_eq!( + buffer[(x, y)].style().fg, + Some(super::COLOR_RED), + "skipped tracks must be red" + ); + // The playing track keeps its red marker when not under the bar. + queue.select(Some(1)); + let (buffer, x, y) = render_and_find(&mut queue, "> artist - one"); + assert_eq!(buffer[(x, y)].style().fg, Some(super::COLOR_RED)); + } + + #[test] + fn colored_rows_darken_under_the_focused_selection_bar() { + // A red (skipped) row under the light focused selection bar was + // unreadable; the foreground switches to the dark tone there + // (architecture/incremental-captures.md D7). + let (tx, _rx) = flume::unbounded(); + let mut queue = Queue::new(tx); + let mut data = queue_data(&["one", "two"], false); + data.tracks[1].is_skipped = true; + queue.update_queue(data); + queue.select(Some(1)); + let (buffer, x, y) = render_and_find(&mut queue, "two"); + assert_eq!( + buffer[(x, y)].style().fg, + Some(super::COLOR_PRIMARY_DARK), + "selected colored rows must use the dark foreground" + ); + } } diff --git a/cbd-tui/src/lib.rs b/cbd-tui/src/lib.rs index 7920e04..89792b7 100644 --- a/cbd-tui/src/lib.rs +++ b/cbd-tui/src/lib.rs @@ -258,6 +258,9 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) { } StreamUpdate::Mute(_) => { /* FIXME: implement */ } StreamUpdate::Volume(_) => { /* FIXME: implement */ } + StreamUpdate::CaptureProgress(progress) => { + app.captures.apply(progress); + } }, } } diff --git a/crabidy-core/crabidy/v1/crabidy.proto b/crabidy-core/crabidy/v1/crabidy.proto index e113bd5..9c49515 100644 --- a/crabidy-core/crabidy/v1/crabidy.proto +++ b/crabidy-core/crabidy/v1/crabidy.proto @@ -23,8 +23,12 @@ service CrabidyService { // Captures the queueable subtree at `path` as the bookmark `name`: a // structure-preserving snapshot under /bookmarks/<name> (folders per // child node, link track files per track). Overwrites an existing - // bookmark of the same name. The client stays where it is — the bookmark - // shows up under /bookmarks on the next visit. + // bookmark of the same name; a *download* capture instead merges into an + // existing /captures/<name>, resuming what is not yet downloaded. + // Returns once the capture is accepted (name, store, and download + // blessing validated); the walk runs server-side and reports through + // CaptureProgress updates on GetUpdateStream, ending in one event with + // `finished` set (and `error` on failure). rpc CaptureLibraryNode(CaptureLibraryNodeRequest) returns (CaptureLibraryNodeResponse); // Queue @@ -173,9 +177,32 @@ message GetUpdateStreamResponse { float volume = 5; bool mute = 6; TrackPosition position = 7; + CaptureProgress capture_progress = 8; } } +// Progress of a running capture (CaptureLibraryNode). Broadcast after each +// processed track; exactly one event per capture sets `finished` (with +// `error` on failure). +message CaptureProgress { + // The capture (or bookmark) name the user chose. + string name = 1; + // True for a download capture (W), false for a bookmark (w). + bool download = 2; + // Tracks settled so far (reused, downloaded, linked, or recorded as + // skipped) — reaches tracks_total on success. + uint32 tracks_done = 3; + // Total tracks discovered by the enumeration; 0 until it completes. + uint32 tracks_total = 4; + // Of the settled tracks, how many were recorded as skipped + // (uncapturable source) this run. + uint32 tracks_skipped = 5; + // Terminal event: the capture is over. + bool finished = 6; + // Why it failed; empty on success. Never carries URLs or file contents. + string error = 7; +} + // Playback message TogglePlayRequest {} message TogglePlayResponse {} @@ -262,6 +289,9 @@ message Track { string title = 3; optional uint32 duration = 4; optional Album album = 5; + // The track has no playable audio (a capture recorded its source as + // uncapturable). Clients mark it; playback skips it. + bool is_skipped = 6; } message LibraryNode { diff --git a/crabidy-core/src/lib.rs b/crabidy-core/src/lib.rs index b8c9c4f..9438c5d 100644 --- a/crabidy-core/src/lib.rs +++ b/crabidy-core/src/lib.rs @@ -360,6 +360,7 @@ mod tests { title: t.to_string(), duration: None, album: None, + is_skipped: false, }) .collect(), is_queable, @@ -394,6 +395,7 @@ mod tests { title: path.to_string(), duration: None, album: None, + is_skipped: false, }) } fn get_lib_root(&self) -> LibraryNode { diff --git a/crabidy-server/src/bookmark_store.rs b/crabidy-server/src/bookmark_store.rs index 390c3f4..eab3325 100644 --- a/crabidy-server/src/bookmark_store.rs +++ b/crabidy-server/src/bookmark_store.rs @@ -52,22 +52,35 @@ impl BookmarkStore { &self.dir } + /// Validates a bookmark request without writing anything: the name + /// must be a legal folder name. The accept-then-stream RPC replies + /// after this and runs [`Self::capture`] detached + /// (architecture/incremental-captures.md D4). + pub fn validate(&self, name: &str) -> Result<(), CaptureError> { + fsdy::validate_folder_name(name, &[]).map_err(CaptureError::InvalidName)?; + Ok(()) + } + /// Captures the subtree at `source_path` as the bookmark `name`, - /// overwriting an existing bookmark of that name. + /// overwriting an existing bookmark of that name. Reports through + /// `progress` (non-terminal events only; the caller sends the + /// terminal one). /// /// Walks `client` (the orchestrator, so any provider is reachable) /// iteratively in pre-order: every child node becomes an /// order-prefixed folder, every track an order-prefixed link file - /// ([`fsdy::TrackFile::from_track`]). A `source_path` that is itself a - /// track captures as a folder with one file. Aborts with - /// [`CaptureError::TooLarge`] beyond [`MAX_CAPTURE_DIRS`] / - /// [`MAX_CAPTURE_TRACKS`]; an unreadable source is - /// [`CaptureError::BadSource`]. Never panics on provider contents. + /// ([`fsdy::TrackFile::from_track`]; a skipped source track writes a + /// skipped toml). A `source_path` that is itself a track captures as + /// a folder with one file. Aborts with [`CaptureError::TooLarge`] + /// beyond [`MAX_CAPTURE_DIRS`] / [`MAX_CAPTURE_TRACKS`]; an + /// unreadable source is [`CaptureError::BadSource`]. Never panics on + /// provider contents. pub async fn capture<C>( &self, client: &C, source_path: &str, name: &str, + progress: &crate::capture::Progress, ) -> Result<(), CaptureError> where C: ProviderClient + Sync, @@ -78,6 +91,7 @@ impl BookmarkStore { name, MAX_CAPTURE_DIRS, MAX_CAPTURE_TRACKS, + progress, ) .await } @@ -90,6 +104,7 @@ impl BookmarkStore { name: &str, max_dirs: usize, max_tracks: usize, + progress: &crate::capture::Progress, ) -> Result<(), CaptureError> where C: ProviderClient + Sync, @@ -106,6 +121,7 @@ impl BookmarkStore { name, caps, &crate::capture::Sink::Link, + progress, ) .await } @@ -118,6 +134,11 @@ mod tests { use std::path::Path; use tempfile::TempDir; + /// A silent progress reporter; bookmark tests assert on disk state. + fn silent() -> crate::capture::Progress { + crate::capture::Progress::silent("test", false) + } + /// A real fsdy instance as the capture source: an artist with two /// albums holding url tracks, plus one link track pointing at Tidal. async fn source() -> (fsdy::Client, TempDir) { @@ -162,7 +183,7 @@ mod tests { let (client, _src) = source().await; let (store, _dir) = store().await; store - .capture(&client, "/fs/artist", "faves") + .capture(&client, "/fs/artist", "faves", &silent()) .await .expect("capture"); @@ -199,7 +220,7 @@ mod tests { let (client, _src) = source().await; let (store, _dir) = store().await; store - .capture(&client, "/fs/artist", "faves") + .capture(&client, "/fs/artist", "faves", &silent()) .await .expect("capture"); @@ -227,6 +248,7 @@ mod tests { &client, "/fs/artist/Album%20One/01%20one.cbd-track.toml", "just one", + &silent(), ) .await .expect("capture track"); @@ -241,18 +263,18 @@ mod tests { for bad in ["", " ", "a/b", ".hidden"] { assert!( matches!( - store.capture(&client, "/fs/artist", bad).await, + store.capture(&client, "/fs/artist", bad, &silent()).await, Err(CaptureError::InvalidName(_)) ), "name {bad:?} must be rejected" ); } store - .capture(&client, "/fs/artist", "faves") + .capture(&client, "/fs/artist", "faves", &silent()) .await .expect("first capture"); store - .capture(&client, "/fs/artist/Album%20Two", "faves") + .capture(&client, "/fs/artist/Album%20Two", "faves", &silent()) .await .expect("overwrite"); // The overwrite fully replaces the older, larger capture. @@ -267,7 +289,7 @@ mod tests { let (client, _src) = source().await; let (store, _dir) = store().await; assert!(matches!( - store.capture(&client, "/fs/nope", "x").await, + store.capture(&client, "/fs/nope", "x", &silent()).await, Err(CaptureError::BadSource(_)) )); } @@ -278,13 +300,20 @@ mod tests { let (store, _dir) = store().await; // The tree has 3 directories (artist + 2 albums); a 2-dir cap trips. let err = store - .capture_with_caps(&client, "/fs/artist", "big", 2, MAX_CAPTURE_TRACKS) + .capture_with_caps( + &client, + "/fs/artist", + "big", + 2, + MAX_CAPTURE_TRACKS, + &silent(), + ) .await .expect_err("over the dir cap"); assert!(matches!(err, CaptureError::TooLarge(_))); // ... same for the track cap. let err = store - .capture_with_caps(&client, "/fs/artist", "big", MAX_CAPTURE_DIRS, 1) + .capture_with_caps(&client, "/fs/artist", "big", MAX_CAPTURE_DIRS, 1, &silent()) .await .expect_err("over the track cap"); assert!(matches!(err, CaptureError::TooLarge(_))); diff --git a/crabidy-server/src/capture.rs b/crabidy-server/src/capture.rs index 60ceadb..4e4d225 100644 --- a/crabidy-server/src/capture.rs +++ b/crabidy-server/src/capture.rs @@ -1,15 +1,26 @@ -//! The shared subtree-capture walk (see `architecture/captures.md` D2). +//! The shared subtree-capture walk +//! (see `architecture/captures.md` D2 and +//! `architecture/incremental-captures.md` D2). //! //! Both bookmark captures (`w`, link files) and download captures (`W`, -//! audio files next to their tomls) mirror a library subtree into a folder: -//! an iterative pre-order walk over [`ProviderClient::get_lib_node`] with -//! order-prefixed names, size caps, a hidden tmp-and-swap write, and -//! all-or-nothing cleanup. This module owns that walk; the two stores only -//! differ in the per-track [`Sink`]. +//! audio files next to their tomls) mirror a library subtree into a folder. +//! The walk runs in two phases: **enumerate** first (every directory and +//! track, enforcing the size caps — this makes the total known before the +//! first download), then **fetch** track by track, reporting [`Progress`] +//! after each one. +//! +//! The per-track [`Sink`] decides the write mode: bookmarks build the whole +//! capture in a hidden tmp sibling and swap it into place (all-or-nothing, +//! overwrite = refresh), download captures write **incrementally** into the +//! final folder — entries that are already satisfied are reused, tracks +//! whose source cannot be captured are recorded as *skipped* tomls, and a +//! real download failure aborts the run but keeps everything written so +//! far, so re-running the same name resumes where it stopped. use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU32, Ordering}; -use crabidy_core::proto::crabidy::Track; +use crabidy_core::proto::crabidy::{CaptureProgress, Track}; use crabidy_core::ProviderClient; use tracing::warn; @@ -29,7 +40,8 @@ pub struct Caps { pub max_dirs: usize, /// Maximum track files. pub max_tracks: usize, - /// Maximum total downloaded bytes (ignored by [`Sink::Link`]). + /// Maximum total bytes downloaded in one run (ignored by + /// [`Sink::Link`]). Reused entries of a resumed capture do not count. pub max_bytes: u64, } @@ -53,7 +65,7 @@ pub const DOWNLOAD_CAPS: Caps = Caps { /// At the RPC boundary: `InvalidName`/`BadSource` → `invalid_argument`, /// `TooLarge`/`Disabled`/`Unsupported` → `failed_precondition`, the rest → /// `internal`. Messages carry names, paths, and counts, never file -/// contents. +/// contents or stream URLs. #[derive(Debug, thiserror::Error)] pub enum CaptureError { #[error("invalid name: {0}")] @@ -74,23 +86,131 @@ pub enum CaptureError { TrackFile(#[from] fsdy::TrackFileError), } +/// Progress reporting for one capture run +/// (`architecture/incremental-captures.md` D4). +/// +/// The walk bumps the counters as it settles tracks; every bump publishes a +/// non-terminal [`CaptureProgress`] snapshot to the (bounded) channel, +/// **lossily** — a full channel drops the snapshot, never blocks the walk. +/// The terminal event is sent exactly once via [`Self::finish`] and is not +/// lossy. A [`Self::silent`] reporter counts without a channel. +#[derive(Debug)] +pub struct Progress { + name: String, + download: bool, + tx: Option<flume::Sender<CaptureProgress>>, + done: AtomicU32, + total: AtomicU32, + skipped: AtomicU32, +} + +impl Progress { + /// A reporter publishing to `tx`. + pub fn new(name: &str, download: bool, tx: flume::Sender<CaptureProgress>) -> Self { + Self { + tx: Some(tx), + ..Self::silent(name, download) + } + } + + /// A reporter that only counts (tests, callers without a stream). + pub fn silent(name: &str, download: bool) -> Self { + Self { + name: name.to_string(), + download, + tx: None, + done: AtomicU32::new(0), + total: AtomicU32::new(0), + skipped: AtomicU32::new(0), + } + } + + /// The wire snapshot of the current counters. + fn snapshot(&self, finished: bool, error: String) -> CaptureProgress { + CaptureProgress { + name: self.name.clone(), + download: self.download, + tracks_done: self.done.load(Ordering::Relaxed), + tracks_total: self.total.load(Ordering::Relaxed), + tracks_skipped: self.skipped.load(Ordering::Relaxed), + finished, + error, + } + } + + /// Publishes a non-terminal snapshot; lossy on a full channel. + fn publish(&self) { + if let Some(tx) = &self.tx { + let _ = tx.try_send(self.snapshot(false, String::new())); + } + } + + fn set_total(&self, total: usize) { + self.total + .store(total.min(u32::MAX as usize) as u32, Ordering::Relaxed); + self.publish(); + } + + /// One track settled with playable data (reused, downloaded, linked). + fn track_done(&self) { + self.done.fetch_add(1, Ordering::Relaxed); + self.publish(); + } + + /// One track settled as skipped (counts toward done — the ratio must + /// reach total on success). + fn track_skipped(&self) { + self.skipped.fetch_add(1, Ordering::Relaxed); + self.done.fetch_add(1, Ordering::Relaxed); + self.publish(); + } + + /// Sends the terminal event: the capture is over, `error` says why it + /// failed (or `None` on success). A vanished receiver is ignored — the + /// capture's outcome is on disk and in the log either way. + pub async fn finish(&self, error: Option<String>) { + if let Some(tx) = &self.tx { + let _ = tx + .send_async(self.snapshot(true, error.unwrap_or_default())) + .await; + } + } +} + /// What happens to each track the walk visits. #[derive(Debug)] pub enum Sink { /// Write an order-prefixed link file ([`fsdy::TrackFile::from_track`]) - /// — the bookmark behavior. + /// — the bookmark behavior: tmp-and-swap, all-or-nothing. Link, /// Download the track's audio next to an order-prefixed toml that - /// points at it — the captures behavior. + /// points at it — the captures behavior: incremental into the final + /// folder, resumable by name. Download(Downloader), } +/// How one visited track settled during the fetch phase. +enum TrackOutcome { + /// Playable data is in place (downloaded now, or already there). + Captured, + /// The source cannot be captured; a skipped toml records the gap. + Skipped, +} + +/// One track discovered by the enumeration phase: what to fetch, where to +/// put it, and its listing position (the order prefix). +struct TrackEntry { + track: Track, + dir: PathBuf, + index: usize, +} + /// Downloads one track's audio via the provider's stream URL. /// /// One shared HTTP client with a connect timeout; each track is bounded by /// [`DOWNLOAD_TRACK_TIMEOUT`] end to end and never retried (a capture is -/// re-runnable; overwrite = refresh). Bodies are streamed to disk against -/// the capture's remaining byte budget. +/// re-runnable; resuming re-attempts what is missing). Bodies are streamed +/// to disk against the capture's remaining byte budget. #[derive(Debug)] pub struct Downloader { http: reqwest::Client, @@ -106,7 +226,7 @@ impl Downloader { Ok(Self { http }) } - /// Downloads one track: audio file first, then the toml pointing at + /// Settles one track: audio file first, then the toml pointing at /// it — a toml never exists without its audio. The whole operation is /// bounded by [`DOWNLOAD_TRACK_TIMEOUT`]. /// @@ -120,7 +240,7 @@ impl Downloader { dir: &Path, index: usize, bytes_left: &mut u64, - ) -> Result<(), CaptureError> + ) -> Result<TrackOutcome, CaptureError> where C: ProviderClient + Sync, { @@ -143,12 +263,13 @@ impl Downloader { /// URL, stream the response to disk against the byte budget, then /// write the toml. /// - /// A track whose source cannot be captured — the stream fails to - /// resolve, or it resolves to something other than an http(s) URL - /// (e.g. a local file playable) — is **skipped** with a warning - /// instead of aborting the capture (architecture/captures.md D3): - /// queue and bookmark captures mix providers, and one local track - /// must not kill the rest. Actual download failures stay fatal. + /// A track whose source cannot be captured — it is itself skipped, its + /// stream fails to resolve, or it resolves to something other than an + /// http(s) URL (e.g. a local file playable) — is recorded as a + /// **skipped toml** instead of aborting the capture + /// (architecture/incremental-captures.md D2): queue and bookmark + /// captures mix providers, and one local track must not kill the rest. + /// Actual download failures stay fatal for the run. async fn fetch_track<C>( &self, client: &C, @@ -156,27 +277,37 @@ impl Downloader { dir: &Path, index: usize, bytes_left: &mut u64, - ) -> Result<(), CaptureError> + ) -> Result<TrackOutcome, CaptureError> where C: ProviderClient + Sync, { let track_path = track.path.as_str(); + if track.is_skipped { + warn!(path = track_path, "recording an already-skipped track"); + return write_skipped(track, dir, index).await; + } let urls = match client.get_urls_for_track(track_path).await { Ok(urls) => urls, Err(err) => { - warn!(path = track_path, "skipping uncapturable track: {err}"); - return Ok(()); + warn!( + path = track_path, + "recording uncapturable track as skipped: {err}" + ); + return write_skipped(track, dir, index).await; } }; let Some(url) = urls.first() else { - warn!(path = track_path, "skipping track without a stream url"); - return Ok(()); + warn!( + path = track_path, + "recording track without a stream url as skipped" + ); + return write_skipped(track, dir, index).await; }; if !(url.starts_with("http://") || url.starts_with("https://")) { // Local file playables (fs tracks, existing captures) resolve // to paths, not URLs — nothing to download. - warn!(path = track_path, "skipping non-http playable"); - return Ok(()); + warn!(path = track_path, "recording non-http playable as skipped"); + return write_skipped(track, dir, index).await; } let download_err = |err: reqwest::Error| { CaptureError::Download(format!("{track_path}: {}", err.without_url())) @@ -211,17 +342,32 @@ impl Downloader { let text = fsdy::TrackFile::from_track_with_file(track, Path::new(&audio_name)).to_toml()?; tokio::fs::write(dir.join(fsdy::track_file_name(index, &track.title)), text).await?; - Ok(()) + Ok(TrackOutcome::Captured) } } -/// Captures the subtree at `source_path` as `dir/<name>/`, overwriting an -/// existing folder of that name. +/// Writes the skipped toml for `track` at listing position `index`, +/// overwriting whatever was there. +async fn write_skipped( + track: &Track, + dir: &Path, + index: usize, +) -> Result<TrackOutcome, CaptureError> { + let text = fsdy::TrackFile::from_track_skipped(track).to_toml()?; + tokio::fs::write(dir.join(fsdy::track_file_name(index, &track.title)), text).await?; + Ok(TrackOutcome::Skipped) +} + +/// Captures the subtree at `source_path` as `dir/<name>/`. /// /// Validates `name` ([`fsdy::validate_folder_name`], nothing reserved), -/// builds the whole capture in a hidden `.tmp-<name>` sibling via -/// [`write_tree`], and swaps it into place. Every failure removes the temp -/// folder — nothing half-written survives, not even hidden. +/// then enumerates the subtree (caps enforced, total reported) and fetches +/// track by track. [`Sink::Link`] builds the whole capture in a hidden +/// `.tmp-<name>` sibling and swaps it into place, removing the temp folder +/// on any failure — the bookmark all-or-nothing. [`Sink::Download`] writes +/// incrementally into `dir/<name>` itself: satisfied entries are reused, +/// uncapturable tracks become skipped tomls, and failures keep everything +/// already written (re-run the same name to resume). pub async fn capture_into<C>( dir: &Path, client: &C, @@ -229,39 +375,68 @@ pub async fn capture_into<C>( name: &str, caps: Caps, sink: &Sink, + progress: &Progress, ) -> Result<(), CaptureError> where C: ProviderClient + Sync, { let name = fsdy::validate_folder_name(name, &[]).map_err(CaptureError::InvalidName)?; - let tmp = dir.join(format!(".tmp-{name}")); - let written = write_tree(client, source_path, &tmp, caps, sink).await; - if let Err(err) = written { - // Every failure path removes the temp folder: nothing - // half-written survives, not even hidden. - let _ = tokio::fs::remove_dir_all(&tmp).await; - return Err(err); + match sink { + Sink::Link => { + let tmp = dir.join(format!(".tmp-{name}")); + let written = write_links(client, source_path, &tmp, caps, progress).await; + if let Err(err) = written { + // Every failure path removes the temp folder: nothing + // half-written survives, not even hidden. + let _ = tokio::fs::remove_dir_all(&tmp).await; + return Err(err); + } + let target = dir.join(name); + if tokio::fs::try_exists(&target).await? { + tokio::fs::remove_dir_all(&target).await?; + } + tokio::fs::rename(&tmp, &target).await?; + Ok(()) + } + Sink::Download(downloader) => { + let target = dir.join(name); + tokio::fs::create_dir_all(&target).await?; + let entries = enumerate(client, source_path, &target, caps).await?; + progress.set_total(entries.len()); + let mut bytes_left = caps.max_bytes; + for entry in &entries { + if existing_is_satisfied(&entry.dir, entry.index, &entry.track.title).await { + progress.track_done(); + continue; + } + let outcome = downloader + .download_track( + client, + &entry.track, + &entry.dir, + entry.index, + &mut bytes_left, + ) + .await?; + match outcome { + TrackOutcome::Captured => progress.track_done(), + TrackOutcome::Skipped => progress.track_skipped(), + } + } + Ok(()) + } } - let target = dir.join(name); - if tokio::fs::try_exists(&target).await? { - tokio::fs::remove_dir_all(&target).await?; - } - tokio::fs::rename(&tmp, &target).await?; - Ok(()) } -/// Builds the mirrored tree inside `tmp`: iterative pre-order over -/// [`ProviderClient::get_lib_node`] (a deep tree must not overflow the -/// stack), every child node an order-prefixed folder ([`fsdy::dir_name`]), -/// every track handed to `sink` with its listing index. A `source_path` -/// that is a track captures as a folder with one entry. All-or-nothing: -/// any provider, download, or write failure aborts the whole capture. -async fn write_tree<C>( +/// The bookmark walk body: fresh tmp folder, enumerate, one link file per +/// track ([`fsdy::TrackFile::from_track`] — skipped source tracks write +/// skipped tomls). The caller owns cleanup on error. +async fn write_links<C>( client: &C, source_path: &str, tmp: &Path, caps: Caps, - sink: &Sink, + progress: &Progress, ) -> Result<(), CaptureError> where C: ProviderClient + Sync, @@ -271,22 +446,50 @@ where tokio::fs::remove_dir_all(tmp).await?; } tokio::fs::create_dir_all(tmp).await?; - let mut bytes_left = caps.max_bytes; + let entries = enumerate(client, source_path, tmp, caps).await?; + progress.set_total(entries.len()); + for entry in &entries { + let text = fsdy::TrackFile::from_track(&entry.track).to_toml()?; + let file = entry + .dir + .join(fsdy::track_file_name(entry.index, &entry.track.title)); + tokio::fs::write(file, text).await?; + progress.track_done(); + } + Ok(()) +} - // A track source captures as a folder with one entry. +/// Phase 1: mirrors the directory structure under `root` and collects +/// every track with its target directory and listing index. Iterative +/// pre-order over [`ProviderClient::get_lib_node`] (a deep tree must not +/// overflow the stack); existing directories are reused +/// (`create_dir_all`), which is what makes download captures resumable. A +/// `source_path` that is a track enumerates as a single entry. Enforces +/// `max_dirs`/`max_tracks`. +async fn enumerate<C>( + client: &C, + source_path: &str, + root: &Path, + caps: Caps, +) -> Result<Vec<TrackEntry>, CaptureError> +where + C: ProviderClient + Sync, +{ if client.is_track_path(source_path) { let track = client .get_metadata_for_track(source_path) .await .map_err(|err| CaptureError::BadSource(format!("{source_path}: {err}")))?; - return write_track(sink, client, &track, tmp, 0, &mut bytes_left).await; + return Ok(vec![TrackEntry { + track, + dir: root.to_path_buf(), + index: 0, + }]); } - + let mut entries = Vec::new(); let mut dirs = 1usize; let mut tracks = 0usize; - // Iterative pre-order: a deep provider tree must not overflow the - // stack. Each entry pairs a library path with its mirror folder. - let mut worklist: Vec<(String, PathBuf)> = vec![(source_path.to_string(), tmp.to_path_buf())]; + let mut worklist: Vec<(String, PathBuf)> = vec![(source_path.to_string(), root.to_path_buf())]; while let Some((lib_path, dir)) = worklist.pop() { let node = client .get_lib_node(&lib_path) @@ -297,7 +500,11 @@ where if tracks > caps.max_tracks { return Err(CaptureError::TooLarge("too many tracks")); } - write_track(sink, client, track, &dir, index, &mut bytes_left).await?; + entries.push(TrackEntry { + track: track.clone(), + dir: dir.clone(), + index, + }); } for (index, child) in node.children.iter().enumerate() { dirs += 1; @@ -305,38 +512,39 @@ where return Err(CaptureError::TooLarge("too many directories")); } let child_dir = dir.join(fsdy::dir_name(index, &child.title)); - tokio::fs::create_dir(&child_dir).await?; + tokio::fs::create_dir_all(&child_dir).await?; worklist.push((child.path.clone(), child_dir)); } } - Ok(()) + Ok(entries) } -/// Writes one visited track into `dir` at listing position `index`, -/// drawing downloaded bytes from `bytes_left`. -async fn write_track<C>( - sink: &Sink, - client: &C, - track: &Track, - dir: &Path, - index: usize, - bytes_left: &mut u64, -) -> Result<(), CaptureError> -where - C: ProviderClient + Sync, -{ - match sink { - Sink::Link => { - let text = fsdy::TrackFile::from_track(track).to_toml()?; - let file = dir.join(fsdy::track_file_name(index, &track.title)); - tokio::fs::write(file, text).await?; - Ok(()) - } - Sink::Download(downloader) => { - downloader - .download_track(client, track, dir, index, bytes_left) - .await +/// Whether the entry for track `title` at listing position `index` in +/// `dir` already has playable data: a parseable toml whose playable is not +/// skipped, and — for a `file` playable — whose audio file exists. +/// Anything else (missing, broken, skipped, audio gone) is re-captured. +async fn existing_is_satisfied(dir: &Path, index: usize, title: &str) -> bool { + let toml_path = dir.join(fsdy::track_file_name(index, title)); + let Ok(text) = tokio::fs::read_to_string(&toml_path).await else { + return false; + }; + let Ok(file) = fsdy::TrackFile::parse(&text) else { + return false; + }; + match file.playable() { + Ok(fsdy::Playable::File(target)) => { + let absolute = if target.is_absolute() { + target + } else { + dir.join(target) + }; + tokio::fs::try_exists(absolute).await.unwrap_or(false) } + Ok(fsdy::Playable::Skipped) => false, + // A url/link playable was not written by this store, but whoever + // put it there gave the entry playable data — keep it. + Ok(_) => true, + Err(_) => false, } } @@ -433,4 +641,36 @@ mod tests { ); assert_eq!(audio_file_name(11, "a/b", "mp3"), "0012 a_b.mp3"); } + + #[test] + fn progress_counts_skipped_toward_done_and_finishes_once() { + let (tx, rx) = flume::bounded(16); + let progress = Progress::new("faves", true, tx); + progress.set_total(3); + progress.track_done(); + progress.track_skipped(); + + let events: Vec<CaptureProgress> = rx.drain().collect(); + let last = events.last().expect("events published"); + assert_eq!(last.tracks_total, 3); + // Skipped counts toward done: the ratio reaches total on success. + assert_eq!(last.tracks_done, 2); + assert_eq!(last.tracks_skipped, 1); + assert!(events.iter().all(|e| !e.finished)); + } + + #[tokio::test] + async fn progress_terminal_event_carries_the_error() { + let (tx, rx) = flume::bounded(16); + let progress = Progress::new("faves", true, tx); + progress.finish(Some("boom".to_string())).await; + let event = rx.recv_async().await.expect("terminal event"); + assert!(event.finished); + assert_eq!(event.error, "boom"); + // A silent reporter must not panic anywhere. + let silent = Progress::silent("x", false); + silent.set_total(1); + silent.track_done(); + silent.finish(None).await; + } } diff --git a/crabidy-server/src/capture_store.rs b/crabidy-server/src/capture_store.rs index 646db8b..7305723 100644 --- a/crabidy-server/src/capture_store.rs +++ b/crabidy-server/src/capture_store.rs @@ -1,11 +1,13 @@ //! Downloaded library subtrees ("captures") on disk -//! (see `architecture/captures.md`). +//! (see `architecture/captures.md` and +//! `architecture/incremental-captures.md`). //! //! Every capture is a folder under `<config>/crabidy/captures/` that //! mirrors the captured subtree like a bookmark, except each track's audio //! is **downloaded** next to its order-prefixed `*.cbd-track.toml`, and the //! toml's playable is a relative `file` pointing at it — replaying a -//! capture needs no provider round trip. The same directory is mounted +//! capture needs no provider round trip. Tracks whose source cannot be +//! captured are recorded as *skipped* tomls. The same directory is mounted //! read-only into the library as `/captures` by an `fsdy` instance (with //! editable top-level folders); this module is the only writer. @@ -13,7 +15,7 @@ use std::path::{Path, PathBuf}; use crabidy_core::ProviderClient; -use crate::capture::{Caps, CaptureError}; +use crate::capture::{Caps, CaptureError, Progress}; /// The library mount point of the captures directory. pub const CAPTURES_PROVIDER_ROOT: &str = "/captures"; @@ -25,9 +27,11 @@ pub fn captures_dir() -> Option<PathBuf> { } /// Writes download captures. All audio is fetched through one shared HTTP -/// client; the whole capture is built as a hidden temp sibling and swapped -/// into place, so a crash or failed download never leaves a half-written -/// capture next to intact ones. +/// client. Captures are **incremental**: the walk writes into the final +/// folder, reuses entries that already have their audio, and keeps +/// everything written so far when a download fails — capturing the same +/// name again resumes (and completes previously skipped entries where the +/// source became capturable). #[derive(Debug)] pub struct CaptureStore { dir: PathBuf, @@ -53,16 +57,14 @@ impl CaptureStore { &self.dir } - /// Captures the subtree at `source_path` as the capture `name`, - /// downloading every track's audio; overwrites an existing capture of - /// that name. - /// - /// The capture **root** must opt in (`architecture/captures.md` D4): a - /// directory source must report `is_downloadable`, a track source's - /// parent node must — otherwise [`CaptureError::Unsupported`]. The - /// walk, caps ([`crate::capture::DOWNLOAD_CAPS`]), tmp-and-swap, and - /// all-or-nothing cleanup are [`crate::capture::capture_into`]'s. - pub async fn capture<C>( + /// Validates a capture request without writing anything: the name must + /// be a legal folder name and the capture **root** must opt in + /// (`architecture/captures.md` D4 — a directory source must report + /// `is_downloadable`, a track source's parent node must). The + /// accept-then-stream RPC replies after this and runs + /// [`Self::capture`] detached (architecture/incremental-captures.md + /// D4). + pub async fn validate<C>( &self, client: &C, source_path: &str, @@ -71,8 +73,38 @@ impl CaptureStore { where C: ProviderClient + Sync, { - self.capture_with_caps(client, source_path, name, crate::capture::DOWNLOAD_CAPS) - .await + fsdy::validate_folder_name(name, &[]).map_err(CaptureError::InvalidName)?; + source_allows_download(client, source_path).await + } + + /// Captures the subtree at `source_path` as the capture `name`, + /// downloading every track's audio — **incrementally**: an existing + /// capture of that name is resumed, not overwritten (satisfied entries + /// are reused, skipped and broken ones re-attempted). Reports through + /// `progress` (non-terminal events only; the caller sends the terminal + /// one). + /// + /// The source must pass [`Self::validate`]. The walk, caps + /// ([`crate::capture::DOWNLOAD_CAPS`]), reuse rule, and skipped tomls + /// are [`crate::capture::capture_into`]'s. + pub async fn capture<C>( + &self, + client: &C, + source_path: &str, + name: &str, + progress: &Progress, + ) -> Result<(), CaptureError> + where + C: ProviderClient + Sync, + { + self.capture_with_caps( + client, + source_path, + name, + crate::capture::DOWNLOAD_CAPS, + progress, + ) + .await } /// [`Self::capture`] with explicit caps — the seam the cap tests use. @@ -82,12 +114,22 @@ impl CaptureStore { source_path: &str, name: &str, caps: Caps, + progress: &Progress, ) -> Result<(), CaptureError> where C: ProviderClient + Sync, { source_allows_download(client, source_path).await?; - crate::capture::capture_into(&self.dir, client, source_path, name, caps, &self.sink).await + crate::capture::capture_into( + &self.dir, + client, + source_path, + name, + caps, + &self.sink, + progress, + ) + .await } } @@ -199,6 +241,7 @@ mod tests { title: title.to_string(), duration: Some(10), album: None, + is_skipped: false, } } } @@ -287,13 +330,27 @@ mod tests { names } + /// A silent progress reporter for tests that do not assert on events. + fn silent() -> Progress { + Progress::silent("test", true) + } + + /// Parses the capture entry's toml and returns its playable. + fn playable_of(dir: &Path, name: &str) -> fsdy::Playable { + let text = fs::read_to_string(dir.join(name)).expect("toml"); + fsdy::TrackFile::parse(&text) + .expect("parses") + .playable() + .expect("playable") + } + #[tokio::test] async fn download_capture_writes_audio_next_to_pointing_tomls() { let url = serve("200 OK", "audio/flac", b"flacbytes".to_vec()).await; let mock = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], true); let (store, _dir) = store().await; store - .capture(&mock, "/mock/a", "faves") + .capture(&mock, "/mock/a", "faves", &silent()) .await .expect("capture"); @@ -312,13 +369,10 @@ mod tests { b"flacbytes" ); // The toml points at its sibling with a *relative* file playable. - let toml = fs::read_to_string(root.join("0002 two.cbd-track.toml")).expect("toml"); - let file = fsdy::TrackFile::parse(&toml).expect("parses"); assert_eq!( - file.playable().expect("playable"), + playable_of(&root, "0002 two.cbd-track.toml"), fsdy::Playable::File("0002 two.flac".into()) ); - assert_eq!(file.title, "two"); // Replay: the capture resolves through a /captures instance and the // audio resolves to the absolute sibling path — no provider round @@ -344,12 +398,18 @@ mod tests { let url = serve("200 OK", "audio/flac", b"x".to_vec()).await; let mock = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], false); let (store, _dir) = store().await; + // validate() is what the accept-then-stream RPC checks up front... assert!(matches!( - store.capture(&mock, "/mock/a", "faves").await, + store.validate(&mock, "/mock/a", "faves").await, + Err(CaptureError::Unsupported) + )); + // ...and the walk itself re-checks, for both node and track roots. + assert!(matches!( + store.capture(&mock, "/mock/a", "faves", &silent()).await, Err(CaptureError::Unsupported) )); assert!(matches!( - store.capture(&mock, "/mock/a/1", "faves").await, + store.capture(&mock, "/mock/a/1", "faves", &silent()).await, Err(CaptureError::Unsupported) )); assert_eq!(fs::read_dir(store.dir()).expect("store dir").count(), 0); @@ -361,7 +421,7 @@ mod tests { let mock = MockProvider::new(&[("/mock/a/1", &url)], true); let (store, _dir) = store().await; store - .capture(&mock, "/mock/a/1", "just one") + .capture(&mock, "/mock/a/1", "just one", &silent()) .await .expect("capture track"); assert_eq!( @@ -371,18 +431,41 @@ mod tests { } #[tokio::test] - async fn download_capture_is_all_or_nothing() { - let ok = serve("200 OK", "audio/flac", b"x".to_vec()).await; + async fn download_failure_keeps_progress_and_resuming_completes() { + let ok = serve("200 OK", "audio/flac", b"first".to_vec()).await; let gone = serve("404 Not Found", "text/plain", Vec::new()).await; let mock = MockProvider::new(&[("/mock/a/1", &ok), ("/mock/a/2", &gone)], true); let (store, _dir) = store().await; let err = store - .capture(&mock, "/mock/a", "faves") + .capture(&mock, "/mock/a", "faves", &silent()) .await - .expect_err("a failed download aborts the capture"); + .expect_err("a failed download aborts the run"); assert!(matches!(err, CaptureError::Download(_)), "got {err:?}"); - // Nothing half-written survives, not even hidden temp folders. - assert_eq!(fs::read_dir(store.dir()).expect("store dir").count(), 0); + // Incremental: what downloaded before the failure survives + // (architecture/incremental-captures.md D2). + let root = store.dir().join("faves"); + assert!(root.join("0001 one.cbd-track.toml").exists()); + assert!(root.join("0001 one.flac").exists()); + assert!(!root.join("0002 two.cbd-track.toml").exists()); + + // Resume with a healthy source: the satisfied entry is reused (the + // sentinel content is not re-downloaded), the missing one arrives. + fs::write(root.join("0001 one.flac"), b"sentinel").expect("stamp"); + let fixed = MockProvider::new(&[("/mock/a/1", &ok), ("/mock/a/2", &ok)], true); + store + .capture(&fixed, "/mock/a", "faves", &silent()) + .await + .expect("resume completes"); + assert_eq!( + fs::read(root.join("0001 one.flac")).expect("audio"), + b"sentinel", + "satisfied entries must not be re-downloaded" + ); + assert!(root.join("0002 two.cbd-track.toml").exists()); + assert_eq!( + fs::read(root.join("0002 two.flac")).expect("audio"), + b"first" + ); } #[tokio::test] @@ -390,31 +473,36 @@ mod tests { let url = serve("200 OK", "audio/flac", b"0123456789".to_vec()).await; let mock = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], true); let (store, _dir) = store().await; - // Byte budget: two 10-byte bodies against a 15-byte budget. + // Byte budget: two 10-byte bodies against a 15-byte budget. The + // first track fits, the second trips the budget — and survives as + // partial progress (no toml, so a resume re-downloads it). let caps = Caps { max_bytes: 15, ..DOWNLOAD_CAPS }; let err = store - .capture_with_caps(&mock, "/mock/a", "big", caps) + .capture_with_caps(&mock, "/mock/a", "big", caps, &silent()) .await .expect_err("over the byte budget"); assert!(matches!(err, CaptureError::TooLarge(_)), "got {err:?}"); - // Track cap. + let root = store.dir().join("big"); + assert!(root.join("0001 one.cbd-track.toml").exists()); + assert!(!root.join("0002 two.cbd-track.toml").exists()); + // Track cap: enumeration fails before anything is fetched. let caps = Caps { max_tracks: 1, ..DOWNLOAD_CAPS }; let err = store - .capture_with_caps(&mock, "/mock/a", "big", caps) + .capture_with_caps(&mock, "/mock/a", "big2", caps, &silent()) .await .expect_err("over the track cap"); assert!(matches!(err, CaptureError::TooLarge(_)), "got {err:?}"); - assert_eq!(fs::read_dir(store.dir()).expect("store dir").count(), 0); + assert_eq!(visible(&store.dir().join("big2")), Vec::<String>::new()); } #[tokio::test] - async fn download_capture_skips_uncapturable_tracks() { + async fn uncapturable_tracks_are_recorded_as_skipped() { // A queue-like source: a downloadable fsdy instance whose folder // mixes an http track with a local-file track — like a persisted // queue linking tidal and fs entries. @@ -439,44 +527,126 @@ mod tests { let (store, _dir) = store().await; store - .capture(&source, "/queues/mix", "mixed") + .capture(&source, "/queues/mix", "mixed", &silent()) .await .expect("capture succeeds despite the local track"); - // The http track downloaded; the local-file track was skipped - // (no pair, no abort), keeping its listing position's prefix. + // The http track downloaded; the local-file track was recorded as + // a skipped toml at its listing position — no silent omission + // (architecture/incremental-captures.md D2). + let root = store.dir().join("mixed"); assert_eq!( - visible(&store.dir().join("mixed")), + visible(&root), vec![ "0001 web.cbd-track.toml".to_string(), - "0001 web.flac".into() + "0001 web.flac".into(), + "0002 local.cbd-track.toml".into(), + ] + ); + assert_eq!( + playable_of(&root, "0002 local.cbd-track.toml"), + fsdy::Playable::Skipped + ); + // The skipped entry lists as a skipped wire track. + let captures = fsdy::Client::new(CAPTURES_PROVIDER_ROOT, store.dir().to_path_buf()) + .expect("captures instance"); + let node = captures + .get_lib_node("/captures/mixed") + .await + .expect("node"); + assert_eq!(node.tracks.len(), 2); + assert!(!node.tracks[0].is_skipped); + assert!(node.tracks[1].is_skipped); + } + + #[tokio::test] + async fn skipped_entries_are_reattempted_on_resume() { + let url = serve("200 OK", "audio/flac", b"flacbytes".to_vec()).await; + // Track two has no stream URL: recorded as skipped. + let mock = MockProvider::new(&[("/mock/a/1", &url)], true); + let (store, _dir) = store().await; + store + .capture(&mock, "/mock/a", "faves", &silent()) + .await + .expect("capture with a skipped entry"); + let root = store.dir().join("faves"); + assert_eq!( + playable_of(&root, "0002 two.cbd-track.toml"), + fsdy::Playable::Skipped + ); + + // The source became capturable: re-capturing the same name + // completes the skipped entry and reuses the satisfied one. + fs::write(root.join("0001 one.flac"), b"sentinel").expect("stamp"); + let fixed = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], true); + store + .capture(&fixed, "/mock/a", "faves", &silent()) + .await + .expect("resume"); + assert_eq!( + playable_of(&root, "0002 two.cbd-track.toml"), + fsdy::Playable::File("0002 two.flac".into()) + ); + assert_eq!( + fs::read(root.join("0001 one.flac")).expect("audio"), + b"sentinel" + ); + } + + #[tokio::test] + async fn capture_validates_names_and_merges_into_existing() { + let url = serve("200 OK", "audio/flac", b"x".to_vec()).await; + let mock = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], true); + let (store, _dir) = store().await; + assert!(matches!( + store.validate(&mock, "/mock/a", "a/b").await, + Err(CaptureError::InvalidName(_)) + )); + assert!(matches!( + store.capture(&mock, "/mock/a", "a/b", &silent()).await, + Err(CaptureError::InvalidName(_)) + )); + store + .capture(&mock, "/mock/a", "faves", &silent()) + .await + .expect("first capture"); + // Capturing a smaller source into the same name merges: the + // single-track capture reuses its entry, the rest stays. + store + .capture(&mock, "/mock/a/1", "faves", &silent()) + .await + .expect("merge"); + assert_eq!( + visible(&store.dir().join("faves")), + vec![ + "0001 one.cbd-track.toml".to_string(), + "0001 one.flac".into(), + "0002 two.cbd-track.toml".into(), + "0002 two.flac".into(), ] ); } #[tokio::test] - async fn capture_validates_names_and_overwrites() { + async fn capture_reports_progress_totals_and_skips() { let url = serve("200 OK", "audio/flac", b"x".to_vec()).await; - let mock = MockProvider::new(&[("/mock/a/1", &url), ("/mock/a/2", &url)], true); + // Track two is uncapturable: one done, one skipped, total two. + let mock = MockProvider::new(&[("/mock/a/1", &url)], true); let (store, _dir) = store().await; - assert!(matches!( - store.capture(&mock, "/mock/a", "a/b").await, - Err(CaptureError::InvalidName(_)) - )); + let (tx, rx) = flume::bounded(64); + let progress = Progress::new("faves", true, tx); store - .capture(&mock, "/mock/a", "faves") + .capture(&mock, "/mock/a", "faves", &progress) .await - .expect("first capture"); - store - .capture(&mock, "/mock/a/1", "faves") - .await - .expect("overwrite"); - // The overwrite fully replaces the older, larger capture. - assert_eq!( - visible(&store.dir().join("faves")), - vec![ - "0001 one.cbd-track.toml".to_string(), - "0001 one.flac".into() - ] - ); + .expect("capture"); + progress.finish(None).await; + let events: Vec<_> = rx.drain().collect(); + assert!(events.iter().any(|e| e.tracks_total == 2 && !e.finished)); + let last = events.last().expect("terminal event"); + assert!(last.finished); + assert!(last.error.is_empty()); + assert_eq!(last.tracks_done, 2, "skipped counts toward done"); + assert_eq!(last.tracks_skipped, 1); + assert!(last.download); + assert_eq!(last.name, "faves"); } } diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index de500e2..12dd1f4 100644 --- a/crabidy-server/src/lib.rs +++ b/crabidy-server/src/lib.rs @@ -261,6 +261,16 @@ impl QueueManager { } } + /// Number of tracks in the queue (playback uses it to bound skip + /// loops: at most one full pass, even with repeat on). + pub fn len(&self) -> usize { + self.tracks.len() + } + + pub fn is_empty(&self) -> bool { + self.tracks.is_empty() + } + pub fn is_last_track(&self) -> bool { !self.tracks.is_empty() && self.current_position() == self.tracks.len() - 1 } @@ -518,6 +528,7 @@ mod tests { title: format!("track {id}"), duration: None, album: None, + is_skipped: false, } } @@ -735,13 +746,19 @@ pub enum ProviderCommand { /// Captures the queueable subtree at `path` as the bookmark `name` /// (see `architecture/bookmarks.md` D1–D3) — or, with `download`, as /// the download capture `name` under `/captures`, fetching every - /// track's audio (see `architecture/captures.md`). Handled on a - /// spawned task — a large walk or download must not block the - /// orchestrator loop. + /// track's audio (see `architecture/captures.md` and + /// `architecture/incremental-captures.md`). Handled on a spawned + /// task — a large walk or download must not block the orchestrator + /// loop. `result_tx` answers once the capture is *accepted* + /// (validation only); the walk then streams `CaptureProgress` events + /// on `progress_tx`, ending in exactly one `finished` event (with + /// `error` set on failure). A rejected capture answers with the error + /// and sends no progress events. CaptureLibraryNode { path: String, name: String, download: bool, + progress_tx: flume::Sender<crabidy_core::proto::crabidy::CaptureProgress>, result_tx: flume::Sender<Result<(), crate::capture::CaptureError>>, }, } diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index b3a2b6d..898e689 100644 --- a/crabidy-server/src/playback.rs +++ b/crabidy-server/src/playback.rs @@ -660,37 +660,67 @@ impl Playback { } } - /// Starts playback of the given track. When fetching stream URLs fails - /// the failing track is skipped and playback continues with the next - /// track in the queue. + /// Finds the stream URLs of the first playable track, starting at + /// `track` and advancing the queue past unplayable ones. Tracks marked + /// `is_skipped` (captures recorded their source as uncapturable) are + /// skipped without a provider round trip; tracks whose stream URLs + /// fail to resolve are skipped with a warning. Bounded by the queue + /// length at entry — one full pass at most — so an all-skipped queue + /// with repeat on returns `None` instead of spinning + /// (architecture/incremental-captures.md D3). + async fn next_playable_urls(&self, mut track: Track) -> Option<Vec<String>> { + let mut attempts_left = { + let Ok(queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return None; + }; + queue.len() + }; + loop { + let path = track.path.as_str(); + if track.is_skipped { + debug!(path, "track is marked skipped, skipping"); + } else { + match self.get_urls_for_track(path).await { + Ok(urls) if !urls.is_empty() => return Some(urls), + Ok(_) => warn!(path, "provider returned no stream urls, skipping track"), + Err(err) => warn!(path, "failed to fetch stream urls ({err}), skipping track"), + } + } + attempts_left = attempts_left.saturating_sub(1); + if attempts_left == 0 { + warn!("no playable track in the queue after a full pass"); + return None; + } + let next = { + let Ok(mut queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return None; + }; + queue.next_track() + }; + match next { + Some(next_track) => track = next_track, + None => { + debug!("reached the end of the queue without a playable track"); + return None; + } + } + } + } + + /// Starts playback of the given track, skipping past unplayable ones + /// (see [`Self::next_playable_urls`]); stops the player when nothing + /// in the queue is playable. #[instrument(skip(self, track), fields(track = track.as_ref().map(|t| t.path.as_str())))] async fn play(&self, track: Option<Track>) { let Some(track) = track else { debug!("nothing to play"); return; }; - let mut path = track.path.clone(); - let urls = loop { - match self.get_urls_for_track(&path).await { - Ok(urls) if !urls.is_empty() => break urls, - Ok(_) => warn!(path, "provider returned no stream urls, skipping track"), - Err(err) => warn!(path, "failed to fetch stream urls ({err}), skipping track"), - } - let next = { - let Ok(mut queue) = self.queue.lock() else { - error!("queue lock poisoned"); - return; - }; - queue.next_track() - }; - match next { - Some(next_track) => path = next_track.path.clone(), - None => { - error!("no playable track left in queue, stopping"); - self.stop_player().await; - return; - } - } + let Some(urls) = self.next_playable_urls(track).await else { + self.stop_player().await; + return; }; { let Ok(queue) = self.queue.lock() else { @@ -724,6 +754,7 @@ mod tests { title: format!("track {i}"), duration: None, album: None, + is_skipped: false, } } @@ -843,6 +874,32 @@ mod tests { assert!(matches!(result, Err(SaveQueueError::EmptyQueue))); } + #[tokio::test] + async fn skipped_tracks_are_skipped_with_a_bounded_pass() { + // All tracks are marked skipped and repeat is on: `next_track` + // cycles forever, so only the one-full-pass bound ends the loop + // with `None`. The marked tracks are skipped without any provider + // round trip (the provider channel is closed — a call would fail, + // not hang). + let playback = playback_with(None); + let tracks: Vec<Track> = (0..3) + .map(|i| Track { + is_skipped: true, + ..track(i) + }) + .collect(); + let first = { + let mut queue = playback.queue.lock().expect("queue lock"); + let first = queue.replace_with_tracks(&tracks); + queue.repeat = true; + first + }; + let urls = playback + .next_playable_urls(first.expect("first track")) + .await; + assert!(urls.is_none(), "an all-skipped queue has nothing playable"); + } + #[tokio::test] async fn queue_mutations_reach_the_persist_channel() { let dir = TempDir::new().expect("tempdir"); diff --git a/crabidy-server/src/provider.rs b/crabidy-server/src/provider.rs index 6d53b92..82afb69 100644 --- a/crabidy-server/src/provider.rs +++ b/crabidy-server/src/provider.rs @@ -188,32 +188,58 @@ impl ProviderOrchestrator { path, name, download, + progress_tx, result_tx, } => { // Spawned: capturing a large artist walks many provider // nodes (and, for downloads, streams audio) and must not // block this loop (the walk itself calls back into - // `get_lib_node` via `this`). + // `get_lib_node` via `this`). Accept-then-stream + // (architecture/incremental-captures.md D4): validation + // answers the RPC, the walk reports through progress + // events, ending in exactly one terminal event. let this = Arc::clone(&self); tokio::spawn( async move { - let result = if download { + let accepted = if download { match &this.capture_store { - Some(store) => store.capture(&*this, &path, &name).await, + Some(store) => store.validate(&*this, &path, &name).await, None => Err(crate::capture::CaptureError::Disabled), } } else { match &this.bookmark_store { - Some(store) => store.capture(&*this, &path, &name).await, + Some(store) => store.validate(&name), + None => Err(crate::capture::CaptureError::Disabled), + } + }; + if let Err(err) = accepted { + warn!(path, name, download, "capture rejected: {err}"); + if let Err(err) = result_tx.send_async(Err(err)).await { + error!("failed to send capture_library_node result: {err}"); + } + return; + } + if let Err(err) = result_tx.send_async(Ok(())).await { + error!("failed to send capture_library_node result: {err}"); + } + let progress = crate::capture::Progress::new(&name, download, progress_tx); + let result = if download { + match &this.capture_store { + Some(store) => store.capture(&*this, &path, &name, &progress).await, + None => Err(crate::capture::CaptureError::Disabled), + } + } else { + match &this.bookmark_store { + Some(store) => store.capture(&*this, &path, &name, &progress).await, None => Err(crate::capture::CaptureError::Disabled), } }; if let Err(err) = &result { warn!(path, name, download, "cannot capture subtree: {err}"); } - if let Err(err) = result_tx.send_async(result).await { - error!("failed to send capture_library_node result: {err}"); - } + progress + .finish(result.err().map(|err| err.to_string())) + .await; } .in_current_span(), ); diff --git a/crabidy-server/src/queue_store.rs b/crabidy-server/src/queue_store.rs index 6d344a3..345ab46 100644 --- a/crabidy-server/src/queue_store.rs +++ b/crabidy-server/src/queue_store.rs @@ -307,6 +307,7 @@ mod tests { title: "album".to_string(), release_date: None, }), + is_skipped: false, } } diff --git a/crabidy-server/src/rpc.rs b/crabidy-server/src/rpc.rs index 8e2f2af..6cfb9a2 100644 --- a/crabidy-server/src/rpc.rs +++ b/crabidy-server/src/rpc.rs @@ -378,12 +378,19 @@ impl CrabidyService for RpcService { /// Captures a queueable subtree as a bookmark via the provider loop /// (structure-preserving snapshot under `/bookmarks/<name>`), or — - /// with `download` — as a download capture under `/captures/<name>`. + /// with `download` — as a download capture under `/captures/<name>` + /// (incremental: an existing name is resumed, not overwritten). + /// + /// Returns once the capture is **accepted**: the reply covers + /// validation only, the walk runs detached and reports through + /// `CaptureProgress` events on the update stream (forwarded here from + /// the provider's progress channel), ending in one `finished` event. /// /// Error mapping is part of the contract: invalid name or - /// uncapturable source → `invalid_argument`; an over-cap subtree, - /// disabled store, or a source that does not allow downloads → - /// `failed_precondition`; walk/write/download failures → `internal`. + /// uncapturable source → `invalid_argument`; a disabled store or a + /// source that does not allow downloads → `failed_precondition`. + /// Walk/write/download failures happen after the reply and surface in + /// the terminal progress event (and the server log). #[instrument(skip(self, request), fields(path, name, download))] async fn capture_library_node( &self, @@ -398,12 +405,24 @@ impl CrabidyService for RpcService { tracing::Span::current().record("name", name.as_str()); tracing::Span::current().record("download", download); debug!("received capture_library_node request"); + // The walk's progress events fan out to every connected client via + // the update broadcast; the forwarder dies with the walk's terminal + // event (the provider drops the sender). + let (progress_tx, progress_rx) = flume::bounded(64); + let update_tx = self.update_tx.clone(); + tokio::spawn(async move { + while let Ok(progress) = progress_rx.recv_async().await { + // No subscribers is normal (e.g. no client connected). + let _ = update_tx.send(StreamUpdate::CaptureProgress(progress)); + } + }); let (result_tx, result_rx) = flume::bounded(1); self.provider_tx .send_async(ProviderMessage::new(ProviderCommand::CaptureLibraryNode { path, name, download, + progress_tx, result_tx, })) .await diff --git a/fsdy/src/lib.rs b/fsdy/src/lib.rs index 6573f06..0694594 100644 --- a/fsdy/src/lib.rs +++ b/fsdy/src/lib.rs @@ -48,7 +48,7 @@ pub struct Settings { pub enum TrackFileError { #[error("not valid TOML: {0}")] Toml(#[from] toml::de::Error), - #[error("[playable] must set exactly one of `file`, `url`, `link`")] + #[error("[playable] must set exactly one of `file`, `url`, `link`, `skipped = true`")] PlayableCardinality, /// Carries only the offending *scheme* — a private stream URL may /// embed a token, and this error's message ends up in logs. @@ -91,7 +91,7 @@ pub struct AlbumMeta { pub release_date: Option<String>, } -/// Raw `[playable]` table: three optional fields so cardinality errors are +/// Raw `[playable]` table: four optional fields so cardinality errors are /// precise. Validated into a [`Playable`] before use. #[derive(Debug, Deserialize, Serialize)] pub struct PlayableSpec { @@ -108,6 +108,12 @@ pub struct PlayableSpec { /// itself a link file fails at play time and cycles cannot recurse. #[serde(skip_serializing_if = "Option::is_none")] pub link: Option<String>, + /// `true` marks the track as having no playable audio: a capture + /// recorded its source as uncapturable (architecture/ + /// incremental-captures.md D1). Must be the only field set; `false` + /// counts as unset and fails cardinality validation. + #[serde(skip_serializing_if = "Option::is_none")] + pub skipped: Option<bool>, } /// A validated playable reference. @@ -116,6 +122,8 @@ pub enum Playable { File(PathBuf), Url(String), Link(String), + /// No playable audio; playback skips the track. + Skipped, } impl TrackFile { @@ -133,9 +141,19 @@ impl TrackFile { /// The validated playable reference. pub fn playable(&self) -> Result<Playable, TrackFileError> { - match (&self.playable.file, &self.playable.url, &self.playable.link) { - (Some(file), None, None) => Ok(Playable::File(file.clone())), - (None, Some(url), None) => { + // `skipped = false` counts as unset — only `true` marks a track. + let skipped = match self.playable.skipped { + Some(true) => Some(()), + _ => None, + }; + match ( + &self.playable.file, + &self.playable.url, + &self.playable.link, + skipped, + ) { + (Some(file), None, None, None) => Ok(Playable::File(file.clone())), + (None, Some(url), None, None) => { let scheme = url::Url::parse(url) .map(|u| u.scheme().to_string()) .unwrap_or_else(|_| "<not a url>".to_string()); @@ -145,12 +163,13 @@ impl TrackFile { } Ok(Playable::Url(url.clone())) } - (None, None, Some(link)) => { + (None, None, Some(link), None) => { if !link.starts_with('/') { return Err(TrackFileError::LinkNotAbsolute(link.clone())); } Ok(Playable::Link(link.clone())) } + (None, None, None, Some(())) => Ok(Playable::Skipped), _ => Err(TrackFileError::PlayableCardinality), } } @@ -159,11 +178,13 @@ impl TrackFile { /// /// For a [`Playable::Link`] the returned track's `path` is the **link /// target**, not `lib_path` — from then on the track routes to the - /// provider that owns the target (architecture D2). File and URL - /// playables keep `lib_path`. + /// provider that owns the target (architecture D2). File, URL, and + /// skipped playables keep `lib_path`; a skipped playable additionally + /// sets the wire track's `is_skipped` flag. pub fn to_track(&self, lib_path: &str) -> Track { - let path = match self.playable() { - Ok(Playable::Link(target)) => target, + let playable = self.playable(); + let path = match &playable { + Ok(Playable::Link(target)) => target.clone(), _ => lib_path.to_string(), }; Track { @@ -175,6 +196,7 @@ impl TrackFile { title: a.title.clone(), release_date: a.release_date.clone(), }), + is_skipped: matches!(playable, Ok(Playable::Skipped)), } } @@ -184,7 +206,27 @@ impl TrackFile { /// `architecture/queue-persistence.md` D2). Round trip: /// `from_track(t).to_track(anywhere)` yields `t` again, because links /// rewrite the path back to the target at listing time. + /// + /// A skipped source track (`is_skipped`) writes a skipped playable + /// instead of a link: there is nothing behind it to link to, and + /// persisted queues and bookmarks must keep the skipped marking + /// (architecture/incremental-captures.md D1). pub fn from_track(track: &Track) -> Self { + let playable = if track.is_skipped { + PlayableSpec { + file: None, + url: None, + link: None, + skipped: Some(true), + } + } else { + PlayableSpec { + file: None, + url: None, + link: Some(track.path.clone()), + skipped: None, + } + }; Self { title: track.title.clone(), artist: track.artist.clone(), @@ -193,11 +235,7 @@ impl TrackFile { title: a.title.clone(), release_date: a.release_date.clone(), }), - playable: PlayableSpec { - file: None, - url: None, - link: Some(track.path.clone()), - }, + playable, } } @@ -213,6 +251,22 @@ impl TrackFile { file: Some(file.to_path_buf()), url: None, link: None, + skipped: None, + }; + this + } + + /// Like [`Self::from_track`], but the playable is [`Playable::Skipped`] + /// regardless of the source track — used by download captures to record + /// an uncapturable track instead of silently omitting it + /// (architecture/incremental-captures.md D2). + pub fn from_track_skipped(track: &Track) -> Self { + let mut this = Self::from_track(track); + this.playable = PlayableSpec { + file: None, + url: None, + link: None, + skipped: Some(true), }; this } @@ -606,6 +660,8 @@ impl ProviderClient for Client { /// [`Playable::Link`] cannot be reached through normal flow (the /// track's path was rewritten at listing time, see [`TrackFile::to_track`]) /// and is [`ProviderError::MalformedPath`] with a warning. + /// [`Playable::Skipped`] has no audio and is [`ProviderError::FetchError`] + /// — playback normally never asks (it skips on the track's wire flag). async fn get_urls_for_track(&self, track_path: &str) -> Result<Vec<String>, ProviderError> { if !self.is_track_path(track_path) { return Err(ProviderError::MalformedPath); @@ -640,6 +696,10 @@ impl ProviderClient for Client { ); Err(ProviderError::MalformedPath) } + Playable::Skipped => { + warn!(path = track_path, "skipped tracks have no audio"); + Err(ProviderError::FetchError) + } } } @@ -824,7 +884,11 @@ mod tests { fn parse_rejects_wrong_playable_cardinality() { let none = "title = \"x\"\n[playable]\n"; let both = "title = \"x\"\n[playable]\nurl = \"https://a\"\nlink = \"/tidal/t\"\n"; - for bad in [none, both] { + // `skipped = false` counts as unset; combined with another field it + // is fine, alone it sets nothing. + let false_only = "title = \"x\"\n[playable]\nskipped = false\n"; + let skipped_and_url = "title = \"x\"\n[playable]\nurl = \"https://a\"\nskipped = true\n"; + for bad in [none, both, false_only, skipped_and_url] { let err = TrackFile::parse(bad) .and_then(|f| f.playable().map(|_| f)) .expect_err("cardinality must be rejected"); @@ -832,6 +896,39 @@ mod tests { } } + #[test] + fn skipped_playable_parses_marks_and_round_trips() { + let file = TrackFile::parse("title = \"gone\"\n[playable]\nskipped = true\n") + .expect("skipped parses"); + assert_eq!(file.playable().expect("playable"), Playable::Skipped); + + // The wire track keeps the lib path and carries the flag. + let lib_path = "/captures/mix/0001 gone.cbd-track.toml"; + let track = file.to_track(lib_path); + assert_eq!(track.path, lib_path); + assert!(track.is_skipped); + + // Persistence keeps skipped-ness: a skipped wire track serializes + // back to a skipped playable, not a dead link. + let persisted = TrackFile::from_track(&track); + assert_eq!(persisted.playable().expect("playable"), Playable::Skipped); + let text = persisted.to_toml().expect("serialize"); + let reparsed = TrackFile::parse(&text).expect("round trip"); + assert!(reparsed.to_track(lib_path).is_skipped); + } + + #[test] + fn from_track_skipped_overrides_any_source_playable() { + // Download captures record an uncapturable track as skipped even + // though the source track itself was a normal (e.g. tidal) track. + let stream = TrackFile::parse(&url_track("t")).expect("valid"); + let track = stream.to_track("/fs/t.cbd-track.toml"); + assert!(!track.is_skipped); + let skipped = TrackFile::from_track_skipped(&track); + assert_eq!(skipped.playable().expect("playable"), Playable::Skipped); + assert_eq!(skipped.title, "t"); + } + #[test] fn parse_rejects_invalid_playables() { let cases = [ @@ -1030,6 +1127,18 @@ mod tests { .await .expect_err("link"); assert_eq!(err, ProviderError::MalformedPath); + // Skipped playables have no audio: a typed error, not empty success + // (playback normally never asks — it skips on the wire flag). + write_track( + &sub, + "gone.cbd-track.toml", + "title = \"g\"\n[playable]\nskipped = true\n", + ); + let err = client + .get_urls_for_track("/fs/mix/gone.cbd-track.toml") + .await + .expect_err("skipped"); + assert_eq!(err, ProviderError::FetchError); } #[tokio::test] @@ -1181,6 +1290,7 @@ mod tests { title: "News of the World".to_string(), release_date: Some("1977-10-28".to_string()), }), + is_skipped: false, }; let toml_text = TrackFile::from_track(&track).to_toml().expect("serialize"); let reparsed = TrackFile::parse(&toml_text).expect("reparse"); @@ -1205,6 +1315,7 @@ mod tests { title: "Greatest".to_string(), release_date: None, }), + is_skipped: false, }; let file = TrackFile::from_track_with_file(&track, Path::new("0001 One.flac")); let toml_text = file.to_toml().expect("serialize"); @@ -1249,6 +1360,7 @@ mod tests { title: "stream".to_string(), duration: None, album: None, + is_skipped: false, }; let toml_text = TrackFile::from_track(&track).to_toml().expect("serialize"); let reparsed = TrackFile::parse(&toml_text).expect("reparse"); diff --git a/plan/incremental-captures.md b/plan/incremental-captures.md new file mode 100644 index 0000000..d8a8e44 --- /dev/null +++ b/plan/incremental-captures.md @@ -0,0 +1,32 @@ +# Plan: incremental-captures + +Ordered tasks; each names its verification (tests and/or gates in +`quality/incremental-captures.md`). + +- [x] **T1 — Skipped playable in fsdy + proto.** `PlayableSpec.skipped`, + `Playable::Skipped`, cardinality validation, `from_track_skipped`, + `from_track` preserves skipped-ness, `to_track` sets `is_skipped`; + `Track.is_skipped = 6` in the proto; fix all struct literals. Verifies: + fsdy parse/round-trip tests; gates "Skipped playable". +- [x] **T2 — Two-phase walk + incremental download sink.** Enumerate + (caps) then fetch; `Sink::Download` writes into the final folder, + reuses satisfied entries, writes skipped tomls for uncapturable tracks, + keeps progress on abort; `Sink::Link` keeps tmp-and-swap. Progress + callback plumbed through. Verifies: capture/capture_store tests + (resume, reuse, skipped toml, abort-keeps-progress, caps); gates + "Incremental capture". +- [x] **T3 — Proto `CaptureProgress` + accept-then-stream RPC.** New + oneof update; provider validates then replies and spawns the walk with + a bounded progress channel; rpc forwards progress into the update + broadcast. Verifies: gates "Progress + RPC". +- [x] **T4 — Playback skip.** Flag-based skip without provider call; + skip loop bounded by queue length. Verifies: playback tests; gate + "Playback". +- [x] **T5 — TUI.** Red skipped tracks (queue + library), capture + progress lines with expiry, warnings in help + input label, focused + selection contrast fix. Verifies: cbd-tui render/dispatch tests; gates + "TUI". +- [x] **T6 — Full verification.** Workspace suite green; clippy/fmt/ + taplo/markdownlint clean; quality gates ticked. +- [x] **T7 — Docs.** `plan/summary.md` section incl. deviations; + reconcile `architecture/incremental-captures.md`. diff --git a/plan/summary.md b/plan/summary.md index 7ddc081..95e89dc 100644 --- a/plan/summary.md +++ b/plan/summary.md @@ -1,5 +1,54 @@ # Implementation summaries +## incremental-captures (2026-07-21) + +Built per `plan/incremental-captures.md`: download captures are now +**incremental and resumable**, uncapturable tracks are first-class +**skipped** entries, and captures stream **progress** to clients. + +- **Skipped playable (fsdy + proto).** `[playable] skipped = true` is a + fourth, mutually exclusive playable; `Playable::Skipped`, + `from_track_skipped`, and a new wire flag `Track.is_skipped` (set by + `to_track`). `from_track` preserves skipped-ness, so persisted queues + and bookmarks keep the marking instead of degrading it into a dead + link. `get_urls_for_track` on a skipped file is a typed `FetchError`. +- **Incremental walk.** The shared capture walk now runs in two phases: + enumerate (dirs + tracks, caps enforced — the total is known before + the first download) then fetch. `Sink::Download` writes straight into + `captures/<name>` (no tmp/swap): satisfied entries — parseable toml, + non-skipped playable, audio present — are reused; skipped, broken, or + audio-less entries are re-captured; uncapturable sources (skipped + source track, unresolvable stream, non-http playable) are recorded as + skipped tomls instead of silently omitted; a real download failure + aborts the run but keeps everything written, so re-capturing the same + name resumes. The byte budget counts only bytes downloaded per run. + Bookmarks keep tmp-and-swap overwrite semantics unchanged. +- **Progress + accept-then-stream RPC.** New `CaptureProgress` update on + the stream (name, download, done/total/skipped, terminal + finished/error). `CaptureLibraryNode` replies once validation (name, + store, download blessing) passes; the walk runs detached and its + bounded progress channel is forwarded into the update broadcast. This + also unfreezes the TUI: its poll loop used to await the whole capture. +- **Playback.** `play` skips `is_skipped` tracks without a provider round + trip and bounds the whole skip loop to one full queue pass — an + all-skipped queue with repeat on now stops instead of hammering the + provider forever (pre-existing spin fixed). +- **TUI.** Skipped tracks render red in queue and library (playing-track + marker keeps precedence). A `CaptureBoard` renders progress lines at + the bottom of the library pane (`capturing faves 3/12 (1 skipped)`), + lingering 5 s on success and 10 s (red) on failure. The `W` help entry + and the capture input label warn that captures are slow and resumable. + Contrast fix: colored items (editable/marked/skipped/current) switch to + the dark foreground under the focused selection bar. + +Deviations from the architecture doc: `tracks_done` counts skipped +entries too (the ratio must reach the total on success) — doc and proto +reconciled; the input-overlay warning was shortened to +"capture (slow, resumable)" to fit narrow panes. `taplo` reports a +pre-existing formatting issue in `.opencode/skills/skill-authoring/` +(not touched here). 167 workspace tests green (13 new); +every gate in `quality/incremental-captures.md` checked. + ## cbd-bundle (2026-07-21) Built per `plan/cbd-bundle.md`: a new **`cbd`** binary bundles server diff --git a/quality/incremental-captures.md b/quality/incremental-captures.md new file mode 100644 index 0000000..44b2bf0 --- /dev/null +++ b/quality/incremental-captures.md @@ -0,0 +1,66 @@ +# Quality gates: incremental-captures + +Criteria beyond the automatic tests. Each gate is pass/fail by reading the +code. Tests live next to the modules they cover (fsdy, capture, +capture_store, playback, cbd-tui). + +## Skipped playable (D1) + +- [x] `[playable] skipped = true` parses; `skipped = false`, `skipped` + combined with another field, and an empty table are typed + `PlayableCardinality`-style errors — never panics. +- [x] `from_track` on an `is_skipped` wire track writes a skipped playable + (queue persistence and bookmarks keep skipped-ness); `to_track` sets + `is_skipped` and keeps the lib path. +- [x] `get_urls_for_track` on a skipped file is a typed error, not empty-vec + success. + +## Incremental capture (D2) + +- [x] Download captures write into the final folder with no tmp/swap and + never delete existing entries; bookmark captures keep tmp-and-swap + byte-identically (their existing tests pass unchanged). +- [x] Enumeration enforces `max_dirs`/`max_tracks` before any download; + the byte budget counts only bytes downloaded this run. +- [x] Reuse rule: parseable toml + non-skipped playable (+ existing file for + `file` playables) is not re-downloaded; skipped/broken/missing-audio + entries are re-captured. +- [x] Uncapturable tracks write skipped tomls (no more silent omission); + real download failures abort the run but keep prior writes. +- [x] Stream URLs and cookies never appear in errors, logs, or progress + events (only names, paths, counts). + +## Progress + RPC (D4) + +- [x] `CaptureLibraryNode` replies after validation (name, store enabled, + download blessing) with the existing status mapping; the walk runs + spawned and reports through `CaptureProgress` updates ending in exactly + one `finished` event (success or error). +- [x] The progress channel is bounded; a vanished receiver does not abort + the capture. + +## Playback (D3) + +- [x] `play` skips `is_skipped` tracks without a provider call and bounds + skipping by the queue length at entry: an all-skipped repeat queue stops + with a warning instead of spinning. + +## TUI (D5, D7) + +- [x] Skipped tracks render red in queue and library; the current-track + marker keeps precedence. +- [x] Progress lines show name, done/total, and skipped count; finished + lines expire on the render tick; errors render red. +- [x] The `W` binding description and the download-capture input label warn + that captures can take long and resume by name. +- [x] The selected row of a focused pane renders colored items with the dark + foreground (readable on the light bar); unfocused panes keep the colored + foregrounds. + +## Hygiene + +- [x] New/changed public items documented, stating error/edge behavior. +- [x] `clippy -D warnings`, `fmt`, `taplo`, `markdownlint` clean; all tests + green. +- [x] `architecture/incremental-captures.md` reconciled where the + implementation diverged. diff --git a/tidaldy/src/models.rs b/tidaldy/src/models.rs index 351c312..e95fba3 100644 --- a/tidaldy/src/models.rs +++ b/tidaldy/src/models.rs @@ -211,6 +211,7 @@ impl Track { .unwrap_or_default(), album: self.album.clone().map(|a| a.into()), duration: self.duration.map(|d| d as u32 * 1000), + is_skipped: false, } } } diff --git a/ytdy/src/lib.rs b/ytdy/src/lib.rs index 049e5e1..2887eb9 100644 --- a/ytdy/src/lib.rs +++ b/ytdy/src/lib.rs @@ -112,6 +112,7 @@ fn entry_to_track(entry: &engine::Entry, node_path: &str) -> Track { title: entry.title.clone().unwrap_or_default(), duration: entry.duration.map(|secs| secs.max(0.0) as u32), album: None, + is_skipped: false, } }