itlxrd/Psychotoxical-psysonic
1
0
Fork 0
mirror of https://github.com/Psychotoxical/psysonic.git synced 2026-07-22 15:25:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(playback): stream buffering UI, M4A moov-at-end streaming, hot-cache spill (#737)

Browse Source 
* feat(playback): stream buffering UI, ranged M4A tail prefetch, demuxer fix

Defer seekbar/progress until HTTP stream is armed for both legacy and
RangedHttpSource; show buffering overlay on cover art. Add MP4 tail
prefetch and Symphonia isomp4 bounded-mdat/moov-at-EOF probing so
moov-at-end M4A can start without reading the full mdat.

* feat(hot-cache): spill large ranged streams to disk for promote

When a ranged HTTP download completes above the 64 MiB RAM promote cap,
write the existing buffer once to app-data stream-spill/ and register it
for hot-cache promote (rename) and replay via fetch_data. Analysis seeds
from the spill file up to the local-file cap (512 MiB).

* fix(ui): stream buffering — grayscale cover and static clock icon

Desaturate player and queue cover art while isPlaybackBuffering; keep a
non-animated clock overlay for visibility without the spinning animation.

* fix(playback): review follow-up — tests, i18n, spill cleanup, changelog

Clippy and test layout fixes; stream spill orphan cleanup on startup;
buffering flag guard in progress handler; bufferingStream in all player
locales; CHANGELOG and contributor credits for stream/M4A work.

* docs: attribute stream buffering and M4A streaming to PR #737

* test(audio): avoid create_engine in stream spill unit test

