fix(analysis): CPU seed queue, single waveform emit, and log URL redaction

Serialize heavy PCM seeding through a dedicated queue with optional priority
for the current track. Emit waveform-updated once per completed seed, fix
Lucky Mix waveform refresh tokens, redact Subsonic URLs in logs, and align
hot-cache prefetch with the queued path.
This commit is contained in:
cucadmuh
2026-04-26 20:59:33 +03:00
committed by Maxim Isaev
parent 5cb233e1c9
commit 218aa00718
7 changed files with 1045 additions and 315 deletions
+5 -54
View File
@@ -1,7 +1,6 @@
use std::collections::HashSet;
use std::path::PathBuf; use std::path::PathBuf;
use std::io::Cursor; use std::io::Cursor;
use std::sync::{Mutex, OnceLock}; use std::sync::Mutex;
use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::time::{Instant, SystemTime, UNIX_EPOCH};
use ebur128::{EbuR128, Mode as Ebur128Mode}; use ebur128::{EbuR128, Mode as Ebur128Mode};
@@ -316,7 +315,7 @@ pub fn recommended_gain_for_target(integrated_lufs: f64, true_peak: f64, target_
recommended_gain_db.clamp(-24.0, 24.0) recommended_gain_db.clamp(-24.0, 24.0)
} }
/// Result of [`seed_from_bytes`]: callers use it to avoid redundant UI events. /// Result of [`seed_from_bytes_execute`] / CPU seed queue: callers use it to avoid redundant UI events.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SeedFromBytesOutcome { pub enum SeedFromBytesOutcome {
/// Wrote waveform (and loudness when PCM decode succeeded). /// Wrote waveform (and loudness when PCM decode succeeded).
@@ -325,63 +324,15 @@ pub enum SeedFromBytesOutcome {
SkippedWaveformCacheHit, SkippedWaveformCacheHit,
/// `AnalysisCache` was not registered on the app handle. /// `AnalysisCache` was not registered on the app handle.
SkippedNoAnalysisCache, SkippedNoAnalysisCache,
/// Another seed for the same `track_id` is already running on a different
/// thread; this caller bails out so the winner's result is used.
SkippedConcurrent,
} }
/// Track-ids whose `seed_from_bytes` is currently running. There are six /// Full Symphonia + (optional) EBU decode for waveform + loudness. Call only from the
/// independent producers (legacy stream capture, ranged stream completion, /// single CPU-seed worker in `lib.rs` (`spawn_blocking`) so at most one heavy decode runs.
/// audio_preload, psysonic-local file read, in-memory play paths, and the pub fn seed_from_bytes_execute(
/// frontend-triggered `analysis_enqueue_seed_from_url` backfill) that can
/// fire for the same id concurrently when LUFS is on. Without this guard
/// the same MP3 gets fully decoded by Symphonia + EBU R128 twice — ~30 s
/// of wasted CPU per cache-miss track. First caller wins; the rest return
/// `SkippedConcurrent` immediately.
static SEEDS_IN_FLIGHT: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
fn seeds_in_flight() -> &'static Mutex<HashSet<String>> {
SEEDS_IN_FLIGHT.get_or_init(|| Mutex::new(HashSet::new()))
}
/// RAII helper: removes the in-flight marker on drop. Survives panics and
/// any early `?`-returns inside `seed_from_bytes`.
struct SeedInFlightGuard {
track_id: String,
}
impl Drop for SeedInFlightGuard {
fn drop(&mut self) {
if let Some(set) = SEEDS_IN_FLIGHT.get() {
if let Ok(mut g) = set.lock() {
g.remove(&self.track_id);
}
}
}
}
pub fn seed_from_bytes(
app: &tauri::AppHandle, app: &tauri::AppHandle,
track_id: &str, track_id: &str,
bytes: &[u8], bytes: &[u8],
) -> Result<SeedFromBytesOutcome, String> { ) -> Result<SeedFromBytesOutcome, String> {
// Reserve the slot for this track_id. Atomic under the mutex: insert()
// returns false if the id was already present.
let track_key = track_id.to_string();
let claimed = {
let mut guard = seeds_in_flight().lock().map_err(|_| "seeds-in-flight lock poisoned".to_string())?;
guard.insert(track_key.clone())
};
if !claimed {
crate::app_deprintln!(
"[analysis] full-track analysis skip track_id={} reason=concurrent_seed_in_flight bytes={}",
track_id,
bytes.len()
);
return Ok(SeedFromBytesOutcome::SkippedConcurrent);
}
let _flight_guard = SeedInFlightGuard { track_id: track_key };
let started = Instant::now(); let started = Instant::now();
let Some(cache) = app.try_state::<AnalysisCache>() else { let Some(cache) = app.try_state::<AnalysisCache>() else {
crate::app_deprintln!( crate::app_deprintln!(
+235 -134
View File
@@ -31,13 +31,6 @@ struct PartialLoudnessPayload {
is_partial: bool, is_partial: bool,
} }
#[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")]
struct WaveformUpdatedPayload {
track_id: String,
is_partial: bool,
}
#[derive(Clone, Serialize)] #[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
struct NormalizationStatePayload { struct NormalizationStatePayload {
@@ -528,6 +521,10 @@ const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024;
const TRACK_STREAM_MAX_BUF_CAPACITY: usize = 32 * 1024 * 1024; 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 memory to promote a completed streamed track for fast replay/seek recovery.
const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024; 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 64MiB;
/// using the stream-promote cap here skipped analysis entirely (empty seekbar).
const LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES: usize = 512 * 1024 * 1024;
/// Consecutive body-stream failures tolerated for track streaming before abort. /// Consecutive body-stream failures tolerated for track streaming before abort.
const TRACK_STREAM_MAX_RECONNECTS: u32 = 3; const TRACK_STREAM_MAX_RECONNECTS: u32 = 3;
/// Seconds at stall threshold while paused before hard-disconnect. /// Seconds at stall threshold while paused before hard-disconnect.
@@ -1226,22 +1223,15 @@ async fn track_download_task(
if !capture_over_limit && !capture.is_empty() { if !capture_over_limit && !capture.is_empty() {
if let Some(track_id) = cache_track_id { if let Some(track_id) = cache_track_id {
crate::app_deprintln!( crate::app_deprintln!(
"[stream] legacy stream: capture complete track_id={} capture_mib={:.2} — invoking full-track analysis (blocks until seed_from_bytes returns)", "[stream] legacy stream: capture complete track_id={} capture_mib={:.2} — full-track analysis (cpu-seed queue)",
track_id, track_id,
capture.len() as f64 / (1024.0 * 1024.0) capture.len() as f64 / (1024.0 * 1024.0)
); );
match crate::analysis_cache::seed_from_bytes(&app, &track_id, &capture) { let high = analysis_seed_high_priority_for_track(&app, &track_id);
Err(e) => crate::app_eprintln!("[analysis] track seed failed for {}: {}", track_id, e), if let Err(e) =
Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => { crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), capture.clone(), high).await
let _ = app.emit( {
"analysis:waveform-updated", crate::app_eprintln!("[analysis] track seed failed for {}: {}", track_id, e);
WaveformUpdatedPayload {
track_id: track_id.clone(),
is_partial: false,
},
);
}
Ok(_) => {}
} }
} }
*promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack {
@@ -1274,7 +1264,25 @@ async fn ranged_download_task(
normalization_target_lufs: Arc<AtomicU32>, normalization_target_lufs: Arc<AtomicU32>,
loudness_pre_analysis_attenuation_db: Arc<AtomicU32>, loudness_pre_analysis_attenuation_db: Arc<AtomicU32>,
cache_track_id: Option<String>, cache_track_id: Option<String>,
// 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<Arc<Mutex<Option<(String, u64)>>>>,
) { ) {
let _ranged_loudness_hold_clear = match (loudness_seed_hold.as_ref(), cache_track_id.as_ref()) {
(Some(slot), Some(tid)) => {
let t = tid.clone();
{
let mut g = slot.lock().unwrap();
*g = Some((t.clone(), gen));
}
Some(RangedLoudnessSeedHoldClear {
slot: Arc::clone(slot),
tid: t,
gen,
})
}
_ => None,
};
let total_size = buf.lock().unwrap().len(); let total_size = buf.lock().unwrap().len();
let mut downloaded: usize = 0; let mut downloaded: usize = 0;
let mut reconnects: u32 = 0; let mut reconnects: u32 = 0;
@@ -1415,7 +1423,7 @@ async fn ranged_download_task(
if downloaded == total_size && total_size > 0 && total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES { if downloaded == total_size && total_size > 0 && total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES {
if let Some(ref tid) = cache_track_id { if let Some(ref tid) = cache_track_id {
crate::app_deprintln!( crate::app_deprintln!(
"[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (blocks this download task until complete)", "[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (cpu-seed queue; this task awaits completion)",
tid, tid,
total_size as f64 / (1024.0 * 1024.0), total_size as f64 / (1024.0 * 1024.0),
total_size total_size
@@ -1430,18 +1438,9 @@ async fn ranged_download_task(
); );
} }
if let Some(track_id) = cache_track_id { if let Some(track_id) = cache_track_id {
match crate::analysis_cache::seed_from_bytes(&app, &track_id, &data) { let high = analysis_seed_high_priority_for_track(&app, &track_id);
Err(e) => crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e), if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await {
Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => { crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e);
let _ = app.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: track_id.clone(),
is_partial: false,
},
);
}
Ok(_) => {}
} }
} }
*promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data }); *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data });
@@ -2285,6 +2284,12 @@ pub struct AudioEngine {
/// Subsonic song id last passed from JS with `audio_play` (trimmed). Used /// Subsonic song id last passed from JS with `audio_play` (trimmed). Used
/// for loudness/waveform cache when the URL is `psysonic-local://…`. /// for loudness/waveform cache when the URL is `psysonic-local://…`.
pub(crate) current_analysis_track_id: Arc<Mutex<Option<String>>>, pub(crate) current_analysis_track_id: Arc<Mutex<Option<String>>>,
/// While a `RangedHttpSource` download task is filling the buffer for this
/// `(track_id, play_generation)`, skip `analysis_enqueue_seed_from_url` for the
/// same id — otherwise a parallel full GET + Symphonia competes with playback
/// decode (ALSA underruns). The ranged task clears this on exit; `gen` avoids a
/// late drop clearing a newer play of the same track.
pub(crate) ranged_loudness_seed_hold: Arc<Mutex<Option<(String, u64)>>>,
} }
pub struct AudioCurrent { pub struct AudioCurrent {
@@ -2559,11 +2564,63 @@ pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) {
radio_state: Mutex::new(None), radio_state: Mutex::new(None),
current_playback_url: Arc::new(Mutex::new(None)), current_playback_url: Arc::new(Mutex::new(None)),
current_analysis_track_id: Arc::new(Mutex::new(None)), current_analysis_track_id: Arc::new(Mutex::new(None)),
ranged_loudness_seed_hold: Arc::new(Mutex::new(None)),
}; };
(engine, thread) (engine, thread)
} }
/// Clears [`AudioEngine::ranged_loudness_seed_hold`] only if it still matches this play.
struct RangedLoudnessSeedHoldClear {
slot: Arc<Mutex<Option<(String, u64)>>>,
tid: String,
gen: u64,
}
impl Drop for RangedLoudnessSeedHoldClear {
fn drop(&mut self) {
if let Ok(mut g) = self.slot.lock() {
if matches!(&*g, Some((t, gen)) if t == &self.tid && *gen == self.gen) {
*g = None;
}
}
}
}
/// `analysis_enqueue_seed_from_url` should bail while this track's ranged HTTP buffer
/// is still filling — playback will seed on completion with the same bytes.
pub(crate) fn ranged_loudness_backfill_should_defer(engine: &AudioEngine, track_id: &str) -> bool {
let tid = track_id.trim();
if tid.is_empty() {
return false;
}
let Ok(g) = engine.ranged_loudness_seed_hold.lock() else {
return false;
};
matches!(&*g, Some((t, _)) if t.as_str() == tid)
}
/// Subsonic id pinned for the playing source (`audio_play`). Used to prioritize
/// HTTP loudness backfill for the track the user is listening to.
pub(crate) fn analysis_track_id_is_current_playback(engine: &AudioEngine, track_id: &str) -> bool {
let needle = track_id.trim();
if needle.is_empty() {
return false;
}
let Ok(guard) = engine.current_analysis_track_id.lock() else {
return false;
};
let Some(cur) = guard.as_deref().map(str::trim).filter(|s| !s.is_empty()) else {
return false;
};
cur == needle
}
fn analysis_seed_high_priority_for_track(app: &AppHandle, track_id: &str) -> bool {
app.try_state::<AudioEngine>()
.is_some_and(|e| analysis_track_id_is_current_playback(&e, track_id))
}
fn audio_http_client(state: &AudioEngine) -> reqwest::Client { fn audio_http_client(state: &AudioEngine) -> reqwest::Client {
state state
.http_client .http_client
@@ -2632,30 +2689,71 @@ fn same_playback_target(a_url: &str, b_url: &str) -> bool {
} }
} }
#[derive(Clone, Copy)]
struct ResolveLoudnessCacheOpts {
/// When false, skip `get_latest_waveform_for_track` — `audio_update_replay_gain` runs
/// on every partial-LUFS tick; loudness gain does not depend on waveform, and the extra
/// SQLite read was pure overhead on the IPC path.
touch_waveform: bool,
/// When false, omit `cache-miss` / `cache-invalid` debug lines (still log hits and errors).
log_soft_misses: bool,
}
impl Default for ResolveLoudnessCacheOpts {
fn default() -> Self {
Self {
touch_waveform: true,
log_soft_misses: true,
}
}
}
fn resolve_loudness_gain_from_cache( fn resolve_loudness_gain_from_cache(
app: &AppHandle, app: &AppHandle,
url: &str, url: &str,
target_lufs: f32, target_lufs: f32,
logical_track_id: Option<&str>, logical_track_id: Option<&str>,
) -> Option<f32> {
resolve_loudness_gain_from_cache_impl(
app,
url,
target_lufs,
logical_track_id,
ResolveLoudnessCacheOpts::default(),
)
}
fn resolve_loudness_gain_from_cache_impl(
app: &AppHandle,
url: &str,
target_lufs: f32,
logical_track_id: Option<&str>,
opts: ResolveLoudnessCacheOpts,
) -> Option<f32> { ) -> Option<f32> {
// Only a SQLite loudness row counts here. Ephemeral JS hints (`analysis:loudness-partial`) // Only a SQLite loudness row counts here. Ephemeral JS hints (`analysis:loudness-partial`)
// are applied in `audio_update_replay_gain` via `loudness_gain_db_or_startup(..., true, _)`. // are applied in `audio_update_replay_gain` via `loudness_gain_db_or_startup(..., true, _)`.
let Some(track_id) = analysis_cache_track_id(logical_track_id, url) else { let Some(track_id) = analysis_cache_track_id(logical_track_id, url) else {
crate::app_deprintln!( if opts.log_soft_misses {
"[normalization] resolve_loudness_gain source=no-identity url_len={}", crate::app_deprintln!(
url.len() "[normalization] resolve_loudness_gain source=no-identity url_len={}",
); url.len()
);
}
return None; return None;
}; };
let Some(cache) = app.try_state::<crate::analysis_cache::AnalysisCache>() else { let Some(cache) = app.try_state::<crate::analysis_cache::AnalysisCache>() else {
crate::app_deprintln!( if opts.log_soft_misses {
"[normalization] resolve_loudness_gain source=no-analysis-cache track_id={}", crate::app_deprintln!(
track_id "[normalization] resolve_loudness_gain source=no-analysis-cache track_id={}",
); track_id
);
}
return None; return None;
}; };
// Also touch waveform row here so playback path verifies current context is present. if opts.touch_waveform {
let _ = cache.get_latest_waveform_for_track(&track_id); // Bind / preload: verify waveform context exists alongside loudness lookup.
let _ = cache.get_latest_waveform_for_track(&track_id);
}
match cache.get_latest_loudness_for_track(&track_id) { match cache.get_latest_loudness_for_track(&track_id) {
Ok(Some(row)) if row.integrated_lufs.is_finite() => { Ok(Some(row)) if row.integrated_lufs.is_finite() => {
let recommended = crate::analysis_cache::recommended_gain_for_target( let recommended = crate::analysis_cache::recommended_gain_for_target(
@@ -2674,18 +2772,22 @@ fn resolve_loudness_gain_from_cache(
Some(recommended) Some(recommended)
} }
Ok(Some(row)) => { Ok(Some(row)) => {
crate::app_deprintln!( if opts.log_soft_misses {
"[normalization] resolve_loudness_gain source=cache-invalid track_id={} integrated_lufs={}", crate::app_deprintln!(
track_id, "[normalization] resolve_loudness_gain source=cache-invalid track_id={} integrated_lufs={}",
row.integrated_lufs track_id,
); row.integrated_lufs
);
}
None None
} }
Ok(None) => { Ok(None) => {
crate::app_deprintln!( if opts.log_soft_misses {
"[normalization] resolve_loudness_gain source=cache-miss track_id={}", crate::app_deprintln!(
track_id "[normalization] resolve_loudness_gain source=cache-miss track_id={}",
); track_id
);
}
None None
} }
Err(e) => { Err(e) => {
@@ -2699,6 +2801,30 @@ fn resolve_loudness_gain_from_cache(
} }
} }
/// LUFS gain after a single `resolve_loudness_gain_from_cache` result (`None` = miss).
/// Keeps `audio_update_replay_gain` / `audio_play` from resolving twice on the same URL.
fn loudness_gain_db_after_resolve(
resolved_from_cache: Option<f32>,
pre_analysis_attenuation_db: f32,
allow_js_when_uncached: bool,
js_gain_db: Option<f32>,
) -> Option<f32> {
let pre = pre_analysis_attenuation_db.clamp(-24.0, 0.0).min(0.0);
match resolved_from_cache {
Some(g) => Some(g),
None => {
if allow_js_when_uncached {
match js_gain_db {
Some(r) if r.is_finite() => Some(r),
_ => Some(pre),
}
} else {
Some(pre)
}
}
}
}
/// LUFS: DB-backed integrated LUFS only at bind time (`allow_js_when_uncached = false`); /// LUFS: DB-backed integrated LUFS only at bind time (`allow_js_when_uncached = false`);
/// after `analysis:loudness-partial`, `audio_update_replay_gain` passes `true` so finite /// after `analysis:loudness-partial`, `audio_update_replay_gain` passes `true` so finite
/// JS gain applies until SQLite catches up. Must never return `None` or `compute_gain` uses unity. /// JS gain applies until SQLite catches up. Must never return `None` or `compute_gain` uses unity.
@@ -2711,20 +2837,19 @@ fn loudness_gain_db_or_startup(
allow_js_when_uncached: bool, allow_js_when_uncached: bool,
js_gain_db: Option<f32>, js_gain_db: Option<f32>,
) -> Option<f32> { ) -> Option<f32> {
let pre = pre_analysis_attenuation_db.clamp(-24.0, 0.0).min(0.0); let resolved = resolve_loudness_gain_from_cache_impl(
match resolve_loudness_gain_from_cache(app, url, target_lufs, logical_track_id) { app,
Some(g) => Some(g), url,
None => { target_lufs,
if allow_js_when_uncached { logical_track_id,
match js_gain_db { ResolveLoudnessCacheOpts::default(),
Some(r) if r.is_finite() => Some(r), );
_ => Some(pre), loudness_gain_db_after_resolve(
} resolved,
} else { pre_analysis_attenuation_db,
Some(pre) allow_js_when_uncached,
} js_gain_db,
} )
}
} }
#[inline] #[inline]
@@ -2831,7 +2956,7 @@ async fn fetch_data(
/// When playback uses full track bytes already in RAM (gapless `reuse_chained_bytes`, /// When playback uses full track bytes already in RAM (gapless `reuse_chained_bytes`,
/// `preloaded`, or `stream_completed_cache` via `fetch_data`), the `psysonic-local` /// `preloaded`, or `stream_completed_cache` via `fetch_data`), the `psysonic-local`
/// disk-read seed path never runs. Spawn the same `seed_from_bytes` work so waveform / /// disk-read seed path never runs. Submit the same full-buffer analysis via the cpu-seed queue so waveform /
/// loudness SQLite can fill **offline** without `analysis_enqueue_seed_from_url` HTTP. /// loudness SQLite can fill **offline** without `analysis_enqueue_seed_from_url` HTTP.
fn spawn_analysis_seed_from_in_memory_bytes( fn spawn_analysis_seed_from_in_memory_bytes(
app: &AppHandle, app: &AppHandle,
@@ -2855,26 +2980,17 @@ fn spawn_analysis_seed_from_in_memory_bytes(
track_id, track_id,
bytes.len() as f64 / (1024.0 * 1024.0) bytes.len() as f64 / (1024.0 * 1024.0)
); );
let high = analysis_seed_high_priority_for_track(&app, &track_id);
tokio::spawn(async move { tokio::spawn(async move {
if gen_arc.load(Ordering::SeqCst) != gen { if gen_arc.load(Ordering::SeqCst) != gen {
return; return;
} }
match crate::analysis_cache::seed_from_bytes(&app, &track_id, &bytes) { if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), bytes, high).await {
Err(e) => crate::app_eprintln!( crate::app_eprintln!(
"[analysis] in-memory play path seed failed for {}: {}", "[analysis] in-memory play path seed failed for {}: {}",
track_id, track_id,
e e
), );
Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => {
let _ = app.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: track_id.clone(),
is_partial: false,
},
);
}
Ok(_) => {}
} }
}); });
} }
@@ -3177,30 +3293,29 @@ pub async fn audio_play(
if gen_arc_seed.load(Ordering::SeqCst) != gen_seed { if gen_arc_seed.load(Ordering::SeqCst) != gen_seed {
return; return;
} }
if data.is_empty() || data.len() > TRACK_STREAM_PROMOTE_MAX_BYTES { if data.is_empty() || data.len() > LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES {
crate::app_deprintln!(
"[stream] psysonic-local: skip analysis seed track_id={} bytes={} (over {} MiB cap)",
seed_id,
data.len(),
LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES / (1024 * 1024)
);
return; return;
} }
crate::app_deprintln!( crate::app_deprintln!(
"[stream] psysonic-local: file read complete track_id={} size_mib={:.2} — invoking full-track analysis (async)", "[stream] psysonic-local: file read complete track_id={} size_mib={:.2} — full-track analysis (cpu-seed queue)",
seed_id, seed_id,
data.len() as f64 / (1024.0 * 1024.0) data.len() as f64 / (1024.0 * 1024.0)
); );
match crate::analysis_cache::seed_from_bytes(&app_seed, &seed_id, &data) { let high = analysis_seed_high_priority_for_track(&app_seed, &seed_id);
Err(e) => crate::app_eprintln!( if let Err(e) =
crate::submit_analysis_cpu_seed(app_seed.clone(), seed_id.clone(), data, high).await
{
crate::app_eprintln!(
"[analysis] local-file seed failed for {}: {}", "[analysis] local-file seed failed for {}: {}",
seed_id, seed_id,
e e
), );
Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => {
let _ = app_seed.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: seed_id.clone(),
is_partial: false,
},
);
}
Ok(_) => {}
} }
}); });
} }
@@ -3250,6 +3365,8 @@ pub async fn audio_play(
let buf = Arc::new(Mutex::new(vec![0u8; total_usize])); let buf = Arc::new(Mutex::new(vec![0u8; total_usize]));
let downloaded_to = Arc::new(AtomicUsize::new(0)); let downloaded_to = Arc::new(AtomicUsize::new(0));
let done = Arc::new(AtomicBool::new(false)); let done = Arc::new(AtomicBool::new(false));
let loudness_hold_for_defer = (total_usize <= TRACK_STREAM_PROMOTE_MAX_BYTES)
.then_some(state.ranged_loudness_seed_hold.clone());
tokio::spawn(ranged_download_task( tokio::spawn(ranged_download_task(
gen, gen,
state.generation.clone(), state.generation.clone(),
@@ -3266,6 +3383,7 @@ pub async fn audio_play(
state.normalization_target_lufs.clone(), state.normalization_target_lufs.clone(),
state.loudness_pre_analysis_attenuation_db.clone(), state.loudness_pre_analysis_attenuation_db.clone(),
cache_id_for_tasks.clone(), cache_id_for_tasks.clone(),
loudness_hold_for_defer,
)); ));
let reader = RangedHttpSource { let reader = RangedHttpSource {
buf, buf,
@@ -3348,26 +3466,19 @@ pub async fn audio_play(
} }
let target_lufs = f32::from_bits(state.normalization_target_lufs.load(Ordering::Relaxed)); let target_lufs = f32::from_bits(state.normalization_target_lufs.load(Ordering::Relaxed));
let resolved_loudness_gain_db = resolve_loudness_gain_from_cache( let cache_loudness = resolve_loudness_gain_from_cache(
&app, &app,
&url, &url,
target_lufs, target_lufs,
logical_trim.as_deref(), logical_trim.as_deref(),
); );
let resolved_loudness_gain_db = cache_loudness;
let norm_mode = state.normalization_engine.load(Ordering::Relaxed); let norm_mode = state.normalization_engine.load(Ordering::Relaxed);
let pre_analysis_db = loudness_pre_analysis_db_for_engine(&state); let pre_analysis_db = loudness_pre_analysis_db_for_engine(&state);
let startup_loudness_gain_db = if norm_mode == 2 { let startup_loudness_gain_db = if norm_mode == 2 {
loudness_gain_db_or_startup( loudness_gain_db_after_resolve(cache_loudness, pre_analysis_db, false, loudness_gain_db)
&app,
&url,
target_lufs,
logical_trim.as_deref(),
pre_analysis_db,
false,
loudness_gain_db,
)
} else { } else {
resolved_loudness_gain_db cache_loudness
}; };
let (gain_linear, effective_volume) = compute_gain( let (gain_linear, effective_volume) = compute_gain(
norm_mode, norm_mode,
@@ -4284,24 +4395,23 @@ pub fn audio_update_replay_gain(
// If `current_playback_url` is not pinned yet, still honour JS `loudness_gain_db` // If `current_playback_url` is not pinned yet, still honour JS `loudness_gain_db`
// so `loudness_ui_current_gain_db` can show a number (otherwise `and_then` // so `loudness_ui_current_gain_db` can show a number (otherwise `and_then`
// drops the requested gain entirely). // drops the requested gain entirely).
let resolved_loudness_gain_db = url_for_loudness let cache_loudness = url_for_loudness.as_deref().and_then(|u| {
.as_deref() resolve_loudness_gain_from_cache_impl(
.and_then(|u| { &app,
resolve_loudness_gain_from_cache( u,
&app, target_lufs,
u, logical_for_loudness.as_deref(),
target_lufs, ResolveLoudnessCacheOpts {
logical_for_loudness.as_deref(), touch_waveform: false,
) log_soft_misses: false,
}) },
.or(loudness_gain_db); )
});
let resolved_loudness_gain_db = cache_loudness.or(loudness_gain_db);
let effective_loudness_db = if norm_mode == 2 { let effective_loudness_db = if norm_mode == 2 {
match url_for_loudness.as_deref() { match url_for_loudness.as_deref() {
Some(u) => loudness_gain_db_or_startup( Some(_u) => loudness_gain_db_after_resolve(
&app, cache_loudness,
u,
target_lufs,
logical_for_loudness.as_deref(),
pre_analysis_db, pre_analysis_db,
true, true,
loudness_gain_db, loudness_gain_db,
@@ -4449,18 +4559,9 @@ pub async fn audio_preload(
track_id, track_id,
data.len() as f64 / (1024.0 * 1024.0) data.len() as f64 / (1024.0 * 1024.0)
); );
match crate::analysis_cache::seed_from_bytes(&app, &track_id, &data) { let high = analysis_track_id_is_current_playback(&state, &track_id);
Err(e) => crate::app_eprintln!("[analysis] preload seed failed for {}: {}", track_id, e), if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await {
Ok(crate::analysis_cache::SeedFromBytesOutcome::Upserted) => { crate::app_eprintln!("[analysis] preload seed failed for {}: {}", track_id, e);
let _ = app.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: track_id.clone(),
is_partial: false,
},
);
}
Ok(_) => {}
} }
} }
let url_for_emit = url.clone(); let url_for_emit = url.clone();
+548 -78
View File
@@ -9,7 +9,7 @@ pub(crate) mod logging;
#[cfg(target_os = "windows")] #[cfg(target_os = "windows")]
mod taskbar_win; mod taskbar_win;
use std::collections::HashMap; use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, OnceLock, RwLock}; use std::sync::{Arc, Mutex, OnceLock, RwLock};
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
@@ -60,10 +60,425 @@ fn sync_cancel_flags() -> &'static Mutex<HashMap<String, Arc<AtomicBool>>> {
FLAGS.get_or_init(|| Mutex::new(HashMap::new())) FLAGS.get_or_init(|| Mutex::new(HashMap::new()))
} }
/// Tracks analysis backfill jobs already running per track_id. #[derive(Debug, Clone, Copy, PartialEq, Eq)]
fn analysis_backfill_inflight() -> &'static Mutex<std::collections::HashSet<String>> { enum AnalysisBackfillEnqueueKind {
static SET: OnceLock<Mutex<std::collections::HashSet<String>>> = OnceLock::new(); /// New job at the tail of the queue.
SET.get_or_init(|| Mutex::new(std::collections::HashSet::new())) NewBack,
/// New job for the currently playing track (head).
NewFront,
/// Same track was already waiting; moved to head with the latest URL.
ReorderedFront,
/// Low-priority duplicate while the track is already queued or running.
DuplicateSkipped,
/// High-priority request but that track is already being downloaded+seeded.
RunningSkipped,
}
#[derive(Default)]
struct AnalysisBackfillQueueState {
deque: VecDeque<(String, String)>,
/// Set while this `track_id` is inside `analysis_backfill_download_and_seed` (not in deque).
in_progress: Option<String>,
}
impl AnalysisBackfillQueueState {
fn is_reserved(&self, tid: &str) -> bool {
self.in_progress.as_deref() == Some(tid)
|| self.deque.iter().any(|(t, _)| t.as_str() == tid)
}
fn try_pop_next(&mut self) -> Option<(String, String)> {
let (tid, url) = self.deque.pop_front()?;
self.in_progress = Some(tid.clone());
Some((tid, url))
}
fn finish_job(&mut self, tid: &str) {
if self.in_progress.as_deref() == Some(tid) {
self.in_progress = None;
}
}
fn enqueue(
&mut self,
tid: String,
url: String,
high_priority: bool,
) -> AnalysisBackfillEnqueueKind {
let tref = tid.as_str();
if self.is_reserved(tref) {
if !high_priority {
return AnalysisBackfillEnqueueKind::DuplicateSkipped;
}
if self.in_progress.as_deref() == Some(tref) {
return AnalysisBackfillEnqueueKind::RunningSkipped;
}
self.deque.retain(|(t, _)| t != &tid);
self.deque.push_front((tid, url));
return AnalysisBackfillEnqueueKind::ReorderedFront;
}
if high_priority {
self.deque.push_front((tid, url));
AnalysisBackfillEnqueueKind::NewFront
} else {
self.deque.push_back((tid, url));
AnalysisBackfillEnqueueKind::NewBack
}
}
}
struct AnalysisBackfillShared {
state: Mutex<AnalysisBackfillQueueState>,
wake_tx: tokio::sync::mpsc::UnboundedSender<()>,
}
impl AnalysisBackfillShared {
fn ping_worker(&self) {
let _ = self.wake_tx.send(());
}
}
static ANALYSIS_BACKFILL: OnceLock<Arc<AnalysisBackfillShared>> = OnceLock::new();
/// Lazily spawns the single backfill worker (first caller supplies `AppHandle`).
fn analysis_backfill_shared(app: &tauri::AppHandle) -> Arc<AnalysisBackfillShared> {
ANALYSIS_BACKFILL
.get_or_init(|| {
let (wake_tx, wake_rx) = tokio::sync::mpsc::unbounded_channel();
let shared = Arc::new(AnalysisBackfillShared {
state: Mutex::new(AnalysisBackfillQueueState::default()),
wake_tx,
});
let app = app.clone();
let sh = shared.clone();
tauri::async_runtime::spawn(analysis_backfill_worker_loop(app, sh, wake_rx));
shared
})
.clone()
}
async fn analysis_backfill_download_and_seed(
app: &tauri::AppHandle,
track_id: &str,
url: &str,
) -> Result<bool, String> {
let client = reqwest::Client::builder()
.user_agent(subsonic_wire_user_agent())
.timeout(std::time::Duration::from_secs(120))
.build()
.map_err(|e| e.to_string())?;
let response = client.get(url).send().await.map_err(|e| e.to_string())?;
if !response.status().is_success() {
return Err(format!("HTTP {}", response.status().as_u16()));
}
let bytes = response.bytes().await.map_err(|e| e.to_string())?;
if bytes.is_empty() {
return Err("empty response".to_string());
}
enqueue_analysis_seed(app, track_id, &bytes).await
}
async fn analysis_backfill_worker_loop(
app: tauri::AppHandle,
shared: Arc<AnalysisBackfillShared>,
mut wake_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
) {
loop {
if wake_rx.recv().await.is_none() {
break;
}
while let Some((track_id, url)) = {
let mut st = shared
.state
.lock()
.unwrap_or_else(|e| e.into_inner());
st.try_pop_next()
} {
crate::app_deprintln!("[analysis] backfill worker: start track_id={}", track_id);
let result = analysis_backfill_download_and_seed(&app, &track_id, &url).await;
match &result {
Ok(has_loudness) => crate::app_deprintln!(
"[analysis] backfill ready: {} (has_loudness={})",
track_id,
has_loudness
),
Err(e) => crate::app_eprintln!("[analysis] backfill failed for {}: {}", track_id, e),
}
let mut st = shared
.state
.lock()
.unwrap_or_else(|e| e.into_inner());
st.finish_job(&track_id);
}
}
}
fn analysis_backfill_is_current_track(app: &tauri::AppHandle, track_id: &str) -> bool {
app.try_state::<crate::audio::AudioEngine>()
.is_some_and(|e| crate::audio::analysis_track_id_is_current_playback(&e, track_id))
}
// ─── Full-track waveform + loudness: single CPU worker (mirrors HTTP backfill queue) ─
// One `spawn_blocking` decode at a time; current playback is high-priority (front + reorder).
// Same `track_id` queued again merges waiters onto one job; while decode runs, same-id
// submitters attach to `running` followers so they all get the same outcome.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AnalysisCpuSeedEnqueueKind {
NewBack,
NewFront,
ReorderedFront,
RunningFollower,
MergedQueued,
}
struct AnalysisCpuSeedJob {
track_id: String,
bytes: Vec<u8>,
waiters: Vec<tokio::sync::oneshot::Sender<Result<analysis_cache::SeedFromBytesOutcome, String>>>,
}
struct AnalysisCpuSeedQueueState {
deque: VecDeque<AnalysisCpuSeedJob>,
/// Decode in progress — same-id callers wait here for the same outcome.
running: Option<(
String,
Arc<Mutex<Vec<tokio::sync::oneshot::Sender<Result<analysis_cache::SeedFromBytesOutcome, String>>>>>,
)>,
}
impl AnalysisCpuSeedQueueState {
fn enqueue(
&mut self,
track_id: String,
bytes: Vec<u8>,
high_priority: bool,
) -> (
AnalysisCpuSeedEnqueueKind,
tokio::sync::oneshot::Receiver<Result<analysis_cache::SeedFromBytesOutcome, String>>,
) {
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
let tid = track_id.as_str();
if let Some((rtid, followers)) = &self.running {
if rtid == tid {
followers
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(done_tx);
return (AnalysisCpuSeedEnqueueKind::RunningFollower, done_rx);
}
}
if let Some(pos) = self.deque.iter().position(|j| j.track_id == track_id) {
let mut job = self.deque.remove(pos).unwrap();
job.bytes = bytes;
job.waiters.push(done_tx);
let kind = if high_priority {
self.deque.push_front(job);
AnalysisCpuSeedEnqueueKind::ReorderedFront
} else {
self.deque.push_back(job);
AnalysisCpuSeedEnqueueKind::MergedQueued
};
return (kind, done_rx);
}
let job = AnalysisCpuSeedJob {
track_id: track_id.clone(),
bytes,
waiters: vec![done_tx],
};
let kind = if high_priority {
self.deque.push_front(job);
AnalysisCpuSeedEnqueueKind::NewFront
} else {
self.deque.push_back(job);
AnalysisCpuSeedEnqueueKind::NewBack
};
(kind, done_rx)
}
}
struct AnalysisCpuSeedShared {
state: Mutex<AnalysisCpuSeedQueueState>,
wake_tx: tokio::sync::mpsc::UnboundedSender<()>,
}
impl Default for AnalysisCpuSeedQueueState {
fn default() -> Self {
Self {
deque: VecDeque::new(),
running: None,
}
}
}
impl AnalysisCpuSeedShared {
fn ping_worker(&self) {
let _ = self.wake_tx.send(());
}
}
static ANALYSIS_CPU_SEED: OnceLock<Arc<AnalysisCpuSeedShared>> = OnceLock::new();
fn analysis_cpu_seed_shared(app: &tauri::AppHandle) -> Arc<AnalysisCpuSeedShared> {
ANALYSIS_CPU_SEED
.get_or_init(|| {
let (wake_tx, wake_rx) = tokio::sync::mpsc::unbounded_channel();
let shared = Arc::new(AnalysisCpuSeedShared {
state: Mutex::new(AnalysisCpuSeedQueueState::default()),
wake_tx,
});
let app = app.clone();
let sh = shared.clone();
tauri::async_runtime::spawn(analysis_cpu_seed_worker_loop(app, sh, wake_rx));
shared
})
.clone()
}
/// HTTP backfill + CPU seed queue sizes (debug log only — `app_deprintln!`).
fn emit_analysis_queue_snapshot_line() {
let http = if let Some(arc) = ANALYSIS_BACKFILL.get() {
let st = arc.state.lock().unwrap_or_else(|e| e.into_inner());
format!(
"http_backfill={{queued:{} download_active:{:?}}}",
st.deque.len(),
st.in_progress.as_deref()
)
} else {
"http_backfill={{not_started}}".to_string()
};
let cpu = if let Some(arc) = ANALYSIS_CPU_SEED.get() {
let st = arc.state.lock().unwrap_or_else(|e| e.into_inner());
let queued_jobs = st.deque.len();
let pending_in_queued_jobs: usize = st.deque.iter().map(|j| j.waiters.len()).sum();
let (decoding_tid, decoding_extra_waiters) = match &st.running {
Some((tid, fl)) => (
Some(tid.as_str()),
fl.lock().map(|g| g.len()).unwrap_or(0),
),
None => (None, 0usize),
};
format!(
"cpu_seed={{queued_jobs:{} pending_channels_in_queue:{} decoding_tid:{:?} extra_waiters_same_id:{}}}",
queued_jobs,
pending_in_queued_jobs,
decoding_tid,
decoding_extra_waiters
)
} else {
"cpu_seed={{not_started}}".to_string()
};
crate::app_deprintln!(
"[analysis] queue_snapshot interval_s=60 note=queues_in_memory_cleared_on_app_restart | {http} | {cpu}"
);
}
async fn analysis_queue_snapshot_loop() {
emit_analysis_queue_snapshot_line();
loop {
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
emit_analysis_queue_snapshot_line();
}
}
async fn analysis_cpu_seed_worker_loop(
app: tauri::AppHandle,
shared: Arc<AnalysisCpuSeedShared>,
mut wake_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
) {
loop {
if wake_rx.recv().await.is_none() {
break;
}
loop {
let (job, followers) = {
let mut st = shared.state.lock().unwrap_or_else(|e| e.into_inner());
let Some(j) = st.deque.pop_front() else {
break;
};
let fl = Arc::new(Mutex::new(Vec::new()));
st.running = Some((j.track_id.clone(), fl.clone()));
(j, fl)
};
let tid_log = job.track_id.clone();
let app2 = app.clone();
let tid = job.track_id.clone();
let bytes = job.bytes;
let outcome = tokio::task::spawn_blocking(move || {
analysis_cache::seed_from_bytes_execute(&app2, &tid, &bytes)
})
.await
.unwrap_or_else(|e| Err(format!("cpu-seed spawn_blocking: {e}")));
let mut extra = followers
.lock()
.unwrap_or_else(|e| e.into_inner())
.drain(..)
.collect::<Vec<_>>();
for tx in job.waiters {
let _ = tx.send(outcome.clone());
}
for tx in extra.drain(..) {
let _ = tx.send(outcome.clone());
}
{
let mut st = shared.state.lock().unwrap_or_else(|e| e.into_inner());
st.running = None;
}
let ok = outcome.as_ref().map(|o| *o == analysis_cache::SeedFromBytesOutcome::Upserted).unwrap_or(false);
crate::app_deprintln!(
"[analysis] cpu-seed worker: done track_id={} upserted={}",
tid_log,
ok
);
}
}
}
/// Submit full-buffer analysis; serializes with other producers. `high_priority` mirrors
/// HTTP backfill head insertion for the currently playing track.
///
/// Emits `analysis:waveform-updated` once here when the DB row is ready (Upserted or cache hit),
/// so `audio` and other callers do not duplicate IPC.
pub(crate) async fn submit_analysis_cpu_seed(
app: tauri::AppHandle,
track_id: String,
bytes: Vec<u8>,
high_priority: bool,
) -> Result<analysis_cache::SeedFromBytesOutcome, String> {
let shared = analysis_cpu_seed_shared(&app);
let rx = {
let mut st = shared.state.lock().unwrap_or_else(|e| e.into_inner());
let (kind, rx) = st.enqueue(track_id.clone(), bytes, high_priority);
crate::app_deprintln!("[analysis] cpu-seed submit: kind={kind:?} high_priority={high_priority}");
drop(st);
shared.ping_worker();
rx
};
let outcome = match rx.await {
Ok(res) => res?,
Err(_) => return Err("cpu-seed: result channel dropped".to_string()),
};
if matches!(
outcome,
analysis_cache::SeedFromBytesOutcome::Upserted
| analysis_cache::SeedFromBytesOutcome::SkippedWaveformCacheHit
) {
let _ = app.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: track_id.clone(),
is_partial: false,
},
);
}
Ok(outcome)
} }
/// Holds the live system-tray icon handle. `None` means the tray is currently hidden/removed. /// Holds the live system-tray icon handle. `None` means the tray is currently hidden/removed.
@@ -1314,6 +1729,15 @@ fn analysis_enqueue_seed_from_url(
if track_id.trim().is_empty() || url.trim().is_empty() { if track_id.trim().is_empty() || url.trim().is_empty() {
return Ok(()); return Ok(());
} }
if let Some(engine) = app.try_state::<crate::audio::AudioEngine>() {
if crate::audio::ranged_loudness_backfill_should_defer(&engine, &track_id) {
crate::app_deprintln!(
"[analysis] backfill skip track_id={} reason=ranged_playback_will_seed",
track_id
);
return Ok(());
}
}
if let Some(cache) = app.try_state::<analysis_cache::AnalysisCache>() { if let Some(cache) = app.try_state::<analysis_cache::AnalysisCache>() {
if cache.get_latest_loudness_for_track(&track_id)?.is_some() { if cache.get_latest_loudness_for_track(&track_id)?.is_some() {
crate::app_deprintln!( crate::app_deprintln!(
@@ -1323,48 +1747,34 @@ fn analysis_enqueue_seed_from_url(
return Ok(()); return Ok(());
} }
} }
{ let tid_log = track_id.clone();
let mut inflight = analysis_backfill_inflight() let high_priority = analysis_backfill_is_current_track(&app, &track_id);
let shared = analysis_backfill_shared(&app);
let kind = {
let mut st = shared
.state
.lock() .lock()
.map_err(|_| "analysis backfill lock poisoned".to_string())?; .map_err(|_| "analysis backfill lock poisoned".to_string())?;
if inflight.contains(&track_id) { st.enqueue(track_id, url, high_priority)
return Ok(()); };
match kind {
AnalysisBackfillEnqueueKind::NewBack | AnalysisBackfillEnqueueKind::NewFront => {
shared.ping_worker();
crate::app_deprintln!(
"[analysis] backfill enqueued: track_id={} position={}",
tid_log,
if high_priority { "front" } else { "back" }
);
} }
inflight.insert(track_id.clone()); AnalysisBackfillEnqueueKind::ReorderedFront => {
shared.ping_worker();
crate::app_deprintln!(
"[analysis] backfill bumped to front (current track) track_id={}",
tid_log
);
}
AnalysisBackfillEnqueueKind::DuplicateSkipped | AnalysisBackfillEnqueueKind::RunningSkipped => {}
} }
let app_clone = app.clone();
tauri::async_runtime::spawn(async move {
crate::app_deprintln!("[analysis] backfill queued: {}", track_id);
let result = async {
let client = reqwest::Client::builder()
.user_agent(subsonic_wire_user_agent())
.timeout(std::time::Duration::from_secs(120))
.build()
.map_err(|e| e.to_string())?;
let response = client.get(&url).send().await.map_err(|e| e.to_string())?;
if !response.status().is_success() {
return Err(format!("HTTP {}", response.status().as_u16()));
}
let bytes = response.bytes().await.map_err(|e| e.to_string())?;
if bytes.is_empty() {
return Err("empty response".to_string());
}
let has_loudness = enqueue_analysis_seed(&app_clone, &track_id, &bytes)?;
Ok::<bool, String>(has_loudness)
}
.await;
match result {
Ok(has_loudness) => crate::app_deprintln!(
"[analysis] backfill ready: {} (has_loudness={})",
track_id,
has_loudness
),
Err(e) => crate::app_eprintln!("[analysis] backfill failed for {}: {}", track_id, e),
}
if let Ok(mut inflight) = analysis_backfill_inflight().lock() {
inflight.remove(&track_id);
}
});
Ok(()) Ok(())
} }
@@ -1388,42 +1798,29 @@ async fn stream_to_file(response: reqwest::Response, dest_path: &std::path::Path
Ok(()) Ok(())
} }
fn enqueue_analysis_seed(app: &tauri::AppHandle, track_id: &str, bytes: &[u8]) -> Result<bool, String> { async fn enqueue_analysis_seed(app: &tauri::AppHandle, track_id: &str, bytes: &[u8]) -> Result<bool, String> {
let outcome = analysis_cache::seed_from_bytes(app, track_id, bytes).map_err(|e| { let high = analysis_backfill_is_current_track(app, track_id);
let outcome = submit_analysis_cpu_seed(
app.clone(),
track_id.to_string(),
bytes.to_vec(),
high,
)
.await
.map_err(|e| {
crate::app_eprintln!("[analysis] failed to seed {}: {}", track_id, e); crate::app_eprintln!("[analysis] failed to seed {}: {}", track_id, e);
e e
})?; })?;
if outcome == analysis_cache::SeedFromBytesOutcome::Upserted {
let _ = app.emit(
"analysis:waveform-updated",
WaveformUpdatedPayload {
track_id: track_id.to_string(),
is_partial: false,
},
);
}
let has_loudness = app let has_loudness = app
.try_state::<analysis_cache::AnalysisCache>() .try_state::<analysis_cache::AnalysisCache>()
.and_then(|cache| cache.get_latest_loudness_for_track(track_id).ok().flatten()) .and_then(|cache| cache.get_latest_loudness_for_track(track_id).ok().flatten())
.is_some(); .is_some();
// SkippedConcurrent gets logged with the actual outcome so the line doesn't crate::app_deprintln!(
// read "seed result has_loudness=false" when in fact another path is mid-seed "[analysis] seed result track_id={} bytes={} has_loudness={} outcome={outcome:?}",
// and will publish the row in seconds. track_id,
if outcome == analysis_cache::SeedFromBytesOutcome::SkippedConcurrent { bytes.len(),
crate::app_deprintln!( has_loudness
"[analysis] seed deferred to in-flight peer track_id={} bytes={} has_loudness_now={}", );
track_id,
bytes.len(),
has_loudness
);
} else {
crate::app_deprintln!(
"[analysis] seed result track_id={} bytes={} has_loudness={}",
track_id,
bytes.len(),
has_loudness
);
}
Ok(has_loudness) Ok(has_loudness)
} }
@@ -1439,7 +1836,7 @@ async fn enqueue_analysis_seed_from_file(
if bytes.is_empty() { if bytes.is_empty() {
return; return;
} }
let _ = enqueue_analysis_seed(app, track_id, &bytes); let _ = enqueue_analysis_seed(app, track_id, &bytes).await;
} }
/// Downloads a single track to the app's offline cache directory. /// Downloads a single track to the app's offline cache directory.
@@ -2031,12 +2428,32 @@ async fn download_track_hot_cache(
.await .await
.map(|m| m.len()) .map(|m| m.len())
.unwrap_or(0); .unwrap_or(0);
crate::app_deprintln!(
"[hot-cache] download disk_hit track_id={} server_id={} bytes={}",
track_id,
server_id,
size
);
// Disk hit: still seed analysis, but do not block the command (full-file read); the
// prefetch worker runs invokes sequentially.
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;
});
return Ok(HotCacheDownloadResult { return Ok(HotCacheDownloadResult {
path: path_str, path: path_str,
size, size,
}); });
} }
crate::app_deprintln!(
"[hot-cache] download http_start track_id={} server_id={}",
track_id,
server_id
);
let client = reqwest::Client::builder() let client = reqwest::Client::builder()
.user_agent(subsonic_wire_user_agent()) .user_agent(subsonic_wire_user_agent())
.timeout(std::time::Duration::from_secs(120)) .timeout(std::time::Duration::from_secs(120))
@@ -2058,12 +2475,23 @@ async fn download_track_hot_cache(
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
enqueue_analysis_seed_from_file(&app, &track_id, &file_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) let size = tokio::fs::metadata(&file_path)
.await .await
.map(|m| m.len()) .map(|m| m.len())
.unwrap_or(0); .unwrap_or(0);
crate::app_deprintln!(
"[hot-cache] download http_done track_id={} server_id={} bytes={}",
track_id,
server_id,
size
);
Ok(HotCacheDownloadResult { Ok(HotCacheDownloadResult {
path: path_str, path: path_str,
size, size,
@@ -2096,12 +2524,30 @@ async fn promote_stream_cache_to_hot_cache(
.await .await
.map(|m| m.len()) .map(|m| m.len())
.unwrap_or(0); .unwrap_or(0);
crate::app_deprintln!(
"[hot-cache] promote disk_hit track_id={} server_id={} bytes={}",
track_id,
server_id,
size
);
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;
});
return Ok(Some(HotCacheDownloadResult { path: path_str, size })); return Ok(Some(HotCacheDownloadResult { path: path_str, size }));
} }
let bytes = match audio::take_stream_completed_for_url(&state, &url) { let bytes = match audio::take_stream_completed_for_url(&state, &url) {
Some(b) => b, Some(b) => b,
None => return Ok(None), None => {
crate::app_deprintln!(
"[hot-cache] promote skip track_id={} reason=no_completed_stream_for_url",
track_id
);
return Ok(None);
}
}; };
let part_path = file_path.with_extension(format!("{suffix}.part")); let part_path = file_path.with_extension(format!("{suffix}.part"));
@@ -2113,12 +2559,18 @@ async fn promote_stream_cache_to_hot_cache(
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
let _ = enqueue_analysis_seed(&app, &track_id, &bytes); let _ = enqueue_analysis_seed(&app, &track_id, &bytes).await;
let size = tokio::fs::metadata(&file_path) let size = tokio::fs::metadata(&file_path)
.await .await
.map(|m| m.len()) .map(|m| m.len())
.unwrap_or(0); .unwrap_or(0);
crate::app_deprintln!(
"[hot-cache] promote from_stream track_id={} server_id={} bytes={}",
track_id,
server_id,
size
);
Ok(Some(HotCacheDownloadResult { path: path_str, size })) Ok(Some(HotCacheDownloadResult { path: path_str, size }))
} }
@@ -2159,11 +2611,20 @@ async fn delete_hot_cache_track(
app: tauri::AppHandle, app: tauri::AppHandle,
) -> Result<(), String> { ) -> Result<(), String> {
let file_path = std::path::PathBuf::from(&local_path); let file_path = std::path::PathBuf::from(&local_path);
if file_path.exists() { let existed = file_path.exists();
if existed {
tokio::fs::remove_file(&file_path) tokio::fs::remove_file(&file_path)
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
} }
crate::app_deprintln!(
"[hot-cache] delete file existed={} path_suffix={}",
existed,
file_path
.file_name()
.and_then(|s| s.to_str())
.unwrap_or("?")
);
let boundary = resolve_hot_cache_root(custom_dir, &app)?; let boundary = resolve_hot_cache_root(custom_dir, &app)?;
@@ -2193,11 +2654,17 @@ async fn delete_hot_cache_track(
#[tauri::command] #[tauri::command]
async fn purge_hot_cache(custom_dir: Option<String>, app: tauri::AppHandle) -> Result<(), String> { async fn purge_hot_cache(custom_dir: Option<String>, app: tauri::AppHandle) -> Result<(), String> {
let dir = resolve_hot_cache_root(custom_dir, &app)?; let dir = resolve_hot_cache_root(custom_dir, &app)?;
if dir.exists() { let existed = dir.exists();
if existed {
tokio::fs::remove_dir_all(&dir) tokio::fs::remove_dir_all(&dir)
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
} }
crate::app_deprintln!(
"[hot-cache] purge root_existed={} dir={}",
existed,
dir.display()
);
Ok(()) Ok(())
} }
@@ -3886,6 +4353,9 @@ pub fn run() {
app.manage(cache); app.manage(cache);
} }
// Periodic analysis queue sizes (debug logging mode only).
tauri::async_runtime::spawn(analysis_queue_snapshot_loop());
// ── Custom title bar on Linux ───────────────────────────────── // ── Custom title bar on Linux ─────────────────────────────────
// Remove OS window decorations on all Linux so the React TitleBar // Remove OS window decorations on all Linux so the React TitleBar
// can take over. The frontend checks is_tiling_wm() to decide // can take over. The frontend checks is_tiling_wm() to decide
+114 -16
View File
@@ -10,6 +10,15 @@ import {
getDeferHotCachePrefetch, getDeferHotCachePrefetch,
} from './utils/hotCacheGate'; } from './utils/hotCacheGate';
/** Settings → Logging → Debug (`frontend_debug_log` → Rust stderr), same as normalization / lucky-mix. */
function hotCacheFrontendDebug(payload: Record<string, unknown>): void {
if (useAuthStore.getState().loggingMode !== 'debug') return;
void invoke('frontend_debug_log', {
scope: 'hot-cache',
message: JSON.stringify(payload),
}).catch(() => {});
}
/** How many upcoming queue tracks may be prefetched (only current + next are eviction-protected). */ /** How many upcoming queue tracks may be prefetched (only current + next are eviction-protected). */
const PREFETCH_AHEAD = 5; const PREFETCH_AHEAD = 5;
@@ -89,11 +98,21 @@ function scheduleEvictAfterPreviousGrace(): void {
function enqueueJobs(jobs: PrefetchJob[]) { function enqueueJobs(jobs: PrefetchJob[]) {
const seen = new Set(pendingQueue.map(j => `${j.serverId}:${j.trackId}`)); const seen = new Set(pendingQueue.map(j => `${j.serverId}:${j.trackId}`));
let merged = 0;
for (const j of jobs) { for (const j of jobs) {
const k = `${j.serverId}:${j.trackId}`; const k = `${j.serverId}:${j.trackId}`;
if (seen.has(k)) continue; if (seen.has(k)) continue;
seen.add(k); seen.add(k);
pendingQueue.push(j); pendingQueue.push(j);
merged++;
}
if (merged > 0) {
hotCacheFrontendDebug({
event: 'prefetch-queue-jobs',
added: merged,
pendingTotal: pendingQueue.length,
trackIds: jobs.map(j => j.trackId),
});
} }
void runWorker(); void runWorker();
} }
@@ -105,6 +124,11 @@ async function runWorker() {
while (pendingQueue.length > 0) { while (pendingQueue.length > 0) {
const auth = useAuthStore.getState(); const auth = useAuthStore.getState();
if (!auth.isLoggedIn || !auth.hotCacheEnabled || !auth.activeServerId) { if (!auth.isLoggedIn || !auth.hotCacheEnabled || !auth.activeServerId) {
hotCacheFrontendDebug({
event: 'prefetch-worker-stop',
reason: 'auth-disabled-or-logged-out',
clearedPending: pendingQueue.length,
});
pendingQueue.length = 0; pendingQueue.length = 0;
break; break;
} }
@@ -117,11 +141,24 @@ async function runWorker() {
if (!job) break; if (!job) break;
const maxBytes = Math.max(0, auth.hotCacheMaxMb) * 1024 * 1024; const maxBytes = Math.max(0, auth.hotCacheMaxMb) * 1024 * 1024;
if (maxBytes <= 0) continue; if (maxBytes <= 0) {
hotCacheFrontendDebug({ event: 'prefetch-skip-job', trackId: job.trackId, reason: 'max-mb-zero' });
continue;
}
const offline = useOfflineStore.getState(); const offline = useOfflineStore.getState();
if (offline.isDownloaded(job.trackId, job.serverId)) continue; if (offline.isDownloaded(job.trackId, job.serverId)) {
if (useHotCacheStore.getState().entries[entryKey(job.serverId, job.trackId)]) continue; hotCacheFrontendDebug({ event: 'prefetch-skip-job', trackId: job.trackId, reason: 'offline-library' });
continue;
}
if (useHotCacheStore.getState().entries[entryKey(job.serverId, job.trackId)]) {
hotCacheFrontendDebug({
event: 'prefetch-skip-job',
trackId: job.trackId,
reason: 'already-in-hot-index',
});
continue;
}
const player = usePlayerStore.getState(); const player = usePlayerStore.getState();
const { queue, queueIndex } = player; const { queue, queueIndex } = player;
@@ -130,19 +167,46 @@ async function runWorker() {
.slice(queueIndex + 1, queueIndex + 1 + PREFETCH_AHEAD) .slice(queueIndex + 1, queueIndex + 1 + PREFETCH_AHEAD)
.map(t => t.id), .map(t => t.id),
); );
if (!wantIds.has(job.trackId)) continue; if (!wantIds.has(job.trackId)) {
hotCacheFrontendDebug({
event: 'prefetch-skip-job',
trackId: job.trackId,
reason: 'not-in-upcoming-window',
queueIndex,
window: PREFETCH_AHEAD,
});
continue;
}
const track = queue.find(t => t.id === job.trackId); const track = queue.find(t => t.id === job.trackId);
if (!track) continue; if (!track) {
hotCacheFrontendDebug({
event: 'prefetch-skip-job',
trackId: job.trackId,
reason: 'track-not-in-queue',
});
continue;
}
const hotEntries = useHotCacheStore.getState().entries; const hotEntries = useHotCacheStore.getState().entries;
const occupied = sumCachedBytesInProtectedWindow(queue, queueIndex, job.serverId, hotEntries); const occupied = sumCachedBytesInProtectedWindow(queue, queueIndex, job.serverId, hotEntries);
const est = estimateTrackHotCacheBytes(track); const est = estimateTrackHotCacheBytes(track);
const isImmediateNext = queue[queueIndex + 1]?.id === job.trackId; const isImmediateNext = queue[queueIndex + 1]?.id === job.trackId;
if (!isImmediateNext && occupied + est > maxBytes) continue; if (!isImmediateNext && occupied + est > maxBytes) {
hotCacheFrontendDebug({
event: 'prefetch-skip-job',
trackId: job.trackId,
reason: 'budget-protected-window-plus-estimate',
occupied,
estimateBytes: est,
maxBytes,
});
continue;
}
const url = buildStreamUrl(job.trackId); const url = buildStreamUrl(job.trackId);
try { try {
const customDir = auth.hotCacheDownloadDir || null; const customDir = auth.hotCacheDownloadDir || null;
hotCacheFrontendDebug({ event: 'prefetch-invoke', trackId: job.trackId });
const res = await invoke<{ path: string; size: number }>('download_track_hot_cache', { const res = await invoke<{ path: string; size: number }>('download_track_hot_cache', {
trackId: job.trackId, trackId: job.trackId,
serverId: job.serverId, serverId: job.serverId,
@@ -150,7 +214,8 @@ async function runWorker() {
suffix: job.suffix, suffix: job.suffix,
customDir, customDir,
}); });
useHotCacheStore.getState().setEntry(job.trackId, job.serverId, res.path, res.size); useHotCacheStore.getState().setEntry(job.trackId, job.serverId, res.path, res.size, 'prefetch');
hotCacheFrontendDebug({ event: 'prefetch-stored', trackId: job.trackId, sizeBytes: res.size });
const fresh = usePlayerStore.getState(); const fresh = usePlayerStore.getState();
const authAfter = useAuthStore.getState(); const authAfter = useAuthStore.getState();
const maxAfter = Math.max(0, authAfter.hotCacheMaxMb) * 1024 * 1024; const maxAfter = Math.max(0, authAfter.hotCacheMaxMb) * 1024 * 1024;
@@ -161,8 +226,8 @@ async function runWorker() {
authAfter.activeServerId ?? '', authAfter.activeServerId ?? '',
authAfter.hotCacheDownloadDir || null, authAfter.hotCacheDownloadDir || null,
); );
} catch { } catch (e: unknown) {
/* network / HTTP — skip */ hotCacheFrontendDebug({ event: 'prefetch-download-failed', trackId: job.trackId, error: String(e) });
} }
} }
} finally { } finally {
@@ -199,30 +264,55 @@ async function replanNow() {
if (maxBytes <= 0) return; if (maxBytes <= 0) return;
const { queue, queueIndex, currentRadio } = usePlayerStore.getState(); const { queue, queueIndex, currentRadio } = usePlayerStore.getState();
if (currentRadio) return; if (currentRadio) {
hotCacheFrontendDebug({ event: 'replan-skip', reason: 'radio-mode' });
return;
}
const offline = useOfflineStore.getState(); const offline = useOfflineStore.getState();
const hot = useHotCacheStore.getState();
await hot.evictToFit(queue, queueIndex, maxBytes, serverId, customDir); await useHotCacheStore.getState().evictToFit(queue, queueIndex, maxBytes, serverId, customDir);
// Must read entries after eviction: the pre-evict snapshot still lists removed keys and would
// skip prefetch for upcoming tracks that no longer have on-disk rows.
const hotEntries = useHotCacheStore.getState().entries;
const targets = queue.slice(queueIndex + 1, queueIndex + 1 + PREFETCH_AHEAD); const targets = queue.slice(queueIndex + 1, queueIndex + 1 + PREFETCH_AHEAD);
const immediateNextId = queue[queueIndex + 1]?.id; const immediateNextId = queue[queueIndex + 1]?.id;
let projectedOccupied = sumCachedBytesInProtectedWindow(queue, queueIndex, serverId, hot.entries); let projectedOccupied = sumCachedBytesInProtectedWindow(queue, queueIndex, serverId, hotEntries);
const jobs: PrefetchJob[] = []; const jobs: PrefetchJob[] = [];
const skipped: { trackId: string; reason: string }[] = [];
for (const t of targets) { for (const t of targets) {
if (offline.isDownloaded(t.id, serverId)) continue; if (offline.isDownloaded(t.id, serverId)) {
if (hot.entries[entryKey(serverId, t.id)]) continue; skipped.push({ trackId: t.id, reason: 'offline-library' });
continue;
}
if (hotEntries[entryKey(serverId, t.id)]) {
skipped.push({ trackId: t.id, reason: 'already-in-hot-index' });
continue;
}
const isImmediateNext = t.id === immediateNextId; const isImmediateNext = t.id === immediateNextId;
if (isImmediateNext) { if (isImmediateNext) {
jobs.push({ trackId: t.id, serverId, suffix: t.suffix || 'mp3' }); jobs.push({ trackId: t.id, serverId, suffix: t.suffix || 'mp3' });
continue; continue;
} }
const est = estimateTrackHotCacheBytes(t); const est = estimateTrackHotCacheBytes(t);
if (projectedOccupied + est > maxBytes) break; if (projectedOccupied + est > maxBytes) {
skipped.push({ trackId: t.id, reason: 'budget-cap-rest-deferred' });
break;
}
projectedOccupied += est; projectedOccupied += est;
jobs.push({ trackId: t.id, serverId, suffix: t.suffix || 'mp3' }); jobs.push({ trackId: t.id, serverId, suffix: t.suffix || 'mp3' });
} }
hotCacheFrontendDebug({
event: 'replan',
queueIndex,
aheadCount: targets.length,
scheduledIds: jobs.map(j => j.trackId),
skipped,
projectedOccupiedBytes: projectedOccupied,
maxBytes,
});
enqueueJobs(jobs); enqueueJobs(jobs);
} }
@@ -266,6 +356,7 @@ export function initHotCachePrefetch(): () => void {
} }
if (!state.hotCacheEnabled || !state.isLoggedIn) { if (!state.hotCacheEnabled || !state.isLoggedIn) {
hotCacheFrontendDebug({ event: 'prefetch-auth-off', clearedPending: pendingQueue.length });
pendingQueue.length = 0; pendingQueue.length = 0;
clearHotCachePreviousGrace(); clearHotCachePreviousGrace();
return; return;
@@ -286,6 +377,13 @@ export function initHotCachePrefetch(): () => void {
if (budgetSettingsChanged) { if (budgetSettingsChanged) {
if (prev && state.hotCacheMaxMb < prev.hotCacheMaxMb) { if (prev && state.hotCacheMaxMb < prev.hotCacheMaxMb) {
hotCacheFrontendDebug({
event: 'prefetch-pending-cleared',
reason: 'hot-cache-max-mb-decreased',
prevMb: prev.hotCacheMaxMb,
nextMb: state.hotCacheMaxMb,
droppedJobs: pendingQueue.length,
});
pendingQueue.length = 0; pendingQueue.length = 0;
} }
void replanNow(); void replanNow();
+90 -4
View File
@@ -20,7 +20,13 @@ interface HotCacheState {
/** Persisted map `${serverId}:${trackId}` → file meta */ /** Persisted map `${serverId}:${trackId}` → file meta */
entries: Record<string, HotCacheEntry>; entries: Record<string, HotCacheEntry>;
getLocalUrl: (trackId: string, serverId: string) => string | null; getLocalUrl: (trackId: string, serverId: string) => string | null;
setEntry: (trackId: string, serverId: string, localPath: string, sizeBytes: number) => void; setEntry: (
trackId: string,
serverId: string,
localPath: string,
sizeBytes: number,
debugSource?: string,
) => void;
/** Bump LRU when the user actually plays this track (if it is in the hot cache). */ /** Bump LRU when the user actually plays this track (if it is in the hot cache). */
touchPlayed: (trackId: string, serverId: string) => void; touchPlayed: (trackId: string, serverId: string) => void;
removeEntry: (trackId: string, serverId: string) => void; removeEntry: (trackId: string, serverId: string) => void;
@@ -51,6 +57,29 @@ function lruStamp(meta: HotCacheEntry | undefined): number {
return meta.lastPlayedAt ?? meta.cachedAt ?? 0; return meta.lastPlayedAt ?? meta.cachedAt ?? 0;
} }
function evictionReasonForTier(tier: number): string {
const labels: Record<number, string> = {
0: 'inactive-server',
1: 'not-in-queue',
2: 'ahead-of-protected-window',
3: 'behind-current-in-queue',
};
return labels[tier] ?? `tier-${tier}`;
}
/** Settings → Logging → Debug, same as `emitNormalizationDebug` / lucky-mix. Dynamic `authStore` import avoids a static cycle (auth → player → hot-cache). */
function hotCacheFrontendDebug(payload: Record<string, unknown>): void {
void import('./authStore')
.then(({ useAuthStore }) => {
if (useAuthStore.getState().loggingMode !== 'debug') return;
return invoke('frontend_debug_log', {
scope: 'hot-cache',
message: JSON.stringify(payload),
});
})
.catch(() => {});
}
export const useHotCacheStore = create<HotCacheState>()( export const useHotCacheStore = create<HotCacheState>()(
persist( persist(
(set, get) => ({ (set, get) => ({
@@ -62,7 +91,7 @@ export const useHotCacheStore = create<HotCacheState>()(
return `psysonic-local://${e.localPath}`; return `psysonic-local://${e.localPath}`;
}, },
setEntry: (trackId, serverId, localPath, sizeBytes) => { setEntry: (trackId, serverId, localPath, sizeBytes, debugSource) => {
const now = Date.now(); const now = Date.now();
set(s => ({ set(s => ({
entries: { entries: {
@@ -75,6 +104,13 @@ export const useHotCacheStore = create<HotCacheState>()(
}, },
}, },
})); }));
hotCacheFrontendDebug({
event: 'index-add',
trackId,
serverId,
sizeBytes,
source: debugSource ?? 'unknown',
});
}, },
touchPlayed: (trackId, serverId) => { touchPlayed: (trackId, serverId) => {
@@ -97,6 +133,12 @@ export const useHotCacheStore = create<HotCacheState>()(
delete next[entryKey(serverId, trackId)]; delete next[entryKey(serverId, trackId)];
return { entries: next }; return { entries: next };
}); });
hotCacheFrontendDebug({
event: 'index-remove',
trackId,
serverId,
reason: 'explicit-removeEntry',
});
emitAnalysisStorageChanged({ trackId, reason: 'hotcache-delete' }); emitAnalysisStorageChanged({ trackId, reason: 'hotcache-delete' });
}, },
@@ -157,8 +199,31 @@ export const useHotCacheStore = create<HotCacheState>()(
return a.lru - b.lru; return a.lru - b.lru;
}); });
for (const { key } of cands) { if (cands.length === 0) {
hotCacheFrontendDebug({
event: 'evict-no-candidates',
sumBytes: sum,
maxBytes,
queueIndex,
entryKeys: keys.length,
reason: 'all-protected-or-grace-or-parse-fail',
});
return;
}
hotCacheFrontendDebug({
event: 'evict-start',
sumBytes: sum,
maxBytes,
queueIndex,
protectLo,
protectHi,
candidateCount: cands.length,
});
for (const cand of cands) {
if (sum <= maxBytes) break; if (sum <= maxBytes) break;
const { key, tier } = cand;
const meta = entries[key]; const meta = entries[key];
if (!meta) continue; if (!meta) continue;
const parsed = parseKey(key); const parsed = parseKey(key);
@@ -166,7 +231,24 @@ export const useHotCacheStore = create<HotCacheState>()(
await invoke('delete_hot_cache_track', { await invoke('delete_hot_cache_track', {
localPath: meta.localPath, localPath: meta.localPath,
customDir: hotCacheCustomDir || null, customDir: hotCacheCustomDir || null,
}).catch(() => {}); }).catch((e: unknown) => {
hotCacheFrontendDebug({
event: 'evict-disk-delete-failed',
trackId: parsed.trackId,
serverId: parsed.serverId,
error: String(e),
});
});
hotCacheFrontendDebug({
event: 'evict-remove',
trackId: parsed.trackId,
serverId: parsed.serverId,
reason: `budget:${evictionReasonForTier(tier)}`,
tier,
bytes: meta.sizeBytes,
sumBytesAfter: sum - (meta.sizeBytes || 0),
maxBytes,
});
sum -= meta.sizeBytes || 0; sum -= meta.sizeBytes || 0;
delete entries[key]; delete entries[key];
emitAnalysisStorageChanged({ trackId: parsed.trackId, reason: 'hotcache-delete' }); emitAnalysisStorageChanged({ trackId: parsed.trackId, reason: 'hotcache-delete' });
@@ -176,6 +258,10 @@ export const useHotCacheStore = create<HotCacheState>()(
}, },
clearAllDisk: async (customDir: string | null) => { clearAllDisk: async (customDir: string | null) => {
hotCacheFrontendDebug({
event: 'purge-all',
customDir: customDir && customDir.length > 0 ? '(custom)' : 'default',
});
await invoke('purge_hot_cache', { customDir: customDir || null }).catch(() => {}); await invoke('purge_hot_cache', { customDir: customDir || null }).catch(() => {});
set({ entries: {} }); set({ entries: {} });
emitAnalysisStorageChanged({ trackId: null, reason: 'hotcache-purge' }); emitAnalysisStorageChanged({ trackId: null, reason: 'hotcache-purge' });
+36 -29
View File
@@ -5,6 +5,7 @@ import { listen } from '@tauri-apps/api/event';
import { showToast } from '../utils/toast'; import { showToast } from '../utils/toast';
import { buildCoverArtUrl, buildStreamUrl, 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, streamUrlTrackId, getPlaybackSourceKind, type PlaybackSourceKind } from '../utils/resolvePlaybackUrl'; import { resolvePlaybackUrl, streamUrlTrackId, getPlaybackSourceKind, type PlaybackSourceKind } from '../utils/resolvePlaybackUrl';
import { redactSubsonicUrlForLog } from '../utils/redactSubsonicUrl';
import { setDeferHotCachePrefetch } from '../utils/hotCacheGate'; import { setDeferHotCachePrefetch } from '../utils/hotCacheGate';
import { lastfmScrobble, lastfmUpdateNowPlaying, lastfmLoveTrack, lastfmUnloveTrack, lastfmGetTrackLoved, lastfmGetAllLovedTracks } from '../api/lastfm'; import { lastfmScrobble, lastfmUpdateNowPlaying, lastfmLoveTrack, lastfmUnloveTrack, lastfmGetTrackLoved, lastfmGetAllLovedTracks } from '../api/lastfm';
import { useAuthStore } from './authStore'; import { useAuthStore } from './authStore';
@@ -608,8 +609,16 @@ function touchHotCacheOnPlayback(trackId: string, serverId: string) {
useHotCacheStore.getState().touchPlayed(trackId, serverId); useHotCacheStore.getState().touchPlayed(trackId, serverId);
} }
/** Coalesce concurrent `analysis_get_waveform_for_track` for one id (StrictMode double mount, init + queue restore). */ /** Last-write-wins generation per track: avoids applying a stale empty waveform read when
const waveformRefreshInflight = new Map<string, Promise<void>>(); * `analysis:waveform-updated` bumps gen after SQLite commit while an older `analysis_get_waveform_for_track`
* is still in flight. Gen is bumped only on explicit invalidation (waveform-updated, analysis storage),
* not on every `refreshWaveformForTrack` call otherwise bursts (Lucky Mix, queue) cancel each other. */
const waveformRefreshGenByTrackId: Record<string, number> = {};
function bumpWaveformRefreshGen(trackId: string) {
if (!trackId) return;
waveformRefreshGenByTrackId[trackId] = (waveformRefreshGenByTrackId[trackId] ?? 0) + 1;
}
/** Coalesce concurrent `analysis_get_loudness_for_track` for one id+mode pair. The /** Coalesce concurrent `analysis_get_loudness_for_track` for one id+mode pair. The
* analysis:waveform-updated listener fires refreshWaveform + refreshLoudness in * analysis:waveform-updated listener fires refreshWaveform + refreshLoudness in
@@ -740,33 +749,25 @@ async function reseedLoudnessForTrackId(trackId: string) {
async function refreshWaveformForTrack(trackId: string) { async function refreshWaveformForTrack(trackId: string) {
if (!trackId) return; if (!trackId) return;
let job = waveformRefreshInflight.get(trackId); const gen = waveformRefreshGenByTrackId[trackId] ?? 0;
if (!job) { try {
const p = (async () => { const row = await invoke<WaveformCachePayload | null>('analysis_get_waveform_for_track', { trackId });
try { if ((waveformRefreshGenByTrackId[trackId] ?? 0) !== gen) return;
const row = await invoke<WaveformCachePayload | null>('analysis_get_waveform_for_track', { trackId }); // Never apply bins for a non-current track (e.g. gapless byte-preload fetches the neighbour).
// Never apply bins for a non-current track (e.g. gapless byte-preload fetches the neighbour). if (usePlayerStore.getState().currentTrack?.id !== trackId) return;
if (usePlayerStore.getState().currentTrack?.id !== trackId) return; const bins = row ? coerceWaveformBins(row.bins) : null;
const bins = row ? coerceWaveformBins(row.bins) : null; if (!bins || bins.length === 0) {
if (!bins || bins.length === 0) { usePlayerStore.setState({
usePlayerStore.setState({ waveformBins: null,
waveformBins: null, });
}); return;
return; }
} usePlayerStore.setState({
usePlayerStore.setState({ waveformBins: bins,
waveformBins: bins,
});
} catch {
// best-effort; seekbar falls back to placeholder waveform
}
})();
job = p.finally(() => {
waveformRefreshInflight.delete(trackId);
}); });
waveformRefreshInflight.set(trackId, job); } catch {
// best-effort; seekbar falls back to placeholder waveform
} }
await job;
} }
/** When `syncPlayingEngine` is false, only update `cachedLoudnessGainByTrackId` (e.g. queue neighbour) — do not call `audio_update_replay_gain` for the already-playing track. */ /** When `syncPlayingEngine` is false, only update `cachedLoudnessGainByTrackId` (e.g. queue neighbour) — do not call `audio_update_replay_gain` for the already-playing track. */
@@ -805,7 +806,11 @@ async function runRefreshLoudnessForTrack(trackId: string, syncEngine: boolean):
analysisBackfillInFlightByTrackId[trackId] = true; analysisBackfillInFlightByTrackId[trackId] = true;
analysisBackfillAttemptsByTrackId[trackId] = attempts + 1; analysisBackfillAttemptsByTrackId[trackId] = attempts + 1;
const url = buildStreamUrl(trackId); const url = buildStreamUrl(trackId);
emitNormalizationDebug('backfill:enqueue', { trackId, url, attempt: attempts + 1 }); emitNormalizationDebug('backfill:enqueue', {
trackId,
url: redactSubsonicUrlForLog(url),
attempt: attempts + 1,
});
void invoke('analysis_enqueue_seed_from_url', { trackId, url }) void invoke('analysis_enqueue_seed_from_url', { trackId, url })
.then(() => emitNormalizationDebug('backfill:queued', { trackId, attempt: attempts + 1 })) .then(() => emitNormalizationDebug('backfill:queued', { trackId, attempt: attempts + 1 }))
.catch((e) => emitNormalizationDebug('backfill:error', { trackId, error: String(e) })) .catch((e) => emitNormalizationDebug('backfill:error', { trackId, error: String(e) }))
@@ -884,7 +889,7 @@ async function promoteCompletedStreamToHotCache(track: Track, serverId: string,
}, },
); );
if (!res || !res.path) return; if (!res || !res.path) return;
useHotCacheStore.getState().setEntry(track.id, serverId, res.path, res.size || 0); useHotCacheStore.getState().setEntry(track.id, serverId, res.path, res.size || 0, 'stream-promote');
} catch { } catch {
// best-effort promotion; normal hot-cache prefetch remains fallback // best-effort promotion; normal hot-cache prefetch remains fallback
} }
@@ -1300,6 +1305,7 @@ export function initAudioListeners(): () => void {
const currentRaw = usePlayerStore.getState().currentTrack?.id; const currentRaw = usePlayerStore.getState().currentTrack?.id;
const currentId = currentRaw ? normalizeAnalysisTrackId(currentRaw) : null; const currentId = currentRaw ? normalizeAnalysisTrackId(currentRaw) : null;
if (currentId && payloadTrackId === currentId) { if (currentId && payloadTrackId === currentId) {
bumpWaveformRefreshGen(currentRaw!);
void refreshWaveformForTrack(currentRaw!); void refreshWaveformForTrack(currentRaw!);
void refreshLoudnessForTrack(currentId); void refreshLoudnessForTrack(currentId);
emitNormalizationDebug('backfill:applied', { trackId: currentId }); emitNormalizationDebug('backfill:applied', { trackId: currentId });
@@ -1468,6 +1474,7 @@ export function initAudioListeners(): () => void {
const currentId = usePlayerStore.getState().currentTrack?.id; const currentId = usePlayerStore.getState().currentTrack?.id;
if (!currentId) return; if (!currentId) return;
if (detail.trackId && detail.trackId !== currentId) return; if (detail.trackId && detail.trackId !== currentId) return;
bumpWaveformRefreshGen(currentId);
void refreshWaveformForTrack(currentId); void refreshWaveformForTrack(currentId);
void refreshLoudnessForTrack(currentId); void refreshLoudnessForTrack(currentId);
}); });
+17
View File
@@ -0,0 +1,17 @@
/**
* Masks Subsonic wire-auth query params so debug logs are safe to copy.
* (`t` salt, `s` token hash, `p` password when present.)
*/
export function redactSubsonicUrlForLog(url: string): string {
if (!url || !url.includes('stream.view')) return url;
try {
const u = new URL(url);
// Placeholder must stay URL-safe (no `<>` — URLSearchParams percent-encodes them).
for (const k of ['t', 's', 'p'] as const) {
if (u.searchParams.has(k)) u.searchParams.set(k, 'REDACTED');
}
return u.toString();
} catch {
return url.replace(/([?&])(t|s|p)=([^&]*)/gi, (_m, sep: string, key: string) => `${sep}${key}=REDACTED`);
}
}