From 689d2dc019294ba51251914e93aa01faf6e00670 Mon Sep 17 00:00:00 2001 From: Maxim Isaev Date: Fri, 17 Apr 2026 13:49:31 +0300 Subject: [PATCH] fix(audio): harden stream playback transitions and cache promotion Add resilient stream-first track playback with reconnect/range recovery, seek fallback handling for non-seekable starts, and promotion of completed streamed bytes into hot cache. Also add crossfade/gapless backup preload windows to keep automatic transitions smooth on imperfect networks. --- src-tauri/src/audio.rs | 411 +++++++++++++++++++++++++++++++++++---- src-tauri/src/lib.rs | 51 +++++ src/store/playerStore.ts | 111 ++++++++++- 3 files changed, 533 insertions(+), 40 deletions(-) diff --git a/src-tauri/src/audio.rs b/src-tauri/src/audio.rs index d7ed1914..fedfc28b 100644 --- a/src-tauri/src/audio.rs +++ b/src-tauri/src/audio.rs @@ -443,6 +443,14 @@ impl> Source for CountingSource { /// Small enough that stale audio drains within a few seconds on reconnect; /// large enough to absorb brief network hiccups without stuttering. const RADIO_BUF_CAPACITY: usize = 256 * 1024; +/// Minimum ring buffer for on-demand track streaming starts. +const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024; +/// Cap ring buffer growth when content-length is known. +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. +const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024; +/// Consecutive body-stream failures tolerated for track streaming before abort. +const TRACK_STREAM_MAX_RECONNECTS: u32 = 3; /// Seconds at stall threshold while paused before hard-disconnect. const RADIO_HARD_PAUSE_SECS: u64 = 5; /// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF. @@ -580,6 +588,11 @@ struct AudioStreamReader { deadline: std::time::Instant, gen_arc: Arc, gen: u64, + /// Diagnostic tag for logs ("radio" or "track-stream"). + source_tag: &'static str, + /// Optional completion marker: when true and the ring buffer is empty, + /// return EOF immediately (used by one-shot track streaming). + eof_when_empty: Option>, /// Monotonic byte offset for SeekFrom::Current(0) "tell" (Symphonia probe). pos: u64, } @@ -615,14 +628,22 @@ impl Read for AudioStreamReader { std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); return Ok(read); } + if self + .eof_when_empty + .as_ref() + .is_some_and(|done| done.load(Ordering::SeqCst)) + { + return Ok(0); + } if std::time::Instant::now() >= self.deadline { eprintln!( - "[radio] AudioStreamReader: {}s without data → EOF", + "[{}] AudioStreamReader: {}s without data → EOF", + self.source_tag, RADIO_READ_TIMEOUT_SECS ); return Err(std::io::Error::new( std::io::ErrorKind::TimedOut, - "radio: no data received", + format!("{}: no data received", self.source_tag), )); } std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS)); @@ -636,7 +657,7 @@ impl Seek for AudioStreamReader { SeekFrom::Current(0) => Ok(self.pos), _ => Err(std::io::Error::new( std::io::ErrorKind::Unsupported, - "radio stream is not seekable", + format!("{} stream is not seekable", self.source_tag), )), } } @@ -853,6 +874,132 @@ async fn radio_download_task( eprintln!("[radio] download task done ({bytes_total} B total)"); } +/// One-shot HTTP downloader for track streaming starts. +/// +/// Pushes response chunks into an SPSC ring buffer consumed by `AudioStreamReader`. +/// Terminates when: +/// - generation changes (track superseded), +/// - response stream ends, or +/// - response emits an error. +async fn track_download_task( + gen: u64, + gen_arc: Arc, + http_client: reqwest::Client, + url: String, + initial_response: reqwest::Response, + mut prod: HeapProducer, + done: Arc, + promote_cache_slot: Arc>>, +) { + let mut downloaded: u64 = 0; + let mut reconnects: u32 = 0; + let mut next_response: Option = Some(initial_response); + let mut capture: Vec = Vec::new(); + let mut capture_over_limit = false; + 'outer: loop { + let response = if let Some(r) = next_response.take() { + r + } else { + let mut req = http_client.get(&url); + if downloaded > 0 { + req = req.header(reqwest::header::RANGE, format!("bytes={downloaded}-")); + } + match req.send().await { + Ok(r) => r, + Err(err) => { + if reconnects >= TRACK_STREAM_MAX_RECONNECTS { + eprintln!( + "[audio] streaming reconnect failed after {} attempts: {}", + reconnects, err + ); + done.store(true, Ordering::SeqCst); + return; + } + reconnects += 1; + tokio::time::sleep(Duration::from_millis(200)).await; + continue 'outer; + } + } + }; + if downloaded > 0 && response.status() != reqwest::StatusCode::PARTIAL_CONTENT { + eprintln!( + "[audio] streaming reconnect returned {}, expected 206 for range resume", + response.status() + ); + done.store(true, Ordering::SeqCst); + return; + } + if downloaded == 0 && !response.status().is_success() { + eprintln!("[audio] streaming HTTP {}", response.status()); + done.store(true, Ordering::SeqCst); + return; + } + + let mut byte_stream = response.bytes_stream(); + while let Some(chunk) = byte_stream.next().await { + if gen_arc.load(Ordering::SeqCst) != gen { + done.store(true, Ordering::SeqCst); + return; + } + let chunk = match chunk { + Ok(c) => c, + Err(e) => { + if reconnects >= TRACK_STREAM_MAX_RECONNECTS { + eprintln!( + "[audio] streaming download error after {} reconnects: {}", + reconnects, e + ); + done.store(true, Ordering::SeqCst); + return; + } + reconnects += 1; + eprintln!( + "[audio] streaming download error (attempt {}/{}): {} — reconnecting", + reconnects, + TRACK_STREAM_MAX_RECONNECTS, + e + ); + next_response = None; + continue 'outer; + } + }; + reconnects = 0; + let mut offset = 0; + while offset < chunk.len() { + if gen_arc.load(Ordering::SeqCst) != gen { + done.store(true, Ordering::SeqCst); + return; + } + let pushed = prod.push_slice(&chunk[offset..]); + if pushed == 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } else { + if !capture_over_limit { + if capture.len().saturating_add(pushed) <= TRACK_STREAM_PROMOTE_MAX_BYTES { + let from = offset; + let to = offset + pushed; + capture.extend_from_slice(&chunk[from..to]); + } else { + capture.clear(); + capture_over_limit = true; + } + } + offset += pushed; + downloaded += pushed as u64; + } + } + } + if !capture_over_limit && !capture.is_empty() { + *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { + url: url.clone(), + data: capture, + }); + } + done.store(true, Ordering::SeqCst); + return; + } +} + fn content_type_to_hint(ct: &str) -> Option { let ct = ct.to_ascii_lowercase(); if ct.contains("mpeg") || ct.contains("mp3") { Some("mp3".into()) } @@ -1429,11 +1576,70 @@ fn build_source( }) } +/// Streaming variant of `build_source`: uses a live `SizedDecoder` source +/// (non-seekable) and skips iTunSMPB parsing, but preserves the same EQ/fade/ +/// counting wrappers and output metadata. +fn build_streaming_source( + decoder: SizedDecoder, + duration_hint: f64, + eq_gains: Arc<[AtomicU32; 10]>, + eq_enabled: Arc, + eq_pre_gain: Arc, + done_flag: Arc, + fade_in_dur: Duration, + sample_counter: Arc, + target_rate: u32, +) -> Result { + let sample_rate = decoder.sample_rate(); + let channels = decoder.channels(); + + // For streaming starts prefer server-provided duration when available. + let effective_dur = if duration_hint > 1.0 { + duration_hint + } else { + decoder + .total_duration() + .map(|d| d.as_secs_f64()) + .unwrap_or(duration_hint) + }; + + let converted = decoder.convert_samples::(); + let dyn_src: DynSource = if target_rate > 0 && sample_rate != target_rate { + DynSource::new(UniformSourceIterator::new(converted, channels, target_rate)) + } else { + DynSource::new(converted) + }; + + let output_rate = if target_rate > 0 && sample_rate != target_rate { + target_rate + } else { + sample_rate + }; + + let fadeout_trigger = Arc::new(AtomicBool::new(false)); + let fadeout_samples = Arc::new(AtomicU64::new(0)); + + let eq_src = EqSource::new(dyn_src, eq_gains, eq_enabled, eq_pre_gain); + 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); + + Ok(BuiltSource { + source: counting, + duration_secs: effective_dur, + output_rate, + output_channels: channels, + fadeout_trigger, + fadeout_samples, + }) +} + // ─── Engine state ───────────────────────────────────────────────────────────── pub(crate) struct PreloadedTrack { - url: String, - data: Vec, + pub(crate) url: String, + pub(crate) data: Vec, } /// Info about the track that has been appended (chained) to the current Sink @@ -1474,6 +1680,9 @@ pub struct AudioEngine { pub eq_enabled: Arc, pub eq_pre_gain: Arc, pub(crate) preloaded: Arc>>, + /// 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>>, pub crossfade_enabled: Arc, pub crossfade_secs: Arc, pub fading_out_sink: Arc>>, @@ -1731,6 +1940,7 @@ pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) { eq_enabled: Arc::new(AtomicBool::new(false)), eq_pre_gain: Arc::new(AtomicU32::new(0f32.to_bits())), preloaded: Arc::new(Mutex::new(None)), + stream_completed_cache: Arc::new(Mutex::new(None)), 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)), @@ -1782,6 +1992,18 @@ fn same_playback_target(a_url: &str, b_url: &str) -> bool { } } +/// Take (consume) completed manual-stream bytes if they correspond to `url`. +pub(crate) fn take_stream_completed_for_url(state: &AudioEngine, url: &str) -> Option> { + let mut guard = state.stream_completed_cache.lock().unwrap(); + if guard + .as_ref() + .is_some_and(|p| same_playback_target(&p.url, url)) + { + return guard.take().map(|p| p.data); + } + None +} + /// Fetch track bytes from the preload cache or via HTTP. async fn fetch_data( url: &str, @@ -1789,7 +2011,20 @@ async fn fetch_data( gen: u64, app: &AppHandle, ) -> Result>, String> { - // Check preload cache first. + // Check completed streamed-track cache first (manual streaming fallback cache). + let streamed_cached = { + let mut streamed = state.stream_completed_cache.lock().unwrap(); + if streamed.as_ref().is_some_and(|p| same_playback_target(&p.url, url)) { + streamed.take().map(|p| p.data) + } else { + None + } + }; + if let Some(data) = streamed_cached { + return Ok(Some(data)); + } + + // Check preload cache next. let cached = { let mut preloaded = state.preloaded.lock().unwrap(); if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, url)) { @@ -1958,15 +2193,104 @@ pub async fn audio_play( old.stop(); } - // Fetch bytes (preload cache) unless we reused the chained download above. - let data = if let Some(d) = reuse_chained_bytes { - Some(d) + // Extract format hint from URL for better symphonia probing. + let format_hint = url.rsplit('.').next() + .and_then(|ext| ext.split('?').next()) + .map(|s| s.to_lowercase()); + + enum PlayInput { + Bytes(Vec), + Streaming { + reader: AudioStreamReader, + format_hint: Option, + }, + } + + // Data source selection: + // 1) Reused chained bytes (manual skip onto pre-chained track) + // 2) Manual uncached remote start -> stream immediately (no full prefetch) + // 3) Existing byte path (preloaded/offline/full HTTP fetch) + let play_input = if let Some(d) = reuse_chained_bytes { + PlayInput::Bytes(d) } else { - fetch_data(&url, &state, gen, &app).await? - }; - let data = match data { - Some(d) => d, - None => return Ok(()), // superseded while downloading + let stream_cache_hit = { + let streamed = state.stream_completed_cache.lock().unwrap(); + streamed + .as_ref() + .is_some_and(|p| same_playback_target(&p.url, &url)) + }; + let preloaded_hit = { + let preloaded = state.preloaded.lock().unwrap(); + preloaded + .as_ref() + .is_some_and(|p| same_playback_target(&p.url, &url)) + }; + let is_local = url.starts_with("psysonic-local://"); + + if manual && !stream_cache_hit && !preloaded_hit && !is_local { + let response = state.http_client.get(&url).send().await.map_err(|e| e.to_string())?; + if !response.status().is_success() { + if state.generation.load(Ordering::SeqCst) != gen { + return Ok(()); // superseded + } + let status = response.status().as_u16(); + let msg = format!("HTTP {status}"); + app.emit("audio:error", &msg).ok(); + return Err(msg); + } + + let stream_hint = content_type_to_hint( + response + .headers() + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""), + ).or_else(|| format_hint.clone()); + + let buffer_cap = response + .content_length() + .map(|n| n as usize) + .unwrap_or(TRACK_STREAM_MIN_BUF_CAPACITY) + .clamp(TRACK_STREAM_MIN_BUF_CAPACITY, TRACK_STREAM_MAX_BUF_CAPACITY); + let rb = HeapRb::::new(buffer_cap); + let (prod, cons) = rb.split(); + let done = Arc::new(AtomicBool::new(false)); + tokio::spawn(track_download_task( + gen, + state.generation.clone(), + state.http_client.clone(), + url.clone(), + response, + prod, + done.clone(), + state.stream_completed_cache.clone(), + )); + + // Track streaming has no reconnect producer; keep an empty channel. + let (_new_cons_tx, new_cons_rx) = std::sync::mpsc::channel::>(); + let reader = AudioStreamReader { + cons, + new_cons_rx: Mutex::new(new_cons_rx), + deadline: std::time::Instant::now() + + Duration::from_secs(RADIO_READ_TIMEOUT_SECS), + gen_arc: state.generation.clone(), + gen, + source_tag: "track-stream", + eof_when_empty: Some(done), + pos: 0, + }; + PlayInput::Streaming { + reader, + format_hint: stream_hint, + } + } else { + let data = fetch_data(&url, &state, gen, &app).await?; + let data = match data { + Some(d) => d, + None => return Ok(()), // superseded while downloading + }; + PlayInput::Bytes(data) + } }; if state.generation.load(Ordering::SeqCst) != gen { @@ -2009,23 +2333,40 @@ pub async fn audio_play( // Always 0 — no application-level resampling. Rodio handles conversion to // the output device rate internally; we let every track play at its native rate. let target_rate: u32 = 0; - // Extract format hint from URL for better symphonia probing. - let format_hint = url.rsplit('.').next() - .and_then(|ext| ext.split('?').next()) - .map(|s| s.to_lowercase()); - let built = build_source( - data, - duration_hint, - state.eq_gains.clone(), - state.eq_enabled.clone(), - state.eq_pre_gain.clone(), - done_flag.clone(), - fade_in_dur, - state.samples_played.clone(), - target_rate, - format_hint.as_deref(), - hi_res_enabled, - ).map_err(|e| { app.emit("audio:error", &e).ok(); e })?; + let built = match play_input { + PlayInput::Bytes(data) => build_source( + data, + duration_hint, + state.eq_gains.clone(), + state.eq_enabled.clone(), + state.eq_pre_gain.clone(), + done_flag.clone(), + fade_in_dur, + state.samples_played.clone(), + target_rate, + format_hint.as_deref(), + hi_res_enabled, + ), + PlayInput::Streaming { reader, format_hint } => { + let decoder = tokio::task::spawn_blocking(move || { + SizedDecoder::new_streaming(Box::new(reader), format_hint.as_deref()) + }) + .await + .map_err(|e| e.to_string())??; + + build_streaming_source( + decoder, + duration_hint, + state.eq_gains.clone(), + state.eq_enabled.clone(), + state.eq_pre_gain.clone(), + done_flag.clone(), + fade_in_dur, + state.samples_played.clone(), + target_rate, + ) + } + }.map_err(|e| { app.emit("audio:error", &e).ok(); e })?; let source = built.source; let duration_secs = built.duration_secs; let output_rate = built.output_rate; @@ -2392,8 +2733,9 @@ fn spawn_progress_task( let mut samples_played = samples_played; loop { - // 500 ms tick — frontend interpolates visually at 60 fps via rAF. - tokio::time::sleep(Duration::from_millis(500)).await; + // 100 ms tick keeps near-end detection timely for crossfade/gapless + // handoff while frontend still interpolates smoothly via rAF. + tokio::time::sleep(Duration::from_millis(100)).await; if gen_counter.load(Ordering::SeqCst) != gen { break; @@ -2596,6 +2938,7 @@ pub async fn audio_resume(state: State<'_, AudioEngine>, app: AppHandle) -> Resu pub fn audio_stop(state: State<'_, AudioEngine>) { state.generation.fetch_add(1, Ordering::SeqCst); *state.chained_info.lock().unwrap() = None; + *state.stream_completed_cache.lock().unwrap() = None; // Drop RadioLiveState → triggers Drop → task.abort() → TCP released. drop(state.radio_state.lock().unwrap().take()); let mut cur = state.current.lock().unwrap(); @@ -2849,6 +3192,8 @@ pub async fn audio_play_radio( deadline: std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS), gen_arc: state.generation.clone(), gen, + source_tag: "radio", + eof_when_empty: None, pos: 0, }; diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 6bb73ca5..ff5d900e 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1302,6 +1302,56 @@ async fn download_track_hot_cache( }) } +/// Promotes bytes captured by the manual streaming path into hot cache on disk. +/// Returns `Ok(None)` when no completed stream cache is available for this URL. +#[tauri::command] +async fn promote_stream_cache_to_hot_cache( + track_id: String, + server_id: String, + url: String, + suffix: String, + custom_dir: Option, + app: tauri::AppHandle, + state: tauri::State<'_, audio::AudioEngine>, +) -> Result, String> { + let root = resolve_hot_cache_root(custom_dir, &app)?; + let cache_dir = root.join(&server_id); + tokio::fs::create_dir_all(&cache_dir) + .await + .map_err(|e| e.to_string())?; + + let file_path = cache_dir.join(format!("{}.{}", track_id, suffix)); + let path_str = file_path.to_string_lossy().to_string(); + + if file_path.exists() { + let size = tokio::fs::metadata(&file_path) + .await + .map(|m| m.len()) + .unwrap_or(0); + return Ok(Some(HotCacheDownloadResult { path: path_str, size })); + } + + let bytes = match audio::take_stream_completed_for_url(&state, &url) { + Some(b) => b, + None => return Ok(None), + }; + + 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 size = tokio::fs::metadata(&file_path) + .await + .map(|m| m.len()) + .unwrap_or(0); + Ok(Some(HotCacheDownloadResult { path: path_str, size })) +} + #[tauri::command] async fn get_hot_cache_size(custom_dir: Option, app: tauri::AppHandle) -> u64 { fn dir_size(root: std::path::PathBuf) -> u64 { @@ -2570,6 +2620,7 @@ pub fn run() { delete_offline_track, get_offline_cache_size, download_track_hot_cache, + promote_stream_cache_to_hot_cache, get_hot_cache_size, delete_hot_cache_track, purge_hot_cache, diff --git a/src/store/playerStore.ts b/src/store/playerStore.ts index bd4e139f..08bff3c0 100644 --- a/src/store/playerStore.ts +++ b/src/store/playerStore.ts @@ -3,7 +3,7 @@ import { persist, createJSONStorage } from 'zustand/middleware'; import { invoke } from '@tauri-apps/api/core'; import { listen } from '@tauri-apps/api/event'; import { showToast } from '../utils/toast'; -import { buildCoverArtUrl, getPlayQueue, savePlayQueue, reportNowPlaying, scrobbleSong, SubsonicSong, getSong, getRandomSongs, getSimilarSongs2, getTopSongs, InternetRadioStation, setRating } from '../api/subsonic'; +import { buildCoverArtUrl, buildStreamUrl, getPlayQueue, savePlayQueue, reportNowPlaying, scrobbleSong, SubsonicSong, getSong, getRandomSongs, getSimilarSongs2, getTopSongs, InternetRadioStation, setRating } from '../api/subsonic'; import { resolvePlaybackUrl } from '../utils/resolvePlaybackUrl'; import { setDeferHotCachePrefetch } from '../utils/hotCacheGate'; import { lastfmScrobble, lastfmUpdateNowPlaying, lastfmLoveTrack, lastfmUnloveTrack, lastfmGetTrackLoved, lastfmGetAllLovedTracks } from '../api/lastfm'; @@ -213,6 +213,10 @@ let seekDebounce: ReturnType | null = null; // Target time of the last seek — blocks stale Rust progress ticks until the // engine has actually caught up to the new position. let seekTarget: number | null = null; +// Streaming fallback seek guard: coalesce repeated "not seekable" recoveries. +let seekFallbackRetryTimer: ReturnType | null = null; +let seekFallbackTrackId: string | null = null; +let seekFallbackRestartAt = 0; // Guard against rapid double-click play/pause sending two state transitions // to the Rust backend before it has finished the previous one. @@ -324,6 +328,25 @@ function touchHotCacheOnPlayback(trackId: string, serverId: string) { useHotCacheStore.getState().touchPlayed(trackId, serverId); } +async function promoteCompletedStreamToHotCache(track: Track, serverId: string, customDir: string | null) { + try { + const res = await invoke<{ path: string; size: number } | null>( + 'promote_stream_cache_to_hot_cache', + { + trackId: track.id, + serverId, + url: buildStreamUrl(track.id), + suffix: track.suffix || 'mp3', + customDir, + }, + ); + if (!res || !res.path) return; + useHotCacheStore.getState().setEntry(track.id, serverId, res.path, res.size || 0); + } catch { + // best-effort promotion; normal hot-cache prefetch remains fallback + } +} + // Track ID that has already been sent to audio_chain_preload (gapless chain). let gaplessPreloadingId: string | null = null; // Track ID that has already been sent to audio_preload (byte pre-download). @@ -380,22 +403,38 @@ function handleAudioProgress(current_time: number, duration: number) { } } - // Pre-buffer / pre-chain next track based on preload mode. - const { gaplessEnabled, preloadMode, preloadCustomSeconds, hotCacheEnabled } = useAuthStore.getState(); + // Pre-buffer / pre-chain next track based on preload mode and crossfade. + const { + gaplessEnabled, + preloadMode, + preloadCustomSeconds, + hotCacheEnabled, + crossfadeEnabled, + crossfadeSecs, + } = useAuthStore.getState(); const remaining = dur - current_time; // Gapless chain: always triggers at 30s regardless of preloadMode. const shouldChainGapless = gaplessEnabled && remaining < 30 && remaining > 0; // Byte pre-download: skip when Hot Cache is active (it already handles buffering). - const shouldBytePreload = !hotCacheEnabled && preloadMode !== 'off' && ( + // Even with preload mode OFF, crossfade needs the next track bytes ready before + // we enter the fade window to avoid a hard gap after track boundary. + const shouldBytePreloadFromMode = preloadMode !== 'off' && ( preloadMode === 'early' ? current_time >= 5 : preloadMode === 'custom' ? remaining < preloadCustomSeconds && remaining > 0 : remaining < 30 && remaining > 0 // balanced (default) ); + const crossfadeWindowSecs = Math.max(8, Math.min(30, crossfadeSecs + 6)); + const shouldBytePreloadForCrossfade = + !gaplessEnabled && crossfadeEnabled && remaining < crossfadeWindowSecs && remaining > 0; + const shouldBytePreload = !hotCacheEnabled && ( + shouldBytePreloadFromMode || + shouldBytePreloadForCrossfade + ); - if (shouldChainGapless || shouldBytePreload) { + if (shouldChainGapless || shouldBytePreload || gaplessEnabled) { const { queue, queueIndex, repeatMode } = store; const nextIdx = queueIndex + 1; const nextTrack = repeatMode === 'one' @@ -403,11 +442,28 @@ function handleAudioProgress(current_time: number, duration: number) { : (nextIdx < queue.length ? queue[nextIdx] : (repeatMode === 'all' ? queue[0] : null)); if (!nextTrack || nextTrack.id === track.id) return; + // Gapless backup: keep next-track bytes ready even if chain/decode misses + // the boundary. Start earlier for larger files / slower conservative link. + const estBytes = (() => { + if (typeof nextTrack.size === 'number' && Number.isFinite(nextTrack.size) && nextTrack.size > 0) { + return nextTrack.size; + } + const kbps = typeof nextTrack.bitRate === 'number' && Number.isFinite(nextTrack.bitRate) && nextTrack.bitRate > 0 + ? nextTrack.bitRate + : 320; + return Math.max(256 * 1024, Math.ceil((nextTrack.duration || 240) * kbps * 1000 / 8)); + })(); + const conservativeBytesPerSec = 300 * 1024; // ~2.4 Mbps effective throughput + const estDownloadSecs = estBytes / conservativeBytesPerSec; + const gaplessBackupWindowSecs = Math.max(15, Math.min(60, Math.ceil(estDownloadSecs * 1.4 + 8))); + const shouldBytePreloadForGaplessBackup = + gaplessEnabled && remaining < gaplessBackupWindowSecs && remaining > 0; + const serverId = useAuthStore.getState().activeServerId ?? ''; const nextUrl = resolvePlaybackUrl(nextTrack.id, serverId); // Byte pre-download — runs early so bytes are cached by chain time. - if (shouldBytePreload && nextTrack.id !== bytePreloadingId) { + if ((shouldBytePreload || shouldBytePreloadForGaplessBackup) && nextTrack.id !== bytePreloadingId) { bytePreloadingId = nextTrack.id; invoke('audio_preload', { url: nextUrl, durationHint: nextTrack.duration }).catch(() => {}); } @@ -912,6 +968,9 @@ export const usePlayerStore = create()( isAudioPaused = false; gaplessPreloadingId = null; bytePreloadingId = null; // new track — allow fresh preload for next if (seekDebounce) { clearTimeout(seekDebounce); seekDebounce = null; } seekTarget = null; + if (seekFallbackRetryTimer) { clearTimeout(seekFallbackRetryTimer); seekFallbackRetryTimer = null; } + seekFallbackTrackId = null; + seekFallbackRestartAt = 0; // If a radio stream is active, stop it before the new track starts so // the PlayerBar clears radio mode immediately and the stream is released. @@ -923,6 +982,7 @@ export const usePlayerStore = create()( } const state = get(); + const prevTrack = state.currentTrack; const newQueue = queue ?? state.queue; const idx = newQueue.findIndex(t => t.id === track.id); @@ -942,6 +1002,18 @@ export const usePlayerStore = create()( }); const authState = useAuthStore.getState(); + if ( + prevTrack + && prevTrack.id !== track.id + && authState.hotCacheEnabled + && authState.activeServerId + ) { + void promoteCompletedStreamToHotCache( + prevTrack, + authState.activeServerId, + authState.hotCacheDownloadDir || null, + ); + } setDeferHotCachePrefetch(true); const url = resolvePlaybackUrl(track.id, authState.activeServerId ?? ''); const replayGainDb = authState.replayGainEnabled @@ -1267,7 +1339,32 @@ export const usePlayerStore = create()( seekDebounce = setTimeout(() => { seekDebounce = null; seekTarget = time; - invoke('audio_seek', { seconds: time }).catch(console.error); + invoke('audio_seek', { seconds: time }).catch((err: unknown) => { + const msg = String(err ?? ''); + if (!msg.includes('not seekable')) { + console.error(err); + return; + } + // Streaming-start path can be non-seekable until the download finishes. + // Fallback: at most one restart burst per track, then keep only the latest retry seek. + const s = get(); + if (!s.currentTrack) return; + const now = Date.now(); + const sameBurst = + seekFallbackTrackId === s.currentTrack.id + && now - seekFallbackRestartAt < 600; + if (!sameBurst) { + seekFallbackTrackId = s.currentTrack.id; + seekFallbackRestartAt = now; + // Keep manual semantics (no crossfade) for seek recovery restarts. + s.playTrack(s.currentTrack, s.queue, true); + } + if (seekFallbackRetryTimer) clearTimeout(seekFallbackRetryTimer); + seekFallbackRetryTimer = setTimeout(() => { + seekFallbackRetryTimer = null; + invoke('audio_seek', { seconds: time }).catch(() => {}); + }, 220); + }); }, 100); },