CI runners have no audio output device; test spill take/consume via
the Mutex slot only, matching install_stream_completed_spill tests.
This commit is contained in:
cucadmuh
2026-05-16 22:56:47 +03:00
committed by GitHub
parent 1ac354fb67
commit 6ea0acede5
45 changed files with 1280 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/crates/psysonic-audio/src/commands.rs
+28 -5
View File
@@ -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(())
src-tauri/crates/psysonic-audio/src/decode.rs
+6 -1
View File
@@ -647,6 +647,7 @@ pub(crate) fn build_streaming_source(
fade_in_dur: Duration,
sample_counter: Arc<AtomicU64>,
target_rate: u32,
count_gate: Option<Arc<AtomicBool>>,
) -> Result<BuiltSource, String> {
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);
src-tauri/crates/psysonic-audio/src/engine.rs
+9 -1
View File
@@ -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<Arc<rodio::MixerDeviceSink>>;
@@ -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<Mutex<Option<PreloadedTrack>>>,
/// On-disk spill for completed ranged streams above `TRACK_STREAM_PROMOTE_MAX_BYTES`.
pub(crate) stream_completed_spill: Arc<Mutex<Option<StreamCompletedSpill>>>,
/// 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<AtomicBool>,
/// 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<AtomicBool>,
pub crossfade_enabled: Arc<AtomicBool>,
pub crossfade_secs: Arc<AtomicU32>,
pub fading_out_sink: Arc<Mutex<Option<Arc<Player>>>>,
@@ -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)),
src-tauri/crates/psysonic-audio/src/helpers.rs
+240
View File
@@ -242,6 +242,9 @@ pub(crate) fn sniff_stream_format_extension(data: &[u8]) -> Option<String> {
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<V
None
}
/// Take (consume) on-disk spill for a completed large ranged stream.
pub fn take_stream_completed_spill_for_url(
state: &AudioEngine,
url: &str,
) -> Option<std::path::PathBuf> {
take_stream_completed_spill_from_slot(&state.stream_completed_spill, url)
}
pub(crate) fn take_stream_completed_spill_from_slot(
slot: &std::sync::Arc<std::sync::Mutex<Option<crate::state::StreamCompletedSpill>>>,
url: &str,
) -> Option<std::path::PathBuf> {
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::path::PathBuf, String> {
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<std::path::PathBuf, String> {
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<std::sync::Mutex<Option<crate::state::StreamCompletedSpill>>>,
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<AtomicU64>,
) {
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<Mutex<Option<crate::state::StreamCompletedSpill>>> =
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<Mutex<Option<crate::state::StreamCompletedSpill>>> =
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);
}
}
src-tauri/crates/psysonic-audio/src/lib.rs
+4 -1
View File
@@ -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, …).
src-tauri/crates/psysonic-audio/src/play_input.rs
+61 -3
View File
@@ -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::<u8>::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::<HeapCons<u8>>();
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<AtomicU64>,
playback_armed: Arc<AtomicBool>,
samples_played: Arc<AtomicU64>,
current: Arc<Mutex<super::engine::AudioCurrent>>,
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()),
)
}
}?;
src-tauri/crates/psysonic-audio/src/progress_task.rs
+67 -2
View File
@@ -67,6 +67,7 @@ pub(super) fn spawn_progress_task<E: ProgressEmitter>(
channels_arc: Arc<AtomicU32>,
gapless_switch_at: Arc<AtomicU64>,
current_playback_url: Arc<Mutex<Option<String>>>,
stream_playback_armed: Arc<AtomicBool>,
) {
// 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<E: ProgressEmitter>(
};
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<E: ProgressEmitter>(
|| 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<f64> {
self.progress
.lock()
.unwrap()
.last()
.map(|p| p.current_time)
}
}
impl ProgressEmitter for Arc<MockEmitter> {
@@ -317,6 +332,7 @@ mod tests {
channels: Arc<AtomicU32>,
gapless_switch_at: Arc<AtomicU64>,
playback_url: Arc<Mutex<Option<String>>>,
stream_playback_armed: Arc<AtomicBool>,
}
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);
src-tauri/crates/psysonic-audio/src/radio_commands.rs
+3
View File
@@ -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(())
src-tauri/crates/psysonic-audio/src/sources.rs
+76 -3
View File
@@ -363,11 +363,31 @@ impl<S: Source<Item = f32>> Source for NotifyingSource<S> {
pub(crate) struct CountingSource<S: Source<Item = f32>> {
inner: S,
counter: Arc<AtomicU64>,
/// When set, count samples only while the flag is true (legacy track stream).
count_gate: Option<Arc<AtomicBool>>,
}
impl<S: Source<Item = f32>> CountingSource<S> {
pub(crate) fn new(inner: S, counter: Arc<AtomicU64>) -> Self {
Self { inner, counter }
Self {
inner,
counter,
count_gate: None,
}
}
pub(crate) fn new_gated(inner: S, counter: Arc<AtomicU64>, gate: Arc<AtomicBool>) -> 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<S: Source<Item = f32>> Iterator for CountingSource<S> {
type Item = f32;
fn next(&mut self) -> Option<f32> {
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<S: Source<Item = f32>> Source for CountingSource<S> {
// 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<S: Source<Item = f32>> Source for PriorityBoostSource<S> {
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<f32> {
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<usize> {
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<Duration> {
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);
}
}
src-tauri/crates/psysonic-audio/src/state.rs
+6
View File
@@ -7,6 +7,12 @@ pub(crate) struct PreloadedTrack {
pub(crate) data: Vec<u8>,
}
/// 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 {
src-tauri/crates/psysonic-audio/src/stream/mod.rs
+23 -2
View File
@@ -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;
src-tauri/crates/psysonic-audio/src/stream/mp4.rs
+139
View File
@@ -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<u64> {
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<u8> {
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));
}
}
src-tauri/crates/psysonic-audio/src/stream/ranged_http.rs
+289 -35
View File
@@ -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<Mutex<Vec<u8>>>,
/// Bytes contiguously downloaded from offset 0.
pub(crate) downloaded_to: Arc<AtomicUsize>,
/// When set, bytes `[tail_filled_from..total_size)` are valid (moov-at-end prefetch).
pub(crate) tail_ready: Arc<AtomicBool>,
pub(crate) tail_filled_from: Arc<AtomicU64>,
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<usize> {
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<F>(
gen: u64,
gen_arc: &Arc<AtomicU64>,
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<Mutex<Vec<u8>>>,
start: u64,
end_inclusive: u64,
gen: u64,
gen_arc: &Arc<AtomicU64>,
) -> Result<usize, ()> {
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<Mutex<Vec<u8>>>,
total_size: usize,
tail_ready: Arc<AtomicBool>,
tail_filled_from: Arc<AtomicU64>,
playback_armed: Arc<AtomicBool>,
gen: u64,
gen_arc: Arc<AtomicU64>,
) {
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<AtomicUsize>,
done: Arc<AtomicBool>,
promote_cache_slot: Arc<Mutex<Option<PreloadedTrack>>>,
spill_cache_slot: Arc<Mutex<Option<StreamCompletedSpill>>>,
normalization_engine: Arc<AtomicU32>,
normalization_target_lufs: Arc<AtomicU32>,
loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
@@ -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<LoudnessSeedHold>,
playback_armed: Arc<AtomicBool>,
format_hint: Option<String>,
tail_ready: Arc<AtomicBool>,
tail_filled_from: Arc<AtomicU64>,
) {
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
src-tauri/crates/psysonic-audio/src/stream/reader.rs
+5 -4
View File
@@ -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<HeapCons<u8>>,
/// 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,
src-tauri/crates/psysonic-audio/src/stream/track_stream.rs
+6 -1
View File
@@ -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<AtomicU32>,
loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
cache_track_id: Option<String>,
playback_armed: Arc<AtomicBool>,
) {
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;
}