From d504ebc85f89dc49d190c411f095a0b3af1c08d3 Mon Sep 17 00:00:00 2001 From: AI User Date: Sun, 19 Jul 2026 22:00:22 +0200 Subject: [PATCH] Fix intermittent playback stops and harden the queue Root causes found and fixed: - QueueManager could panic and kill the playback task permanently: is_last_track() underflowed on an empty queue, remove_tracks accepted pos == len (Vec::remove panic) and corrupted positions when removing multiple tracks (indices shifted mid-loop), shuffle_behind indexed out of range on an empty play order, insert_tracks shifted play-order entries by the queue length instead of the inserted count and then assert!()ed on the resulting inconsistency, and clear() left play_order stale. All mutation methods are now guarded, multi-remove works highest-position-first, and an inconsistent play order is rebuilt instead of panicking. Regression tests cover these cases. - The tidal access token was only obtained at startup and never refreshed, so long-running sessions ended with every track fetch failing (playback just stopped at the next track boundary). Login state now lives behind a lock; tokens are refreshed proactively before expiry (5 min margin) and once reactively on a 401, and all API responses are status-checked (new ClientError::ApiError) instead of being fed to the JSON decoder blind. The http client also got a 30s timeout so a hung connection cannot wedge the provider loop. - (from the rodio rewrite, same bug class) end of stream used to be detected by string-comparing an io::Error message; any other decode or network error ended the stream silently without an EndOfStream message, so playback never advanced. EOS is now a guaranteed callback with a generation counter. Plus workspace-wide clippy cleanup (zero warnings), cargo-machete cleanup, and fmt. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 2 - Cargo.toml | 1 - audio-player/examples/basic.rs | 12 +- cbd-tui/src/app/library.rs | 9 +- cbd-tui/src/app/list.rs | 6 - cbd-tui/src/app/mod.rs | 2 +- cbd-tui/src/app/now_playing.rs | 6 +- cbd-tui/src/app/queue.rs | 11 +- cbd-tui/src/main.rs | 31 +-- crabidy-core/Cargo.toml | 6 +- crabidy-core/src/lib.rs | 62 +++--- crabidy-server/Cargo.toml | 1 - crabidy-server/src/lib.rs | 179 +++++++++++++---- tidaldy/src/config.rs | 2 +- tidaldy/src/lib.rs | 339 ++++++++++++++++----------------- tidaldy/src/models.rs | 3 + 16 files changed, 394 insertions(+), 278 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7b203e0..9185b0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -721,7 +721,6 @@ name = "crabidy-core" version = "0.1.0" dependencies = [ "async-trait", - "clap", "clap-serde-derive", "dirs", "prost", @@ -743,7 +742,6 @@ dependencies = [ "dirs", "flume", "futures", - "log", "rand 0.10.2", "tidaldy", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 4b2e18f..736d9d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,6 @@ crossterm = "0.29" dirs = "6" flume = "0.12" futures = "0.3" -log = "0.4" notify-rust = "4" prost = "0.14" rand = "0.10" diff --git a/audio-player/examples/basic.rs b/audio-player/examples/basic.rs index 90491c8..5e94f01 100644 --- a/audio-player/examples/basic.rs +++ b/audio-player/examples/basic.rs @@ -1,5 +1,3 @@ -use std::{thread, time::Duration}; - use audio_player::{Player, PlayerMessage}; #[tokio::main] @@ -13,7 +11,10 @@ async fn main() { loop { match player.messages.recv_async().await { - Ok(PlayerMessage::Elapsed { duration, elapsed }) => { + Ok(PlayerMessage::Elapsed { + duration: _, + elapsed, + }) => { println!("ELAPSED: {:?}", elapsed); } Ok(PlayerMessage::EndOfStream) => { @@ -30,7 +31,10 @@ async fn main() { loop { match player.messages.recv_async().await { - Ok(PlayerMessage::Elapsed { duration, elapsed }) => { + Ok(PlayerMessage::Elapsed { + duration: _, + elapsed, + }) => { println!("ELAPSED: {:?}", elapsed); } Ok(PlayerMessage::EndOfStream) => { diff --git a/cbd-tui/src/app/library.rs b/cbd-tui/src/app/library.rs index e6c2436..b123828 100644 --- a/cbd-tui/src/app/library.rs +++ b/cbd-tui/src/app/library.rs @@ -54,14 +54,15 @@ impl Library { } pub fn ascend(&mut self) { if let Some(parent) = self.parent.as_ref() { - self.tx.send(MessageFromUi::GetLibraryNode(parent.clone())); + let _ = self.tx.send(MessageFromUi::GetLibraryNode(parent.clone())); } } pub fn dive(&mut self) { if let Some(idx) = self.list_state.selected() { let item = &self.list[idx]; if let UiItemKind::Node = item.kind { - self.tx + let _ = self + .tx .send(MessageFromUi::GetLibraryNode(item.path.clone())); } } @@ -103,7 +104,7 @@ impl Library { } pub fn toggle_mark(&mut self) { if let Some(idx) = self.list_state.selected() { - let mut item = &mut self.list[idx]; + let item = &mut self.list[idx]; if !item.is_queable { return; } @@ -176,7 +177,7 @@ impl Library { } else { Style::default() }; - return ListItem::new(Span::from(text)).style(style); + ListItem::new(Span::from(text)).style(style) }) .collect(); diff --git a/cbd-tui/src/app/list.rs b/cbd-tui/src/app/list.rs index 6a4b63e..a39876e 100644 --- a/cbd-tui/src/app/list.rs +++ b/cbd-tui/src/app/list.rs @@ -1,5 +1,3 @@ -pub use ratatui::widgets::ListState; - // FIXME: Move marking stuff here, to be able to use it in queue as well pub trait StatefulList { fn get_size(&self) -> usize; @@ -72,10 +70,6 @@ pub trait StatefulList { } } - fn is_selected(&self) -> bool { - self.selected().is_some() - } - fn is_empty(&self) -> bool { self.get_size() == 0 } diff --git a/cbd-tui/src/app/mod.rs b/cbd-tui/src/app/mod.rs index 58b758e..7edeb80 100644 --- a/cbd-tui/src/app/mod.rs +++ b/cbd-tui/src/app/mod.rs @@ -105,7 +105,7 @@ impl App { } pub fn render(&mut self, f: &mut Frame) { - let full_screen = f.area(); + let _full_screen = f.area(); let library_focused = matches!(self.focus, UiFocus::Library); let queue_focused = matches!(self.focus, UiFocus::Queue); diff --git a/cbd-tui/src/app/now_playing.rs b/cbd-tui/src/app/now_playing.rs index e94d325..119a1e2 100644 --- a/cbd-tui/src/app/now_playing.rs +++ b/cbd-tui/src/app/now_playing.rs @@ -68,7 +68,7 @@ impl NowPlaying { self.track = active; } pub fn update_modifiers(&mut self, mods: &QueueModifiers) { - self.modifiers = mods.clone(); + self.modifiers = *mods; } pub fn render(&self, f: &mut Frame, area: Rect) { @@ -129,7 +129,7 @@ impl NowPlaying { f.render_widget(media_info_p, now_playing_layout[0]); - if let (Some(position), Some(duration), Some(track)) = + if let (Some(position), Some(duration), Some(_track)) = (self.position, self.duration, &self.track) { let pos = position.as_secs(); @@ -151,7 +151,7 @@ impl NowPlaying { let progress = LineGauge::default() .label("") .block(Block::default().borders(Borders::NONE)) - .gauge_style(Style::default().fg(COLOR_SECONDARY).bg(Color::Black)) + .filled_style(Style::default().fg(COLOR_SECONDARY).bg(Color::Black)) .ratio(ratio); f.render_widget(progress, elapsed_layout[0]); diff --git a/cbd-tui/src/app/queue.rs b/cbd-tui/src/app/queue.rs index db59489..542d394 100644 --- a/cbd-tui/src/app/queue.rs +++ b/cbd-tui/src/app/queue.rs @@ -30,14 +30,14 @@ impl Queue { } } pub fn play_next(&self) { - self.tx.send(MessageFromUi::NextTrack); + let _ = self.tx.send(MessageFromUi::NextTrack); } pub fn play_prev(&self) { - self.tx.send(MessageFromUi::PrevTrack); + let _ = self.tx.send(MessageFromUi::PrevTrack); } pub fn play_selected(&self) { if let Some(pos) = self.selected() { - self.tx.send(MessageFromUi::SetCurrentTrack(pos)); + let _ = self.tx.send(MessageFromUi::SetCurrentTrack(pos)); } } pub fn select_current(&mut self) { @@ -46,7 +46,7 @@ impl Queue { pub fn remove_track(&mut self) { if let Some(pos) = self.selected() { // FIXME: mark multiple tracks on queue and remove them - self.tx.send(MessageFromUi::RemoveTracks(vec![pos])); + let _ = self.tx.send(MessageFromUi::RemoveTracks(vec![pos])); } } pub fn update_position(&mut self, pos: usize) { @@ -57,8 +57,7 @@ impl Queue { self.list = queue .tracks .iter() - .enumerate() - .map(|(i, t)| UiItem { + .map(|t| UiItem { path: t.path.clone(), title: format!("{} - {}", t.artist, t.title), kind: UiItemKind::Track, diff --git a/cbd-tui/src/main.rs b/cbd-tui/src/main.rs index 7df8af2..381410e 100644 --- a/cbd-tui/src/main.rs +++ b/cbd-tui/src/main.rs @@ -82,7 +82,7 @@ async fn main() -> Result<(), Box> { Ok(()) } -async fn orchestrate<'a>( +async fn orchestrate( config: &'static Config, (tx, rx): (Sender, Receiver), ) -> Result<(), Box> { @@ -114,7 +114,7 @@ async fn poll( match msg { MessageFromUi::GetLibraryNode(path) => { if let Some(node) = rpc_client.get_library_node(&path).await? { - tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); + let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); } }, MessageFromUi::AppendTracks(uuids) => { @@ -211,7 +211,7 @@ fn run_ui(tx: Sender, rx: Receiver) { app.now_playing.update_track(track.track); app.queue.update_position(track.queue_position as usize); } - if let Some(ps) = PlayState::try_from(init_data.play_state).ok() { + if let Ok(ps) = PlayState::try_from(init_data.play_state) { app.now_playing.update_play_state(ps); } if let Some(mods) = init_data.mods { @@ -228,7 +228,7 @@ fn run_ui(tx: Sender, rx: Receiver) { } StreamUpdate::Position(pos) => app.now_playing.update_position(pos), StreamUpdate::PlayState(play_state) => { - if let Some(ps) = PlayState::try_from(play_state).ok() { + if let Ok(ps) = PlayState::try_from(play_state) { app.now_playing.update_play_state(ps); } } @@ -241,7 +241,10 @@ fn run_ui(tx: Sender, rx: Receiver) { } } - terminal.draw(|f| app.render(f)); + if let Err(err) = terminal.draw(|f| app.render(f)) { + error!("failed to draw frame: {err}"); + break; + } let timeout = tick_rate .checked_sub(last_tick.elapsed()) @@ -256,25 +259,25 @@ fn run_ui(tx: Sender, rx: Receiver) { } (_, KeyModifiers::NONE, KeyCode::Tab) => app.cycle_active(), (_, KeyModifiers::NONE, KeyCode::Char(' ')) => { - tx.send(MessageFromUi::TogglePlay); + let _ = tx.send(MessageFromUi::TogglePlay); } (_, KeyModifiers::NONE, KeyCode::Char('r')) => { - tx.send(MessageFromUi::RestartTrack); + let _ = tx.send(MessageFromUi::RestartTrack); } (_, KeyModifiers::SHIFT, KeyCode::Char('J')) => { - tx.send(MessageFromUi::ChangeVolume(-0.1)); + let _ = tx.send(MessageFromUi::ChangeVolume(-0.1)); } (_, KeyModifiers::SHIFT, KeyCode::Char('K')) => { - tx.send(MessageFromUi::ChangeVolume(0.1)); + let _ = tx.send(MessageFromUi::ChangeVolume(0.1)); } (_, KeyModifiers::NONE, KeyCode::Char('m')) => { - tx.send(MessageFromUi::ToggleMute); + let _ = tx.send(MessageFromUi::ToggleMute); } (_, KeyModifiers::NONE, KeyCode::Char('z')) => { - tx.send(MessageFromUi::ToggleShuffle); + let _ = tx.send(MessageFromUi::ToggleShuffle); } (_, KeyModifiers::NONE, KeyCode::Char('x')) => { - tx.send(MessageFromUi::ToggleRepeat); + let _ = tx.send(MessageFromUi::ToggleRepeat); } (_, KeyModifiers::CONTROL, KeyCode::Char('n')) => { app.queue.play_next(); @@ -351,10 +354,10 @@ fn run_ui(tx: Sender, rx: Receiver) { app.queue.remove_track(); } (UiFocus::Queue, KeyModifiers::NONE, KeyCode::Char('c')) => { - tx.send(MessageFromUi::ClearQueue(true)); + let _ = tx.send(MessageFromUi::ClearQueue(true)); } (UiFocus::Queue, KeyModifiers::SHIFT, KeyCode::Char('C')) => { - tx.send(MessageFromUi::ClearQueue(false)); + let _ = tx.send(MessageFromUi::ClearQueue(false)); } _ => {} } diff --git a/crabidy-core/Cargo.toml b/crabidy-core/Cargo.toml index d031a17..7e9d71b 100644 --- a/crabidy-core/Cargo.toml +++ b/crabidy-core/Cargo.toml @@ -5,7 +5,6 @@ edition.workspace = true [dependencies] async-trait.workspace = true -clap.workspace = true clap-serde-derive.workspace = true dirs.workspace = true prost.workspace = true @@ -16,3 +15,8 @@ tonic-prost.workspace = true [build-dependencies] tonic-prost-build.workspace = true + +# prost and tonic-prost are used by the code generated from the proto files, +# which cargo-machete cannot see. +[package.metadata.cargo-machete] +ignored = ["prost", "tonic-prost"] diff --git a/crabidy-core/src/lib.rs b/crabidy-core/src/lib.rs index cb68ea6..6316146 100644 --- a/crabidy-core/src/lib.rs +++ b/crabidy-core/src/lib.rs @@ -101,37 +101,6 @@ impl LibraryNodeChild { } } -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn parent_path_walks_up_to_root() { - assert_eq!( - parent_path("/tidal/playlists/abc"), - Some("/tidal/playlists") - ); - assert_eq!(parent_path("/tidal/playlists"), Some("/tidal")); - assert_eq!(parent_path("/tidal"), Some("/")); - assert_eq!(parent_path("/"), None); - } - - #[test] - fn join_path_appends_segments() { - assert_eq!(join_path("/", "tidal"), "/tidal"); - assert_eq!(join_path("/tidal", "playlists"), "/tidal/playlists"); - } - - #[test] - fn path_segments_splits() { - assert_eq!(path_segments("/"), Vec::<&str>::new()); - assert_eq!( - path_segments("/tidal/artists/1/2"), - vec!["tidal", "artists", "1", "2"] - ); - } -} - pub enum QueueError { NotQueable, } @@ -164,3 +133,34 @@ where } T::default().merge_clap() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parent_path_walks_up_to_root() { + assert_eq!( + parent_path("/tidal/playlists/abc"), + Some("/tidal/playlists") + ); + assert_eq!(parent_path("/tidal/playlists"), Some("/tidal")); + assert_eq!(parent_path("/tidal"), Some("/")); + assert_eq!(parent_path("/"), None); + } + + #[test] + fn join_path_appends_segments() { + assert_eq!(join_path("/", "tidal"), "/tidal"); + assert_eq!(join_path("/tidal", "playlists"), "/tidal/playlists"); + } + + #[test] + fn path_segments_splits() { + assert_eq!(path_segments("/"), Vec::<&str>::new()); + assert_eq!( + path_segments("/tidal/artists/1/2"), + vec!["tidal", "artists", "1", "2"] + ); + } +} diff --git a/crabidy-server/Cargo.toml b/crabidy-server/Cargo.toml index c488282..6094e07 100644 --- a/crabidy-server/Cargo.toml +++ b/crabidy-server/Cargo.toml @@ -15,7 +15,6 @@ crabidy-core.workspace = true dirs.workspace = true flume.workspace = true futures.workspace = true -log.workspace = true rand.workspace = true tidaldy.workspace = true tokio = { workspace = true, features = ["full"] } diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index 88d7415..9193775 100644 --- a/crabidy-server/src/lib.rs +++ b/crabidy-server/src/lib.rs @@ -28,6 +28,12 @@ impl From for Queue { } } +impl Default for QueueManager { + fn default() -> Self { + Self::new() + } +} + impl QueueManager { pub fn new() -> Self { Self { @@ -48,7 +54,7 @@ impl QueueManager { } pub fn is_last_track(&self) -> bool { - self.current_position() == self.tracks.len() - 1 + !self.tracks.is_empty() && self.current_position() == self.tracks.len() - 1 } pub fn shuffle_on(&mut self) { @@ -69,11 +75,15 @@ impl QueueManager { } pub fn shuffle_before(&mut self, pos: usize) { - self.play_order[..pos].shuffle(&mut rng()); + if let Some(slice) = self.play_order.get_mut(..pos) { + slice.shuffle(&mut rng()); + } } pub fn shuffle_behind(&mut self, pos: usize) { - self.play_order[pos + 1..].shuffle(&mut rng()); + if let Some(slice) = self.play_order.get_mut(pos + 1..) { + slice.shuffle(&mut rng()); + } } pub fn current_track(&self) -> Option { @@ -114,7 +124,7 @@ impl QueueManager { pub fn prev_track(&mut self) -> Option { if 0 < self.current_offset { self.current_offset -= 1; - Some(self.tracks[self.current_position()].clone()) + self.current_track() } else { None } @@ -178,28 +188,37 @@ impl QueueManager { pub fn remove_tracks(&mut self, positions: &[u32]) -> Option { let mut play_next = false; + // Remove highest positions first so earlier removals don't shift the + // positions that are still to be removed. + let mut positions: Vec = positions.iter().map(|p| *p as usize).collect(); + positions.sort_unstable_by(|a, b| b.cmp(a)); + positions.dedup(); for pos in positions { - if (self.tracks.len() as u32) < *pos { - return None; - }; - if *pos == self.current_position() as u32 { + if pos >= self.tracks.len() { + debug!(pos, len = self.tracks.len(), "ignoring out-of-range remove"); + continue; + } + if pos == self.current_position() { play_next = true; } - let Some(offset) = self.play_order.iter().position(|&i| i == *pos as usize) else { - error!("invalid current position"); - error!("queue: {:#?}", self); + let Some(offset) = self.play_order.iter().position(|&i| i == pos) else { + error!(pos, "track position missing from play order, rebuilding"); + self.rebuild_play_order(); return None; }; if offset < self.current_offset { self.current_offset -= 1; } - self.tracks.remove(*pos as usize); + self.tracks.remove(pos); self.play_order.remove(offset); self.play_order .iter_mut() - .filter(|i| (*pos as usize) < **i) + .filter(|i| pos < **i) .for_each(|i| *i -= 1); } + if self.current_offset >= self.play_order.len() { + self.current_offset = 0; + } if play_next { self.current_track() } else { @@ -212,38 +231,36 @@ impl QueueManager { if len == 0 { return self.replace_with_tracks(tracks); } - let order_additions: Vec = (len..len + tracks.len()).collect(); + let inserted = tracks.len(); + let position = (position as usize).min(len - 1); + let order_additions: Vec = (len..len + inserted).collect(); self.play_order.extend(order_additions); let tail: Vec = self .tracks - .splice((position as usize + 1).., tracks.to_vec()) + .splice(position + 1.., tracks.to_vec()) .collect(); self.tracks.extend(tail); let mut changed: Vec = Vec::new(); - // in shuffle mode, it might be that we played already postions which are behind - // the insertion point and which postions are shifted by the lenght of the inserted - // track + // In shuffle mode we may already have played positions that are + // behind the insertion point; those shift by the number of inserted + // tracks. for i in self .play_order .iter_mut() .take(self.current_offset) - .filter(|i| (position as usize) < **i) + .filter(|i| position < **i) { - *i += len; + *i += inserted; changed.push(*i); } - if !self.shuffle { - // if we don't shuffle, there should be no positions alredy played behind the - // current track - assert!(changed.is_empty()); - } - // the newly inserted indices need to replaced with the ones that we already handled + // The freshly appended order entries need to swap with the shifted + // ones so every index stays unique. self.play_order .iter_mut() .skip(self.current_offset) .for_each(|i| { if changed.contains(i) { - *i -= len; + *i -= inserted; } }); @@ -262,28 +279,126 @@ impl QueueManager { let current_track = self.current_track(); self.current_offset = 0; self.tracks.clear(); + self.play_order.clear(); if exclude_current { if let Some(track) = current_track { self.tracks.push(track); + self.play_order.push(0); } } !exclude_current } + + /// Restores play_order to a consistent state after an inconsistency was + /// detected. Loses shuffle history but keeps the queue playable. + fn rebuild_play_order(&mut self) { + self.play_order = (0..self.tracks.len()).collect(); + self.current_offset = 0; + } } #[cfg(test)] - mod tests { use super::*; + fn track(id: usize) -> Track { + Track { + path: format!("/tidal/playlists/p/{id}"), + artist: "artist".to_string(), + title: format!("track {id}"), + duration: None, + album: None, + } + } + + fn queue_with(n: usize) -> QueueManager { + let mut q = QueueManager::new(); + let tracks: Vec = (0..n).map(track).collect(); + q.replace_with_tracks(&tracks); + q + } + #[test] - fn random_delete_before() {} + fn empty_queue_operations_do_not_panic() { + let mut q = QueueManager::new(); + assert!(!q.is_last_track()); + assert!(q.current_track().is_none()); + assert!(q.next_track().is_none()); + assert!(q.prev_track().is_none()); + assert!(q.remove_tracks(&[0]).is_none()); + q.shuffle_on(); + q.shuffle_off(); + q.clear(true); + q.clear(false); + } + #[test] - fn random_delete_track() {} + fn remove_out_of_range_is_ignored() { + let mut q = queue_with(2); + assert!(q.remove_tracks(&[5]).is_none()); + assert_eq!(q.tracks.len(), 2); + // pos == len used to panic via Vec::remove + assert!(q.remove_tracks(&[2]).is_none()); + assert_eq!(q.tracks.len(), 2); + } + #[test] - fn random_delete_after() {} + fn remove_multiple_positions() { + let mut q = queue_with(4); + q.remove_tracks(&[1, 3]); + assert_eq!(q.tracks.len(), 2); + assert_eq!(q.play_order.len(), 2); + assert_eq!(q.current_track().unwrap().title, "track 0"); + } + #[test] - fn random_select_track() {} + fn remove_current_returns_successor() { + let mut q = queue_with(3); + let next = q.remove_tracks(&[0]); + assert_eq!(next.unwrap().title, "track 1"); + assert_eq!(q.tracks.len(), 2); + } + + #[test] + fn clear_keeps_play_order_consistent() { + let mut q = queue_with(3); + q.next_track(); + q.clear(true); + assert_eq!(q.tracks.len(), 1); + assert_eq!(q.play_order.len(), 1); + assert!(q.current_track().is_some()); + assert!(q.next_track().is_none()); + } + + #[test] + fn next_track_advances_and_repeats() { + let mut q = queue_with(2); + assert_eq!(q.next_track().unwrap().title, "track 1"); + assert!(q.next_track().is_none()); + q.repeat = true; + assert_eq!(q.next_track().unwrap().title, "track 0"); + } + + #[test] + fn insert_past_end_appends() { + let mut q = queue_with(2); + q.insert_tracks(99, &[track(2)]); + assert_eq!(q.tracks.len(), 3); + assert_eq!(q.play_order.len(), 3); + assert_eq!(q.tracks.last().unwrap().title, "track 2"); + } + + #[test] + fn shuffle_insert_keeps_order_unique() { + let mut q = queue_with(5); + q.shuffle_on(); + q.next_track(); + q.next_track(); + q.queue_tracks(&[track(5), track(6)]); + let mut order = q.play_order.clone(); + order.sort_unstable(); + assert_eq!(order, (0..7).collect::>()); + } } diff --git a/tidaldy/src/config.rs b/tidaldy/src/config.rs index 7834d9e..05b1219 100644 --- a/tidaldy/src/config.rs +++ b/tidaldy/src/config.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; use std::iter::zip; use thiserror::Error; -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Settings { pub base_url: String, pub hifi_url: String, diff --git a/tidaldy/src/lib.rs b/tidaldy/src/lib.rs index c1df883..b0849d3 100644 --- a/tidaldy/src/lib.rs +++ b/tidaldy/src/lib.rs @@ -13,8 +13,14 @@ pub use models::*; pub struct Client { http_client: HttpClient, settings: config::Settings, + /// Login state changes at runtime when tokens are refreshed, while the + /// client is shared immutably, hence the lock. Never held across awaits. + login: std::sync::RwLock, } +/// Refresh the access token this long before it actually expires. +const TOKEN_REFRESH_MARGIN_SECS: u64 = 300; + #[async_trait] impl crabidy_core::ProviderClient for Client { #[instrument(skip(raw_toml_settings))] @@ -27,17 +33,19 @@ impl crabidy_core::ProviderClient for Client { }; let mut client = Self::new(settings)?; - if let Ok(_) = client.login_config().await { + if client.login_config().await.is_ok() { return Ok(client); } - if let Ok(_) = client.login_web().await { + if client.login_web().await.is_ok() { return Ok(client); } Err(crabidy_core::ProviderError::CouldNotLogin) } #[instrument(skip(self))] fn settings(&self) -> String { - toml::to_string_pretty(&self.settings).unwrap_or_default() + let mut settings = self.settings.clone(); + settings.login = self.login_snapshot(); + toml::to_string_pretty(&settings).unwrap_or_default() } fn is_track_path(&self, path: &str) -> bool { matches!( @@ -113,7 +121,7 @@ impl crabidy_core::ProviderClient for Client { &self, path: &str, ) -> Result { - let Some(user_id) = self.settings.login.user_id.clone() else { + let Some(user_id) = self.get_user_id() else { return Err(crabidy_core::ProviderError::UnknownUser); }; let parsed = parse_path(path)?; @@ -293,17 +301,94 @@ impl Client { pub fn new(settings: config::Settings) -> Result { let http_client = HttpClient::builder() .user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 Edg/91.0.864.59") + .timeout(std::time::Duration::from_secs(30)) .build()?; + let login = std::sync::RwLock::new(settings.login.clone()); Ok(Self { http_client, settings, + login, }) } + /// A consistent copy of the current login state. + fn login_snapshot(&self) -> config::LoginConfig { + match self.login.read() { + Ok(login) => login.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + } + + fn store_refresh(&self, refresh: RefreshResponse) { + let now = chrono::Utc::now().timestamp() as u64; + let mut login = match self.login.write() { + Ok(login) => login, + Err(poisoned) => poisoned.into_inner(), + }; + login.expires_after = Some(now + refresh.expires_in); + login.access_token = Some(refresh.access_token); + if let Some(refresh_token) = refresh.refresh_token { + login.refresh_token = Some(refresh_token); + } + } + + /// Refreshes the access token unconditionally and stores the result. + async fn force_refresh_token(&self) -> Result<(), ClientError> { + let refresh = self.refresh_access_token().await?; + self.store_refresh(refresh); + info!("access token refreshed"); + Ok(()) + } + + /// Refreshes the access token if it is expired or about to expire. + /// Without this, long-running sessions ended up with an expired token + /// and every track fetch failed, silently stopping playback. + async fn ensure_fresh_token(&self) -> Result<(), ClientError> { + let login = self.login_snapshot(); + let Some(expires_after) = login.expires_after else { + return Ok(()); + }; + let now = chrono::Utc::now().timestamp() as u64; + if now + TOKEN_REFRESH_MARGIN_SECS < expires_after { + return Ok(()); + } + info!("access token expired or expiring soon"); + self.force_refresh_token().await + } + + /// Performs an authenticated GET against the hifi API. + async fn authed_get( + &self, + uri: &str, + query: Option<&[(&str, String)]>, + ) -> Result { + let login = self.login_snapshot(); + let Some(access_token) = login.access_token else { + return Err(ClientError::AuthError("No access token found".to_string())); + }; + let Some(country_code) = login.country_code else { + return Err(ClientError::AuthError("No country code found".to_string())); + }; + let mut params: Vec<(&str, String)> = vec![("countryCode", country_code)]; + if let Some(query) = query { + params.extend(query.iter().cloned()); + } + self.http_client + .get(format!("{}/{}", self.settings.hifi_url, uri)) + .bearer_auth(access_token) + .query(¶ms) + .send() + .await + .map_err(|e| { + warn!(uri, "tidal api request failed: {e}"); + ClientError::from(e) + }) + } + #[instrument(skip(self))] pub fn get_user_id(&self) -> Option { - self.settings.login.user_id.clone() + self.login_snapshot().user_id } #[instrument(skip(self))] @@ -312,37 +397,24 @@ impl Client { uri: &str, query: Option<&[(&str, String)]>, ) -> Result { - debug!("make_request {}", uri); - let Some(ref access_token) = self.settings.login.access_token.clone() else { - return Err(ClientError::AuthError("No access token found".to_string())); - }; - let Some(country_code) = self.settings.login.country_code.clone() else { - return Err(ClientError::AuthError("No country code found".to_string())); - }; - let country_param = ("countryCode", country_code); - let mut params: Vec<&(&str, String)> = vec![&country_param]; - if let Some(query) = query { - params.extend(query); + trace!(uri, "make_request"); + self.ensure_fresh_token().await?; + let mut response = self.authed_get(uri, query).await?; + if response.status() == reqwest::StatusCode::UNAUTHORIZED { + // The token may have been revoked or the clock may be off: + // refresh once and retry (GETs are idempotent). + info!(uri, "got 401, refreshing access token and retrying once"); + self.force_refresh_token().await?; + response = self.authed_get(uri, query).await?; } - - let response: T = self - .http_client - .get(format!("{}/{}", self.settings.hifi_url, uri)) - .bearer_auth(access_token) - .query(¶ms) - .send() - .await - .map_err(|e| { - error!("{:?}", e); - e - })? - .json() - .await - .map_err(|e| { - error!("{:?}", e); - e - })?; - Ok(response) + if !response.status().is_success() { + warn!(uri, status = %response.status(), "tidal api request failed"); + return Err(ClientError::ApiError(response.status().as_u16())); + } + response.json().await.map_err(|e| { + error!(uri, "failed to decode tidal api response: {e}"); + ClientError::from(e) + }) } #[instrument(skip(self))] @@ -351,68 +423,25 @@ impl Client { uri: &str, query: Option<&[(&str, String)]>, ) -> Result, ClientError> { - debug!("make_paginated_request {}", uri); - let Some(ref access_token) = self.settings.login.access_token.clone() else { - return Err(ClientError::AuthError("No access token found".to_string())); - }; - let Some(country_code) = self.settings.login.country_code.clone() else { - return Err(ClientError::AuthError("No country code found".to_string())); - }; - let country_param = ("countryCode", country_code); - let limit = 50; - let mut offset = 0; - let limit_param = ("limit", limit.to_string()); - let mut params: Vec<&(&str, String)> = vec![&country_param, &limit_param]; - if let Some(query) = query { - params.extend(query); - } - - let mut response: Page = self - .http_client - .get(format!("{}/{}", self.settings.hifi_url, uri)) - .bearer_auth(access_token) - .query(¶ms) - .send() - .await - .map_err(|e| { - error!("{:?}", e); - e - })? - .json() - .await - .map_err(|e| { - error!("{:?}", e); - e - })?; - let mut items = Vec::with_capacity(response.total_number_of_items); - items.extend(response.items); - while response.offset + limit < response.total_number_of_items { - offset += limit; - let offset_param = ("offset", offset.to_string()); - let mut params: Vec<&(&str, String)> = - vec![&country_param, &limit_param, &offset_param]; + trace!(uri, "make_paginated_request"); + let limit: usize = 50; + let mut offset: usize = 0; + let mut items = Vec::new(); + loop { + let mut params: Vec<(&str, String)> = + vec![("limit", limit.to_string()), ("offset", offset.to_string())]; if let Some(query) = query { - params.extend(query); + params.extend(query.iter().cloned()); + } + let page: Page = self.make_request(uri, Some(¶ms)).await?; + let fetched = page.items.len(); + items.extend(page.items); + offset += fetched; + if fetched == 0 || offset >= page.total_number_of_items { + break; } - response = self - .http_client - .get(format!("{}/{}", self.settings.hifi_url, uri)) - .bearer_auth(access_token) - .query(¶ms) - .send() - .await - .map_err(|e| { - error!("{:?}", e); - e - })? - .json() - .await - .map_err(|e| { - error!("{:?}", e); - e - })?; - items.extend(response.items); } + debug!(uri, count = items.len(), "fetched paginated collection"); Ok(items) } @@ -422,35 +451,8 @@ impl Client { uri: &str, query: Option<&[(&str, String)]>, ) -> Result<(), ClientError> { - let Some(ref access_token) = self.settings.login.access_token.clone() else { - return Err(ClientError::AuthError("No access token found".to_string())); - }; - let Some(country_code) = self.settings.login.country_code.clone() else { - return Err(ClientError::AuthError("No country code found".to_string())); - }; - let country_param = ("countryCode", country_code); - let mut params: Vec<&(&str, String)> = vec![&country_param]; - if let Some(query) = query { - params.extend(query); - } - - let response = self - .http_client - .get(format!("{}/{}", self.settings.hifi_url, uri)) - .bearer_auth(access_token) - .query(¶ms) - .send() - .await - .map_err(|e| { - error!("{:?}", e); - e - })? - .text() - .await - .map_err(|e| { - error!("{:?}", e); - e - })?; + self.ensure_fresh_token().await?; + let response = self.authed_get(uri, query).await?.text().await?; debug!(?response, "explorer response"); Ok(()) } @@ -458,7 +460,7 @@ impl Client { #[instrument(skip(self))] pub async fn search(&self, query: &str) -> Result<(), ClientError> { let query = vec![("query", query.to_string())]; - self.make_explorer_request(&format!("search/artists"), Some(&query)) + self.make_explorer_request("search/artists", Some(&query)) .await?; Ok(()) } @@ -468,37 +470,32 @@ impl Client { &self, playlist_uuid: &str, ) -> Result, ClientError> { - Ok(self - .make_paginated_request(&format!("playlists/{}/tracks", playlist_uuid), None) - .await?) + self.make_paginated_request(&format!("playlists/{}/tracks", playlist_uuid), None) + .await } #[instrument(skip(self))] pub async fn get_playlist(&self, playlist_uuid: &str) -> Result { - Ok(self - .make_request(&format!("playlists/{}", playlist_uuid), None) - .await?) + self.make_request(&format!("playlists/{}", playlist_uuid), None) + .await } #[instrument(skip(self))] pub async fn get_artist(&self, artist_uuid: &str) -> Result { - Ok(self - .make_request(&format!("artists/{}", artist_uuid), None) - .await?) + self.make_request(&format!("artists/{}", artist_uuid), None) + .await } #[instrument(skip(self))] pub async fn get_artist_albums(&self, artist_uuid: &str) -> Result, ClientError> { - Ok(self - .make_paginated_request(&format!("artists/{}/albums", artist_uuid), None) - .await?) + self.make_paginated_request(&format!("artists/{}/albums", artist_uuid), None) + .await } #[instrument(skip(self))] pub async fn get_users_playlists(&self, user_id: u64) -> Result, ClientError> { - Ok(self - .make_paginated_request(&format!("users/{}/playlists", user_id), None) - .await?) + self.make_paginated_request(&format!("users/{}/playlists", user_id), None) + .await } #[instrument(skip(self))] @@ -506,12 +503,11 @@ impl Client { &self, user_id: &str, ) -> Result, ClientError> { - Ok(self - .make_paginated_request( - &format!("users/{}/playlistsAndFavoritePlaylists", user_id), - None, - ) - .await?) + self.make_paginated_request( + &format!("users/{}/playlistsAndFavoritePlaylists", user_id), + None, + ) + .await } #[instrument(skip(self))] @@ -527,13 +523,12 @@ impl Client { #[instrument(skip(self))] pub async fn get_users_artists(&self, user_id: &str) -> Result, ClientError> { - Ok(self - .make_paginated_request( - &format!("users/{}/favorites/artists", user_id), - None, - // Some(&query), - ) - .await?) + self.make_paginated_request( + &format!("users/{}/favorites/artists", user_id), + None, + // Some(&query), + ) + .await } #[instrument(skip(self))] @@ -599,12 +594,18 @@ impl Client { let timestamp = chrono::Utc::now().timestamp() as u64; let login_results = login?; - self.settings.login.device_code = Some(code_response.device_code); - self.settings.login.access_token = Some(login_results.access_token); - self.settings.login.refresh_token = login_results.refresh_token; - self.settings.login.expires_after = Some(login_results.expires_in + timestamp); - self.settings.login.user_id = Some(login_results.user.user_id.to_string()); - self.settings.login.country_code = Some(login_results.user.country_code); + { + let mut login = match self.login.write() { + Ok(login) => login, + Err(poisoned) => poisoned.into_inner(), + }; + login.device_code = Some(code_response.device_code); + login.access_token = Some(login_results.access_token); + login.refresh_token = login_results.refresh_token; + login.expires_after = Some(login_results.expires_in + timestamp); + login.user_id = Some(login_results.user.user_id.to_string()); + login.country_code = Some(login_results.user.country_code); + } info!("device login succeeded"); return Ok(()); } @@ -614,10 +615,11 @@ impl Client { #[instrument(skip(self))] pub async fn login_config(&mut self) -> Result<(), ClientError> { - let Some(access_token) = self.settings.login.access_token.clone() else { + let login = self.login_snapshot(); + let Some(access_token) = login.access_token else { return Err(ClientError::AuthError("No access token found".to_string())); }; - //return if our session is still valid + // Return if our session is still valid. if self .http_client .get(format!("{}/sessions", self.settings.base_url)) @@ -625,27 +627,23 @@ impl Client { .send() .await .map_err(|e| { - error!("{:?}", e); + warn!("session check failed: {e}"); e })? .status() .is_success() { + debug!("existing session still valid"); return Ok(()); } - //otherwise refresh our token - let refresh = self.refresh_access_token().await?; - let now = chrono::Utc::now().timestamp() as u64; - - self.settings.login.expires_after = Some(refresh.expires_in + now); - self.settings.login.access_token = Some(refresh.access_token); - Ok(()) + // Otherwise refresh our token. + self.force_refresh_token().await } #[instrument(skip(self))] pub async fn refresh_access_token(&self) -> Result { - let Some(refresh_token) = self.settings.login.refresh_token.clone() else { + let Some(refresh_token) = self.login_snapshot().refresh_token else { return Err(ClientError::AuthError("No refresh token found".to_string())); }; let data = DeviceAuthRequest { @@ -693,7 +691,7 @@ impl Client { .http_client .post(format!( "{}/device_authorization", - &self.settings.oauth.base_url + self.settings.oauth.base_url )) .header("Content-Type", "application/x-www-form-urlencoded") .body(payload) @@ -779,6 +777,5 @@ mod tests { println!("{:?}", result); let result = client.get_album("244167550").await.unwrap(); println!("{:?}", result); - assert!(false); } } diff --git a/tidaldy/src/models.rs b/tidaldy/src/models.rs index a44c593..930c2e8 100644 --- a/tidaldy/src/models.rs +++ b/tidaldy/src/models.rs @@ -51,6 +51,8 @@ pub enum ClientError { SerdeUrlError(#[from] serde_urlencoded::ser::Error), #[error("authentication failed")] AuthError(String), + #[error("tidal api returned status {0}")] + ApiError(u16), #[error("base64 decoding failed")] Base64DecodeError(#[from] base64::DecodeError), #[error("utf8 decoding failed")] @@ -65,6 +67,7 @@ impl From for crabidy_core::ProviderError { ClientError::ConnectionError => Self::FetchError, ClientError::HttpClientError(_) => Self::FetchError, ClientError::SerdeUrlError(_) => Self::FetchError, + ClientError::ApiError(_) => Self::FetchError, _ => Self::Other, } }