Something went wrong. Try again.
A decentralized music tracking and discovery platform built on AT Protocol 🎵 rocksky.app
spotify atproto lastfm musicbrainz scrobbling listenbrainz
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570//! Thread confinement for the playback engine.//!//! `rockbox_playback::Player` owns the cpal output stream and is `!Send`, so//! it lives on a dedicated thread. `Engine` is the `Send + Sync` handle the//! rest of the daemon holds: commands go over a channel, and the thread//! refreshes a shared status snapshot after every command and on a 250 ms//! tick.
use std::sync::mpsc::{channel, Sender};use std::sync::{Arc, Mutex};use std::time::{Duration, Instant};
use rockbox_playback::{ BassEnhancement, ChannelMode, Compressor, CrossfadeSettings, Crossfeed, Equalizer, InsertPosition, PlaybackState, Player, PlayerConfig, RepeatMode, ReplayGainMode, Status, Surround, ToneControls,};
pub enum EngineCmd { Execute(Box<EngineCmd>, Sender<Result<(), String>>), Stop, Clear, InsertAt { paths: Vec<String>, index: usize, }, PlayAt { index: usize, position: Duration, }, /// Replace the queue and start playback at `start_index` ("play now"). OpenAt { paths: Vec<String>, start_index: usize, shuffle: bool, }, Append(Vec<String>), InsertNext(Vec<String>), Play, Pause, Next, Previous, Seek(Duration), SkipTo(usize), Remove(usize), RemoveRange { start: usize, end: usize, }, /// Move a queue entry while preserving the playing track and position. Move { from: usize, to: usize, }, SetEqualizer(Equalizer), SetTone(ToneControls), SetBalance(i32), SetChannelMode(ChannelMode), SetCrossfade(CrossfadeSettings), SetReplaygain { mode: ReplayGainMode, preamp_db: f32, noclip: bool, }, SetStereoWidth(i32), SetCrossfeed(Crossfeed), SetCompressor(Compressor), SetSurround(Surround), SetBassEnhancement(BassEnhancement), /// 0.0..=1.0. SetVolume(f32), SetShuffle(bool), SetRepeat(RepeatMode), /// Reply with a snapshot taken NOW, on the engine thread — the live read /// path behind [`Engine::snapshot`]. GetSnapshot(Sender<Snapshot>), /// Restore the persisted queue + position from the resume file. Cues the /// track paused with the seek deferred until it opens — a bare `Seek` would /// be dropped, since there is no decoder open yet. Resume,}
/// A `Send + Clone` snapshot of the engine, refreshed by the engine thread.#[derive(Clone)]pub struct Snapshot { pub status: Status, /// The engine queue's URIs, verbatim and in the ENGINE's order — the /// single source of truth for queue order. Everything shown to /// controllers derives from this list (URIs mapped to display metadata), /// so no external mirror can drift from what actually plays. pub queue: Vec<String>, pub next_track: Option<String>, /// Output volume 0.0..=1.0. pub volume: f32, /// Changes on every explicit seek, including sub-second seeks. pub seek_revision: u64, /// When `status` was captured, so [`Engine::snapshot`] can extrapolate the /// position to read time. taken: Instant,}
impl Default for Snapshot { fn default() -> Self { Snapshot { status: Status { state: PlaybackState::Stopped, index: None, position: Duration::ZERO, duration: Duration::ZERO, metadata: None, queue_len: 0, shuffle: false, repeat: RepeatMode::Off, }, queue: Vec::new(), next_track: None, volume: 1.0, seek_revision: 0, taken: Instant::now(), } }}
pub struct Engine { tx: Mutex<Sender<EngineCmd>>, snapshot: Arc<Mutex<Snapshot>>,}
impl Engine { /// Spawn the engine thread. Fails when the output backend is unusable. pub fn start(config: PlayerConfig) -> Result<Self, String> { let (tx, rx) = channel::<EngineCmd>(); let (ready_tx, ready_rx) = channel::<Result<(), String>>(); let snapshot: Arc<Mutex<Snapshot>> = Arc::new(Mutex::new(Snapshot::default())); let shared = snapshot.clone();
std::thread::Builder::new() .name("rockbox-engine".into()) .spawn(move || { let player = match Player::with_config(config) { Ok(p) => { let _ = ready_tx.send(Ok(())); p } Err(e) => { let _ = ready_tx.send(Err(format!("audio engine: {e:?}"))); return; } }; let mut smoother = BoundarySmoother::default(); loop { match rx.recv_timeout(Duration::from_millis(250)) { Ok(EngineCmd::Execute(cmd, reply)) => { let jump = intentional_jump(&cmd); let seek = matches!(*cmd, EngineCmd::Seek(_)); apply(&player, *cmd); let result = if player.synchronize(Duration::from_secs(10)) { if jump { smoother.expect_jump = true; } Ok(()) } else { Err("playback engine did not acknowledge command".into()) }; if seek && result.is_ok() { smoother.seek_revision = smoother.seek_revision.wrapping_add(1); } *shared.lock().unwrap() = snapshot_of(&player, &mut smoother); let _ = reply.send(result); } Ok(EngineCmd::GetSnapshot(reply)) => { let _ = reply.send(snapshot_of(&player, &mut smoother)); } Ok(cmd) => { let seek = matches!(cmd, EngineCmd::Seek(_)); if intentional_jump(&cmd) { smoother.expect_jump = true; } apply(&player, cmd); if seek && player.synchronize(Duration::from_secs(10)) { smoother.seek_revision = smoother.seek_revision.wrapping_add(1); } } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, } *shared.lock().unwrap() = snapshot_of(&player, &mut smoother); } }) .map_err(|e| e.to_string())?;
ready_rx .recv() .map_err(|_| "engine thread died during startup".to_string())??;
Ok(Engine { tx: Mutex::new(tx), snapshot, }) }
pub fn send(&self, cmd: EngineCmd) { let _ = self.tx.lock().unwrap().send(cmd); }
pub async fn execute(self: &Arc<Self>, cmd: EngineCmd) -> Result<(), String> { let engine = self.clone(); tokio::task::spawn_blocking(move || { let (tx, rx) = channel(); engine.send(EngineCmd::Execute(Box::new(cmd), tx)); rx.recv_timeout(Duration::from_secs(12)) .map_err(|_| "playback engine unavailable".to_string())? }) .await .map_err(|e| e.to_string())? }
/// The engine state, read LIVE on the engine thread — no cache. The shared /// snapshot is only a fallback for when the thread cannot answer within /// 100ms, extrapolated to read time so even the fallback ticks evenly (see /// the desktop engine for the full story). pub fn snapshot(&self) -> Snapshot { let (reply_tx, reply_rx) = channel(); if self .tx .lock() .unwrap() .send(EngineCmd::GetSnapshot(reply_tx)) .is_ok() { if let Ok(snap) = reply_rx.recv_timeout(Duration::from_millis(100)) { return snap; } } let mut snap = self.snapshot.lock().unwrap().clone(); if snap.status.state == PlaybackState::Playing { let pos = snap.status.position + snap.taken.elapsed(); snap.status.position = if snap.status.duration > Duration::ZERO { pos.min(snap.status.duration) } else { pos }; } snap }}
/// A snapshot computed right now. Runs on the engine thread only.fn snapshot_of(player: &Player, smoother: &mut BoundarySmoother) -> Snapshot { Snapshot { status: smoother.filter(player.status()), queue: player .queue() .iter() .map(|p| p.to_string_lossy().into_owned()) .collect(), next_track: player .next_track() .map(|p| p.to_string_lossy().into_owned()), volume: player.volume(), seek_revision: smoother.seek_revision, taken: Instant::now(), }}
/// Smooths over the playback crate's decode-side track flip.////// `Player::status()` reports the DECODER's track. With a deep decode-ahead/// buffer the index, duration and position all flip to the next track/// ~buffer_seconds before the current one audibly ends, and the position/// collapses to zero — measured: a 15s track "flipped" at wall 5.0s with the/// 10s buffer (see the desktop crate's examples/boundary_probe.rs). Read raw,/// every track change made the clock jump back several seconds, seconds early.////// So until the audible track has run its duration out, keep reporting it:/// same index, same duration, position projected on the wall clock from the/// last pre-flip sample. Release at the old duration — the real boundary,/// where the raw position is ≈0 again and correct. Deliberate jumps (seek,/// skip, queue edits) are flagged by the command loop and pass through.#[derive(Default)]struct BoundarySmoother { seek_revision: u64, /// The next discontinuity is intentional (a transport/queue command). expect_jump: bool, last: Option<LastSample>, hold: Option<Hold>,}
struct LastSample { index: Option<usize>, position: Duration, at: Instant, duration: Duration,}
struct Hold { index: Option<usize>, duration: Duration, position: Duration, since: Instant,}
impl BoundarySmoother { fn filter(&mut self, mut status: Status) -> Status { let now = Instant::now(); let playing = status.state == PlaybackState::Playing;
if status.state == PlaybackState::Stopped { self.last = None; self.hold = None; self.expect_jump = false; return status; } if self.expect_jump { self.expect_jump = false; self.hold = None; self.remember(&status, now); return status; }
if let Some(hold) = self.hold.as_mut() { if playing { hold.position += now - hold.since; } hold.since = now; // The engine caught back up on the held track (a decode hiccup, // not a flip) — believe it again. let caught_up = status.index == hold.index && status.position + Duration::from_millis(1000) >= hold.position; // The audible track has run out; from here the raw status (≈0 into // the next track) is the truth. let ended = hold.position + Duration::from_millis(150) >= hold.duration; if caught_up || ended { self.hold = None; self.remember(&status, now); return status; } status.index = hold.index; status.position = hold.position.min(hold.duration); status.duration = hold.duration; self.remember(&status, now); return status; }
if let Some(last) = &self.last { let projected = if playing { last.position + (now - last.at) } else { last.position }; let dropped = status.position + Duration::from_millis(1000) < projected; if playing && dropped && last.duration > Duration::ZERO { let position = projected.min(last.duration); self.hold = Some(Hold { index: last.index, duration: last.duration, position, since: now, }); status.index = self.hold.as_ref().unwrap().index; status.position = position; status.duration = self.hold.as_ref().unwrap().duration; self.remember(&status, now); return status; } } self.remember(&status, now); status }
fn remember(&mut self, status: &Status, at: Instant) { self.last = Some(LastSample { index: status.index, position: status.position, at, duration: status.duration, }); }}
/// Commands after which a position/index discontinuity is the user's own doing/// and must pass through the smoother untouched.fn intentional_jump(cmd: &EngineCmd) -> bool { matches!( cmd, EngineCmd::Seek(_) | EngineCmd::Stop | EngineCmd::Clear | EngineCmd::PlayAt { .. } | EngineCmd::SkipTo(_) | EngineCmd::Next | EngineCmd::Previous | EngineCmd::OpenAt { .. } | EngineCmd::Remove(_) | EngineCmd::RemoveRange { .. } | EngineCmd::Move { .. } | EngineCmd::Resume )}
fn apply(player: &Player, cmd: EngineCmd) { match cmd { EngineCmd::Execute(_, _) => unreachable!("handled by engine loop"), EngineCmd::Stop => player.stop(), EngineCmd::Clear => player.clear_queue(), EngineCmd::RemoveRange { start, end } => { for index in (start..end).rev() { player.remove(index); } } EngineCmd::InsertAt { paths, index } => { player.insert_tracks(paths, InsertPosition::Index(index)) } EngineCmd::PlayAt { index, position } => player.play_at(index, position), EngineCmd::OpenAt { paths, start_index, shuffle, } => { tracing::debug!( count = paths.len(), start_index, first = paths.first().map(String::as_str), "engine: open" ); player.set_queue(paths.iter().map(String::as_str)); player.set_shuffle(shuffle); if start_index > 0 { player.skip_to(start_index); } player.play(); } EngineCmd::Append(paths) => player.insert_tracks_last(paths.iter().map(String::as_str)), EngineCmd::InsertNext(paths) => player.insert_tracks_next(paths.iter().map(String::as_str)), EngineCmd::Play => player.play(), EngineCmd::Pause => player.pause(), EngineCmd::Next => player.next(), EngineCmd::Previous => player.previous(), EngineCmd::Seek(pos) => player.seek(pos), EngineCmd::SkipTo(index) => player.skip_to(index), EngineCmd::Remove(index) => player.remove(index), EngineCmd::Move { from, to } => player.move_track(from, to), EngineCmd::SetEqualizer(equalizer) => player.set_equalizer(equalizer), EngineCmd::SetTone(tone) => player.set_tone(tone), EngineCmd::SetBalance(balance) => player.set_balance(balance), EngineCmd::SetChannelMode(mode) => player.set_channel_mode(mode), EngineCmd::SetCrossfade(settings) => player.set_crossfade(settings), EngineCmd::SetReplaygain { mode, preamp_db, noclip, } => player.set_replaygain(mode, preamp_db, noclip), EngineCmd::SetStereoWidth(percent) => player.set_stereo_width(percent), EngineCmd::SetCrossfeed(crossfeed) => player.set_crossfeed(crossfeed), EngineCmd::SetCompressor(compressor) => player.set_compressor(compressor), EngineCmd::SetSurround(surround) => player.set_surround(surround), EngineCmd::SetBassEnhancement(pbe) => player.set_bass_enhancement(pbe), EngineCmd::SetVolume(volume) => player.set_volume(volume), EngineCmd::SetShuffle(enabled) => player.set_shuffle(enabled), EngineCmd::SetRepeat(mode) => player.set_repeat(mode), // Answered in the loop before apply() is reached; kept here only so the // match stays exhaustive. EngineCmd::GetSnapshot(_) => {} EngineCmd::Resume => { match player.resume() { Some(state) => tracing::info!( tracks = state.tracks.len(), index = state.index, elapsed_ms = state.elapsed.as_millis() as u64, "resumed queue" ), None => tracing::debug!("nothing to resume"), }; } }}
#[cfg(test)]mod boundary_tests { use super::*; use rockbox_playback::{PlaybackState, RepeatMode};
fn status(state: PlaybackState, index: Option<usize>, pos_ms: u64, dur_ms: u64) -> Status { Status { state, index, position: Duration::from_millis(pos_ms), duration: Duration::from_millis(dur_ms), metadata: None, queue_len: 2, shuffle: false, repeat: RepeatMode::Off, } }
/// The measured pathology: mid-track, the crate flips to the next track's /// index with position ≈0. The smoother must keep reporting the audible /// track — same index, same duration, position still advancing. #[test] fn decode_side_flip_is_held() { let mut sm = BoundarySmoother::default(); sm.filter(status(PlaybackState::Playing, Some(0), 5_000, 15_000)); std::thread::sleep(Duration::from_millis(30)); let out = sm.filter(status(PlaybackState::Playing, Some(1), 0, 12_000)); assert_eq!(out.index, Some(0), "index must stay on the audible track"); assert_eq!(out.duration.as_millis(), 15_000, "duration must stay too"); assert!( out.position.as_millis() >= 5_000, "position went backwards: {}ms", out.position.as_millis() ); }
/// Once the held track's duration runs out, the raw status (≈0 into the /// next track) is the truth and must be released to. #[test] fn hold_releases_at_the_audible_boundary() { let mut sm = BoundarySmoother::default(); sm.filter(status(PlaybackState::Playing, Some(0), 14_950, 15_000)); std::thread::sleep(Duration::from_millis(30)); // Flip arrives; held for the last few ms... let held = sm.filter(status(PlaybackState::Playing, Some(1), 0, 12_000)); assert_eq!(held.index, Some(0)); std::thread::sleep(Duration::from_millis(120)); // ...and released once the old duration is spent. let out = sm.filter(status(PlaybackState::Playing, Some(1), 150, 12_000)); assert_eq!(out.index, Some(1), "must release at the boundary"); assert_eq!(out.duration.as_millis(), 12_000); }
/// A user seek backwards is intentional and must NOT be held. #[test] fn an_intentional_jump_passes_through() { let mut sm = BoundarySmoother::default(); sm.filter(status(PlaybackState::Playing, Some(0), 60_000, 180_000)); sm.expect_jump = true; // what the command loop sets on Seek let out = sm.filter(status(PlaybackState::Playing, Some(0), 5_000, 180_000)); assert_eq!(out.position.as_millis(), 5_000, "seek must be followed"); }
/// Pausing while held freezes the projected position. #[test] fn pause_freezes_a_hold() { let mut sm = BoundarySmoother::default(); sm.filter(status(PlaybackState::Playing, Some(0), 5_000, 15_000)); std::thread::sleep(Duration::from_millis(20)); sm.filter(status(PlaybackState::Playing, Some(1), 0, 12_000)); let a = sm .filter(status(PlaybackState::Paused, Some(1), 0, 12_000)) .position; std::thread::sleep(Duration::from_millis(40)); let b = sm .filter(status(PlaybackState::Paused, Some(1), 0, 12_000)) .position; assert_eq!(a.as_millis(), b.as_millis(), "held clock ran while paused"); }
/// Stopped clears everything — the next play starts from the raw truth. #[test] fn stopped_resets() { let mut sm = BoundarySmoother::default(); sm.filter(status(PlaybackState::Playing, Some(0), 5_000, 15_000)); sm.filter(status(PlaybackState::Playing, Some(1), 0, 12_000)); // held sm.filter(status(PlaybackState::Stopped, None, 0, 0)); let out = sm.filter(status(PlaybackState::Playing, Some(1), 100, 12_000)); assert_eq!(out.index, Some(1)); assert_eq!(out.position.as_millis(), 100); }}