473 lines
18 KiB
Rust
473 lines
18 KiB
Rust
//! 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> {
|
|
// The shared fs-provider naming rules, with the auto-persisted
|
|
// queue's folder reserved.
|
|
fsdy::validate_folder_name(name, &[CURRENT_QUEUE_NAME]).map_err(SaveQueueError::InvalidName)
|
|
}
|
|
|
|
/// 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,
|
|
}),
|
|
is_skipped: false,
|
|
}
|
|
}
|
|
|
|
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;
|
|
}
|
|
}
|
|
}
|