diff --git a/CHANGELOG.md b/CHANGELOG.md index d69a42bd..b388ac9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -335,6 +335,20 @@ Foundational work: faster reviews, narrower diffs, and a safety net under the pa * Rolled out across the main **card-grid library surfaces** — **Albums**, **Random Albums**, **New Releases**, **Lossless Albums**, **Playlists**, **Composers**, **Composer detail**, **Genre detail**, **Label albums**, **Album detail** (similar / same-artist rails), **Artist detail** (album / appearance / compilation grids), **Internet Radio**, **Offline Library** (albums + playlists), and **Artists** grid mode (virtual rows driven by the same column metrics). * **Settings → Appearance → Library card grids:** persisted **maximum columns** (**4–12**, default **6**) with translated copy calling out **performance** trade-offs. Settings search index updated. +### Playback — stream buffering indicator on cover art + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#737](https://github.com/Psychotoxical/psysonic/pull/737)** + +* While an **HTTP stream** is still opening, the **player bar** and **queue** cover art is **greyscaled** with a static **clock** overlay; the seekbar and timer stay at **0** until the Rust engine arms playback (no optimistic drift). +* **`audio:progress`** carries an optional **`buffering`** flag; the UI only updates **`isPlaybackBuffering`** when the flag changes to avoid redundant store writes. + +### Hot cache — promote completed ranged streams larger than 64 MiB + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#737](https://github.com/Psychotoxical/psysonic/pull/737)** + +* Completed **ranged HTTP** downloads above the in-RAM promote cap (including long **M4A** / **ALAC** albums) are **spilled to disk** under app-data **`stream-spill/`**, then **renamed into hot cache** on promote instead of being skipped. +* Orphan spill files from prior sessions are **removed on startup** (best-effort). + ## Removed ### Settings — Animations 3-state setting under Seekbar Style @@ -346,6 +360,15 @@ Foundational work: faster reviews, narrower diffs, and a safety net under the pa ## Fixed +### Playback — M4A / MP4 streaming (moov-at-end) and seekbar during buffer + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#737](https://github.com/Psychotoxical/psysonic/pull/737)** + +* **M4A** and **MP4** tracks streamed from the server (including **AAC** and **ALAC** in `.m4a` / `.mp4` containers) with **`moov` at the end of the file** — typical for many Navidrome / iTunes-style encoders — **start audibly sooner**: playback no longer waits for the entire **`mdat`** blob to download before Symphonia can open the file. +* **Tail prefetch** on the ranged HTTP path pulls the end of the file first so metadata is available while the linear download still fills from byte 0. +* **Symphonia `isomp4` patch** (vendored): when a top-level **`mdat`** atom spans to EOF, the demuxer **scans the file tail for `moov`** instead of seeking past end-of-stream during probe — fixes failures such as **`format probe failed: end of stream`** and long stalls before the first sample on large moov-at-end files. +* While the stream is still opening, the **seekbar and elapsed time stay at 0** (and the cover shows the buffering state — see **Added** above) instead of advancing ahead of decoded audio — the same guard applies to **legacy** HTTP readers and **`RangedHttpSource`**. + ### Mixes — rating filter and Lucky Mix queue fill **By [@cucadmuh](https://github.com/cucadmuh), PR [#714](https://github.com/Psychotoxical/psysonic/pull/714)** diff --git a/src-tauri/crates/psysonic-audio/src/commands.rs b/src-tauri/crates/psysonic-audio/src/commands.rs index 6e9087ad..4947e045 100644 --- a/src-tauri/crates/psysonic-audio/src/commands.rs +++ b/src-tauri/crates/psysonic-audio/src/commands.rs @@ -15,8 +15,8 @@ use super::engine::{audio_http_client, AudioEngine}; use super::helpers::*; use super::ipc::{maybe_emit_normalization_state, NormalizationStatePayload}; use super::play_input::{ - build_source_from_play_input, select_play_input, swap_in_new_sink, url_format_hint, - PlayInputContext, SinkSwapInputs, + build_source_from_play_input, select_play_input, spawn_legacy_stream_start_when_armed, + swap_in_new_sink, url_format_hint, PlayInputContext, SinkSwapInputs, }; use super::preview::preview_clear_for_new_main_playback; use super::progress_task::spawn_progress_task; @@ -97,6 +97,8 @@ pub async fn audio_play( // Bump generation first so the old progress task stops before we peel // chained_info (avoids a race where it sees current_done + empty chain). let gen = state.generation.fetch_add(1, Ordering::SeqCst) + 1; + // Ranged/legacy HTTP paths reset this to false in `select_play_input`. + state.stream_playback_armed.store(true, Ordering::SeqCst); // Manual skip onto the gapless-pre-chained track: reuse raw bytes (no HTTP; // preload cache was already consumed when the chain was built). Otherwise @@ -325,7 +327,8 @@ pub async fn audio_play( // without an underrun on the very first period. // Standard mode: no pre-fill needed — default 44.1/48 kHz quantum is small. let needs_prefill = hi_res_enabled && output_rate > 48_000; - if needs_prefill { + let defer_playback_start = !state.stream_playback_armed.load(Ordering::Relaxed); + if needs_prefill || defer_playback_start { sink.pause(); } @@ -372,7 +375,9 @@ pub async fn audio_play( if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); // skipped during pre-fill — abort silently } - sink.play(); + if !defer_playback_start { + sink.play(); + } } swap_in_new_sink(&state, SinkSwapInputs { @@ -386,7 +391,24 @@ pub async fn audio_play( actual_fade_secs, }); - app.emit("audio:playing", duration_secs).ok(); + if defer_playback_start { + { + let mut cur = state.current.lock().unwrap(); + cur.play_started = None; + cur.paused_at = Some(0.0); + } + spawn_legacy_stream_start_when_armed( + gen, + state.generation.clone(), + state.stream_playback_armed.clone(), + state.samples_played.clone(), + state.current.clone(), + app.clone(), + duration_secs, + ); + } else { + app.emit("audio:playing", duration_secs).ok(); + } // ── Progress + ended detection ──────────────────────────────────────────── spawn_progress_task( @@ -403,6 +425,7 @@ pub async fn audio_play( state.current_channels.clone(), state.gapless_switch_at.clone(), state.current_playback_url.clone(), + state.stream_playback_armed.clone(), ); Ok(()) diff --git a/src-tauri/crates/psysonic-audio/src/decode.rs b/src-tauri/crates/psysonic-audio/src/decode.rs index 503a8ba8..4fc19983 100644 --- a/src-tauri/crates/psysonic-audio/src/decode.rs +++ b/src-tauri/crates/psysonic-audio/src/decode.rs @@ -647,6 +647,7 @@ pub(crate) fn build_streaming_source( fade_in_dur: Duration, sample_counter: Arc, target_rate: u32, + count_gate: Option>, ) -> Result { let sample_rate = decoder.sample_rate(); let channels = decoder.channels(); @@ -685,7 +686,10 @@ pub(crate) fn build_streaming_source( let fade_in = EqualPowerFadeIn::new(eq_src, fade_in_dur); let fade_out = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone()); let notifying = NotifyingSource::new(fade_out, done_flag); - let counting = CountingSource::new(notifying, sample_counter); + let counting = match count_gate { + Some(gate) => CountingSource::new_gated(notifying, sample_counter, gate), + None => CountingSource::new(notifying, sample_counter), + }; let boosted = PriorityBoostSource::new(counting); Ok(BuiltSource { @@ -980,6 +984,7 @@ mod build_source_tests { Duration::ZERO, sample_counter, 0, + None, ) .expect("build_streaming_source must succeed for a valid WAV decoder"); assert_eq!(built.output_channels, 1); diff --git a/src-tauri/crates/psysonic-audio/src/engine.rs b/src-tauri/crates/psysonic-audio/src/engine.rs index 02afef1b..b8c3fac7 100644 --- a/src-tauri/crates/psysonic-audio/src/engine.rs +++ b/src-tauri/crates/psysonic-audio/src/engine.rs @@ -6,7 +6,7 @@ use std::time::{Duration, Instant}; use rodio::Player; use tauri::{AppHandle, Manager}; -use super::state::{ChainedInfo, PreloadedTrack}; +use super::state::{ChainedInfo, PreloadedTrack, StreamCompletedSpill}; /// Reply channel handed back to the audio-stream thread once a re-open finishes. pub type StreamReopenReply = std::sync::mpsc::SyncSender>; @@ -36,11 +36,17 @@ pub struct AudioEngine { /// Last fully downloaded manual-stream track bytes (same playback identity), /// used to recover seek/replay without waiting for network again. pub(crate) stream_completed_cache: Arc>>, + /// On-disk spill for completed ranged streams above `TRACK_STREAM_PROMOTE_MAX_BYTES`. + pub(crate) stream_completed_spill: Arc>>, /// True when the currently playing source supports seeking (in-memory bytes /// or `RangedHttpSource`); false for the legacy non-seekable streaming /// fallback (`AudioStreamReader`). `audio_seek` rejects with a "not /// seekable" error when false so the frontend restart-fallback can engage. pub(crate) current_is_seekable: Arc, + /// HTTP stream paths (`RangedHttpSource`, legacy `AudioStreamReader`): false + /// until `TRACK_STREAM_PLAY_START_BYTES` are buffered (or download ends). + /// Bytes / local file / radio keep true. + pub(crate) stream_playback_armed: Arc, pub crossfade_enabled: Arc, pub crossfade_secs: Arc, pub fading_out_sink: Arc>>>, @@ -357,7 +363,9 @@ pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) { eq_pre_gain: Arc::new(AtomicU32::new(0f32.to_bits())), preloaded: Arc::new(Mutex::new(None)), stream_completed_cache: Arc::new(Mutex::new(None)), + stream_completed_spill: Arc::new(Mutex::new(None)), current_is_seekable: Arc::new(AtomicBool::new(true)), + stream_playback_armed: Arc::new(AtomicBool::new(true)), crossfade_enabled: Arc::new(AtomicBool::new(false)), crossfade_secs: Arc::new(AtomicU32::new(3.0f32.to_bits())), fading_out_sink: Arc::new(Mutex::new(None)), diff --git a/src-tauri/crates/psysonic-audio/src/helpers.rs b/src-tauri/crates/psysonic-audio/src/helpers.rs index 8cf88133..0b30fe19 100644 --- a/src-tauri/crates/psysonic-audio/src/helpers.rs +++ b/src-tauri/crates/psysonic-audio/src/helpers.rs @@ -242,6 +242,9 @@ pub(crate) fn sniff_stream_format_extension(data: &[u8]) -> Option { pub struct ProgressPayload { pub current_time: f64, pub duration: f64, + /// HTTP stream still filling its play buffer — UI must not extrapolate + /// progress until this clears. + pub buffering: bool, } // ─── Helpers ────────────────────────────────────────────────────────────────── @@ -528,6 +531,90 @@ pub fn take_stream_completed_for_url(state: &AudioEngine, url: &str) -> Option Option { + take_stream_completed_spill_from_slot(&state.stream_completed_spill, url) +} + +pub(crate) fn take_stream_completed_spill_from_slot( + slot: &std::sync::Arc>>, + url: &str, +) -> Option { + let mut guard = slot.lock().unwrap(); + if guard + .as_ref() + .is_some_and(|p| same_playback_target(&p.url, url)) + { + return guard.take().map(|p| p.path); + } + None +} + +/// Atomically write completed stream bytes under `dir` (`{track_id}.complete.part` → rename). +pub(crate) fn write_stream_spill_bytes_in_dir( + dir: &std::path::Path, + track_id: &str, + bytes: &[u8], +) -> Result { + std::fs::create_dir_all(dir).map_err(|e| e.to_string())?; + let path = dir.join(format!("{track_id}.complete")); + let part = dir.join(format!("{track_id}.complete.part")); + std::fs::write(&part, bytes).map_err(|e| e.to_string())?; + std::fs::rename(&part, &path).map_err(|e| e.to_string())?; + Ok(path) +} + +/// Atomically write completed stream bytes to app-data `stream-spill/` (sync; no await while holding `buf`). +pub(crate) fn write_stream_spill_file( + app: &AppHandle, + track_id: &str, + bytes: &[u8], +) -> Result { + let dir = app + .path() + .app_data_dir() + .map_err(|e| e.to_string())? + .join("stream-spill"); + write_stream_spill_bytes_in_dir(&dir, track_id, bytes) +} + +/// Remove leftover `stream-spill/*.complete*` from prior sessions (best-effort). +pub fn cleanup_orphan_stream_spill_dir(app: &AppHandle) { + let Ok(dir) = app.path().app_data_dir().map(|d| d.join("stream-spill")) else { + return; + }; + if !dir.is_dir() { + return; + } + let Ok(entries) = std::fs::read_dir(&dir) else { + return; + }; + for entry in entries.flatten() { + let name = entry.file_name(); + let lossy = name.to_string_lossy(); + if lossy.ends_with(".complete") || lossy.ends_with(".complete.part") { + let _ = std::fs::remove_file(entry.path()); + } + } +} + +pub(crate) fn install_stream_completed_spill( + slot: &std::sync::Arc>>, + url: String, + path: std::path::PathBuf, +) { + let mut guard = slot.lock().unwrap(); + if let Some(old) = guard.take() { + if old.path != path { + let _ = std::fs::remove_file(&old.path); + } + } + *guard = Some(crate::state::StreamCompletedSpill { url, path }); +} + /// Fetch track bytes from the preload cache or via HTTP. pub(crate) async fn fetch_data( url: &str, @@ -548,6 +635,27 @@ pub(crate) async fn fetch_data( return Ok(Some(data)); } + // Spill path is cloned (not taken) so replay of the same URL can still read from disk + // until hot-cache promote consumes the file via `take_stream_completed_spill_for_url`. + let spill_path = { + let guard = state.stream_completed_spill.lock().unwrap(); + guard + .as_ref() + .filter(|p| same_playback_target(&p.url, url)) + .map(|p| p.path.clone()) + }; + if let Some(path) = spill_path { + let data = tokio::fs::read(&path).await.map_err(|e| e.to_string())?; + if !data.is_empty() { + crate::app_deprintln!( + "[stream] fetch_data from spill path={} bytes={}", + path.display(), + data.len() + ); + return Ok(Some(data)); + } + } + // Check preload cache next. let cached = { let mut preloaded = state.preloaded.lock().unwrap(); @@ -648,6 +756,72 @@ pub(crate) fn spawn_analysis_seed_from_in_memory_bytes( }); } +/// Full-track analysis for a completed ranged stream spilled to disk (> RAM promote cap). +pub(crate) fn spawn_analysis_seed_from_spill_file( + app: &AppHandle, + track_id: &str, + spill_path: std::path::PathBuf, + gen: u64, + gen_arc: &Arc, +) { + let track_id = track_id.trim().to_string(); + if track_id.is_empty() { + return; + } + let app = app.clone(); + let gen_arc = gen_arc.clone(); + let max_bytes = crate::stream::LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES; + tokio::spawn(async move { + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + let bytes = match tokio::fs::read(&spill_path).await { + Ok(b) if b.is_empty() => return, + Ok(b) if b.len() > max_bytes => { + crate::app_deprintln!( + "[stream] spill analysis skip track_id={} bytes={} max={}", + track_id, + b.len(), + max_bytes + ); + return; + } + Ok(b) => b, + Err(e) => { + crate::app_eprintln!( + "[stream] spill analysis read failed track_id={}: {}", + track_id, + e + ); + return; + } + }; + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + crate::app_deprintln!( + "[stream] spill path: scheduling full-track analysis track_id={} size_mib={:.2}", + track_id, + bytes.len() as f64 / (1024.0 * 1024.0) + ); + let high = crate::engine::analysis_seed_high_priority_for_track(&app, &track_id); + if let Err(e) = psysonic_analysis::analysis_runtime::submit_analysis_cpu_seed( + app, + track_id.clone(), + bytes, + high, + ) + .await + { + crate::app_eprintln!( + "[analysis] spill path seed failed for {}: {}", + track_id, + e + ); + } + }); +} + /// -1 dB headroom applied at full scale to prevent inter-sample clipping. /// Modern masters are often at 0 dBFS; the EQ biquad chain and resampler /// can produce inter-sample peaks slightly above ±1.0 → audible distortion. @@ -1298,3 +1472,69 @@ mod tests { assert!(g.is_some()); } } + +#[cfg(test)] +mod stream_spill_tests { + use super::*; + use std::sync::{Arc, Mutex}; + + fn scratch_dir(label: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "psysonic-audio-spill-{label}-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("scratch dir"); + dir + } + + #[test] + fn write_stream_spill_bytes_in_dir_creates_complete_file() { + let dir = scratch_dir("write"); + let path = + write_stream_spill_bytes_in_dir(&dir, "track-1", b"hello").expect("write spill"); + assert!(path.exists()); + assert_eq!(std::fs::read(&path).unwrap(), b"hello"); + assert!(!dir.join("track-1.complete.part").exists()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn install_stream_completed_spill_replaces_prior_file() { + let dir = scratch_dir("install"); + let old_path = dir.join("old.complete"); + let new_path = dir.join("new.complete"); + std::fs::write(&old_path, b"old").unwrap(); + std::fs::write(&new_path, b"new").unwrap(); + let slot: Arc>> = + Arc::new(Mutex::new(None)); + install_stream_completed_spill( + &slot, + "http://example/a".into(), + old_path.clone(), + ); + install_stream_completed_spill( + &slot, + "http://example/b".into(), + new_path.clone(), + ); + assert!(!old_path.exists(), "previous spill file must be removed"); + assert!(new_path.exists()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn take_stream_completed_spill_for_url_consumes_slot() { + let dir = scratch_dir("take"); + let path = dir.join("t.complete"); + std::fs::write(&path, b"x").unwrap(); + let slot: Arc>> = + Arc::new(Mutex::new(None)); + let url = "https://server/stream?id=1"; + install_stream_completed_spill(&slot, url.into(), path.clone()); + let taken = take_stream_completed_spill_from_slot(&slot, url); + assert_eq!(taken.as_deref(), Some(path.as_path())); + assert!(take_stream_completed_spill_from_slot(&slot, url).is_none()); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/src-tauri/crates/psysonic-audio/src/lib.rs b/src-tauri/crates/psysonic-audio/src/lib.rs index 8bcb7137..2d090fc0 100644 --- a/src-tauri/crates/psysonic-audio/src/lib.rs +++ b/src-tauri/crates/psysonic-audio/src/lib.rs @@ -36,7 +36,10 @@ mod stream; pub use device_commands::{audio_default_output_device_name, audio_list_devices_for_engine}; pub use device_watcher::start_device_watcher; pub use engine::{create_engine, refresh_http_user_agent, AudioEngine}; -pub use helpers::take_stream_completed_for_url; +pub use helpers::{ + cleanup_orphan_stream_spill_dir, take_stream_completed_for_url, + take_stream_completed_spill_for_url, +}; /// Register platform-specific listeners so the output stream is reopened after sleep/resume /// when the device name may be unchanged (Windows WASAPI, Linux PipeWire, …). diff --git a/src-tauri/crates/psysonic-audio/src/play_input.rs b/src-tauri/crates/psysonic-audio/src/play_input.rs index ea9dfa87..febf8e1f 100644 --- a/src-tauri/crates/psysonic-audio/src/play_input.rs +++ b/src-tauri/crates/psysonic-audio/src/play_input.rs @@ -2,7 +2,7 @@ //! Subsonic hints, decide whether to play from in-memory bytes, a seekable //! local file, a seekable RangedHttpSource, or a non-seekable streaming reader. -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -22,7 +22,7 @@ use super::helpers::{ use super::stream::{ ranged_download_task, track_download_task, AudioStreamReader, LocalFileSource, RangedHttpSource, LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES, - RADIO_READ_TIMEOUT_SECS, TRACK_STREAM_MAX_BUF_CAPACITY, TRACK_STREAM_MIN_BUF_CAPACITY, + TRACK_READ_TIMEOUT_SECS, TRACK_STREAM_MAX_BUF_CAPACITY, TRACK_STREAM_MIN_BUF_CAPACITY, }; /// What `audio_play` will hand to `build_source` / `build_streaming_source`. @@ -284,6 +284,10 @@ async fn open_ranged_or_streaming_input( let buf = Arc::new(Mutex::new(vec![0u8; total_usize])); let downloaded_to = Arc::new(AtomicUsize::new(0)); let done = Arc::new(AtomicBool::new(false)); + state.stream_playback_armed.store(false, Ordering::SeqCst); + let playback_armed = state.stream_playback_armed.clone(); + let tail_ready = Arc::new(AtomicBool::new(false)); + let tail_filled_from = Arc::new(AtomicU64::new(0)); let loudness_hold_for_defer = (total_usize <= super::stream::TRACK_STREAM_PROMOTE_MAX_BYTES) .then_some(state.ranged_loudness_seed_hold.clone()); tokio::spawn(ranged_download_task( @@ -298,15 +302,22 @@ async fn open_ranged_or_streaming_input( downloaded_to.clone(), done.clone(), state.stream_completed_cache.clone(), + state.stream_completed_spill.clone(), state.normalization_engine.clone(), state.normalization_target_lufs.clone(), state.loudness_pre_analysis_attenuation_db.clone(), ctx.cache_id_for_tasks.map(|s| s.to_string()), loudness_hold_for_defer, + playback_armed, + stream_hint.clone(), + tail_ready.clone(), + tail_filled_from.clone(), )); let reader = RangedHttpSource { buf, downloaded_to, + tail_ready, + tail_filled_from, total_size: total, pos: 0, done, @@ -332,6 +343,8 @@ async fn open_ranged_or_streaming_input( let rb = HeapRb::::new(buffer_cap); let (prod, cons) = rb.split(); let done = Arc::new(AtomicBool::new(false)); + state.stream_playback_armed.store(false, Ordering::SeqCst); + let playback_armed = state.stream_playback_armed.clone(); tokio::spawn(track_download_task( ctx.gen, state.generation.clone(), @@ -346,14 +359,16 @@ async fn open_ranged_or_streaming_input( state.normalization_target_lufs.clone(), state.loudness_pre_analysis_attenuation_db.clone(), ctx.cache_id_for_tasks.map(|s| s.to_string()), + playback_armed, )); let (_new_cons_tx, new_cons_rx) = std::sync::mpsc::channel::>(); let reader = AudioStreamReader { + read_timeout_secs: TRACK_READ_TIMEOUT_SECS, cons: Mutex::new(cons), new_cons_rx: Mutex::new(new_cons_rx), deadline: std::time::Instant::now() - + Duration::from_secs(RADIO_READ_TIMEOUT_SECS), + + Duration::from_secs(TRACK_READ_TIMEOUT_SECS), gen_arc: state.generation.clone(), gen: ctx.gen, source_tag: "track-stream", @@ -366,6 +381,47 @@ async fn open_ranged_or_streaming_input( })) } +/// Legacy `AudioStreamReader`: keep the sink paused until the download task arms +/// playback, then reset counters and emit `audio:playing` so the UI does not +/// extrapolate ahead of audible output. +pub(super) fn spawn_legacy_stream_start_when_armed( + gen: u64, + gen_arc: Arc, + playback_armed: Arc, + samples_played: Arc, + current: Arc>, + app: AppHandle, + duration_secs: f64, +) { + tokio::spawn(async move { + loop { + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + if playback_armed.load(Ordering::Relaxed) { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + samples_played.store(0, Ordering::Relaxed); + let sink = current.lock().unwrap().sink.clone(); + if let Some(sink) = sink { + { + let mut cur = current.lock().unwrap(); + cur.play_started = Some(std::time::Instant::now()); + cur.paused_at = None; + cur.seek_offset = 0.0; + } + sink.play(); + app.emit("audio:playing", duration_secs).ok(); + crate::app_deprintln!("[stream] legacy track-stream: playback started after buffer ready"); + } + }); +} + /// Pulled out of the format_hint extraction block in `audio_play` — strip the /// query string first so Subsonic-style URLs (`stream.view?...&v=1.16.1&...`) /// don't latch onto random query-param substrings; only accept short @@ -517,6 +573,7 @@ pub(super) async fn build_source_from_play_input( fade_in_dur, state.samples_played.clone(), target_rate, + None, ) } PlayInput::Streaming { reader, format_hint: stream_hint } => { @@ -536,6 +593,7 @@ pub(super) async fn build_source_from_play_input( fade_in_dur, state.samples_played.clone(), target_rate, + Some(state.stream_playback_armed.clone()), ) } }?; diff --git a/src-tauri/crates/psysonic-audio/src/progress_task.rs b/src-tauri/crates/psysonic-audio/src/progress_task.rs index 6acadc07..f48399d9 100644 --- a/src-tauri/crates/psysonic-audio/src/progress_task.rs +++ b/src-tauri/crates/psysonic-audio/src/progress_task.rs @@ -67,6 +67,7 @@ pub(super) fn spawn_progress_task( channels_arc: Arc, gapless_switch_at: Arc, current_playback_url: Arc>>, + stream_playback_armed: Arc, ) { // Keep progress aligned with audible output (ALSA/PipeWire/Pulse queue) on // Linux; mirrors the quantum policy used for stream open/reopen plus a small @@ -201,7 +202,9 @@ pub(super) fn spawn_progress_task( }; let is_paused = paused_at.is_some(); - let pos_raw = if let Some(p) = paused_at { + let pos_raw = if !stream_playback_armed.load(Ordering::Relaxed) { + 0.0 + } else if let Some(p) = paused_at { p } else { (samples / divisor).min(dur.max(0.001)) @@ -218,7 +221,12 @@ pub(super) fn spawn_progress_task( || now.duration_since(last_progress_emit_at) >= Duration::from_millis(PROGRESS_EMIT_MIN_MS) || (pos - last_progress_emit_pos).abs() >= PROGRESS_EMIT_MIN_DELTA_SECS; if should_emit_progress { - emitter.emit_progress(ProgressPayload { current_time: pos, duration: dur }); + let buffering = !stream_playback_armed.load(Ordering::Relaxed); + emitter.emit_progress(ProgressPayload { + current_time: pos, + duration: dur, + buffering, + }); last_progress_emit_at = now; last_progress_emit_pos = pos; last_progress_emit_paused = is_paused; @@ -289,6 +297,13 @@ mod tests { fn track_switched_count(&self) -> usize { self.track_switched.lock().unwrap().len() } + fn last_progress_time(&self) -> Option { + self.progress + .lock() + .unwrap() + .last() + .map(|p| p.current_time) + } } impl ProgressEmitter for Arc { @@ -317,6 +332,7 @@ mod tests { channels: Arc, gapless_switch_at: Arc, playback_url: Arc>>, + stream_playback_armed: Arc, } impl TaskHarness { @@ -345,6 +361,7 @@ mod tests { channels: Arc::new(AtomicU32::new(2)), gapless_switch_at: Arc::new(AtomicU64::new(0)), playback_url: Arc::new(Mutex::new(None)), + stream_playback_armed: Arc::new(AtomicBool::new(true)), } } @@ -363,12 +380,60 @@ mod tests { self.channels.clone(), self.gapless_switch_at.clone(), self.playback_url.clone(), + self.stream_playback_armed.clone(), ); } } // ── tests ───────────────────────────────────────────────────────────────── + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn progress_emits_buffering_while_stream_not_armed() { + let h = TaskHarness::new(240.0); + h.stream_playback_armed.store(false, Ordering::SeqCst); + h.samples_played.store(441_000, Ordering::SeqCst); + + let emitter = Arc::new(MockEmitter::default()); + h.spawn_with(emitter.clone()); + + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + emitter.progress.lock().unwrap().iter().any(|p| p.buffering), + "progress payload must flag HTTP stream buffering before armed" + ); + + h.gen_counter.store(99, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn legacy_stream_holds_progress_at_zero_until_armed() { + let h = TaskHarness::new(240.0); + h.stream_playback_armed.store(false, Ordering::SeqCst); + h.samples_played.store(441_000, Ordering::SeqCst); + + let emitter = Arc::new(MockEmitter::default()); + h.spawn_with(emitter.clone()); + + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + emitter.last_progress_time().unwrap_or(0.0) < 0.01, + "progress must stay at 0 while legacy stream is buffering" + ); + assert!( + emitter.progress.lock().unwrap().iter().any(|p| p.buffering), + "progress payload must flag legacy stream buffering" + ); + + h.stream_playback_armed.store(true, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + emitter.last_progress_time().unwrap_or(0.0) > 4.0, + "progress should follow samples once armed (got {:?})", + emitter.last_progress_time() + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn task_breaks_immediately_when_generation_already_changed() { let h = TaskHarness::new(120.0); diff --git a/src-tauri/crates/psysonic-audio/src/radio_commands.rs b/src-tauri/crates/psysonic-audio/src/radio_commands.rs index b4c98efa..e50c1952 100644 --- a/src-tauri/crates/psysonic-audio/src/radio_commands.rs +++ b/src-tauri/crates/psysonic-audio/src/radio_commands.rs @@ -108,6 +108,7 @@ pub async fn audio_play_radio( // ── Build Symphonia decoder in a blocking thread ────────────────────────── let reader = AudioStreamReader { + read_timeout_secs: RADIO_READ_TIMEOUT_SECS, cons: Mutex::new(cons), new_cons_rx: Mutex::new(new_cons_rx), deadline: std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS), @@ -173,6 +174,7 @@ pub async fn audio_play_radio( app.emit("audio:playing", 0.0f64).ok(); + state.stream_playback_armed.store(true, Ordering::SeqCst); spawn_progress_task( gen, state.generation.clone(), @@ -187,6 +189,7 @@ pub async fn audio_play_radio( state.current_channels.clone(), state.gapless_switch_at.clone(), state.current_playback_url.clone(), + state.stream_playback_armed.clone(), ); Ok(()) diff --git a/src-tauri/crates/psysonic-audio/src/sources.rs b/src-tauri/crates/psysonic-audio/src/sources.rs index d1ff6452..fcf59867 100644 --- a/src-tauri/crates/psysonic-audio/src/sources.rs +++ b/src-tauri/crates/psysonic-audio/src/sources.rs @@ -363,11 +363,31 @@ impl> Source for NotifyingSource { pub(crate) struct CountingSource> { inner: S, counter: Arc, + /// When set, count samples only while the flag is true (legacy track stream). + count_gate: Option>, } impl> CountingSource { pub(crate) fn new(inner: S, counter: Arc) -> Self { - Self { inner, counter } + Self { + inner, + counter, + count_gate: None, + } + } + + pub(crate) fn new_gated(inner: S, counter: Arc, gate: Arc) -> Self { + Self { + inner, + counter, + count_gate: Some(gate), + } + } + + fn should_count(&self) -> bool { + self.count_gate + .as_ref() + .is_none_or(|g| g.load(Ordering::Relaxed)) } } @@ -375,7 +395,7 @@ impl> Iterator for CountingSource { type Item = f32; fn next(&mut self) -> Option { let sample = self.inner.next(); - if sample.is_some() { + if sample.is_some() && self.should_count() { self.counter.fetch_add(1, Ordering::Relaxed); } sample @@ -393,7 +413,7 @@ impl> Source for CountingSource { // new position while the decoder is still at the old one — causing // a permanent desync between displayed time and actual audio. let result = self.inner.try_seek(pos); - if result.is_ok() { + if result.is_ok() && self.should_count() { let samples = (pos.as_secs_f64() * self.inner.sample_rate().get() as f64 * self.inner.channels().get() as f64) as u64; self.counter.store(samples, Ordering::Relaxed); @@ -481,3 +501,56 @@ impl> Source for PriorityBoostSource { self.inner.try_seek(pos) } } + +#[cfg(test)] +mod counting_source_tests { + use super::*; + use rodio::Source; + use std::time::Duration; + + struct TwoSamples(u8); + impl Iterator for TwoSamples { + type Item = f32; + fn next(&mut self) -> Option { + match self.0 { + 0 => { + self.0 = 1; + Some(0.1) + } + 1 => { + self.0 = 2; + Some(0.2) + } + _ => None, + } + } + } + impl Source for TwoSamples { + fn current_span_len(&self) -> Option { + Some(1) + } + fn channels(&self) -> rodio::ChannelCount { + std::num::NonZero::new(1).unwrap() + } + fn sample_rate(&self) -> rodio::SampleRate { + std::num::NonZero::new(44_100).unwrap() + } + fn total_duration(&self) -> Option { + Some(Duration::from_secs_f32(2.0 / 44_100.0)) + } + } + + #[test] + fn gated_counter_skips_samples_until_gate_is_set() { + let counter = Arc::new(AtomicU64::new(0)); + let gate = Arc::new(AtomicBool::new(false)); + let mut src = CountingSource::new_gated(TwoSamples(0), counter.clone(), gate.clone()); + assert_eq!(src.next(), Some(0.1)); + assert_eq!(src.next(), Some(0.2)); + assert_eq!(counter.load(Ordering::Relaxed), 0); + gate.store(true, Ordering::SeqCst); + let mut src2 = CountingSource::new_gated(TwoSamples(0), counter.clone(), gate); + assert_eq!(src2.next(), Some(0.1)); + assert_eq!(counter.load(Ordering::Relaxed), 1); + } +} diff --git a/src-tauri/crates/psysonic-audio/src/state.rs b/src-tauri/crates/psysonic-audio/src/state.rs index 805cf108..4e0abff5 100644 --- a/src-tauri/crates/psysonic-audio/src/state.rs +++ b/src-tauri/crates/psysonic-audio/src/state.rs @@ -7,6 +7,12 @@ pub(crate) struct PreloadedTrack { pub(crate) data: Vec, } +/// Completed ranged stream too large for `stream_completed_cache`; bytes live on disk. +pub(crate) struct StreamCompletedSpill { + pub(crate) url: String, + pub(crate) path: std::path::PathBuf, +} + /// Info about the track that has been appended (chained) to the current Sink /// but whose source has not yet started playing (gapless mode only). pub(crate) struct ChainedInfo { diff --git a/src-tauri/crates/psysonic-audio/src/stream/mod.rs b/src-tauri/crates/psysonic-audio/src/stream/mod.rs index 401fb5c2..e2417b3e 100644 --- a/src-tauri/crates/psysonic-audio/src/stream/mod.rs +++ b/src-tauri/crates/psysonic-audio/src/stream/mod.rs @@ -11,6 +11,7 @@ mod icy; mod local_file; +mod mp4; mod radio; mod ranged_http; mod reader; @@ -32,7 +33,8 @@ pub(crate) const RADIO_BUF_CAPACITY: usize = 256 * 1024; pub(crate) const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024; /// Cap ring buffer growth when content-length is known. pub(crate) const TRACK_STREAM_MAX_BUF_CAPACITY: usize = 32 * 1024 * 1024; -/// Max bytes kept in memory to promote a completed streamed track for fast replay/seek recovery. +/// Max bytes kept in RAM (`stream_completed_cache`) for fast replay; larger completed +/// ranged streams are spilled under app-data `stream-spill/` for hot-cache promote. pub(crate) const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024; /// Hot/offline `psysonic-local://` files are read from disk for waveform/LUFS seeding — not the /// same heap pressure as retaining a full HTTP capture. FLAC/DSD tracks often exceed 64 MiB; @@ -42,7 +44,26 @@ pub(crate) const LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES: usize = 512 * 1024 * 1024; pub(crate) const TRACK_STREAM_MAX_RECONNECTS: u32 = 3; /// Seconds at stall threshold while paused before hard-disconnect. pub(crate) const RADIO_HARD_PAUSE_SECS: u64 = 5; -/// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF. +/// Live radio: if no audio bytes arrive for this long → EOF. pub(crate) const RADIO_READ_TIMEOUT_SECS: u64 = 15; +/// On-demand tracks (`track-stream`, `RangedHttpSource`): allow long gaps while a +/// large file is still downloading (format probe may read/seek ahead of the filler). +pub(crate) const TRACK_READ_TIMEOUT_SECS: u64 = 120; +/// HTTP track paths (`AudioStreamReader`, `RangedHttpSource`): minimum linear +/// download before audible playback and seekbar progress (demux probe may read +/// far ahead of the play cursor). +pub(crate) const TRACK_STREAM_PLAY_START_BYTES: u64 = 384 * 1024; + +/// Arm deferred playback / progress once enough of the file is buffered. +pub(crate) fn maybe_arm_stream_playback(downloaded: u64, playback_armed: &std::sync::atomic::AtomicBool) { + use std::sync::atomic::Ordering; + if !playback_armed.load(Ordering::Relaxed) && downloaded >= TRACK_STREAM_PLAY_START_BYTES { + playback_armed.store(true, Ordering::SeqCst); + crate::app_deprintln!( + "[stream] playback armed after {} KiB buffered", + downloaded / 1024 + ); + } +} /// Sleep interval when ring buffer is empty (prevents CPU spin). pub(crate) const RADIO_YIELD_MS: u64 = 2; diff --git a/src-tauri/crates/psysonic-audio/src/stream/mp4.rs b/src-tauri/crates/psysonic-audio/src/stream/mp4.rs new file mode 100644 index 00000000..a46c2e2e --- /dev/null +++ b/src-tauri/crates/psysonic-audio/src/stream/mp4.rs @@ -0,0 +1,139 @@ +//! MP4/M4A layout helpers for HTTP streaming path selection. + +/// True when the Subsonic / sniffed container hint is ISO-BMFF (m4a, mp4, …). +pub(crate) fn container_hint_is_mp4(hint: Option<&str>) -> bool { + let Some(h) = hint else { return false }; + matches!( + h.to_ascii_lowercase().as_str(), + "m4a" | "m4af" | "mp4" | "m4b" | "mov" | "mp4a" | "isom" + ) +} + +/// Walk top-level atoms in `prefix` and return true when `mdat` appears before `moov` +/// (classic non–fast-start layout — Symphonia must read the `moov` near EOF). +pub(crate) fn mp4_moov_follows_mdat(prefix: &[u8]) -> bool { + let mut pos = 0usize; + let mut saw_mdat = false; + while pos + 8 <= prefix.len() { + let atom_size = match read_mp4_atom_size(prefix, pos) { + Some(s) => s, + None => break, + }; + if atom_size < 8 { + break; + } + let atom_type = &prefix[pos + 4..pos + 8]; + if atom_type == b"mdat" { + saw_mdat = true; + } + if atom_type == b"moov" { + return saw_mdat; + } + let advance = atom_size.min((prefix.len() - pos) as u64) as usize; + if advance < 8 { + break; + } + pos += advance; + } + false +} + +/// True when we should prefetch the file tail before linear fill (moov-at-end). +pub(crate) fn mp4_needs_tail_prefetch(prefix: &[u8], hint: Option<&str>) -> bool { + if !container_hint_is_mp4(hint) { + return false; + } + if prefix.is_empty() { + return true; + } + if mp4_moov_follows_mdat(prefix) { + return true; + } + // mdat seen but no moov in the scanned prefix — moov is likely at EOF. + let mut pos = 0usize; + let mut saw_mdat = false; + let mut saw_moov = false; + while pos + 8 <= prefix.len() { + let atom_size = match read_mp4_atom_size(prefix, pos) { + Some(s) => s, + None => break, + }; + if atom_size < 8 { + break; + } + let atom_type = &prefix[pos + 4..pos + 8]; + if atom_type == b"mdat" { + saw_mdat = true; + } + if atom_type == b"moov" { + saw_moov = true; + break; + } + let advance = atom_size.min((prefix.len() - pos) as u64) as usize; + if advance < 8 { + break; + } + pos += advance; + } + saw_mdat && !saw_moov +} + +fn read_mp4_atom_size(data: &[u8], pos: usize) -> Option { + if pos + 8 > data.len() { + return None; + } + let size32 = u32::from_be_bytes(data[pos..pos + 4].try_into().ok()?) as u64; + if size32 == 1 { + if pos + 16 > data.len() { + return None; + } + Some(u64::from_be_bytes(data[pos + 8..pos + 16].try_into().ok()?)) + } else { + Some(size32) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn atom(typ: &[u8; 4], payload_len: usize) -> Vec { + let size = (8 + payload_len) as u32; + let mut v = Vec::with_capacity(8 + payload_len); + v.extend_from_slice(&size.to_be_bytes()); + v.extend_from_slice(typ); + v.resize(8 + payload_len, 0); + v + } + + #[test] + fn moov_after_mdat_detected() { + let mut buf = Vec::new(); + buf.extend(atom(b"ftyp", 4)); + buf.extend(atom(b"mdat", 100)); + buf.extend(atom(b"moov", 40)); + assert!(mp4_moov_follows_mdat(&buf)); + assert!(mp4_needs_tail_prefetch(&buf, Some("m4a"))); + } + + #[test] + fn moov_before_mdat_no_tail_prefetch() { + let mut buf = Vec::new(); + buf.extend(atom(b"ftyp", 4)); + buf.extend(atom(b"moov", 40)); + buf.extend(atom(b"mdat", 100)); + assert!(!mp4_moov_follows_mdat(&buf)); + assert!(!mp4_needs_tail_prefetch(&buf, Some("m4a"))); + } + + #[test] + fn empty_prefix_with_m4a_hint_needs_tail_prefetch() { + assert!(mp4_needs_tail_prefetch(&[], Some("m4a"))); + } + + #[test] + fn empty_prefix_without_mp4_hint_skips_tail_prefetch() { + assert!(!mp4_needs_tail_prefetch(&[], Some("mp3"))); + assert!(!mp4_needs_tail_prefetch(&[], None)); + } +} diff --git a/src-tauri/crates/psysonic-audio/src/stream/ranged_http.rs b/src-tauri/crates/psysonic-audio/src/stream/ranged_http.rs index 0b8c92f8..02aecc8a 100644 --- a/src-tauri/crates/psysonic-audio/src/stream/ranged_http.rs +++ b/src-tauri/crates/psysonic-audio/src/stream/ranged_http.rs @@ -23,9 +23,13 @@ use tauri::{AppHandle, Emitter}; use super::super::state::PreloadedTrack; use super::{ - RADIO_READ_TIMEOUT_SECS, RADIO_YIELD_MS, TRACK_STREAM_MAX_RECONNECTS, + RADIO_YIELD_MS, TRACK_READ_TIMEOUT_SECS, TRACK_STREAM_MAX_RECONNECTS, TRACK_STREAM_PROMOTE_MAX_BYTES, }; +use crate::helpers::{ + install_stream_completed_spill, spawn_analysis_seed_from_spill_file, write_stream_spill_file, +}; +use crate::state::StreamCompletedSpill; /// Clears `AudioEngine::ranged_loudness_seed_hold` only if it still matches this play. struct RangedLoudnessSeedHoldClear { @@ -49,6 +53,9 @@ pub(crate) struct RangedHttpSource { pub(crate) buf: Arc>>, /// Bytes contiguously downloaded from offset 0. pub(crate) downloaded_to: Arc, + /// When set, bytes `[tail_filled_from..total_size)` are valid (moov-at-end prefetch). + pub(crate) tail_ready: Arc, + pub(crate) tail_filled_from: Arc, pub(crate) total_size: u64, pub(crate) pos: u64, /// Set when the download task terminates (success or hard error). @@ -57,6 +64,22 @@ pub(crate) struct RangedHttpSource { pub(crate) gen: u64, } +impl RangedHttpSource { + fn region_ready(&self, start: u64, end: u64) -> bool { + let dl = self.downloaded_to.load(Ordering::Relaxed) as u64; + if end <= dl { + return true; + } + if self.tail_ready.load(Ordering::Relaxed) { + let tail_from = self.tail_filled_from.load(Ordering::Relaxed); + if start >= tail_from && end <= self.total_size { + return true; + } + } + false + } +} + impl Read for RangedHttpSource { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { if self.gen_arc.load(Ordering::SeqCst) != self.gen { @@ -75,7 +98,9 @@ impl Read for RangedHttpSource { } let target_end = self.pos + max_read as u64; - let deadline = Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); + let stall_timeout = Duration::from_secs(TRACK_READ_TIMEOUT_SECS); + let mut deadline = Instant::now() + stall_timeout; + let mut last_dl_seen = self.downloaded_to.load(Ordering::Relaxed) as u64; loop { if self.gen_arc.load(Ordering::SeqCst) != self.gen { crate::app_deprintln!( @@ -85,13 +110,20 @@ impl Read for RangedHttpSource { ); return Ok(0); } - let dl = self.downloaded_to.load(Ordering::SeqCst) as u64; - if dl >= target_end { + if self.region_ready(self.pos, target_end) { break; } + let dl = self.downloaded_to.load(Ordering::SeqCst) as u64; + if dl > last_dl_seen { + last_dl_seen = dl; + deadline = Instant::now() + stall_timeout; + } // Download finished but our cursor is past downloaded_to (e.g. seek // beyond a partial download that aborted). Return what we have. if self.done.load(Ordering::SeqCst) { + if self.region_ready(self.pos, target_end) { + break; + } if dl > self.pos { let avail = (dl - self.pos) as usize; let src = self.buf.lock().unwrap(); @@ -188,6 +220,7 @@ pub(crate) async fn ranged_http_download_loop( gen: u64, gen_arc: &Arc, mut on_partial: F, + playback_armed: Option<&AtomicBool>, ) -> (usize, RangedHttpLoopOutcome) where F: FnMut(usize, usize), @@ -274,6 +307,9 @@ where } downloaded += n; downloaded_to.store(downloaded, Ordering::SeqCst); + if let Some(armed) = playback_armed { + super::maybe_arm_stream_playback(downloaded as u64, armed); + } on_partial(downloaded, total_size); let mb = downloaded / (1024 * 1024); while mb >= next_progress_mb { @@ -302,10 +338,111 @@ where } } +/// Fetch `bytes=start-end` into `buf[start..=end]` (inclusive HTTP Range). +async fn ranged_write_http_range( + http_client: &reqwest::Client, + url: &str, + buf: &Arc>>, + start: u64, + end_inclusive: u64, + gen: u64, + gen_arc: &Arc, +) -> Result { + if gen_arc.load(Ordering::SeqCst) != gen { + return Err(()); + } + let response = http_client + .get(url) + .header(reqwest::header::RANGE, format!("bytes={start}-{end_inclusive}")) + .send() + .await + .map_err(|_| ())?; + if gen_arc.load(Ordering::SeqCst) != gen { + return Err(()); + } + if !(response.status() == reqwest::StatusCode::PARTIAL_CONTENT + || response.status() == reqwest::StatusCode::OK) + { + return Err(()); + } + let mut written = 0usize; + let start_usize = start as usize; + let mut byte_stream = response.bytes_stream(); + while let Some(chunk) = byte_stream.next().await { + if gen_arc.load(Ordering::SeqCst) != gen { + return Err(()); + } + let chunk = chunk.map_err(|_| ())?; + if chunk.is_empty() { + continue; + } + let mut b = buf.lock().unwrap(); + let end = (start_usize + written + chunk.len()).min(b.len()); + let n = end.saturating_sub(start_usize + written); + b[start_usize + written..start_usize + written + n] + .copy_from_slice(&chunk[..n]); + written += n; + if start_usize + written > end_inclusive as usize { + break; + } + } + Ok(written) +} + +/// Prefetch the tail of a moov-at-end MP4 so Symphonia can parse metadata while +/// the linear download still fills `mdat` from offset 0. +#[allow(clippy::too_many_arguments)] +async fn ranged_prefetch_mp4_tail( + http_client: reqwest::Client, + url: String, + buf: Arc>>, + total_size: usize, + tail_ready: Arc, + tail_filled_from: Arc, + playback_armed: Arc, + gen: u64, + gen_arc: Arc, +) { + const MIN_TAIL: u64 = 256 * 1024; + const MAX_TAIL: u64 = 8 * 1024 * 1024; + let total = total_size as u64; + if total < MIN_TAIL + 64 * 1024 { + return; + } + let tail_len = MAX_TAIL.min(total / 2).max(MIN_TAIL); + let tail_from = total.saturating_sub(tail_len); + let end_inclusive = total.saturating_sub(1); + match ranged_write_http_range( + &http_client, + &url, + &buf, + tail_from, + end_inclusive, + gen, + &gen_arc, + ) + .await + { + Ok(written) if written > 0 => { + tail_filled_from.store(tail_from, Ordering::Relaxed); + tail_ready.store(true, Ordering::SeqCst); + super::maybe_arm_stream_playback(tail_from + written as u64, &playback_armed); + crate::app_deprintln!( + "[stream] ranged: moov-at-end tail prefetch {} KiB (from byte {})", + written / 1024, + tail_from / 1024 + ); + } + _ => { + crate::app_deprintln!("[stream] ranged: moov-at-end tail prefetch failed"); + } + } +} + /// Linear downloader for `RangedHttpSource`: fills the pre-allocated buffer /// from offset 0 to total_size. Reconnects via HTTP Range from the current /// `downloaded` offset on transient errors. On completion (full track) the -/// data is promoted to `stream_completed_cache` for fast replay. +/// data is promoted to `stream_completed_cache` (≤ 64 MiB) or spilled to disk for hot cache. #[allow(clippy::too_many_arguments)] pub(crate) async fn ranged_download_task( gen: u64, @@ -319,6 +456,7 @@ pub(crate) async fn ranged_download_task( downloaded_to: Arc, done: Arc, promote_cache_slot: Arc>>, + spill_cache_slot: Arc>>, normalization_engine: Arc, normalization_target_lufs: Arc, loudness_pre_analysis_attenuation_db: Arc, @@ -326,6 +464,10 @@ pub(crate) async fn ranged_download_task( // When `Some`, ranged playback seeds on completion — defer HTTP backfill for that // track; `None` for large files where ranged skips seed (needs backfill). loudness_seed_hold: Option, + playback_armed: Arc, + format_hint: Option, + tail_ready: Arc, + tail_filled_from: Arc, ) { let _ranged_loudness_hold_clear = match (loudness_seed_hold.as_ref(), cache_track_id.as_ref()) { (Some(slot), Some(tid)) => { @@ -393,6 +535,33 @@ pub(crate) async fn ranged_download_task( ); }; + let tail_prefetch = super::mp4::mp4_needs_tail_prefetch(&[], format_hint.as_deref()); + let tail_handle = if tail_prefetch { + let client = http_client.clone(); + let url_tail = url.clone(); + let buf_tail = buf.clone(); + let tail_ready_bg = tail_ready.clone(); + let tail_from_bg = tail_filled_from.clone(); + let armed_bg = playback_armed.clone(); + let gen_bg = gen_arc.clone(); + Some(tokio::spawn(async move { + ranged_prefetch_mp4_tail( + client, + url_tail, + buf_tail, + total_size, + tail_ready_bg, + tail_from_bg, + armed_bg, + gen, + gen_bg, + ) + .await; + })) + } else { + None + }; + let (downloaded, outcome) = ranged_http_download_loop( http_client, &url, @@ -402,9 +571,15 @@ pub(crate) async fn ranged_download_task( gen, &gen_arc, on_partial, + Some(&playback_armed), ) .await; + if let Some(handle) = tail_handle { + let _ = handle.await; + } + + playback_armed.store(true, Ordering::SeqCst); done.store(true, Ordering::SeqCst); if matches!(outcome, RangedHttpLoopOutcome::Superseded) { @@ -428,34 +603,76 @@ pub(crate) async fn ranged_download_task( ); } - if downloaded == total_size && total_size > 0 && total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES { - if let Some(ref tid) = cache_track_id { - crate::app_deprintln!( - "[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (cpu-seed queue; this task awaits completion)", - tid, - total_size as f64 / (1024.0 * 1024.0), - total_size - ); - } - let t_clone = Instant::now(); - let data = buf.lock().unwrap().clone(); - if total_size > 32 * 1024 * 1024 { - crate::app_deprintln!( - "[stream] ranged: buffer cloned in_ms={}", - t_clone.elapsed().as_millis() - ); - } - if let Some(track_id) = cache_track_id { - let high = crate::engine::analysis_seed_high_priority_for_track(&app, &track_id); - if let Err(e) = psysonic_analysis::analysis_runtime::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await { - crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e); + if downloaded == total_size && total_size > 0 { + if total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES { + if let Some(ref tid) = cache_track_id { + crate::app_deprintln!( + "[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (cpu-seed queue; this task awaits completion)", + tid, + total_size as f64 / (1024.0 * 1024.0), + total_size + ); + } + let t_clone = Instant::now(); + let data = buf.lock().unwrap().clone(); + if total_size > 32 * 1024 * 1024 { + crate::app_deprintln!( + "[stream] ranged: buffer cloned in_ms={}", + t_clone.elapsed().as_millis() + ); + } + if let Some(track_id) = cache_track_id { + let high = crate::engine::analysis_seed_high_priority_for_track(&app, &track_id); + if let Err(e) = psysonic_analysis::analysis_runtime::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await { + crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e); + } + } + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data }); + crate::app_deprintln!("[stream] promoted to stream_completed_cache for replay"); + } else if let Some(track_id) = cache_track_id.clone() { + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + let spill_result = { + let spill_bytes = buf.lock().unwrap(); + if gen_arc.load(Ordering::SeqCst) != gen { + return; + } + write_stream_spill_file(&app, &track_id, &spill_bytes) + }; + match spill_result { + Ok(path) => { + crate::app_deprintln!( + "[stream] ranged: spilled to disk track_id={} size_mib={:.2} path={}", + track_id, + total_size as f64 / (1024.0 * 1024.0), + path.display() + ); + if gen_arc.load(Ordering::SeqCst) != gen { + let _ = std::fs::remove_file(&path); + return; + } + install_stream_completed_spill(&spill_cache_slot, url, path.clone()); + spawn_analysis_seed_from_spill_file( + &app, + &track_id, + path, + gen, + &gen_arc, + ); + } + Err(e) => { + crate::app_eprintln!( + "[stream] ranged: spill write failed track_id={}: {}", + track_id, + e + ); + } } } - if gen_arc.load(Ordering::SeqCst) != gen { - return; - } - *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data }); - crate::app_deprintln!("[stream] promoted to stream_completed_cache for replay"); } } @@ -474,6 +691,8 @@ mod tests { RangedHttpSource { buf, downloaded_to, + tail_ready: Arc::new(AtomicBool::new(true)), + tail_filled_from: Arc::new(AtomicU64::new(0)), total_size: total, pos: 0, done, @@ -541,6 +760,8 @@ mod tests { let mut src = RangedHttpSource { buf, downloaded_to, + tail_ready: Arc::new(AtomicBool::new(false)), + tail_filled_from: Arc::new(AtomicU64::new(0)), total_size: total, pos: 0, done, @@ -554,6 +775,35 @@ mod tests { assert_eq!(src.pos, 5); } + #[test] + fn read_blocks_until_download_progress_reaches_seek_target() { + let total: u64 = 8; + let buf = Arc::new(Mutex::new(vec![1, 2, 3, 4, 5, 6, 7, 8])); + let downloaded_to = Arc::new(AtomicUsize::new(2)); + let done = Arc::new(AtomicBool::new(false)); + let gen_arc = Arc::new(AtomicU64::new(1)); + let dl_bg = downloaded_to.clone(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(80)); + dl_bg.store(8, Ordering::SeqCst); + }); + let mut src = RangedHttpSource { + buf, + downloaded_to, + tail_ready: Arc::new(AtomicBool::new(false)), + tail_filled_from: Arc::new(AtomicU64::new(0)), + total_size: total, + pos: 6, + done, + gen_arc, + gen: 1, + }; + let mut out = [0u8; 2]; + let n = src.read(&mut out).unwrap(); + assert_eq!(n, 2); + assert_eq!(out, [7, 8]); + } + #[test] fn read_returns_zero_when_done_with_no_data_ahead_of_cursor() { let total: u64 = 8; @@ -564,6 +814,8 @@ mod tests { let mut src = RangedHttpSource { buf: src_buf, downloaded_to, + tail_ready: Arc::new(AtomicBool::new(false)), + tail_filled_from: Arc::new(AtomicU64::new(0)), total_size: total, pos: 5, // past downloaded_to done, @@ -676,6 +928,7 @@ mod tests { 1, &gen_arc, |_, _| {}, + None, ) .await; @@ -710,6 +963,7 @@ mod tests { 1, &gen_arc, |downloaded, total| calls.lock().unwrap().push((downloaded, total)), + None, ) .await; @@ -736,7 +990,7 @@ mod tests { let (buf, dl, gen_arc) = loop_state(1024); let (downloaded, outcome) = - ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}) + ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}, None) .await; assert_eq!(outcome, RangedHttpLoopOutcome::Aborted); @@ -767,7 +1021,7 @@ mod tests { gen_arc.store(99, Ordering::SeqCst); let (downloaded, outcome) = - ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}) + ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}, None) .await; assert_eq!(outcome, RangedHttpLoopOutcome::Superseded); @@ -826,7 +1080,7 @@ mod tests { let (buf, dl, gen_arc) = loop_state(body.len()); let (downloaded, outcome) = - ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}) + ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}, None) .await; // Stream finishes via a Range-resumed second request. @@ -868,7 +1122,7 @@ mod tests { let (buf, dl, gen_arc) = loop_state(body.len()); let (downloaded, outcome) = - ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}) + ranged_http_download_loop(client, &url, initial, &buf, &dl, 1, &gen_arc, |_, _| {}, None) .await; // Reconnect server returned 200 instead of 206 → Aborted, downloaded diff --git a/src-tauri/crates/psysonic-audio/src/stream/reader.rs b/src-tauri/crates/psysonic-audio/src/stream/reader.rs index 93c7e017..4cbe74ef 100644 --- a/src-tauri/crates/psysonic-audio/src/stream/reader.rs +++ b/src-tauri/crates/psysonic-audio/src/stream/reader.rs @@ -18,9 +18,10 @@ use ringbuf::HeapCons; use ringbuf::traits::{Consumer, Observer}; use symphonia::core::io::MediaSource; -use super::{RADIO_READ_TIMEOUT_SECS, RADIO_YIELD_MS}; +use super::{RADIO_YIELD_MS}; pub(crate) struct AudioStreamReader { + pub(crate) read_timeout_secs: u64, pub(crate) cons: Mutex>, /// Delivers fresh consumers on hard-pause reconnect (unbounded; drain to latest). /// Wrapped in Mutex so AudioStreamReader is Sync (required by symphonia::MediaSource). @@ -53,7 +54,7 @@ impl Read for AudioStreamReader { if let Some(c) = newest { *self.cons.lock().unwrap() = c; self.deadline = - std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); + std::time::Instant::now() + Duration::from_secs(self.read_timeout_secs); } loop { if self.gen_arc.load(Ordering::SeqCst) != self.gen { @@ -66,7 +67,7 @@ impl Read for AudioStreamReader { self.pos += read as u64; // Reset deadline: data arrived, so connection is alive. self.deadline = - std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); + std::time::Instant::now() + Duration::from_secs(self.read_timeout_secs); return Ok(read); } if self @@ -80,7 +81,7 @@ impl Read for AudioStreamReader { crate::app_eprintln!( "[{}] AudioStreamReader: {}s without data → EOF", self.source_tag, - RADIO_READ_TIMEOUT_SECS + self.read_timeout_secs ); return Err(std::io::Error::new( std::io::ErrorKind::TimedOut, diff --git a/src-tauri/crates/psysonic-audio/src/stream/track_stream.rs b/src-tauri/crates/psysonic-audio/src/stream/track_stream.rs index 7e3c9feb..bc220e09 100644 --- a/src-tauri/crates/psysonic-audio/src/stream/track_stream.rs +++ b/src-tauri/crates/psysonic-audio/src/stream/track_stream.rs @@ -16,7 +16,9 @@ use ringbuf::traits::Producer; use tauri::AppHandle; use super::super::state::PreloadedTrack; -use super::{TRACK_STREAM_MAX_RECONNECTS, TRACK_STREAM_PROMOTE_MAX_BYTES}; +use super::{ + maybe_arm_stream_playback, TRACK_STREAM_MAX_RECONNECTS, TRACK_STREAM_PROMOTE_MAX_BYTES, +}; #[allow(clippy::too_many_arguments)] pub(crate) async fn track_download_task( @@ -33,6 +35,7 @@ pub(crate) async fn track_download_task( normalization_target_lufs: Arc, loudness_pre_analysis_attenuation_db: Arc, cache_track_id: Option, + playback_armed: Arc, ) { let mut downloaded: u64 = 0; let mut reconnects: u32 = 0; @@ -147,6 +150,7 @@ pub(crate) async fn track_download_task( } offset += pushed; downloaded += pushed as u64; + maybe_arm_stream_playback(downloaded, &playback_armed); } } } @@ -177,6 +181,7 @@ pub(crate) async fn track_download_task( data: capture, }); } + playback_armed.store(true, Ordering::SeqCst); done.store(true, Ordering::SeqCst); return; } diff --git a/src-tauri/crates/psysonic-syncfs/src/cache/hot.rs b/src-tauri/crates/psysonic-syncfs/src/cache/hot.rs index b69297c0..8ff6625a 100644 --- a/src-tauri/crates/psysonic-syncfs/src/cache/hot.rs +++ b/src-tauri/crates/psysonic-syncfs/src/cache/hot.rs @@ -141,39 +141,63 @@ pub async fn promote_stream_cache_to_hot_cache( return Ok(Some(HotCacheDownloadResult { path: path_str, size })); } - let bytes = match audio::take_stream_completed_for_url(&state, &url) { - Some(b) => b, - None => { - crate::app_deprintln!( - "[hot-cache] promote skip track_id={} reason=no_completed_stream_for_url", - track_id - ); - return Ok(None); + if let Some(bytes) = audio::take_stream_completed_for_url(&state, &url) { + let part_path = file_path.with_extension(format!("{suffix}.part")); + if let Err(e) = tokio::fs::write(&part_path, &bytes).await { + let _ = tokio::fs::remove_file(&part_path).await; + return Err(e.to_string()); } - }; + tokio::fs::rename(&part_path, &file_path) + .await + .map_err(|e| e.to_string())?; - let part_path = file_path.with_extension(format!("{suffix}.part")); - if let Err(e) = tokio::fs::write(&part_path, &bytes).await { - let _ = tokio::fs::remove_file(&part_path).await; - return Err(e.to_string()); + let _ = enqueue_analysis_seed(&app, &track_id, &bytes).await; + + let size = tokio::fs::metadata(&file_path) + .await + .map(|m| m.len()) + .unwrap_or(0); + crate::app_deprintln!( + "[hot-cache] promote from_stream_ram track_id={} server_id={} bytes={}", + track_id, + server_id, + size + ); + return Ok(Some(HotCacheDownloadResult { path: path_str, size })); } - tokio::fs::rename(&part_path, &file_path) - .await - .map_err(|e| e.to_string())?; - let _ = enqueue_analysis_seed(&app, &track_id, &bytes).await; + if let Some(spill_path) = audio::take_stream_completed_spill_for_url(&state, &url) { + if let Err(e) = tokio::fs::rename(&spill_path, &file_path).await { + if let Err(copy_err) = tokio::fs::copy(&spill_path, &file_path).await { + let _ = tokio::fs::remove_file(&spill_path).await; + return Err(format!("promote spill rename: {e}; copy: {copy_err}")); + } + let _ = tokio::fs::remove_file(&spill_path).await; + } + let app_seed = app.clone(); + let tid = track_id.clone(); + let fp = file_path.clone(); + tokio::spawn(async move { + enqueue_analysis_seed_from_file(&app_seed, &tid, &fp).await; + }); + let size = tokio::fs::metadata(&file_path) + .await + .map(|m| m.len()) + .unwrap_or(0); + crate::app_deprintln!( + "[hot-cache] promote from_stream_spill track_id={} server_id={} bytes={}", + track_id, + server_id, + size + ); + return Ok(Some(HotCacheDownloadResult { path: path_str, size })); + } - let size = tokio::fs::metadata(&file_path) - .await - .map(|m| m.len()) - .unwrap_or(0); crate::app_deprintln!( - "[hot-cache] promote from_stream track_id={} server_id={} bytes={}", - track_id, - server_id, - size + "[hot-cache] promote skip track_id={} reason=no_completed_stream_for_url", + track_id ); - Ok(Some(HotCacheDownloadResult { path: path_str, size })) + Ok(None) } #[tauri::command] diff --git a/src-tauri/patches/symphonia-format-isomp4/src/atoms/mod.rs b/src-tauri/patches/symphonia-format-isomp4/src/atoms/mod.rs index b23bd87f..c6788822 100644 --- a/src-tauri/patches/symphonia-format-isomp4/src/atoms/mod.rs +++ b/src-tauri/patches/symphonia-format-isomp4/src/atoms/mod.rs @@ -414,6 +414,11 @@ impl AtomIterator { &mut self.reader } + /// Exclusive end offset of the current atom in the stream. + pub fn current_atom_end(&self) -> u64 { + self.next_atom_pos + } + pub fn next(&mut self) -> Result> { // Ignore any remaining data in the current atom that was not read. let cur_pos = self.reader.pos(); diff --git a/src-tauri/patches/symphonia-format-isomp4/src/demuxer.rs b/src-tauri/patches/symphonia-format-isomp4/src/demuxer.rs index dbe780b3..64a311d4 100644 --- a/src-tauri/patches/symphonia-format-isomp4/src/demuxer.rs +++ b/src-tauri/patches/symphonia-format-isomp4/src/demuxer.rs @@ -26,6 +26,29 @@ use crate::stream::*; use log::{debug, info, trace, warn}; +/// When `mdat` claims the rest of the file (common moov-at-end M4A), locate `moov` in the +/// tail instead of seeking to EOF (which yields end-of-stream during probe). +fn find_moov_atom_offset_in_tail( + mss: &mut MediaSourceStream, + total_len: u64, +) -> Result> { + const MAX_SCAN: u64 = 8 * 1024 * 1024; + if total_len < 16 { + return Ok(None); + } + let scan_len = MAX_SCAN.min(total_len) as usize; + let scan_start = total_len - scan_len as u64; + mss.seek(SeekFrom::Start(scan_start))?; + let mut buf = vec![0u8; scan_len]; + mss.read_buf_exact(&mut buf).map_err(Error::from)?; + for i in 0..buf.len().saturating_sub(8) { + if &buf[i + 4..i + 8] == b"moov" { + return Ok(Some(scan_start + i as u64)); + } + } + Ok(None) +} + pub struct TrackState { codec_params: CodecParameters, /// The track number. @@ -397,6 +420,29 @@ impl FormatReader for IsoMp4Reader { // The remainder of the stream will be read incrementally. break; } + let end = iter.current_atom_end(); + let file_len = total_len.unwrap_or(end); + if moov.is_none() { + let mut mss = iter.into_inner(); + let resume_at = if end < file_len.saturating_sub(8) { + // Bounded mdat — `moov` is the next top-level sibling. + end + } else if let Some(tl) = total_len { + match find_moov_atom_offset_in_tail(&mut mss, tl)? { + Some(off) => off, + None => return unsupported_error("isomp4: missing moov atom"), + } + } else { + end + }; + mss.seek(SeekFrom::Start(resume_at))?; + iter = AtomIterator::new_root(mss, total_len); + } else if end < file_len.saturating_sub(8) { + // Fast-start: skip a bounded mdat without linear read. + let mut mss = iter.into_inner(); + mss.seek(SeekFrom::Start(end))?; + iter = AtomIterator::new_root(mss, total_len); + } } AtomType::Meta => { // Read the metadata atom and append it to the log. diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 0f30217e..fc9ec00b 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -103,6 +103,8 @@ pub fn run() { app.manage(cache); } + audio::cleanup_orphan_stream_spill_dir(app.handle()); + // ── Playback-query port (analysis → audio back-edge) ────────── // Two closures, each capturing an AppHandle, so analysis_runtime // can ask AudioEngine playback questions without depending on the diff --git a/src/components/playback/PlaybackBufferingOverlay.test.tsx b/src/components/playback/PlaybackBufferingOverlay.test.tsx new file mode 100644 index 00000000..a7ab42b5 --- /dev/null +++ b/src/components/playback/PlaybackBufferingOverlay.test.tsx @@ -0,0 +1,13 @@ +import { describe, expect, it } from 'vitest'; +import { screen } from '@testing-library/react'; +import { renderWithProviders } from '@/test/helpers/renderWithProviders'; +import { PlaybackBufferingOverlay } from './PlaybackBufferingOverlay'; + +describe('PlaybackBufferingOverlay', () => { + it('exposes buffering status for assistive tech', () => { + renderWithProviders(); + expect( + screen.getByRole('status', { name: 'Loading track from server' }), + ).toBeInTheDocument(); + }); +}); diff --git a/src/components/playback/PlaybackBufferingOverlay.tsx b/src/components/playback/PlaybackBufferingOverlay.tsx new file mode 100644 index 00000000..82d090c2 --- /dev/null +++ b/src/components/playback/PlaybackBufferingOverlay.tsx @@ -0,0 +1,17 @@ +import { Clock } from 'lucide-react'; +import { useTranslation } from 'react-i18next'; + +/** Shown over cover art while an HTTP stream is still opening / buffering. */ +export function PlaybackBufferingOverlay() { + const { t } = useTranslation(); + return ( +
+ +
+ ); +} diff --git a/src/components/playerBar/PlayerTrackInfo.tsx b/src/components/playerBar/PlayerTrackInfo.tsx index 5b1efcca..7b94e4c0 100644 --- a/src/components/playerBar/PlayerTrackInfo.tsx +++ b/src/components/playerBar/PlayerTrackInfo.tsx @@ -10,6 +10,8 @@ import LastfmIcon from '../LastfmIcon'; import MarqueeText from '../MarqueeText'; import { OpenArtistRefInline } from '../OpenArtistRefInline'; import StarRating from '../StarRating'; +import { PlaybackBufferingOverlay } from '../playback/PlaybackBufferingOverlay'; +import { usePlayerStore } from '../../store/playerStore'; import { usePlayerBarLayoutStore, type PlayerBarLayoutItemId, @@ -52,6 +54,7 @@ export function PlayerTrackInfo({ userRatingOverrides, setUserRatingOverride, toggleFullscreen, navigate, openContextMenu, t, }: Props) { + const showBufferingOverlay = usePlayerStore(s => s.isPlaybackBuffering); const layoutItems = usePlayerBarLayoutStore(s => s.items); const isLayoutVisible = (id: PlayerBarLayoutItemId) => layoutItems.find(i => i.id === id)?.visible !== false; @@ -59,7 +62,7 @@ export function PlayerTrackInfo({ return (
!isRadio && !showPreviewMeta && currentTrack && toggleFullscreen()} data-tooltip={!isRadio && !showPreviewMeta && currentTrack ? t('player.openFullscreen') : undefined} > @@ -93,6 +96,9 @@ export function PlayerTrackInfo({
)} + {showBufferingOverlay && !isRadio && !showPreviewMeta && ( + + )}
{showPreviewMeta && ( diff --git a/src/components/queuePanel/QueueCurrentTrack.tsx b/src/components/queuePanel/QueueCurrentTrack.tsx index d6552a5b..3a8ced00 100644 --- a/src/components/queuePanel/QueueCurrentTrack.tsx +++ b/src/components/queuePanel/QueueCurrentTrack.tsx @@ -11,6 +11,8 @@ import { import { loudnessGainPlaceholderUntilCacheDb } from '../../utils/audio/loudnessPlaceholder'; import { effectiveLoudnessPreAnalysisAttenuationDb } from '../../utils/audio/loudnessPreAnalysisSlider'; import { QueueLufsTargetMenu } from './QueueLufsTargetMenu'; +import { PlaybackBufferingOverlay } from '../playback/PlaybackBufferingOverlay'; +import { usePlayerStore } from '../../store/playerStore'; interface Props { currentTrack: Track; @@ -45,6 +47,7 @@ export function QueueCurrentTrack({ reanalyzeLoudnessForTrack, setLoudnessTargetLufs, lufsTgtOpen, setLufsTgtOpen, lufsTgtBtnRef, lufsTgtMenuRef, lufsTgtPopStyle, t, }: Props) { + const showBufferingOverlay = usePlayerStore(s => s.isPlaybackBuffering); return (
{(() => { @@ -191,12 +194,13 @@ export function QueueCurrentTrack({ ); })()}
-
+
{currentTrack.coverArt ? ( ) : (
)} + {showBufferingOverlay && }

{currentTrack.title}

diff --git a/src/config/settingsCredits.ts b/src/config/settingsCredits.ts index 90411af0..54ee1bfc 100644 --- a/src/config/settingsCredits.ts +++ b/src/config/settingsCredits.ts @@ -113,6 +113,8 @@ const CONTRIBUTOR_ENTRIES = [ 'Now Playing: composite list keys on similar artists, album-card tracklist, and top songs — avoids duplicate React keys when Subsonic repeats ids (PR #703)', 'Search: share links in live/mobile search + queue preview modal (PR #716)', 'Multi-server: pin queue streams, cover art, links, context menu, and Now Playing to queue server (PR #717)', + 'M4A/MP4 streaming: moov-at-end tail prefetch and Symphonia isomp4 probe fix (PR #737)', + 'HTTP stream buffering — seekbar/timer at zero and cover overlay until playback arms (PR #737)', ], }, { diff --git a/src/hooks/useWaveformInterpolation.ts b/src/hooks/useWaveformInterpolation.ts index 492b9b84..ffe4556a 100644 --- a/src/hooks/useWaveformInterpolation.ts +++ b/src/hooks/useWaveformInterpolation.ts @@ -42,7 +42,7 @@ export function useWaveformInterpolation({ // On resume the first `tick` would add the entire pause duration to `elapsedSec` and // overshoot the playhead until the next transport heartbeat corrects it. const snap = getPlaybackProgressSnapshot(); - const raw = snap.progress; + const raw = snap.buffering || snap.currentTime < 0.005 ? 0 : snap.progress; progressRef.current = raw; progressAnchorRef.current = { progress: raw, @@ -74,6 +74,15 @@ export function useWaveformInterpolation({ rafId = requestAnimationFrame(tick); return; } + const snap = getPlaybackProgressSnapshot(); + if (snap.buffering || snap.currentTime < 0.005) { + progressRef.current = 0; + visualTargetProgressRef.current = 0; + visualProgressRef.current = 0; + progressAnchorRef.current = { progress: 0, atMs: now }; + rafId = requestAnimationFrame(tick); + return; + } const anchor = progressAnchorRef.current; const elapsedSec = Math.max(0, (now - anchor.atMs) / 1000); const predicted = Math.max(0, Math.min(1, anchor.progress + elapsedSec / duration)); diff --git a/src/locales/de/player.ts b/src/locales/de/player.ts index b50875f0..87b9369e 100644 --- a/src/locales/de/player.ts +++ b/src/locales/de/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Vorheriger Titel', play: 'Play', pause: 'Pause', + bufferingStream: 'Loading track from server', previewActive: 'Vorschau läuft', previewLabel: 'Vorschau', delayModalTitle: 'Timer', diff --git a/src/locales/en/player.ts b/src/locales/en/player.ts index d92b5562..175d74b0 100644 --- a/src/locales/en/player.ts +++ b/src/locales/en/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Previous Track', play: 'Play', pause: 'Pause', + bufferingStream: 'Loading track from server', previewActive: 'Preview playing', previewLabel: 'Preview', delayModalTitle: 'Timer', diff --git a/src/locales/es/player.ts b/src/locales/es/player.ts index 64e8638f..a489cbf2 100644 --- a/src/locales/es/player.ts +++ b/src/locales/es/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Pista Anterior', play: 'Reproducir', pause: 'Pausa', + bufferingStream: 'Loading track from server', previewActive: 'Vista previa activa', previewLabel: 'Vista previa', delayModalTitle: 'Temporizador', diff --git a/src/locales/fr/player.ts b/src/locales/fr/player.ts index 6e7e5b3d..aea1df3f 100644 --- a/src/locales/fr/player.ts +++ b/src/locales/fr/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Piste précédente', play: 'Lecture', pause: 'Pause', + bufferingStream: 'Loading track from server', previewActive: 'Aperçu en cours', previewLabel: 'Aperçu', delayModalTitle: 'Minuteur', diff --git a/src/locales/nb/player.ts b/src/locales/nb/player.ts index 1c317332..bb7f9f9b 100644 --- a/src/locales/nb/player.ts +++ b/src/locales/nb/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Forrige spor', play: 'Spill av', pause: 'Pause', + bufferingStream: 'Loading track from server', previewActive: 'Forhåndsvisning spilles av', previewLabel: 'Forhåndsvisning', delayModalTitle: 'Tidsur', diff --git a/src/locales/nl/player.ts b/src/locales/nl/player.ts index d9dec61d..e5cda5af 100644 --- a/src/locales/nl/player.ts +++ b/src/locales/nl/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Vorig nummer', play: 'Afspelen', pause: 'Pauzeren', + bufferingStream: 'Loading track from server', previewActive: 'Voorbeeld speelt af', previewLabel: 'Voorbeeld', delayModalTitle: 'Timer', diff --git a/src/locales/ro/player.ts b/src/locales/ro/player.ts index 49ad7f77..b8d94a61 100644 --- a/src/locales/ro/player.ts +++ b/src/locales/ro/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Piesa anterioară', play: 'Redă', pause: 'Pauză', + bufferingStream: 'Loading track from server', previewActive: 'Previzualizează redarea', previewLabel: 'Previzualizare', delayModalTitle: 'Temporizator', diff --git a/src/locales/ru/player.ts b/src/locales/ru/player.ts index 532bf8ee..eeaba83d 100644 --- a/src/locales/ru/player.ts +++ b/src/locales/ru/player.ts @@ -9,6 +9,7 @@ export const player = { prev: 'Предыдущий трек', play: 'Играть', pause: 'Пауза', + bufferingStream: 'Загрузка трека с сервера', previewActive: 'Превью воспроизводится', previewLabel: 'Превью', delayModalTitle: 'Таймер', diff --git a/src/locales/zh/player.ts b/src/locales/zh/player.ts index 6f7f84f4..4ef43759 100644 --- a/src/locales/zh/player.ts +++ b/src/locales/zh/player.ts @@ -9,6 +9,7 @@ export const player = { prev: '上一首', play: '播放', pause: '暂停', + bufferingStream: 'Loading track from server', previewActive: '正在试听', previewLabel: '试听', delayModalTitle: '定时', diff --git a/src/store/audioEventHandlers.ts b/src/store/audioEventHandlers.ts index dcbd52d7..f36e370f 100644 --- a/src/store/audioEventHandlers.ts +++ b/src/store/audioEventHandlers.ts @@ -77,10 +77,14 @@ export type NormalizationStatePayload = { export function handleAudioPlaying(_duration: number): void { setDeferHotCachePrefetch(false); resetProgressEmitThrottles(); - usePlayerStore.setState({ isPlaying: true }); + usePlayerStore.setState({ isPlaying: true, isPlaybackBuffering: false }); } -export function handleAudioProgress(current_time: number, duration: number): void { +export function handleAudioProgress( + current_time: number, + duration: number, + buffering = false, +): void { bumpPerfCounter('audioProgressEvents'); const perfFlags = getPerfProbeFlags(); const progressUiDisabled = perfFlags.disablePlayerProgressUi; @@ -105,6 +109,9 @@ export function handleAudioProgress(current_time: number, duration: number): voi const store = usePlayerStore.getState(); const track = store.currentTrack; if (!track) return; + if (!store.currentRadio && store.isPlaybackBuffering !== buffering) { + usePlayerStore.setState({ isPlaybackBuffering: buffering }); + } // Some backends can emit stale progress ticks shortly after pause/stop. // Ignoring them avoids reactivating UI redraw loops while transport is idle. const transportActive = store.isPlaying || store.currentRadio != null; @@ -114,7 +121,7 @@ export function handleAudioProgress(current_time: number, duration: number): voi setSeekFallbackVisualTarget(null); visualTarget = null; } - let displayTime = current_time; + let displayTime = buffering ? 0 : current_time; if (visualTarget && visualTarget.trackId === track.id) { const nearTarget = Math.abs(current_time - visualTarget.seconds) <= 2.0; if (nearTarget) { @@ -142,8 +149,9 @@ export function handleAudioProgress(current_time: number, duration: number): voi ) { emitPlaybackProgress({ currentTime: displayTime, - progress, + progress: buffering ? 0 : progress, buffered: 0, + buffering, }); markLiveProgressEmit(nowLive); } @@ -315,6 +323,7 @@ export function handleAudioEnded(): void { setIsAudioPaused(false); usePlayerStore.setState({ isPlaying: false, + isPlaybackBuffering: false, progress: 0, currentTime: 0, buffered: 0, @@ -386,6 +395,7 @@ export function handleAudioTrackSwitched(_duration: number): void { normalizationDbgTrackId: nextTrack.id, queueIndex: newIndex, isPlaying: true, + isPlaybackBuffering: switchPlaybackSource === 'stream', progress: 0, currentTime: 0, buffered: 0, @@ -427,7 +437,7 @@ export function handleAudioError(message: string): void { showToast(`Couldn't play track — skipping. ${detail}`, 8000, 'error'); const gen = getPlayGeneration(); - usePlayerStore.setState({ isPlaying: false }); + usePlayerStore.setState({ isPlaying: false, isPlaybackBuffering: false }); setTimeout(() => { if (getPlayGeneration() !== gen) return; usePlayerStore.getState().next(false); diff --git a/src/store/audioListenerSetup/audioEngineListeners.ts b/src/store/audioListenerSetup/audioEngineListeners.ts index 0e9cac82..17f8eaf0 100644 --- a/src/store/audioListenerSetup/audioEngineListeners.ts +++ b/src/store/audioListenerSetup/audioEngineListeners.ts @@ -38,7 +38,7 @@ export function setupAudioEngineListeners(): () => void { const pending = [ listen('audio:playing', ({ payload }) => handleAudioPlaying(payload)), - listen<{ current_time: number; duration: number }>('audio:progress', ({ payload }) => { + listen<{ current_time: number; duration: number; buffering?: boolean }>('audio:progress', ({ payload }) => { if (import.meta.env.DEV) { _devEventCount++; const now = Date.now(); @@ -51,7 +51,7 @@ export function setupAudioEngineListeners(): () => void { _devWindowStart = now; } } - handleAudioProgress(payload.current_time, payload.duration); + handleAudioProgress(payload.current_time, payload.duration, payload.buffering ?? false); }), listen('audio:ended', () => handleAudioEnded()), listen('audio:error', ({ payload }) => handleAudioError(payload)), diff --git a/src/store/playTrackAction.ts b/src/store/playTrackAction.ts index c61b9488..fecb74a5 100644 --- a/src/store/playTrackAction.ts +++ b/src/store/playTrackAction.ts @@ -254,7 +254,10 @@ export function runPlayTrack( currentTime: initialTime, scrobbled: false, lastfmLoved: false, - isPlaying: true, // optimistic — reverted on error + // HTTP stream: wait for Rust `audio:playing` so the seekbar does not + // extrapolate while RangedHttpSource / legacy reader is still buffering. + isPlaying: playbackSourceHint !== 'stream', + isPlaybackBuffering: playbackSourceHint === 'stream', currentPlaybackSource: playbackSourceHint, enginePreloadedTrackId: keepPreloadHint ? track.id : null, }); diff --git a/src/store/playbackProgress.test.ts b/src/store/playbackProgress.test.ts index 38c73081..6b0bd592 100644 --- a/src/store/playbackProgress.test.ts +++ b/src/store/playbackProgress.test.ts @@ -19,12 +19,14 @@ afterEach(() => { describe('getPlaybackProgressSnapshot', () => { it('starts at zero', () => { const snap = getPlaybackProgressSnapshot(); - expect(snap).toEqual({ currentTime: 0, progress: 0, buffered: 0 }); + expect(snap).toEqual({ currentTime: 0, progress: 0, buffered: 0, buffering: false }); }); it('returns the latest snapshot after an emit', () => { emitPlaybackProgress({ currentTime: 42, progress: 0.5, buffered: 0.7 }); - expect(getPlaybackProgressSnapshot()).toEqual({ currentTime: 42, progress: 0.5, buffered: 0.7 }); + expect(getPlaybackProgressSnapshot()).toEqual({ + currentTime: 42, progress: 0.5, buffered: 0.7, buffering: false, + }); }); }); @@ -34,8 +36,10 @@ describe('subscribePlaybackProgress', () => { subscribePlaybackProgress(cb); emitPlaybackProgress({ currentTime: 1, progress: 0.1, buffered: 0.2 }); expect(cb).toHaveBeenCalledTimes(1); - expect(cb.mock.calls[0][0]).toEqual({ currentTime: 1, progress: 0.1, buffered: 0.2 }); - expect(cb.mock.calls[0][1]).toEqual({ currentTime: 0, progress: 0, buffered: 0 }); + expect(cb.mock.calls[0][0]).toEqual({ + currentTime: 1, progress: 0.1, buffered: 0.2, buffering: false, + }); + expect(cb.mock.calls[0][1]).toEqual({ currentTime: 0, progress: 0, buffered: 0, buffering: false }); }); it('returns an unsubscribe that detaches the listener', () => { diff --git a/src/store/playbackProgress.ts b/src/store/playbackProgress.ts index 6d0f6e1c..d06d6faa 100644 --- a/src/store/playbackProgress.ts +++ b/src/store/playbackProgress.ts @@ -12,12 +12,15 @@ export type PlaybackProgressSnapshot = { currentTime: number; progress: number; buffered: number; + /** Legacy HTTP stream still filling — do not extrapolate the seekbar. */ + buffering?: boolean; }; let playbackProgressSnapshot: PlaybackProgressSnapshot = { currentTime: 0, progress: 0, buffered: 0, + buffering: false, }; const playbackProgressListeners = new Set<( @@ -26,16 +29,21 @@ const playbackProgressListeners = new Set<( ) => void>(); export function emitPlaybackProgress(next: PlaybackProgressSnapshot): void { + const normalized: PlaybackProgressSnapshot = { + ...next, + buffering: next.buffering ?? false, + }; const prev = playbackProgressSnapshot; if ( - Math.abs(prev.currentTime - next.currentTime) < 0.005 && - Math.abs(prev.progress - next.progress) < 0.0002 && - Math.abs(prev.buffered - next.buffered) < 0.0002 + Math.abs(prev.currentTime - normalized.currentTime) < 0.005 && + Math.abs(prev.progress - normalized.progress) < 0.0002 && + Math.abs(prev.buffered - normalized.buffered) < 0.0002 && + (prev.buffering ?? false) === normalized.buffering ) { return; } - playbackProgressSnapshot = next; - playbackProgressListeners.forEach(cb => cb(next, prev)); + playbackProgressSnapshot = normalized; + playbackProgressListeners.forEach(cb => cb(normalized, prev)); } export function getPlaybackProgressSnapshot(): PlaybackProgressSnapshot { @@ -53,6 +61,6 @@ export function subscribePlaybackProgress( /** Test-only: reset module state between specs so suites stay isolated. */ export function _resetPlaybackProgressForTest(): void { - playbackProgressSnapshot = { currentTime: 0, progress: 0, buffered: 0 }; + playbackProgressSnapshot = { currentTime: 0, progress: 0, buffered: 0, buffering: false }; playbackProgressListeners.clear(); } diff --git a/src/store/playerStore.progress.test.ts b/src/store/playerStore.progress.test.ts index 06cabe20..9e8f7cea 100644 --- a/src/store/playerStore.progress.test.ts +++ b/src/store/playerStore.progress.test.ts @@ -238,3 +238,53 @@ describe('audio:progress throttling (live-emit guard)', () => { unsub(); }); }); + +describe('audio:progress buffering flag', () => { + it('sets isPlaybackBuffering from the optional buffering field', () => { + const track = makeTrack({ duration: 100 }); + usePlayerStore.setState({ + currentTrack: track, + isPlaying: true, + isPlaybackBuffering: false, + }); + + emitTauriEvent('audio:progress', { + current_time: 0, + duration: 100, + buffering: true, + }); + expect(usePlayerStore.getState().isPlaybackBuffering).toBe(true); + + emitTauriEvent('audio:playing', 100); + expect(usePlayerStore.getState().isPlaybackBuffering).toBe(false); + }); + + it('does not rewrite isPlaybackBuffering when the flag is unchanged', () => { + const track = makeTrack({ duration: 100 }); + usePlayerStore.setState({ + currentTrack: track, + isPlaying: true, + isPlaybackBuffering: true, + }); + const setStateSpy = vi.spyOn(usePlayerStore, 'setState'); + + emitTauriEvent('audio:progress', { + current_time: 0, + duration: 100, + buffering: true, + }); + vi.advanceTimersByTime(2000); + emitTauriEvent('audio:progress', { + current_time: 0.9, + duration: 100, + buffering: true, + }); + + const bufferingWrites = setStateSpy.mock.calls.filter( + call => typeof call[0] === 'object' && call[0] !== null && 'isPlaybackBuffering' in call[0], + ); + expect(bufferingWrites).toHaveLength(0); + + setStateSpy.mockRestore(); + }); +}); diff --git a/src/store/playerStore.ts b/src/store/playerStore.ts index ece2963f..c355581f 100644 --- a/src/store/playerStore.ts +++ b/src/store/playerStore.ts @@ -39,6 +39,7 @@ export const usePlayerStore = create()( queueServerId: null, queueIndex: 0, isPlaying: false, + isPlaybackBuffering: false, progress: 0, buffered: 0, currentTime: 0, diff --git a/src/store/playerStoreTypes.ts b/src/store/playerStoreTypes.ts index b0f82f0a..8318dbcb 100644 --- a/src/store/playerStoreTypes.ts +++ b/src/store/playerStoreTypes.ts @@ -59,6 +59,8 @@ export interface PlayerState { queueServerId: string | null; queueIndex: number; isPlaying: boolean; + /** HTTP stream still buffering (network / demux probe) — show loading on cover art. */ + isPlaybackBuffering: boolean; progress: number; // 0–1 buffered: number; // 0–1 (unused in Rust backend, kept for UI compat) currentTime: number; diff --git a/src/styles/components/index.css b/src/styles/components/index.css index 4a0565db..ea5a24f5 100644 --- a/src/styles/components/index.css +++ b/src/styles/components/index.css @@ -19,6 +19,7 @@ @import './modal.css'; @import './playback-delay-sleep-delayed-start-modal.css'; @import './lucky-mix-pips-shared-by-the-inline-queue-lucky-cube-indicator.css'; +@import './playback-buffering-overlay.css'; @import './playlist-edit-modal.css'; @import './artists-page.css'; @import './years-page.css'; diff --git a/src/styles/components/playback-buffering-overlay.css b/src/styles/components/playback-buffering-overlay.css new file mode 100644 index 00000000..dc1fa285 --- /dev/null +++ b/src/styles/components/playback-buffering-overlay.css @@ -0,0 +1,32 @@ +/* Cover art while `isPlaybackBuffering` (player bar + queue current track). */ +.player-album-art-wrap.playback-buffering .player-album-art, +.player-album-art-wrap.playback-buffering .player-album-art-placeholder { + filter: grayscale(1); + transition: filter 0.2s ease; +} + +.queue-current-cover.playback-buffering img, +.queue-current-cover.playback-buffering .fallback { + filter: grayscale(1); + transition: filter 0.2s ease; +} + +.playback-buffering-overlay { + position: absolute; + inset: 0; + z-index: 3; + display: flex; + align-items: center; + justify-content: center; + border-radius: inherit; + pointer-events: none; + color: #fff; +} + +.playback-buffering-overlay__icon { + filter: drop-shadow(0 1px 3px rgba(0, 0, 0, 0.45)); +} + +.player-album-art-wrap .playback-buffering-overlay { + border-radius: var(--radius-md); +}