From 4e44f672e815b8f7335240f8c1ed532a6ed47d79 Mon Sep 17 00:00:00 2001 From: Test User Date: Tue, 21 Jul 2026 10:41:49 +0200 Subject: [PATCH] Persist queues as fs-provider folders Queues now survive restarts, built entirely on the fs provider: fsdy::Client is instance-mountable and a second, read-only instance serves /crabidy/queues/ as /queues. Every queue is a folder of order-prefixed link track files plus a hidden state sidecar, written only by the new QueueStore (tmp-and-swap). The playback loop streams every queue change through a latest-wins watch channel to a debouncing persister task and restores queues/current/ (tracks, position, modifiers) at startup without autoplay. w on the queue pane asks for a name and drives the previously stubbed SaveQueue rpc; reloading a saved queue is just queueing /queues/, since link entries rewrite to their targets at listing time. The old "no links into /fs" parse rejection gave way to one-hop link semantics so queues can reference fs tracks. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 4 + architecture/fs-provider.md | 21 +- architecture/queue-persistence.md | 225 ++++++++++++++ cbd-tui/src/app/bindings.rs | 30 ++ cbd-tui/src/app/mod.rs | 71 +++++ cbd-tui/src/main.rs | 7 + cbd-tui/src/rpc.rs | 10 +- crabidy-server/Cargo.toml | 6 + crabidy-server/src/lib.rs | 2 + crabidy-server/src/main.rs | 34 ++- crabidy-server/src/playback.rs | 258 +++++++++++++++- crabidy-server/src/provider.rs | 81 +++++ crabidy-server/src/queue_store.rs | 488 ++++++++++++++++++++++++++++++ crabidy-server/src/rpc.rs | 36 ++- fsdy/src/lib.rs | 330 ++++++++++++++++++-- plan/queue-persistence.md | 68 +++++ plan/summary.md | 55 ++++ quality/queue-persistence.md | 92 ++++++ 18 files changed, 1770 insertions(+), 48 deletions(-) create mode 100644 architecture/queue-persistence.md create mode 100644 crabidy-server/src/queue_store.rs create mode 100644 plan/queue-persistence.md create mode 100644 quality/queue-persistence.md diff --git a/Cargo.lock b/Cargo.lock index 0c38446..b6485c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -748,9 +748,13 @@ dependencies = [ "fsdy", "futures", "rand 0.10.2", + "serde", + "tempfile", + "thiserror 2.0.19", "tidaldy", "tokio", "tokio-stream", + "toml", "tonic", "tracing", "tracing-appender", diff --git a/architecture/fs-provider.md b/architecture/fs-provider.md index 4227780..8dec7d2 100644 --- a/architecture/fs-provider.md +++ b/architecture/fs-provider.md @@ -78,8 +78,12 @@ Consequences, accepted deliberately: - `get_urls_for_track` on an fs path whose playable is a link cannot occur through normal flow (the path was rewritten before it could be queued); if it happens anyway it is `MalformedPath` with a warning, not a chain - resolution. **Link chains are structurally impossible** — see D3's "no - links into `/fs`" rule. + resolution. **Links therefore resolve one hop by construction**: a link + whose target is itself a link file dies at play time with a warning, and + cycles cannot recurse. (Amended by queue-persistence D2: the original + "no links into `/fs`" parse-time rejection was dropped — persisted + queues must link to `/fs` tracks — and this one-hop argument replaces + it.) ### D3 — On-disk schema: TOML, extension `.cbd-track.toml`, exactly one playable @@ -114,9 +118,10 @@ file = "../flac/we-will-rock-you.flac" music folder stays relocatable). Existence is **not** checked at listing time (TOCTOU; the player produces a good error at play time). - `url`: must parse as `http`/`https` (matching what the player accepts). -- `link`: must be an absolute crabidy path (`/`-prefixed) and must **not** - point into `/fs` — self-links would allow chains/cycles; other providers - are one hop away by construction. +- `link`: must be an absolute crabidy path (`/`-prefixed). Links into fs + instances (including `/fs` itself) are legal — persisted queues rely on + it (queue-persistence D2); safety comes from links resolving one hop + only (see D2 above). - A file that fails to parse or validate is **skipped with a warning** at listing time; it never panics and never poisons its directory (hard rule: no panic on user input). @@ -165,7 +170,11 @@ file = "../flac/we-will-rock-you.flac" - Reading audio-file tags (ID3 etc.) to synthesize track nodes for plain `.mp3` files sitting in the tree: future work — this feature is about the serialized-node format. -- Multiple roots, file watching, link chains: rejected above. +- Multiple roots, file watching: rejected above. Link chains resolve at + most one hop (D2); deeper chains fail at play time by design. +- Since queue-persistence D1, `fsdy::Client::new(provider_root, disk_root)` + can mount additional instances (the server mounts `/queues` over the + persisted-queues folder); `fsdy.toml` still configures only `/fs`. ## Structure diff --git a/architecture/queue-persistence.md b/architecture/queue-persistence.md new file mode 100644 index 0000000..a4017b7 --- /dev/null +++ b/architecture/queue-persistence.md @@ -0,0 +1,225 @@ +# Queue persistence + +## Context and problem statement + +The queue lives only in the playback loop's memory: restarting +`crabidy-server` loses it. The user wants + +1. an **automatically maintained current queue**, persisted on every queue + operation and reloaded when the server starts, and +2. **named saved queues**: pressing `w` on the queue pane asks for a name + and stores the current entries under it. + +The explicit framing: realize this **completely with the fs provider** — a +second `fsdy` instance pointed at a `queues/` folder inside the crabidy +config directory, one subfolder per queue, each holding serialized track +files. + +## Assumptions (confirmed against the code) + +- The proto already declares `SaveQueue(SaveQueueRequest{name})`; the + server handler is a no-op stub (`rpc.rs`). No wire change is needed. +- The playback loop is the single writer of queue state + (`Playback.queue: Mutex`); every content change funnels + through `broadcast_queue`, every current-track change through `play` + (plus the shuffle/repeat toggles). Hooking those sites observes every + queue operation. +- `fsdy` track files carry metadata plus one playable; a `link` playable + rewrites `Track.path` to its target at listing time + (architecture/fs-provider.md D2). Queueing a folder of link files + therefore reconstructs the original tracks with zero new mechanisms. +- `w` is unbound in the TUI's `Queue` scope; the input overlay + (`InputState`/`InputPurpose`) already handles ask-for-a-name flows. + +## Decisions + +### D1 — Mount a second `fsdy` instance at `/queues` + +Options considered: + +- *(a)* A new provider crate (`queuedy`) that owns the queues folder. +- *(b)* Parameterize `fsdy::Client` with its provider root and mount a + second instance at `/queues` over `/crabidy/queues`. + +**Decision: (b)** — the user's framing, and the listing/parsing/routing +logic is byte-for-byte the same. `fsdy::Client` gains a constructor +`Client::new(provider_root, disk_root)`; the `ProviderClient::init` path +keeps building the `/fs` instance from `fsdy.toml`. The hardcoded +`"/fs/"` prefixes in `disk_path`/`is_track_path`/`list_dir` become +instance state. The orchestrator gains `queues_client` and `/queues` +routing arms; init creates the folder (`create_dir_all`) and is non-fatal +like `/fs` (a failure costs persistence, never the server). Loading a +saved queue is just browsing `/queues` and queueing a folder — no new +RPCs, no new TUI flows. + +### D2 — Persist a queue as a folder of order-prefixed **link** files + +Every queue entry becomes `NNNN .cbd-track.toml` with the entry's +metadata (title, artist, duration, album) and `playable.link = +Track.path` — uniformly, for every entry. The 4-digit zero-padded prefix +makes the case-insensitive listing sort reproduce queue order; the +sanitized title keeps the files human-readable. Round trip: listing +rewrites each link track's path back to its target, so reloading yields +the original tracks with the persisted metadata. + +**Consequence — the "no links into `/fs`" rule falls.** A queue may +contain `/fs/...` tracks (file/url playables keep their fs path), so +persisted files must be able to link into an fs-provider instance. The +original rejection (fs-provider D3) existed to prevent chains; it is +replaced by the stronger structural argument: **links are one hop by +construction** — `get_urls_for_track` never follows a link (a +link-playable target is `MalformedPath`), so a link whose target is +itself a link file dies at play time with a warning, and cycles cannot +recurse anywhere. `TrackFileError::LinkIntoFs` is removed; a link must +merely be an absolute path. `architecture/fs-provider.md` D2/D3 are +reconciled with this. + +Not chosen: inlining the target's `file`/`url` playable into the saved +file — the persister only has the wire `Track` (path + metadata), and +links keep the saved queue pointing at the *node*, surviving edits to the +underlying track file. + +### D3 — Layout: `<config>/crabidy/queues/<name>/`, current queue = `current` + +- The automatically maintained queue lives in `queues/current/` — a + visible, ordinary queue folder (it shows up under `/queues` like any + saved queue). The name is **reserved**: `SaveQueue("current")` is + rejected so a named save is never silently clobbered by auto-persist. +- Each queue folder carries a hidden sidecar `.queue-state.toml` + (`current_position`, `repeat`, `shuffle`). Dot-prefixed → invisible to + the provider listing. It is written for every queue and read only when + restoring `current` at startup. +- Writes go to a hidden sibling temp dir (`.tmp-<name>`), then the old + folder is removed and the temp renamed into place. Not atomic (rename + over a non-empty dir is impossible); the crash window can lose the + folder — accepted for a local music queue, and a warning covers it. +- Saving an existing name overwrites it (same temp-and-swap). + +### D4 — Auto-persist through a latest-wins channel and one persister task + +The playback loop must never block on disk. Every queue-state change +sends a snapshot (`tracks`, `current_position`, `repeat`, `shuffle`) +into a `tokio::sync::watch` channel (bounded, single slot, latest wins — +a burst of resolve chunks coalesces naturally). A dedicated persister +task awaits changes, debounces briefly, skips writes whose snapshot +equals the last one written (broadcasts that only toggled the +`resolving` flag stay free), and rewrites `queues/current/` per D3. Disk +failures are warnings; playback is never affected. Send sites: the +`broadcast_queue` funnel, the current-track broadcast in `play`, and the +shuffle/repeat toggle handlers. + +### D5 — Restore at startup, bespoke, never autoplay + +`Replace(["/queues/current"])` through the normal resolve flow was +rejected: it starts playback (a restarted server must stay silent), and +it cannot restore the queue position. Instead, before the loops start +serving, the server reads `queues/current/` directly — sorted listing, +`TrackFile::parse`, `to_track` (identical semantics to the provider) — +applies the tracks to the `QueueManager`, restores +`current_position`/`repeat`/`shuffle` from the sidecar, and leaves +`PlayState::Stopped`. A missing folder is a fresh start; a broken file is +skipped with a warning like any listing. + +### D6 — `SaveQueue` wiring + +`rpc save_queue` sends `PlaybackCommand::SaveQueue { name, result_tx }` +to the playback loop (single-writer discipline: only the loop may +snapshot). The loop validates and snapshots, then hands the write to a +spawned task so it never blocks on disk; the RPC reply reports the actual +write result. Errors: invalid name (empty after trim, contains a path +separator or NUL, starts with `.`, or is `current`) → +`invalid_argument`; empty queue → `failed_precondition`; I/O → +`internal`. + +### D7 — TUI: `w` on the queue pane + +New `Action::QueueSaveAs` bound to `w` in `Scope::Queue` ("Save queue +as…"). It opens the existing input overlay with a new +`InputPurpose::SaveQueue` (label `save queue`), no-op while the queue +is empty. Submit sends `MessageFromUi::SaveQueue(name)` → the +`SaveQueue` RPC. The saved queue appears under `/queues` on the next +library visit — no push update needed. + +### D8 — Out of scope (explicitly) + +- Renaming/deleting saved queues from the TUI (`fsdy` keeps + create/rename/delete `NotSupported`); file tools work today, marking + `/queues` children deletable is future work. +- Making the queues directory configurable; it is derived from the + config dir. +- Persisting the playback *position within the track*, autoplay on + restore, or multiple current queues. + +## Structure + +```d2 +direction: right + +server: crabidy-server { + pb: Playback loop { + q: "QueueManager (single writer)" + } + persister: "persister task" { + w: "debounce, skip unchanged,\nwrite current/" + } + store: QueueStore { + s: "validate name, tmp-and-swap" + } + orch: ProviderOrchestrator +} + +fs: "fsdy /fs\n(music root)" +qfs: "fsdy /queues\n(config queues dir)" + +disk: "config/crabidy/queues" { + shape: cylinder + cur: "current/ + .queue-state.toml" + saved: "<name>/ per saved queue" +} + +server.pb -> server.persister: "watch channel\n(latest snapshot wins)" +server.persister -> server.store: persist current +server.pb -> server.store: "SaveQueue(name)\n(spawned write)" +server.store -> disk +server.orch -> qfs: "/queues/..." +server.orch -> fs: "/fs/..." +qfs -> disk: "list + parse (read only)" +``` + +## Key flow: save, restart, reload + +```d2 +shape: sequence_diagram +tui: TUI +rpc: gRPC +pb: Playback loop +store: QueueStore +orch: Orchestrator + +tui -> rpc: "SaveQueue(road trip)" +rpc -> pb: "PlaybackCommand::SaveQueue" +pb -> store: "snapshot -> spawned write" +store -> rpc: "queues/road trip/ written" +rpc -> tui: OK +tui -> pb: "(server restarts; restore reads current/)" +tui -> orch: "GetLibraryNode(/queues)" +orch -> tui: "children: [current, road trip]" +tui -> pb: "ReplaceQueue([/queues/road%20trip])" +pb -> orch: "resolve: links rewritten to targets" +``` + +## Risks and open questions + +- **Hand-written files in `queues/`** behave like any fs tree (broken + files skipped with warnings). A hand-written `url`/`file` track keeps + its `/queues/...` path when queued; persisting then links to that + file — one hop, resolves fine. +- **Queues past 9999 tracks** sort wrong beyond the 4-digit prefix; + accepted (prefix width is a constant). +- **Concurrent saves to the same name** race on the temp dir; last + writer wins. Accepted for a single-user local server. +- **Metadata drift**: a saved queue replays the metadata captured at + save time, not the target's live metadata — consistent with + fs-provider D2. +- Open (future): deletable saved queues in the TUI; a `SaveQueue` + confirmation/overwrite prompt; persisting the in-track position. diff --git a/cbd-tui/src/app/bindings.rs b/cbd-tui/src/app/bindings.rs index 86bf079..2c54282 100644 --- a/cbd-tui/src/app/bindings.rs +++ b/cbd-tui/src/app/bindings.rs @@ -81,6 +81,9 @@ pub enum Action { QueueRemoveTrack, QueueClearKeepCurrent, QueueClearAll, + /// Open the input overlay asking for a name to save the queue under + /// (persisted as `/queues/<name>`). No-op while the queue is empty. + QueueSaveAs, // Help modal CloseHelp, } @@ -379,6 +382,13 @@ pub const BINDINGS: &[Binding] = &[ action: Action::QueueClearAll, description: "Clear entire queue", }, + Binding { + scope: Scope::Queue, + mods: KeyModifiers::NONE, + code: KeyCode::Char('w'), + action: Action::QueueSaveAs, + description: "Save queue under a name", + }, // -- Help modal ------------------------------------------------------ Binding { scope: Scope::Help, @@ -524,6 +534,26 @@ mod tests { ); } + #[test] + fn save_binds_in_the_queue_scope_only() { + assert_eq!( + lookup( + UiFocus::Queue, + false, + key(KeyCode::Char('w'), KeyModifiers::NONE) + ), + Some(Action::QueueSaveAs) + ); + assert_eq!( + lookup( + UiFocus::Library, + false, + key(KeyCode::Char('w'), KeyModifiers::NONE) + ), + None + ); + } + #[test] fn same_chord_resolves_per_pane() { assert_eq!( diff --git a/cbd-tui/src/app/mod.rs b/cbd-tui/src/app/mod.rs index 6979633..1802ad7 100644 --- a/cbd-tui/src/app/mod.rs +++ b/cbd-tui/src/app/mod.rs @@ -86,6 +86,9 @@ pub enum MessageFromUi { DeleteNode { path: String, }, + /// Save the current queue under a name (server-side snapshot; appears + /// as `/queues/<name>` in the library on the next visit). + SaveQueue(String), AppendTracks(Vec<String>), QueueTracks(Vec<String>), InsertTracks(Vec<String>, usize), @@ -123,6 +126,8 @@ pub enum InputPurpose { /// current title, so "rename" degrades to "retype" (append-only editing, /// no cursor movement). Rename { path: String }, + /// `w`: save the current queue under the entered name. + SaveQueue, } /// State of the one-line text input overlay (node creation and rename). @@ -196,6 +201,9 @@ impl App { new_title: title, }); } + InputPurpose::SaveQueue => { + let _ = self.tx.send(MessageFromUi::SaveQueue(title)); + } } } self.input = None; @@ -304,6 +312,16 @@ impl App { Action::QueueClearAll => { let _ = self.tx.send(MessageFromUi::ClearQueue(false)); } + Action::QueueSaveAs => { + // Nothing to save from an empty queue; silently ignored + // like the other capability-gated openers. + if !self.queue.is_empty() { + self.input = Some(InputState { + purpose: InputPurpose::SaveQueue, + buffer: String::new(), + }); + } + } } DispatchResult::Continue } @@ -345,6 +363,7 @@ impl App { let label = match &input.purpose { InputPurpose::Create { .. } => "new node", InputPurpose::Rename { .. } => "rename", + InputPurpose::SaveQueue => "save queue", }; let line = Rect::new(area.x + 1, area.y + area.height - 2, area.width - 2, 1); f.render_widget(Clear, line); @@ -717,6 +736,58 @@ mod tests { assert!(text.contains("rename: abba"), "rename overlay label"); } + /// A one-track queue update, as the server would broadcast it. + fn one_track_queue() -> crabidy_core::proto::crabidy::Queue { + use crabidy_core::proto::crabidy::{Queue as ProtoQueue, Track}; + ProtoQueue { + timestamp: 0, + current_position: 0, + tracks: vec![Track { + path: "/tidal/playlists/p/1".to_string(), + artist: "artist".to_string(), + title: "track".to_string(), + duration: None, + album: None, + }], + resolving: false, + } + } + + #[test] + fn save_queue_opens_the_overlay_only_with_a_non_empty_queue() { + let (mut app, _rx) = app(); + // Nothing to save from an empty queue: 'w' must be a no-op. + assert_eq!(app.dispatch(Action::QueueSaveAs), DispatchResult::Continue); + assert!(app.input.is_none()); + + app.queue.update_queue(one_track_queue()); + let _ = app.dispatch(Action::QueueSaveAs); + let input = app.input.as_ref().expect("save overlay open"); + assert!(matches!(input.purpose, InputPurpose::SaveQueue)); + assert_eq!(input.buffer, ""); + } + + #[test] + fn save_queue_submits_the_trimmed_name_and_esc_cancels() { + let (mut app, rx) = app(); + app.queue.update_queue(one_track_queue()); + + let _ = app.dispatch(Action::QueueSaveAs); + type_str(&mut app, " road trip "); + app.handle_input_key(key(crossterm::event::KeyCode::Enter)); + assert!(app.input.is_none()); + match rx.try_recv() { + Ok(MessageFromUi::SaveQueue(name)) => assert_eq!(name, "road trip"), + other => panic!("expected SaveQueue, got {:?}", other.is_ok()), + } + + let _ = app.dispatch(Action::QueueSaveAs); + type_str(&mut app, "discard"); + app.handle_input_key(key(crossterm::event::KeyCode::Esc)); + assert!(app.input.is_none()); + assert!(rx.try_recv().is_err(), "Esc must not save"); + } + #[test] fn queue_clear_actions_carry_the_keep_current_flag() { let (mut app, rx) = app(); diff --git a/cbd-tui/src/main.rs b/cbd-tui/src/main.rs index a30cc5d..30d2ad7 100644 --- a/cbd-tui/src/main.rs +++ b/cbd-tui/src/main.rs @@ -196,6 +196,13 @@ async fn poll( MessageFromUi::ClearQueue(exclude_current) => { rpc_client.clear_queue(exclude_current).await? } + MessageFromUi::SaveQueue(name) => { + // A rejected save (bad name, empty queue) must not tear + // down the poll loop; the server logs the cause. + if let Err(err) = rpc_client.save_queue(name.clone()).await { + error!(name, "failed to save queue: {err}"); + } + } } } Some(resp) = rpc_client.update_stream.next() => { diff --git a/cbd-tui/src/rpc.rs b/cbd-tui/src/rpc.rs index 86a93ea..1d50929 100644 --- a/cbd-tui/src/rpc.rs +++ b/cbd-tui/src/rpc.rs @@ -3,8 +3,8 @@ use crabidy_core::proto::crabidy::{ ClearQueueRequest, CreateLibraryNodeRequest, DeleteLibraryNodeRequest, GetLibraryNodeRequest, GetUpdateStreamRequest, GetUpdateStreamResponse, InitRequest, InitResponse, InsertRequest, LibraryNode, NextRequest, PrevRequest, QueueRequest, RemoveRequest, RenameLibraryNodeRequest, - ReplaceRequest, RestartTrackRequest, SetCurrentRequest, ToggleMuteRequest, TogglePlayRequest, - ToggleRepeatRequest, ToggleShuffleRequest, + ReplaceRequest, RestartTrackRequest, SaveQueueRequest, SetCurrentRequest, ToggleMuteRequest, + TogglePlayRequest, ToggleRepeatRequest, ToggleShuffleRequest, }; use std::{collections::HashMap, error::Error, fmt, time::Duration}; @@ -214,6 +214,12 @@ impl RpcClient { Ok(()) } + pub async fn save_queue(&mut self, name: String) -> Result<(), Box<dyn Error>> { + let save_queue_request = Request::new(SaveQueueRequest { name }); + self.client.save_queue(save_queue_request).await?; + Ok(()) + } + pub async fn replace_queue(&mut self, paths: Vec<String>) -> Result<(), Box<dyn Error>> { let replace_request = Request::new(ReplaceRequest { paths }); self.client.replace(replace_request).await?; diff --git a/crabidy-server/Cargo.toml b/crabidy-server/Cargo.toml index 27dc2e8..8dd2bd8 100644 --- a/crabidy-server/Cargo.toml +++ b/crabidy-server/Cargo.toml @@ -17,10 +17,16 @@ flume.workspace = true fsdy.workspace = true futures.workspace = true rand.workspace = true +serde.workspace = true +thiserror.workspace = true tidaldy.workspace = true tokio = { workspace = true, features = ["full"] } +toml.workspace = true tokio-stream = { workspace = true, features = ["sync"] } tonic.workspace = true tracing.workspace = true tracing-appender.workspace = true tracing-subscriber.workspace = true + +[dev-dependencies] +tempfile.workspace = true diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index 058d9f8..8b51b64 100644 --- a/crabidy-server/src/lib.rs +++ b/crabidy-server/src/lib.rs @@ -1,3 +1,5 @@ +pub mod queue_store; + use crabidy_core::proto::crabidy::{Queue, Track}; use rand::{rng, seq::SliceRandom}; use std::sync::{atomic::AtomicBool, Arc}; diff --git a/crabidy-server/src/main.rs b/crabidy-server/src/main.rs index ed7b05b..43fe112 100644 --- a/crabidy-server/src/main.rs +++ b/crabidy-server/src/main.rs @@ -27,7 +27,30 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> { err })?; - let playback = Playback::new(update_tx.clone(), orchestrator.provider_tx.clone()); + // Queue persistence is optional: without a usable queues directory the + // server runs with an in-memory queue only. + let queue_store = match crabidy_server::queue_store::queues_dir() { + Some(dir) => match crabidy_server::queue_store::QueueStore::open(dir).await { + Ok(store) => Some(std::sync::Arc::new(store)), + Err(err) => { + warn!("queue persistence disabled: {err}"); + None + } + }, + None => { + warn!("queue persistence disabled: no config directory"); + None + } + }; + + let playback = Playback::new( + update_tx.clone(), + orchestrator.provider_tx.clone(), + queue_store, + ); + // Reload the persisted current queue before anything can observe or + // mutate state; never starts playback. + playback.restore_current().await; let playback_tx = playback.playback_tx.clone(); let player_msg = playback.player.messages.clone(); @@ -252,6 +275,14 @@ pub enum PlaybackCommand { SetCurrent { position: u32, }, + /// Saves the current queue under a name (see + /// `architecture/queue-persistence.md` D6). Handled on the loop so the + /// snapshot is consistent; the disk write happens on a spawned task and + /// reports through `result_tx`. + SaveQueue { + name: String, + result_tx: flume::Sender<Result<(), crabidy_server::queue_store::SaveQueueError>>, + }, ToggleShuffle, ToggleRepeat, TogglePlay, @@ -291,6 +322,7 @@ impl PlaybackCommand { Self::ResolveFinished { .. } => "resolve_finished", Self::Clear { .. } => "clear", Self::SetCurrent { .. } => "set_current", + Self::SaveQueue { .. } => "save_queue", Self::ToggleShuffle => "toggle_shuffle", Self::ToggleRepeat => "toggle_repeat", Self::TogglePlay => "toggle_play", diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index f8997f0..6eab118 100644 --- a/crabidy-server/src/playback.rs +++ b/crabidy-server/src/playback.rs @@ -6,11 +6,12 @@ use crabidy_core::proto::crabidy::{ Queue as ProtoQueue, QueueTrack, Track, TrackPosition, }; use crabidy_core::ProviderError; +use crabidy_server::queue_store::{self, QueueSnapshot, QueueStore, SaveQueueError}; use crabidy_server::{PendingResolve, QueueManager, ResolveKind}; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Mutex; -use tracing::{debug, debug_span, error, instrument, trace, warn, Instrument}; +use std::sync::{Arc, Mutex}; +use tracing::{debug, debug_span, error, info, instrument, trace, warn, Instrument}; pub struct Playback { update_tx: tokio::sync::broadcast::Sender<StreamUpdate>, @@ -24,6 +25,12 @@ pub struct Playback { /// touches this map (same single-writer discipline as `queue`). pending: Mutex<HashMap<u64, PendingResolve>>, next_op_id: AtomicU64, + /// `None` when queue persistence is disabled (no usable queues + /// directory) — the queue then lives in memory only. + store: Option<Arc<QueueStore>>, + /// Feeds the persister task; latest snapshot wins, so the loop never + /// waits on disk (architecture/queue-persistence.md D4). + persist_tx: tokio::sync::watch::Sender<Option<QueueSnapshot>>, pub player: Player, } @@ -31,10 +38,12 @@ impl Playback { pub fn new( update_tx: tokio::sync::broadcast::Sender<StreamUpdate>, provider_tx: flume::Sender<ProviderMessage>, + store: Option<Arc<QueueStore>>, ) -> Self { let (playback_tx, playback_rx) = flume::bounded(64); let queue = Mutex::new(QueueManager::new()); let state = Mutex::new(PlayState::Stopped); + let (persist_tx, _) = tokio::sync::watch::channel(None); let player = Player::default(); Self { update_tx, @@ -45,11 +54,50 @@ impl Playback { state, pending: Mutex::new(HashMap::new()), next_op_id: AtomicU64::new(0), + store, + persist_tx, player, } } + /// Reloads the persisted current queue: tracks, position, and + /// shuffle/repeat. Never starts playback — a restarted server stays + /// silent. Call before [`Self::run`] so nothing observes the empty + /// queue first. + pub async fn restore_current(&self) { + let Some(store) = &self.store else { + return; + }; + let Some(snapshot) = store.load_current().await else { + debug!("no persisted queue, starting fresh"); + return; + }; + let Ok(mut queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return; + }; + // No autoplay: the track the replace would start is ignored. + let _ = queue.replace_with_tracks(&snapshot.tracks); + queue.repeat = snapshot.repeat; + // An out-of-range position (edited folder) is refused by + // `set_current_position` and playback starts at the first track. + let _ = queue.set_current_position(snapshot.current_position); + if snapshot.shuffle { + // The play order is not persisted; restoring shuffle reshuffles + // around the restored current track. + queue.shuffle_on(); + } + info!( + tracks = snapshot.tracks.len(), + position = snapshot.current_position, + "restored the persisted queue" + ); + } + pub fn run(self) { + if let Some(store) = &self.store { + queue_store::spawn_persister(Arc::clone(store), self.persist_tx.subscribe()); + } tokio::spawn(async move { while let Ok(PlaybackMessage { span, command }) = self.playback_rx.recv_async().await { // Attribute all handler events to a span that is a child of @@ -204,6 +252,39 @@ impl Playback { self.play(track).await; } + PlaybackCommand::SaveQueue { name, result_tx } => { + debug!(name, "saving the queue"); + // Snapshot on the loop (single-writer discipline), write on + // a spawned task — the loop never waits on disk. + let snapshot = { + let Ok(queue) = self.queue.lock() else { + error!("queue lock poisoned"); + return; + }; + let proto: ProtoQueue = queue.clone().into(); + QueueSnapshot { + tracks: proto.tracks, + current_position: proto.current_position, + repeat: queue.repeat, + shuffle: queue.shuffle, + } + }; + let store = self.store.clone(); + tokio::spawn( + async move { + let result = match &store { + Some(store) => store.save(&name, &snapshot).await, + None => Err(SaveQueueError::Disabled), + }; + if let Err(err) = &result { + warn!(name, "cannot save queue: {err}"); + } + let _ = result_tx.send_async(result).await; + } + .in_current_span(), + ); + } + PlaybackCommand::ToggleShuffle => { let (shuffle, repeat) = { let Ok(mut queue) = self.queue.lock() else { @@ -215,6 +296,7 @@ impl Playback { } else { queue.shuffle_on() } + self.send_persist_snapshot(&queue); (queue.shuffle, queue.repeat) }; debug!(shuffle, "toggled shuffle"); @@ -228,6 +310,7 @@ impl Playback { return; }; queue.repeat = !queue.repeat; + self.send_persist_snapshot(&queue); (queue.shuffle, queue.repeat) }; debug!(repeat, "toggled repeat"); @@ -513,11 +596,30 @@ impl Playback { } /// Broadcasts the current queue snapshot. All queue broadcasts go - /// through here so the `resolving` flag can never be forgotten. + /// through here so the `resolving` flag can never be forgotten — and + /// every queue-content change reaches the persister the same way. fn broadcast_queue(&self, queue: &QueueManager) { + self.send_persist_snapshot(queue); self.broadcast(StreamUpdate::Queue(self.queue_snapshot(queue))); } + /// Hands the queue's persistable state to the persister task; a no-op + /// when persistence is disabled. Latest snapshot wins, so calling this + /// on every mutation is free of backpressure (the persister skips + /// writes for unchanged snapshots). + fn send_persist_snapshot(&self, queue: &QueueManager) { + if self.store.is_none() { + return; + } + let proto: ProtoQueue = queue.clone().into(); + self.persist_tx.send_replace(Some(QueueSnapshot { + tracks: proto.tracks, + current_position: proto.current_position, + repeat: queue.repeat, + shuffle: queue.shuffle, + })); + } + #[instrument(skip(self))] async fn get_urls_for_track(&self, path: &str) -> Result<Vec<String>, ProviderError> { let (result_tx, result_rx) = flume::bounded(1); @@ -595,6 +697,9 @@ impl Playback { error!("queue lock poisoned"); return; }; + // Current-track moves (Next/Prev/SetCurrent/skips) change the + // persisted position without a queue broadcast. + self.send_persist_snapshot(&queue); self.broadcast(StreamUpdate::QueueTrack(QueueTrack { queue_position: queue.current_position() as u32, track: queue.current_track(), @@ -606,3 +711,150 @@ impl Playback { } } } + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn track(i: usize) -> Track { + Track { + path: format!("/tidal/playlists/p/{i}"), + artist: "artist".to_string(), + title: format!("track {i}"), + duration: None, + album: None, + } + } + + async fn store_in(dir: &TempDir) -> Arc<QueueStore> { + Arc::new( + QueueStore::open(dir.path().join("queues")) + .await + .expect("open store"), + ) + } + + fn playback_with(store: Option<Arc<QueueStore>>) -> Playback { + let (update_tx, _) = tokio::sync::broadcast::channel(64); + let (provider_tx, _provider_rx) = flume::bounded(16); + Playback::new(update_tx, provider_tx, store) + } + + fn fill_queue(playback: &Playback, n: usize) { + let tracks: Vec<Track> = (0..n).map(track).collect(); + let mut queue = playback.queue.lock().expect("queue lock"); + let _ = queue.replace_with_tracks(&tracks); + } + + #[tokio::test] + async fn restore_fills_the_queue_without_starting_playback() { + let dir = TempDir::new().expect("tempdir"); + let store = store_in(&dir).await; + store + .persist_current(&QueueSnapshot { + tracks: (0..3).map(track).collect(), + current_position: 1, + repeat: true, + shuffle: false, + }) + .await + .expect("persist"); + + let playback = playback_with(Some(store)); + playback.restore_current().await; + + let queue = playback.queue.lock().expect("queue lock"); + let snapshot: ProtoQueue = queue.clone().into(); + assert_eq!(snapshot.tracks.len(), 3); + assert_eq!(queue.current_position(), 1); + assert!(queue.repeat); + // A restarted server stays silent: restoring must not play. + assert_eq!( + *playback.state.lock().expect("state lock"), + PlayState::Stopped + ); + } + + #[tokio::test] + async fn restore_survives_an_out_of_range_position() { + let dir = TempDir::new().expect("tempdir"); + let store = store_in(&dir).await; + store + .persist_current(&QueueSnapshot { + tracks: vec![track(0)], + current_position: 99, // hand-edited folder + repeat: false, + shuffle: false, + }) + .await + .expect("persist"); + let playback = playback_with(Some(store)); + playback.restore_current().await; + let queue = playback.queue.lock().expect("queue lock"); + assert_eq!(queue.current_position(), 0); + } + + #[tokio::test] + async fn save_queue_command_snapshots_the_live_queue() { + let dir = TempDir::new().expect("tempdir"); + let store = store_in(&dir).await; + let playback = playback_with(Some(Arc::clone(&store))); + fill_queue(&playback, 2); + + let (result_tx, result_rx) = flume::bounded(1); + playback + .handle_command(PlaybackCommand::SaveQueue { + name: "road trip".to_string(), + result_tx, + }) + .await; + result_rx + .recv_async() + .await + .expect("reply") + .expect("save succeeds"); + + let entries = std::fs::read_dir(store.dir().join("road trip")) + .expect("saved queue folder") + .filter(|e| { + !e.as_ref() + .expect("entry") + .file_name() + .to_string_lossy() + .starts_with('.') + }) + .count(); + assert_eq!(entries, 2); + } + + #[tokio::test] + async fn save_queue_rejects_an_empty_queue() { + let dir = TempDir::new().expect("tempdir"); + let playback = playback_with(Some(store_in(&dir).await)); + let (result_tx, result_rx) = flume::bounded(1); + playback + .handle_command(PlaybackCommand::SaveQueue { + name: "empty".to_string(), + result_tx, + }) + .await; + let result = result_rx.recv_async().await.expect("reply"); + assert!(matches!(result, Err(SaveQueueError::EmptyQueue))); + } + + #[tokio::test] + async fn queue_mutations_reach_the_persist_channel() { + let dir = TempDir::new().expect("tempdir"); + let playback = playback_with(Some(store_in(&dir).await)); + fill_queue(&playback, 2); + let rx = playback.persist_tx.subscribe(); + + playback + .handle_command(PlaybackCommand::Remove { positions: vec![1] }) + .await; + + let snapshot = rx.borrow().clone().expect("snapshot sent"); + assert_eq!(snapshot.tracks.len(), 1); + } +} diff --git a/crabidy-server/src/provider.rs b/crabidy-server/src/provider.rs index 6911d85..85d2030 100644 --- a/crabidy-server/src/provider.rs +++ b/crabidy-server/src/provider.rs @@ -4,6 +4,7 @@ use crabidy_core::{ proto::crabidy::{LibraryNode, LibraryNodeChild, Track}, ProviderClient, ProviderError, }; +use crabidy_server::queue_store::QUEUES_PROVIDER_ROOT; use std::{fs, path::PathBuf, sync::Arc}; use tracing::{debug, debug_span, error, instrument, warn, Instrument}; @@ -15,6 +16,10 @@ pub struct ProviderOrchestrator { /// `None` when the filesystem provider failed to initialize — the /// server runs without `/fs` instead of dying (architecture D5). fs_client: Option<Arc<fsdy::Client>>, + /// Second `fsdy` instance over the persisted-queues folder, mounted at + /// `/queues` (architecture/queue-persistence.md D1). `None` without a + /// config directory — the server then runs without `/queues`. + queues_client: Option<Arc<fsdy::Client>>, } /// Whether a path belongs to the filesystem provider. @@ -22,6 +27,12 @@ fn fs_owns(path: &str) -> bool { path == fsdy::PROVIDER_ROOT || path.starts_with("/fs/") } +/// Whether a path belongs to the persisted-queues provider instance. +fn queues_owns(path: &str) -> bool { + // The literal prefix mirrors `fs_owns`; QUEUES_PROVIDER_ROOT is "/queues". + path == QUEUES_PROVIDER_ROOT || path.starts_with("/queues/") +} + impl ProviderOrchestrator { /// The fs client, or `MalformedPath` (with a warning) when the /// provider is disabled — a `/fs` path then has no owner. @@ -31,6 +42,15 @@ impl ProviderOrchestrator { ProviderError::MalformedPath }) } + + /// The queues client, or `MalformedPath` (with a warning) when the + /// instance is disabled — a `/queues` path then has no owner. + fn queues_provider(&self) -> Result<&fsdy::Client, ProviderError> { + self.queues_client.as_deref().ok_or_else(|| { + warn!("queues library is disabled"); + ProviderError::MalformedPath + }) + } pub fn run(self) { tokio::spawn(async move { // Behind an Arc so long-running resolves can be spawned onto @@ -152,12 +172,29 @@ impl ProviderClient for ProviderOrchestrator { None } }; + // The queues instance mounts the folder the playback side persists + // into; a folder that does not exist yet lists as empty-on-arrival + // (created by QueueStore::open in main). + let queues_client = match crabidy_server::queue_store::queues_dir() { + Some(dir) => match fsdy::Client::new(QUEUES_PROVIDER_ROOT, dir) { + Ok(client) => Some(Arc::new(client)), + Err(err) => { + warn!("queues library disabled: {err}"); + None + } + }, + None => { + warn!("queues library disabled: no config directory"); + None + } + }; let (provider_tx, provider_rx) = flume::bounded(100); Ok(Self { provider_rx, provider_tx, tidal_client, fs_client, + queues_client, }) } @@ -176,6 +213,12 @@ impl ProviderClient for ProviderOrchestrator { .as_ref() .is_some_and(|fs| fs.is_track_path(path)); } + if queues_owns(path) { + return self + .queues_client + .as_ref() + .is_some_and(|queues| queues.is_track_path(path)); + } false } @@ -187,6 +230,9 @@ impl ProviderClient for ProviderOrchestrator { if fs_owns(track_path) { return self.fs_provider()?.get_urls_for_track(track_path).await; } + if queues_owns(track_path) { + return self.queues_provider()?.get_urls_for_track(track_path).await; + } warn!(path = track_path, "no provider owns this track path"); Err(ProviderError::MalformedPath) } @@ -199,6 +245,12 @@ impl ProviderClient for ProviderOrchestrator { if fs_owns(track_path) { return self.fs_provider()?.get_metadata_for_track(track_path).await; } + if queues_owns(track_path) { + return self + .queues_provider()? + .get_metadata_for_track(track_path) + .await; + } warn!(path = track_path, "no provider owns this track path"); Err(ProviderError::MalformedPath) } @@ -213,6 +265,11 @@ impl ProviderClient for ProviderOrchestrator { LibraryNodeChild::new(fsdy::PROVIDER_ROOT.to_owned(), "fs".to_owned(), false); root_node.children.push(child); } + if self.queues_client.is_some() { + let child = + LibraryNodeChild::new(QUEUES_PROVIDER_ROOT.to_owned(), "queues".to_owned(), false); + root_node.children.push(child); + } root_node } @@ -228,6 +285,9 @@ impl ProviderClient for ProviderOrchestrator { if fs_owns(path) { return self.fs_provider()?.get_lib_node(path).await; } + if queues_owns(path) { + return self.queues_provider()?.get_lib_node(path).await; + } warn!(path, "no provider owns this path"); Err(ProviderError::MalformedPath) } @@ -249,6 +309,12 @@ impl ProviderClient for ProviderOrchestrator { .create_lib_node(parent_path, title) .await; } + if queues_owns(parent_path) { + return self + .queues_provider()? + .create_lib_node(parent_path, title) + .await; + } warn!(parent_path, "no provider supports creating nodes here"); Err(ProviderError::NotSupported) } @@ -267,6 +333,12 @@ impl ProviderClient for ProviderOrchestrator { if fs_owns(path) { return self.fs_provider()?.rename_lib_node(path, new_title).await; } + if queues_owns(path) { + return self + .queues_provider()? + .rename_lib_node(path, new_title) + .await; + } warn!(path, "no provider supports renaming this node"); Err(ProviderError::NotSupported) } @@ -288,6 +360,12 @@ impl ProviderClient for ProviderOrchestrator { .resolve_tracks_into(path, chunk_tx) .await; } + if queues_owns(path) { + return self + .queues_provider()? + .resolve_tracks_into(path, chunk_tx) + .await; + } warn!(path, "no provider owns this path"); Err(ProviderError::MalformedPath) } @@ -302,6 +380,9 @@ impl ProviderClient for ProviderOrchestrator { if fs_owns(path) { return self.fs_provider()?.delete_lib_node(path).await; } + if queues_owns(path) { + return self.queues_provider()?.delete_lib_node(path).await; + } warn!(path, "no provider supports deleting this node"); Err(ProviderError::NotSupported) } diff --git a/crabidy-server/src/queue_store.rs b/crabidy-server/src/queue_store.rs new file mode 100644 index 0000000..e24e302 --- /dev/null +++ b/crabidy-server/src/queue_store.rs @@ -0,0 +1,488 @@ +//! Persisted queues on disk (see `architecture/queue-persistence.md`). +//! +//! Every queue is a folder under the store directory +//! (`<config>/crabidy/queues/`) holding one order-prefixed +//! `*.cbd-track.toml` **link** file per entry, plus a hidden +//! [`STATE_FILE_NAME`] sidecar. The automatically maintained queue lives in +//! [`CURRENT_QUEUE_NAME`]; every other folder is a named save. The same +//! directory is mounted read-only into the library as `/queues` by a second +//! `fsdy` instance — this module is the only writer. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use crabidy_core::proto::crabidy::Track; +use serde::{Deserialize, Serialize}; +use tracing::{debug, warn}; + +/// The library mount point of the queues directory (second `fsdy` +/// instance, see `architecture/queue-persistence.md` D1). +pub const QUEUES_PROVIDER_ROOT: &str = "/queues"; + +/// Reserved folder name of the automatically maintained queue. +pub const CURRENT_QUEUE_NAME: &str = "current"; + +/// Hidden per-queue sidecar carrying [`QueueState`]. Dot-prefixed, so +/// library listings never show it. +pub const STATE_FILE_NAME: &str = ".queue-state.toml"; + +/// The queues directory: `queues/` inside the crabidy config directory. +/// `None` when the platform has no config directory. +pub fn queues_dir() -> Option<PathBuf> { + dirs::config_dir().map(|d| d.join("crabidy").join("queues")) +} + +/// Everything the playback loop knows about the queue that is worth +/// persisting. Sent through the persister's `watch` channel (latest wins) +/// and written by [`QueueStore`]. +#[derive(Clone, Debug, PartialEq)] +pub struct QueueSnapshot { + /// Queue entries in track order (not play order — shuffle order is + /// deliberately not persisted). + pub tracks: Vec<Track>, + /// Index of the current track in `tracks`. + pub current_position: u32, + pub repeat: bool, + pub shuffle: bool, +} + +/// The on-disk schema of the [`STATE_FILE_NAME`] sidecar. +#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)] +pub struct QueueState { + pub current_position: u32, + pub repeat: bool, + pub shuffle: bool, +} + +/// Errors from validating or writing a persisted queue. +/// +/// At the RPC boundary: `InvalidName` → `invalid_argument`, `EmptyQueue` → +/// `failed_precondition`, the rest → `internal`. Messages carry names and +/// paths, never file contents. +#[derive(Debug, thiserror::Error)] +pub enum SaveQueueError { + #[error("invalid queue name: {0}")] + InvalidName(&'static str), + #[error("the queue is empty")] + EmptyQueue, + #[error("queue persistence is disabled")] + Disabled, + #[error("cannot write queue: {0}")] + Io(#[from] std::io::Error), + #[error(transparent)] + TrackFile(#[from] fsdy::TrackFileError), + #[error("cannot serialize queue state: {0}")] + State(#[from] toml::ser::Error), +} + +/// Reads and writes persisted queue folders. Cheap to clone behind an +/// `Arc`; all I/O is `tokio::fs`. +#[derive(Debug)] +pub struct QueueStore { + dir: PathBuf, +} + +impl QueueStore { + /// Opens the store at `dir`, creating the directory (and parents) if + /// missing. + pub async fn open(dir: PathBuf) -> Result<Self, std::io::Error> { + tokio::fs::create_dir_all(&dir).await?; + Ok(Self { dir }) + } + + /// The store directory (what the `/queues` provider instance mounts). + pub fn dir(&self) -> &Path { + &self.dir + } + + /// Validates a user-supplied queue name, returning the trimmed name. + /// + /// Rejected: empty after trimming, containing `/`, `\` or NUL, starting + /// with a dot (hidden folders are invisible to listings), and the + /// reserved [`CURRENT_QUEUE_NAME`]. + pub fn validate_name(name: &str) -> Result<&str, SaveQueueError> { + let name = name.trim(); + if name.is_empty() { + return Err(SaveQueueError::InvalidName("must not be empty")); + } + if name.contains(['/', '\\', '\0']) { + return Err(SaveQueueError::InvalidName( + "must not contain a path separator or NUL", + )); + } + if name.starts_with('.') { + return Err(SaveQueueError::InvalidName( + "must not start with a dot (hidden folders are invisible)", + )); + } + if name == CURRENT_QUEUE_NAME { + return Err(SaveQueueError::InvalidName( + "is reserved for the automatically maintained queue", + )); + } + Ok(name) + } + + /// Saves `snapshot` as the named queue, overwriting an existing one. + /// + /// Validates `name` per [`Self::validate_name`] and rejects an empty + /// snapshot with [`SaveQueueError::EmptyQueue`]. The folder is written + /// to a hidden temp sibling first, then swapped into place (remove old, + /// rename) — a crash can lose the folder, never corrupt it half-written + /// next to intact files. + pub async fn save(&self, name: &str, snapshot: &QueueSnapshot) -> Result<(), SaveQueueError> { + let name = Self::validate_name(name)?; + if snapshot.tracks.is_empty() { + return Err(SaveQueueError::EmptyQueue); + } + self.write_queue_dir(name, snapshot).await + } + + /// Persists `snapshot` as the current queue ([`CURRENT_QUEUE_NAME`]). + /// + /// Same write path as [`Self::save`] but without name validation and + /// with an empty snapshot allowed — clearing the queue must persist as + /// cleared. + pub async fn persist_current(&self, snapshot: &QueueSnapshot) -> Result<(), SaveQueueError> { + self.write_queue_dir(CURRENT_QUEUE_NAME, snapshot).await + } + + /// The shared write path: build the whole folder as a hidden temp + /// sibling, then swap it into place (remove old, rename). A crash can + /// lose the folder, never leave it half-written next to intact files + /// (architecture/queue-persistence.md D3). + async fn write_queue_dir( + &self, + name: &str, + snapshot: &QueueSnapshot, + ) -> Result<(), SaveQueueError> { + let tmp = self.dir.join(format!(".tmp-{name}")); + // A leftover temp folder from a crashed or racing write is stale. + if tokio::fs::try_exists(&tmp).await? { + tokio::fs::remove_dir_all(&tmp).await?; + } + tokio::fs::create_dir_all(&tmp).await?; + for (index, track) in snapshot.tracks.iter().enumerate() { + let text = fsdy::TrackFile::from_track(track).to_toml()?; + let file = tmp.join(fsdy::track_file_name(index, &track.title)); + tokio::fs::write(file, text).await?; + } + let state = QueueState { + current_position: snapshot.current_position, + repeat: snapshot.repeat, + shuffle: snapshot.shuffle, + }; + tokio::fs::write(tmp.join(STATE_FILE_NAME), toml::to_string_pretty(&state)?).await?; + + let target = self.dir.join(name); + if tokio::fs::try_exists(&target).await? { + tokio::fs::remove_dir_all(&target).await?; + } + tokio::fs::rename(&tmp, &target).await?; + Ok(()) + } + + /// Loads the persisted current queue for the startup restore. + /// + /// Reads the folder like a library listing (sorted case-insensitively, + /// broken/hidden/foreign entries skipped with warnings) plus the + /// [`QueueState`] sidecar (missing or broken sidecar → default state). + /// `None` when the folder does not exist — a fresh start. Never fails + /// the server; every defect is a warning and degrades to less state. + pub async fn load_current(&self) -> Option<QueueSnapshot> { + let dir = self.dir.join(CURRENT_QUEUE_NAME); + let mut read_dir = match tokio::fs::read_dir(&dir).await { + Ok(read_dir) => read_dir, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None, + Err(err) => { + warn!(dir = %dir.display(), "cannot read the persisted queue: {err}"); + return None; + } + }; + + // Mirror the provider listing: visible regular `*.cbd-track.toml` + // files, sorted case-insensitively — restore order == listing order. + let mut names: Vec<String> = Vec::new(); + loop { + let entry = match read_dir.next_entry().await { + Ok(Some(entry)) => entry, + Ok(None) => break, + Err(err) => { + warn!(dir = %dir.display(), "error while reading the persisted queue: {err}"); + break; + } + }; + let is_file = entry + .file_type() + .await + .is_ok_and(|file_type| file_type.is_file()); + let Some(name) = entry.file_name().to_str().map(str::to_string) else { + warn!(dir = %dir.display(), "skipping queue entry with non-UTF-8 name"); + continue; + }; + if is_file && !name.starts_with('.') && name.ends_with(fsdy::TRACK_FILE_SUFFIX) { + names.push(name); + } + } + names.sort_by_key(|name| name.to_lowercase()); + + let mut tracks = Vec::new(); + for name in names { + let file = dir.join(&name); + let text = match tokio::fs::read_to_string(&file).await { + Ok(text) => text, + Err(err) => { + warn!(file = %file.display(), "cannot read queue entry: {err}"); + continue; + } + }; + match fsdy::TrackFile::parse(&text) { + Ok(track_file) => { + // The same library path the /queues listing would give + // the entry, so non-link playables behave identically. + let lib_path = crabidy_core::join_path( + &crabidy_core::join_path(QUEUES_PROVIDER_ROOT, CURRENT_QUEUE_NAME), + &crabidy_core::encode_segment(&name), + ); + tracks.push(track_file.to_track(&lib_path)); + } + Err(err) => { + warn!(file = %file.display(), "skipping invalid queue entry: {err}"); + } + } + } + + let state_file = dir.join(STATE_FILE_NAME); + let state = match tokio::fs::read_to_string(&state_file).await { + Ok(text) => toml::from_str(&text).unwrap_or_else(|err| { + warn!(file = %state_file.display(), "broken queue state, using defaults: {err}"); + QueueState::default() + }), + Err(err) => { + debug!(file = %state_file.display(), "no queue state, using defaults: {err}"); + QueueState::default() + } + }; + Some(QueueSnapshot { + tracks, + current_position: state.current_position, + repeat: state.repeat, + shuffle: state.shuffle, + }) + } +} + +/// Spawns the auto-persist task: awaits snapshot changes on `rx`, debounces +/// briefly to coalesce bursts (resolve chunks), skips writes whose snapshot +/// equals the last one written, and rewrites the current queue folder. +/// Write failures are warnings; the task never affects playback. Exits when +/// the sender side is dropped. +pub fn spawn_persister( + store: Arc<QueueStore>, + mut rx: tokio::sync::watch::Receiver<Option<QueueSnapshot>>, +) { + tokio::spawn(async move { + let mut last_written: Option<QueueSnapshot> = None; + while rx.changed().await.is_ok() { + // Debounce: a resolve streams many chunks in quick succession; + // the watch channel keeps only the newest snapshot, so waiting + // briefly coalesces the burst into one write. + tokio::time::sleep(Duration::from_millis(200)).await; + let Some(snapshot) = rx.borrow_and_update().clone() else { + continue; + }; + // Broadcasts that only toggled the `resolving` flag carry an + // unchanged snapshot — skip the write. + if last_written.as_ref() == Some(&snapshot) { + continue; + } + match store.persist_current(&snapshot).await { + Ok(()) => last_written = Some(snapshot), + Err(err) => warn!("cannot persist the current queue: {err}"), + } + } + debug!("queue snapshot channel closed, persister exiting"); + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crabidy_core::proto::crabidy::Album; + use std::sync::Arc; + use std::time::Duration; + use tempfile::TempDir; + + fn track(i: usize) -> Track { + Track { + path: format!("/tidal/playlists/p/{i}"), + artist: "artist".to_string(), + title: format!("track {i}"), + duration: Some(60 + i as u32), + album: Some(Album { + title: "album".to_string(), + release_date: None, + }), + } + } + + fn snapshot(n: usize) -> QueueSnapshot { + QueueSnapshot { + tracks: (0..n).map(track).collect(), + current_position: 0, + repeat: false, + shuffle: false, + } + } + + async fn store() -> (QueueStore, TempDir) { + let dir = TempDir::new().expect("tempdir"); + let store = QueueStore::open(dir.path().join("queues")) + .await + .expect("open creates the directory"); + (store, dir) + } + + /// Sorted visible file names of a queue folder. + fn visible_files(dir: &Path) -> Vec<String> { + let mut names: Vec<String> = std::fs::read_dir(dir) + .expect("queue folder") + .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned()) + .filter(|n| !n.starts_with('.')) + .collect(); + names.sort_by_key(|n| n.to_lowercase()); + names + } + + #[tokio::test] + async fn save_writes_ordered_link_files_and_the_sidecar() { + let (store, _dir) = store().await; + let mut snap = snapshot(3); + // A queue may hold fs tracks; persisting must link to them too. + snap.tracks[1].path = "/fs/mix/song.cbd-track.toml".to_string(); + store.save("road trip", &snap).await.expect("save"); + + let queue_dir = store.dir().join("road trip"); + let names = visible_files(&queue_dir); + assert_eq!(names.len(), 3); + for (i, name) in names.iter().enumerate() { + assert!(name.starts_with(&format!("{:04} ", i + 1)), "{name}"); + let text = std::fs::read_to_string(queue_dir.join(name)).expect("read entry"); + let file = fsdy::TrackFile::parse(&text).expect("entry parses"); + // The listing rewrite restores the original track exactly. + assert_eq!(file.to_track("/queues/irrelevant"), snap.tracks[i]); + } + assert!( + queue_dir.join(STATE_FILE_NAME).exists(), + "sidecar written (hidden from listings by its dot prefix)" + ); + } + + #[tokio::test] + async fn save_validates_names_and_rejects_an_empty_queue() { + let (store, _dir) = store().await; + for bad in ["", " ", "a/b", "a\\b", ".hidden", CURRENT_QUEUE_NAME] { + assert!( + matches!( + store.save(bad, &snapshot(1)).await, + Err(SaveQueueError::InvalidName(_)) + ), + "name {bad:?} must be rejected" + ); + } + assert!(matches!( + store.save("fine", &snapshot(0)).await, + Err(SaveQueueError::EmptyQueue) + )); + // A valid name is used trimmed. + store + .save(" padded ", &snapshot(1)) + .await + .expect("trimmed name saves"); + assert!(store.dir().join("padded").is_dir()); + } + + #[tokio::test] + async fn save_overwrites_an_existing_queue_completely() { + let (store, _dir) = store().await; + store.save("mix", &snapshot(3)).await.expect("first save"); + store.save("mix", &snapshot(1)).await.expect("overwrite"); + // No stale entries from the longer first save survive. + assert_eq!(visible_files(&store.dir().join("mix")).len(), 1); + } + + #[tokio::test] + async fn persist_current_and_load_round_trip() { + let (store, _dir) = store().await; + let snap = QueueSnapshot { + current_position: 2, + repeat: true, + shuffle: true, + ..snapshot(4) + }; + store.persist_current(&snap).await.expect("persist"); + let loaded = store.load_current().await.expect("load"); + assert_eq!(loaded, snap); + } + + #[tokio::test] + async fn persist_current_accepts_an_empty_queue() { + // Clearing the queue must persist as cleared, not keep yesterday's + // tracks for the next restart. + let (store, _dir) = store().await; + store + .persist_current(&snapshot(2)) + .await + .expect("non-empty"); + store.persist_current(&snapshot(0)).await.expect("empty"); + let loaded = store.load_current().await.expect("load"); + assert!(loaded.tracks.is_empty()); + } + + #[tokio::test] + async fn load_current_without_a_folder_is_a_fresh_start() { + let (store, _dir) = store().await; + assert!(store.load_current().await.is_none()); + } + + #[tokio::test] + async fn load_current_skips_broken_entries_and_survives_a_broken_sidecar() { + let (store, _dir) = store().await; + store.persist_current(&snapshot(2)).await.expect("persist"); + let current = store.dir().join(CURRENT_QUEUE_NAME); + std::fs::write(current.join("0000 broken.cbd-track.toml"), "not [ toml") + .expect("write broken entry"); + std::fs::write(current.join(STATE_FILE_NAME), "also not [ toml") + .expect("break the sidecar"); + let loaded = store.load_current().await.expect("load"); + // The two good tracks load; the broken entry is skipped and the + // broken sidecar degrades to default state instead of failing. + assert_eq!(loaded.tracks.len(), 2); + assert_eq!(loaded.current_position, 0); + } + + #[tokio::test] + async fn persister_writes_the_latest_snapshot() { + let (store, _dir) = store().await; + let store = Arc::new(store); + let (tx, rx) = tokio::sync::watch::channel(None); + spawn_persister(Arc::clone(&store), rx); + // A burst: only the newest snapshot matters (latest-wins channel). + tx.send(Some(snapshot(5))).expect("send"); + tx.send(Some(snapshot(3))).expect("send"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + if let Some(loaded) = store.load_current().await { + if loaded.tracks.len() == 3 { + break; + } + } + assert!( + tokio::time::Instant::now() < deadline, + "persister never wrote the latest snapshot" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } +} diff --git a/crabidy-server/src/rpc.rs b/crabidy-server/src/rpc.rs index 4f724cc..545299f 100644 --- a/crabidy-server/src/rpc.rs +++ b/crabidy-server/src/rpc.rs @@ -14,6 +14,7 @@ use crabidy_core::proto::crabidy::{ ToggleRepeatResponse, ToggleShuffleRequest, ToggleShuffleResponse, }; use crabidy_core::ProviderError; +use crabidy_server::queue_store::SaveQueueError; use std::pin::Pin; use tokio_stream::StreamExt; use tonic::{Request, Response, Status}; @@ -373,13 +374,40 @@ impl CrabidyService for RpcService { Ok(Response::new(Box::pin(output_stream))) } - #[instrument(skip(self, _request))] + /// Saves the current queue under a name (persisted queues, visible as + /// `/queues/<name>` in the library). + /// + /// Error mapping is part of the contract: an invalid name → + /// `invalid_argument`; an empty queue or disabled persistence → + /// `failed_precondition`; I/O and serialization failures → `internal`. + #[instrument(skip(self, request), fields(name))] async fn save_queue( &self, - _request: Request<SaveQueueRequest>, + request: Request<SaveQueueRequest>, ) -> Result<Response<SaveQueueResponse>, Status> { - debug!("received save_queue request (not implemented)"); - Ok(Response::new(SaveQueueResponse {})) + let name = request.into_inner().name; + tracing::Span::current().record("name", name.as_str()); + debug!("received save_queue request"); + let (result_tx, result_rx) = flume::bounded(1); + self.send_playback(PlaybackCommand::SaveQueue { name, result_tx }) + .await?; + let result = result_rx.recv_async().await.map_err(|err| { + error!("no reply from playback loop: {err}"); + Status::internal("playback loop did not reply") + })?; + match result { + Ok(()) => Ok(Response::new(SaveQueueResponse {})), + Err(err @ SaveQueueError::InvalidName(_)) => { + Err(Status::invalid_argument(err.to_string())) + } + Err(err @ (SaveQueueError::EmptyQueue | SaveQueueError::Disabled)) => { + Err(Status::failed_precondition(err.to_string())) + } + Err(err) => { + error!("save_queue failed: {err}"); + Err(Status::internal("cannot save the queue")) + } + } } #[instrument(skip(self, _request))] diff --git a/fsdy/src/lib.rs b/fsdy/src/lib.rs index 9623c9a..0dc7f38 100644 --- a/fsdy/src/lib.rs +++ b/fsdy/src/lib.rs @@ -56,13 +56,17 @@ pub enum TrackFileError { UrlScheme(String), #[error("playable link must be an absolute crabidy path: {0}")] LinkNotAbsolute(String), - #[error("playable link must not point into {PROVIDER_ROOT}: {0}")] - LinkIntoFs(String), + #[error("cannot serialize track file: {0}")] + Serialize(#[from] toml::ser::Error), } /// The on-disk schema of a `*.cbd-track.toml` file. See /// `architecture/fs-provider.md` (D3) for the format documentation. -#[derive(Debug, Deserialize)] +/// +/// Serializable both ways: queue persistence writes these files (see +/// `architecture/queue-persistence.md` D2). `Option` fields are skipped on +/// serialization — TOML has no null. +#[derive(Debug, Deserialize, Serialize)] pub struct TrackFile { /// Track title. Required. pub title: String, @@ -70,31 +74,39 @@ pub struct TrackFile { #[serde(default)] pub artist: String, /// Optional duration in seconds. + #[serde(skip_serializing_if = "Option::is_none")] pub duration: Option<u32>, /// Optional album metadata. + #[serde(skip_serializing_if = "Option::is_none")] pub album: Option<AlbumMeta>, /// The playable reference; exactly one of its fields must be set. pub playable: PlayableSpec, } /// Optional `[album]` table of a track file. -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Serialize)] pub struct AlbumMeta { pub title: String, + #[serde(skip_serializing_if = "Option::is_none")] pub release_date: Option<String>, } /// Raw `[playable]` table: three optional fields so cardinality errors are /// precise. Validated into a [`Playable`] before use. -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Serialize)] pub struct PlayableSpec { /// Local audio file; absolute, or relative to the track file's /// directory. + #[serde(skip_serializing_if = "Option::is_none")] pub file: Option<PathBuf>, /// http(s) URL streamed by the player. + #[serde(skip_serializing_if = "Option::is_none")] pub url: Option<String>, - /// Absolute crabidy track path owned by another provider - /// (e.g. `/tidal/...`). Links into `/fs` are rejected (no chains). + /// Absolute crabidy track path, usually owned by another provider + /// (e.g. `/tidal/...`). Links resolve **one hop** by construction: + /// `get_urls_for_track` never follows a link, so a link whose target is + /// itself a link file fails at play time and cycles cannot recurse. + #[serde(skip_serializing_if = "Option::is_none")] pub link: Option<String>, } @@ -137,9 +149,6 @@ impl TrackFile { if !link.starts_with('/') { return Err(TrackFileError::LinkNotAbsolute(link.clone())); } - if link == PROVIDER_ROOT || link.starts_with("/fs/") { - return Err(TrackFileError::LinkIntoFs(link.clone())); - } Ok(Playable::Link(link.clone())) } _ => Err(TrackFileError::PlayableCardinality), @@ -168,6 +177,66 @@ impl TrackFile { }), } } + + /// The inverse of [`Self::to_track`] for persistence: a track file whose + /// metadata is copied from `track` and whose playable is a + /// [`Playable::Link`] to `track.path` — uniformly, for every track (see + /// `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. + pub fn from_track(track: &Track) -> Self { + Self { + title: track.title.clone(), + artist: track.artist.clone(), + duration: track.duration, + album: track.album.as_ref().map(|a| AlbumMeta { + title: a.title.clone(), + release_date: a.release_date.clone(), + }), + playable: PlayableSpec { + file: None, + url: None, + link: Some(track.path.clone()), + }, + } + } + + /// Serializes this file to TOML text. + /// + /// Only fails when TOML cannot represent the value + /// ([`TrackFileError::Serialize`]); never panics. + pub fn to_toml(&self) -> Result<String, TrackFileError> { + Ok(toml::to_string_pretty(self)?) + } +} + +/// File name for the serialized queue entry at zero-based `index`: +/// a zero-padded four-digit one-based order prefix, a sanitized `title`, +/// and [`TRACK_FILE_SUFFIX`] — e.g. `0001 Bohemian Rhapsody.cbd-track.toml`. +/// +/// The prefix makes the case-insensitive listing sort reproduce queue order +/// (wrong beyond 9999 tracks — accepted, see +/// `architecture/queue-persistence.md`). Sanitizing replaces path +/// separators and NUL, strips leading dots (hidden files are skipped by +/// listings), and falls back to `track` for an empty result. Never panics. +pub fn track_file_name(index: usize, title: &str) -> String { + let sanitized: String = title + .chars() + .map(|c| { + if matches!(c, '/' | '\\' | '\0') { + '_' + } else { + c + } + }) + .collect(); + let sanitized = sanitized.trim_start_matches('.').trim(); + let name = if sanitized.is_empty() { + "track" + } else { + sanitized + }; + format!("{:04} {name}{TRACK_FILE_SUFFIX}", index + 1) } /// The filesystem provider. @@ -175,13 +244,60 @@ impl TrackFile { /// All I/O is `tokio::fs`; nothing is cached — every node visit reads the /// directory fresh, so edits made with normal file tools appear on the next /// navigation. The client never panics on the contents of the tree. +/// +/// A process may mount several instances, each serving one disk root under +/// its own provider root: the configured `/fs` instance (built by +/// [`ProviderClient::init`] from `fsdy.toml`) and the server's `/queues` +/// instance over the persisted-queues folder (see +/// `architecture/queue-persistence.md` D1). #[derive(Debug)] pub struct Client { root: PathBuf, + /// First library path segment this instance owns, e.g. `/fs`. + provider_root: String, } impl Client { - /// Maps a `/fs/...` library path to the on-disk path it addresses. + /// Builds an instance serving `disk_root` under `provider_root`. + /// + /// `provider_root` must be a single absolute segment (`/name`, no + /// trailing slash, no inner slash) and `disk_root` an absolute path; + /// anything else is [`ProviderError::Config`]. A `disk_root` that does + /// not exist (yet) is accepted — listing it just fails until it appears. + pub fn new(provider_root: &str, disk_root: PathBuf) -> Result<Self, ProviderError> { + let is_single_absolute_segment = provider_root.len() > 1 + && provider_root.starts_with('/') + && !provider_root[1..].contains('/'); + if !is_single_absolute_segment { + return Err(ProviderError::Config(format!( + "provider root must be a single absolute segment like /fs: {provider_root}" + ))); + } + if !disk_root.is_absolute() { + return Err(ProviderError::Config(format!( + "disk root must be an absolute path: {}", + disk_root.display() + ))); + } + Ok(Self { + root: disk_root, + provider_root: provider_root.to_string(), + }) + } + + /// The provider root this instance owns (e.g. `/fs`), as configured + /// through [`Self::new`]. + pub fn provider_root(&self) -> &str { + &self.provider_root + } + + /// The display title of this instance's root node: the provider root + /// without its leading slash (`fs`, `queues`). + fn root_title(&self) -> &str { + self.provider_root.trim_start_matches('/') + } + /// Maps a library path under this instance's provider root to the + /// on-disk path it addresses. /// /// Segments are decoded with [`crabidy_core::decode_segment`] and each /// must be a plain file name: decoded segments that are empty, `.`, @@ -189,11 +305,12 @@ impl Client { /// [`ProviderError::MalformedPath`] — the result is always a pure /// descent from the root, so client-supplied paths cannot escape it. fn disk_path(&self, lib_path: &str) -> Result<PathBuf, ProviderError> { - let rest = if lib_path == PROVIDER_ROOT { + let rest = if lib_path == self.provider_root { "" } else { lib_path - .strip_prefix("/fs/") + .strip_prefix(self.provider_root.as_str()) + .and_then(|rest| rest.strip_prefix('/')) .ok_or(ProviderError::MalformedPath)? }; let mut disk = self.root.clone(); @@ -283,8 +400,8 @@ impl Client { }) .collect(); - let title = if lib_path == PROVIDER_ROOT { - "fs".to_string() + let title = if lib_path == self.provider_root { + self.root_title().to_string() } else { crabidy_core::path_segments(lib_path) .last() @@ -332,13 +449,8 @@ impl ProviderClient for Client { "no `root` configured and no platform music directory".to_string(), ) })?; - if !root.is_absolute() { - return Err(ProviderError::Config(format!( - "`root` must be an absolute path: {}", - root.display() - ))); - } - Ok(Self { root }) + // `new` enforces the absolute-path requirement for `root`. + Self::new(PROVIDER_ROOT, root) } /// Serializes the effective settings back for the config write-back. @@ -352,9 +464,12 @@ impl ProviderClient for Client { }) } - /// `/fs/...` paths ending in [`TRACK_FILE_SUFFIX`] address tracks. + /// Paths under this instance's provider root ending in + /// [`TRACK_FILE_SUFFIX`] address tracks. fn is_track_path(&self, path: &str) -> bool { - path.starts_with("/fs/") && path.ends_with(TRACK_FILE_SUFFIX) + path.strip_prefix(self.provider_root.as_str()) + .is_some_and(|rest| rest.starts_with('/')) + && path.ends_with(TRACK_FILE_SUFFIX) } /// Resolves the playable reference of a track file. @@ -411,12 +526,13 @@ impl ProviderClient for Client { Ok(file.to_track(track_path)) } - /// A minimal `/fs` node; children are discovered via [`Self::get_lib_node`] - /// (the trait method is synchronous, directory listing is not). + /// A minimal root node for this instance; children are discovered via + /// [`Self::get_lib_node`] (the trait method is synchronous, directory + /// listing is not). fn get_lib_root(&self) -> LibraryNode { LibraryNode { - path: PROVIDER_ROOT.to_string(), - title: "fs".to_string(), + path: self.provider_root.clone(), + title: self.root_title().to_string(), children: Vec::new(), parent: Some(crabidy_core::ROOT_PATH.to_string()), tracks: Vec::new(), @@ -551,10 +667,6 @@ mod tests { "title = \"x\"\n[playable]\nlink = \"tidal/relative\"\n", "relative link", ), - ( - "title = \"x\"\n[playable]\nlink = \"/fs/other.cbd-track.toml\"\n", - "link into /fs", - ), ("title = \"x\"\nnot toml at all [", "invalid toml"), ]; for (bad, what) in cases { @@ -567,6 +679,26 @@ mod tests { } } + #[test] + fn links_into_fs_instances_are_legal_and_one_hop() { + // Persisted queues link to whatever the queue held — including + // `/fs/...` tracks — so links into fs-provider instances must parse + // (architecture/queue-persistence.md D2). Safety comes from links + // being one hop by construction: `get_urls_for_track` never follows + // a link (covered by `urls_resolve_per_playable_kind`), so chains + // die at play time and cycles cannot recurse. + let file = TrackFile::parse("title = \"t\"\n[playable]\nlink = \"/fs/a.cbd-track.toml\"\n") + .expect("link into /fs parses"); + assert_eq!( + file.playable().expect("playable"), + Playable::Link("/fs/a.cbd-track.toml".into()) + ); + assert_eq!( + file.to_track("/queues/mix/0001 t.cbd-track.toml").path, + "/fs/a.cbd-track.toml" + ); + } + #[test] fn to_track_rewrites_the_path_only_for_links() { let lib_path = "/fs/mix/song.cbd-track.toml"; @@ -789,4 +921,138 @@ mod tests { let err = client.get_lib_node("/fs").await.expect_err("listing fails"); assert_eq!(err, ProviderError::MalformedPath); } + + // ---- multiple instances (queue persistence) -------------------------- + + #[tokio::test] + async fn instances_serve_their_own_provider_root() { + let dir = TempDir::new().expect("tempdir"); + fs::create_dir(dir.path().join("road trip")).expect("mkdir"); + write_track( + &dir.path().join("road trip"), + "0001 one.cbd-track.toml", + "title = \"one\"\n[playable]\nlink = \"/tidal/artists/1/2\"\n", + ); + let client = Client::new("/queues", dir.path().to_path_buf()).expect("instance"); + assert_eq!(client.provider_root(), "/queues"); + + // The whole path scheme follows the instance root... + let root = client.get_lib_node("/queues").await.expect("root"); + assert_eq!(root.title, "queues"); + assert_eq!(root.parent.as_deref(), Some(crabidy_core::ROOT_PATH)); + assert_eq!(root.children[0].path, "/queues/road%20trip"); + let node = client + .get_lib_node("/queues/road%20trip") + .await + .expect("queue folder"); + assert_eq!(node.parent.as_deref(), Some("/queues")); + assert_eq!(node.tracks[0].path, "/tidal/artists/1/2"); + assert!( + client.is_track_path("/queues/road%20trip/0001%20one.cbd-track.toml"), + "track paths under the instance root" + ); + + // ... and paths of other instances are foreign to this one. + assert!(!client.is_track_path("/fs/x.cbd-track.toml")); + assert!(client.get_lib_node("/fs").await.is_err()); + let lib_root = ProviderClient::get_lib_root(&client); + assert_eq!(lib_root.path, "/queues"); + assert_eq!(lib_root.title, "queues"); + } + + #[tokio::test] + async fn instance_paths_cannot_escape_their_root_either() { + let dir = TempDir::new().expect("tempdir"); + let client = Client::new("/queues", dir.path().to_path_buf()).expect("instance"); + for evil in ["/queues/..", "/queues/%2E%2E", "/queues//x"] { + let err = client.get_lib_node(evil).await.expect_err(evil); + assert_eq!(err, ProviderError::MalformedPath, "{evil}"); + } + } + + #[test] + fn new_rejects_malformed_roots() { + let dir = TempDir::new().expect("tempdir"); + let disk = dir.path().to_path_buf(); + for bad_provider_root in ["queues", "/", "/a/b", "/queues/", ""] { + assert!( + matches!( + Client::new(bad_provider_root, disk.clone()), + Err(ProviderError::Config(_)) + ), + "provider root {bad_provider_root:?} must be rejected" + ); + } + assert!( + matches!( + Client::new("/queues", PathBuf::from("relative/dir")), + Err(ProviderError::Config(_)) + ), + "relative disk roots must be rejected" + ); + } + + // ---- serialization (queue persistence) ------------------------------- + + #[test] + fn from_track_round_trips_through_a_link_file() { + let track = Track { + path: "/fs/mix/song.cbd-track.toml".to_string(), + artist: "Queen".to_string(), + title: "We Will Rock You".to_string(), + duration: Some(122), + album: Some(Album { + title: "News of the World".to_string(), + release_date: Some("1977-10-28".to_string()), + }), + }; + let toml_text = TrackFile::from_track(&track).to_toml().expect("serialize"); + let reparsed = TrackFile::parse(&toml_text).expect("reparse"); + assert_eq!( + reparsed.playable().expect("playable"), + Playable::Link("/fs/mix/song.cbd-track.toml".into()) + ); + // The listing rewrite turns the link back into the original track, + // wherever the file lives. + let restored = reparsed.to_track("/queues/current/0001 We Will Rock You.cbd-track.toml"); + assert_eq!(restored, track); + } + + #[test] + fn from_track_serializes_sparse_metadata() { + // No artist/duration/album: serialization must not fail on `None` + // (TOML has no null) and the round trip stays lossless. + let track = Track { + path: "/tidal/artists/1/2".to_string(), + artist: String::new(), + title: "stream".to_string(), + duration: None, + album: None, + }; + let toml_text = TrackFile::from_track(&track).to_toml().expect("serialize"); + let reparsed = TrackFile::parse(&toml_text).expect("reparse"); + assert_eq!(reparsed.to_track("/queues/x.cbd-track.toml"), track); + } + + #[test] + fn track_file_names_sort_in_queue_order_and_stay_plain() { + // Zero-padded one-based prefix, suffix appended. + assert_eq!( + track_file_name(0, "Bohemian Rhapsody"), + "0001 Bohemian Rhapsody.cbd-track.toml" + ); + assert_eq!(track_file_name(9, "x"), "0010 x.cbd-track.toml"); + // Sanitized: no separators/NUL (they would corrupt the folder), no + // leading dot (hidden files are invisible to listings), never empty. + let tricky = track_file_name(1, "../a/b\\c\0"); + assert!(!tricky.contains(['/', '\\', '\0']), "{tricky}"); + assert!(!tricky.starts_with('.'), "{tricky}"); + assert_eq!(track_file_name(2, ""), "0003 track.cbd-track.toml"); + assert_eq!(track_file_name(3, "..."), "0004 track.cbd-track.toml"); + // The case-insensitive listing sort reproduces queue order. + let names: Vec<String> = (0..12).map(|i| track_file_name(i, "Song")).collect(); + let mut sorted = names.clone(); + sorted.sort_by_key(|n| n.to_lowercase()); + assert_eq!(names, sorted); + } } diff --git a/plan/queue-persistence.md b/plan/queue-persistence.md new file mode 100644 index 0000000..9e5f3a6 --- /dev/null +++ b/plan/queue-persistence.md @@ -0,0 +1,68 @@ +# Plan: queue-persistence + +Ordered tasks; each names its verification (tests in `fsdy/src/lib.rs` / +`crabidy-server/src/queue_store.rs` and/or gates in +`quality/queue-persistence.md`). Stubs and workspace wiring exist; the new +fsdy/queue-store tests fail on `todo!()` at plan time. + +- [x] **T1 — fsdy instance parameterization.** `Client::new(provider_root, + disk_root)` with validation; replace every hardcoded `"/fs"`/`"/fs/"` + inside `Client` methods with instance state (`disk_path`, + `is_track_path`, `list_dir` title/paths, `get_lib_root`); `init` builds + the `/fs` instance via `new`. Verifies: + `instances_serve_their_own_provider_root`, + `instance_paths_cannot_escape_their_root_either`, + `new_rejects_malformed_roots`; gates "fsdy instances" (prefix + derivation, single traversal site). +- [x] **T2 — fsdy serialization.** `TrackFile::from_track` (uniform link + playable), `to_toml`, `track_file_name` (zero-padded prefix + + sanitized title). Verifies: `from_track_round_trips_through_a_link_file`, + `from_track_serializes_sparse_metadata`, + `track_file_names_sort_in_queue_order_and_stay_plain`, + `links_into_fs_instances_are_legal_and_one_hop` (rule relaxation, done + at stub time). +- [x] **T3 — QueueStore.** `open` (create_dir_all), `validate_name`, + `save`/`persist_current` via one tmp-and-swap writer (entries + + `.queue-state.toml` sidecar), `load_current` (sorted listing, skip + broken, default state on sidecar defects). Verifies: all + `queue_store::tests` except `persister_writes_the_latest_snapshot`; + gates "Queue store". +- [x] **T4 — persister task.** `spawn_persister`: watch changes → + debounce → skip-if-equal → `persist_current`, warn on failure, exit on + sender drop. Verifies: `persister_writes_the_latest_snapshot`; gates + "playback wiring" (debounce/skip, warnings only). +- [x] **T5 — playback loop wiring.** `Playback::new` takes + `Option<Arc<QueueStore>>`; watch sender + persist sends from + `broadcast_queue`, `play`, and the shuffle/repeat toggles; `run` + spawns the persister; `PlaybackCommand::SaveQueue { name, result_tx }` + handled on the loop (snapshot, spawn write, reply); + `restore_current` applies tracks + position + mods before `run`, no + autoplay. Playback-level tests (restore → Init snapshot; SaveQueue + ok/empty; a queue mutation reaches the store). Verifies: new + `playback::tests`; gates "Playback wiring". +- [x] **T6 — orchestrator + server startup.** `queues_client: + Option<Arc<fsdy::Client>>` mounted over `QueueStore::dir` at + `/queues`; routing arms in every trait method; root child when + present; `main.rs` builds the store (non-fatal), restores, spawns + everything in order. Verifies: gates "RPC and orchestrator"; + workspace build. +- [x] **T7 — rpc `save_queue`.** Send `SaveQueue` to the playback loop, + map `SaveQueueError` → `Status` (invalid_argument / + failed_precondition / internal). Verifies: gate "RPC and + orchestrator" (error mapping). +- [x] **T8 — TUI save flow.** `Action::QueueSaveAs` (`w`, `Scope::Queue`), + `InputPurpose::SaveQueue` (+ overlay label), `MessageFromUi::SaveQueue`, + orchestrator arm → `rpc::save_queue`. TUI tests: binding lookup, + overlay open only with a non-empty queue, submit sends the trimmed + name. Verifies: new `cbd-tui` tests; gates "TUI". +- [x] **T9 — full verification.** Whole workspace suite green; + clippy/fmt/taplo/markdownlint clean; walk every gate in + `quality/queue-persistence.md` and tick it; no `todo!()` left. +- [x] **T10 — live smoke test.** Temp store + real tree: persist a mixed + queue (tidal + fs tracks), reload it, browse `/queues` through the + provider instance, queue a saved folder via the resolve walk — remove + any temporary probe afterwards. Verifies: end-to-end D2/D5 behavior + outside unit scope. +- [x] **T11 — docs.** `plan/summary.md` section incl. deviations; + reconcile `architecture/fs-provider.md` (one-hop links) and + `architecture/queue-persistence.md` if the implementation diverged. diff --git a/plan/summary.md b/plan/summary.md index 5de0b2b..1e9ae6e 100644 --- a/plan/summary.md +++ b/plan/summary.md @@ -1,5 +1,60 @@ # Implementation summaries +## queue-persistence (2026-07-21) + +Built per `plan/queue-persistence.md`: queues now survive server restarts, +realized entirely on top of the fs provider. `fsdy::Client` became +instance-mountable (`Client::new(provider_root, disk_root)`); the +orchestrator mounts a second, read-only instance at `/queues` over +`<config>/crabidy/queues/`, so saved queues are ordinary browsable, +queueable library folders. Every queue is a folder of order-prefixed +(`0001 <title>.cbd-track.toml`) **link** files — metadata copied from the +queue entry, `playable.link = Track.path` — written only by the new +`crabidy_server::queue_store::QueueStore` (tmp-and-swap, hidden +`.queue-state.toml` sidecar for position/repeat/shuffle). The playback +loop feeds every queue-state change into a latest-wins `watch` channel; a +persister task debounces, skips unchanged snapshots, and rewrites +`queues/current/`. On startup the server restores tracks, position, and +modifiers from `current/` without ever starting playback. `w` on the TUI +queue pane opens the existing input overlay (`save queue`) and drives the +previously stubbed `SaveQueue` rpc (invalid name → `invalid_argument`, +empty queue/disabled persistence → `failed_precondition`). Reloading a +saved queue is just queueing `/queues/<name>` — the listing rewrites each +link back to its target, so zero new resolve mechanisms. All 116 +workspace tests green (10 new in `fsdy`, 9 in `queue_store`, 5 playback, +3 TUI); every gate in `quality/queue-persistence.md` checked. A temporary +live probe (removed after passing) round-tripped a mixed queue — a track +fetched from the live Tidal API plus an fs url track — through persist, +reload, `/queues` listing, and the resolve walk, and the reloaded Tidal +path still yielded a stream URL. + +The whole feature ran autonomously per standing instruction; decisions +are recorded in `architecture/queue-persistence.md` (options + rationale). + +### Deviations from plan / architecture (queue-persistence) + +- **The "no links into `/fs`" rule was dropped** (fs-provider D3): queue + entries persist as links to whatever path the queue held, including + `/fs/...` tracks. Replaced by the one-hop argument — + `get_urls_for_track` never follows a link, so chains die at play time + and cycles cannot recurse. `architecture/fs-provider.md` reconciled. +- **`SaveQueueError` gained `Disabled` and `State` variants** beyond the + stub: `Disabled` (no usable queues directory) maps to + `failed_precondition` instead of masquerading as I/O; `State` covers + sidecar serialization. +- **The orchestrator mounts `/queues` independently of `QueueStore`**: + both derive the directory from `queue_store::queues_dir()`, so a + mount over a not-yet-created folder simply lists as missing until the + store (created in `main`) writes it. No plumbing between the two. +- **Shuffle order is not persisted** (documented in D3/D4 but worth + repeating): restoring `shuffle = true` reshuffles around the restored + current track. +- **The live probe needed no bespoke server run**: provider-layer clients + plus `QueueStore` cover the full D2/D5 story; the gRPC and TUI layers + above are unit-tested. +- **Environment note**: builds/tests again ran with a session-local + `CARGO_TARGET_DIR`; no repo change. + ## fs-provider (2026-07-21) Built per `plan/fs-provider.md`: a second media provider (crate `fsdy`, diff --git a/quality/queue-persistence.md b/quality/queue-persistence.md new file mode 100644 index 0000000..7fa2356 --- /dev/null +++ b/quality/queue-persistence.md @@ -0,0 +1,92 @@ +# Quality gates: queue-persistence + +Criteria the implementation must satisfy beyond the automatic tests +(`fsdy/src/lib.rs`, `crabidy-server/src/queue_store.rs`, plus the playback +and TUI tests added during implementation). Each gate is pass/fail by +reading the code. + +## fsdy instances and serialization + +- [x] Every occurrence of the hardcoded `"/fs"`/`"/fs/"` prefix inside + `fsdy::Client` methods now derives from the instance's `provider_root`; + `PROVIDER_ROOT` remains only as the default instance's constant + (`init`) and for external callers. +- [x] Path traversal validation still happens in exactly one place + (`disk_path`) and applies to every instance. +- [x] `TrackFile::from_track` is the single Track→file conversion site, + and it always emits a `link` playable (no special cases per provider). +- [x] Serialization never panics: `to_toml` returns + `TrackFileError::Serialize`, and every `Option` field is + skip-serialized (TOML cannot represent `None`). +- [x] The one-hop link argument holds in code: `get_urls_for_track` still + returns `MalformedPath` for a link playable, so removing the + `LinkIntoFs` rejection cannot introduce recursion anywhere. + +## Queue store + +- [x] `QueueStore` is the only writer of the queues directory; the + `/queues` provider instance only reads. +- [x] Writes are tmp-and-swap: entries are written to a hidden + (dot-prefixed) temp sibling, then the old folder is removed and the + temp renamed. No code path writes entries into the live folder + directly. +- [x] `save` validates the name first and never touches disk for an + invalid name or empty snapshot. +- [x] `load_current` never fails the server: missing folder → `None`, + broken entry → skip with a warning naming the file (never its + contents), broken sidecar → default state with a warning. +- [x] File names come from `fsdy::track_file_name` — no second naming + scheme. +- [x] No file contents in logs anywhere in the store (paths and names + only). + +## Playback wiring + +- [x] The playback loop never blocks on disk: auto-persist goes through + the `watch` channel (latest wins), `SaveQueue` writes on a spawned + task that reports back through the command's result channel. +- [x] Every queue-state change reaches the persist channel: queue + content changes (the `broadcast_queue` funnel), current-track changes + (`play`), and the shuffle/repeat toggles. +- [x] The persister task debounces and skips snapshots equal to the last + one written (pure `resolving`-flag broadcasts must not rewrite the + folder). +- [x] Persist failures are warnings; no persist error can stop playback + or crash the loop. +- [x] The startup restore runs before the playback loop serves commands, + restores tracks + position + repeat/shuffle, and never starts + playback (`PlayState::Stopped`). +- [x] Restore tolerates a corrupt position (out of range → clamped or + reset, never a panic). +- [x] A server without a usable queues directory (no config dir, mkdir + fails) runs without persistence after a warning — never dies. + +## RPC and orchestrator + +- [x] `save_queue` maps errors: invalid name → `invalid_argument`, empty + queue → `failed_precondition`, I/O → `internal`; no `color-eyre`/debug + reports leak to clients. +- [x] The orchestrator routes `/queues` in **every** `ProviderClient` + method (same completeness as `/fs`), and `get_lib_root` lists the + `queues` child only when the instance exists. +- [x] `/queues` mutations via the library stay `NotSupported` + (create/rename/delete unchanged). + +## TUI + +- [x] `w` is bound in `Scope::Queue` only, has a help description, and + passes the existing bindings-table invariant tests unchanged. +- [x] The save overlay reuses `InputState` (Esc cancels, Enter submits + trimmed, empty submit closes silently) and is a no-op while the queue + is empty. +- [x] `MessageFromUi::SaveQueue` reaches the `SaveQueue` RPC; a failed + save must not crash the TUI. + +## Hygiene + +- [x] New public items are documented; docs state error/edge behavior. +- [x] `clippy -D warnings`, `fmt`, `taplo`, `markdownlint` clean on the + whole workspace; all tests green. +- [x] `architecture/fs-provider.md` reconciled: the "no links into /fs" + rule replaced by the one-hop semantics, D2's "chains structurally + impossible" wording updated.