fix(analysis): persist library backfill scan phase across coordinator ticks (#882)

* fix(analysis): persist backfill scan phase and cursor across coordinator ticks

Keep HashBpmGaps progress when the candidate SQL page is empty only because
id > cursor; store scan phase in the native worker so each tick does not
restart from Candidates and rescan the first ~10k ready tracks.

* docs: CHANGELOG and credits for PR #882 backfill scan phase fix

* fix(analysis): move backfill tests below production code for clippy

Clippy items_after_test_module requires all non-test items before mod tests.
This commit is contained in:
cucadmuh
2026-05-28 11:26:24 +03:00
committed by GitHub
parent b24a7fc5cb
commit 8004ec559c
5 changed files with 115 additions and 23 deletions
+8
View File
@@ -423,6 +423,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Analytics — advanced backfill scan no longer replays the first chunk
**By [@cucadmuh](https://github.com/cucadmuh), PR [#882](https://github.com/Psychotoxical/psysonic/pull/882)**
* **Settings → Library → Analytics → Advanced** on large libraries no longer stalls mid-pass when most early tracks are already analyzed: the native coordinator keeps hash/BPM gap scan phase and cursor across ticks instead of restarting from the first ids every cycle.
## [1.46.0] - 2026-05-18 ## [1.46.0] - 2026-05-18
> **🙏 Special thanks to [@zz5zz](https://github.com/zz5zz)** for his tireless quirk-spotting and bug reports on the [Psysonic Discord](https://discord.gg/AMnDRErm4u) — several of the polish fixes in this release landed directly off the back of his messages. > **🙏 Special thanks to [@zz5zz](https://github.com/zz5zz)** for his tireless quirk-spotting and bug reports on the [Psysonic Discord](https://discord.gg/AMnDRErm4u) — several of the polish fixes in this release landed directly off the back of his messages.
@@ -29,23 +29,64 @@ pub struct LibraryAnalysisProgressDto {
pub done_tracks: i64, pub done_tracks: i64,
} }
enum ScanMode { /// Persisted across native coordinator ticks (see `library_analysis_backfill` worker).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum AnalysisBackfillScanPhase {
#[default]
Candidates, Candidates,
/// Tracks with hash + BPM that may still need waveform/LUFS/enrichment gaps. /// Tracks with hash + BPM that may still need waveform/LUFS/enrichment gaps.
HashBpmGaps, HashBpmGaps,
} }
fn begin_hash_bpm_gap_scan() -> (ScanMode, Option<String>) { enum ScanMode {
Candidates,
HashBpmGaps,
}
impl From<AnalysisBackfillScanPhase> for ScanMode {
fn from(p: AnalysisBackfillScanPhase) -> Self {
match p {
AnalysisBackfillScanPhase::Candidates => ScanMode::Candidates,
AnalysisBackfillScanPhase::HashBpmGaps => ScanMode::HashBpmGaps,
}
}
}
impl From<ScanMode> for AnalysisBackfillScanPhase {
fn from(m: ScanMode) -> Self {
match m {
ScanMode::Candidates => AnalysisBackfillScanPhase::Candidates,
ScanMode::HashBpmGaps => AnalysisBackfillScanPhase::HashBpmGaps,
}
}
}
/// End of the cheap candidate SQL pass — start hash/BPM gap scan from the first id.
fn begin_hash_bpm_gap_scan_from_start() -> (ScanMode, Option<String>) {
(ScanMode::HashBpmGaps, None) (ScanMode::HashBpmGaps, None)
} }
/// Candidate page empty only because `id > cursor`; continue the id walk in hash/BPM phase.
fn begin_hash_bpm_gap_scan_from_cursor(cursor: String) -> (ScanMode, Option<String>) {
(ScanMode::HashBpmGaps, Some(cursor))
}
fn advance_after_empty_candidate_page(after: Option<String>) -> (ScanMode, Option<String>) {
match after {
None => begin_hash_bpm_gap_scan_from_start(),
Some(cursor) => begin_hash_bpm_gap_scan_from_cursor(cursor),
}
}
pub fn collect_analysis_backfill_batch( pub fn collect_analysis_backfill_batch(
app: &AppHandle, app: &AppHandle,
runtime: &LibraryRuntime, runtime: &LibraryRuntime,
server_id: &str, server_id: &str,
phase: AnalysisBackfillScanPhase,
cursor: Option<&str>, cursor: Option<&str>,
limit: Option<u32>, limit: Option<u32>,
) -> Result<LibraryAnalysisBackfillBatchDto, String> { ) -> Result<(LibraryAnalysisBackfillBatchDto, AnalysisBackfillScanPhase), String> {
let want = limit.unwrap_or(DEFAULT_BATCH).min(MAX_BATCH) as usize; let want = limit.unwrap_or(DEFAULT_BATCH).min(MAX_BATCH) as usize;
let needs_work = app let needs_work = app
.try_state::<TrackAnalysisNeedsWorkQuery>() .try_state::<TrackAnalysisNeedsWorkQuery>()
@@ -54,7 +95,7 @@ pub fn collect_analysis_backfill_batch(
let repo = TrackRepository::new(&runtime.store); let repo = TrackRepository::new(&runtime.store);
let mut found = Vec::with_capacity(want); let mut found = Vec::with_capacity(want);
let mut after = cursor.map(str::to_string); let mut after = cursor.map(str::to_string);
let mut mode = ScanMode::Candidates; let mut mode = ScanMode::from(phase);
let mut scanned = 0usize; let mut scanned = 0usize;
while found.len() < want && scanned < MAX_SCAN_IDS_PER_CALL { while found.len() < want && scanned < MAX_SCAN_IDS_PER_CALL {
@@ -70,15 +111,16 @@ pub fn collect_analysis_backfill_batch(
if page.is_empty() { if page.is_empty() {
match mode { match mode {
ScanMode::Candidates => { ScanMode::Candidates => {
(mode, after) = begin_hash_bpm_gap_scan(); (mode, after) = advance_after_empty_candidate_page(after);
continue; continue;
} }
ScanMode::HashBpmGaps => { ScanMode::HashBpmGaps => {
return Ok(LibraryAnalysisBackfillBatchDto { let dto = LibraryAnalysisBackfillBatchDto {
track_ids: found, track_ids: found,
next_cursor: after, next_cursor: after,
exhausted: true, exhausted: true,
}); };
return Ok((dto, AnalysisBackfillScanPhase::HashBpmGaps));
} }
} }
} }
@@ -105,24 +147,26 @@ pub fn collect_analysis_backfill_batch(
if page_len < SCAN_CHUNK { if page_len < SCAN_CHUNK {
match mode { match mode {
ScanMode::Candidates => { ScanMode::Candidates => {
(mode, after) = begin_hash_bpm_gap_scan(); (mode, after) = begin_hash_bpm_gap_scan_from_start();
} }
ScanMode::HashBpmGaps => { ScanMode::HashBpmGaps => {
return Ok(LibraryAnalysisBackfillBatchDto { let dto = LibraryAnalysisBackfillBatchDto {
track_ids: found, track_ids: found,
next_cursor: after, next_cursor: after,
exhausted: true, exhausted: true,
}); };
return Ok((dto, AnalysisBackfillScanPhase::HashBpmGaps));
} }
} }
} }
} }
Ok(LibraryAnalysisBackfillBatchDto { let dto = LibraryAnalysisBackfillBatchDto {
track_ids: found, track_ids: found,
next_cursor: after, next_cursor: after,
exhausted: false, exhausted: false,
}) };
Ok((dto, ScanMode::into(mode)))
} }
pub fn collect_analysis_progress( pub fn collect_analysis_progress(
@@ -170,3 +214,29 @@ pub fn collect_analysis_progress(
done_tracks: done, done_tracks: done,
}) })
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn empty_candidate_page_with_cursor_keeps_id_walk() {
let (mode, after) = advance_after_empty_candidate_page(Some("track-z".to_string()));
assert!(matches!(mode, ScanMode::HashBpmGaps));
assert_eq!(after.as_deref(), Some("track-z"));
}
#[test]
fn empty_candidate_page_without_cursor_starts_gap_scan_at_beginning() {
let (mode, after) = advance_after_empty_candidate_page(None);
assert!(matches!(mode, ScanMode::HashBpmGaps));
assert!(after.is_none());
}
#[test]
fn finished_candidate_phase_resets_gap_scan_cursor() {
let (mode, after) = begin_hash_bpm_gap_scan_from_start();
assert!(matches!(mode, ScanMode::HashBpmGaps));
assert!(after.is_none());
}
}
@@ -82,13 +82,15 @@ pub fn library_analysis_backfill_batch(
cursor: Option<String>, cursor: Option<String>,
limit: Option<u32>, limit: Option<u32>,
) -> Result<LibraryAnalysisBackfillBatchDto, String> { ) -> Result<LibraryAnalysisBackfillBatchDto, String> {
analysis_backfill::collect_analysis_backfill_batch( let (dto, _) = analysis_backfill::collect_analysis_backfill_batch(
&app, &app,
&runtime, &runtime,
server_id.trim(), server_id.trim(),
analysis_backfill::AnalysisBackfillScanPhase::Candidates,
cursor.as_deref().filter(|s| !s.is_empty()), cursor.as_deref().filter(|s| !s.is_empty()),
limit, limit,
) )?;
Ok(dto)
} }
#[tauri::command] #[tauri::command]
@@ -13,7 +13,8 @@ use psysonic_analysis::analysis_runtime::{
}; };
use psysonic_integration::subsonic::build_stream_view_url; use psysonic_integration::subsonic::build_stream_view_url;
use psysonic_library::analysis_backfill::{ use psysonic_library::analysis_backfill::{
collect_analysis_backfill_batch, collect_analysis_progress, LibraryAnalysisBackfillBatchDto, collect_analysis_backfill_batch, collect_analysis_progress, AnalysisBackfillScanPhase,
LibraryAnalysisBackfillBatchDto,
}; };
use psysonic_library::analysis_backfill_policy::{ use psysonic_library::analysis_backfill_policy::{
library_backfill_needs_top_up, library_backfill_top_up_limit, PipelineBacklogCounts, library_backfill_needs_top_up, library_backfill_top_up_limit, PipelineBacklogCounts,
@@ -49,6 +50,7 @@ pub struct LibraryAnalysisBackfillWorker {
pub enabled: AtomicBool, pub enabled: AtomicBool,
session: Mutex<Option<LibraryAnalysisBackfillSession>>, session: Mutex<Option<LibraryAnalysisBackfillSession>>,
cursor: Mutex<Option<String>>, cursor: Mutex<Option<String>>,
scan_phase: Mutex<AnalysisBackfillScanPhase>,
completed_total: Mutex<Option<i64>>, completed_total: Mutex<Option<i64>>,
exhausted_streak: Mutex<u32>, exhausted_streak: Mutex<u32>,
} }
@@ -59,6 +61,7 @@ impl LibraryAnalysisBackfillWorker {
enabled: AtomicBool::new(false), enabled: AtomicBool::new(false),
session: Mutex::new(None), session: Mutex::new(None),
cursor: Mutex::new(None), cursor: Mutex::new(None),
scan_phase: Mutex::new(AnalysisBackfillScanPhase::Candidates),
completed_total: Mutex::new(None), completed_total: Mutex::new(None),
exhausted_streak: Mutex::new(0), exhausted_streak: Mutex::new(0),
} }
@@ -69,6 +72,7 @@ impl LibraryAnalysisBackfillWorker {
*self.session.lock().await = session; *self.session.lock().await = session;
if !enabled { if !enabled {
*self.cursor.lock().await = None; *self.cursor.lock().await = None;
*self.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
*self.completed_total.lock().await = None; *self.completed_total.lock().await = None;
*self.exhausted_streak.lock().await = 0; *self.exhausted_streak.lock().await = 0;
} }
@@ -157,6 +161,7 @@ async fn coordinator_tick(
} }
*worker.completed_total.lock().await = None; *worker.completed_total.lock().await = None;
*worker.cursor.lock().await = None; *worker.cursor.lock().await = None;
*worker.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
*worker.exhausted_streak.lock().await = 0; *worker.exhausted_streak.lock().await = 0;
} }
@@ -196,10 +201,11 @@ async fn coordinator_tick(
} }
let cursor = worker.cursor.lock().await.clone(); let cursor = worker.cursor.lock().await.clone();
let phase = *worker.scan_phase.lock().await;
let app_for_batch = app.clone(); let app_for_batch = app.clone();
let lib_id = session.library_server_id.clone(); let lib_id = session.library_server_id.clone();
let batch: Option<LibraryAnalysisBackfillBatchDto> = tauri::async_runtime::spawn_blocking( let batch: Option<(LibraryAnalysisBackfillBatchDto, AnalysisBackfillScanPhase)> =
move || { tauri::async_runtime::spawn_blocking(move || {
let runtime = app_for_batch let runtime = app_for_batch
.try_state::<LibraryRuntime>() .try_state::<LibraryRuntime>()
.ok_or_else(|| "LibraryRuntime not available".to_string())?; .ok_or_else(|| "LibraryRuntime not available".to_string())?;
@@ -207,20 +213,21 @@ async fn coordinator_tick(
&app_for_batch, &app_for_batch,
&runtime, &runtime,
lib_id.trim(), lib_id.trim(),
phase,
cursor.as_deref().filter(|s| !s.is_empty()), cursor.as_deref().filter(|s| !s.is_empty()),
Some(fetch_limit), Some(fetch_limit),
) )
}, })
) .await
.await .ok()
.ok() .and_then(|r| r.ok());
.and_then(|r| r.ok());
let Some(batch) = batch else { let Some((batch, next_phase)) = batch else {
return CoordinatorTick { sleep_ms: TOP_UP_POLL_MS }; return CoordinatorTick { sleep_ms: TOP_UP_POLL_MS };
}; };
*worker.cursor.lock().await = batch.next_cursor.clone(); *worker.cursor.lock().await = batch.next_cursor.clone();
*worker.scan_phase.lock().await = next_phase;
let enqueued_count = batch.track_ids.len(); let enqueued_count = batch.track_ids.len();
let track_ids = batch.track_ids.clone(); let track_ids = batch.track_ids.clone();
@@ -271,6 +278,7 @@ async fn coordinator_tick(
*worker.completed_total.lock().await = Some(p.total_tracks); *worker.completed_total.lock().await = Some(p.total_tracks);
} }
*worker.cursor.lock().await = None; *worker.cursor.lock().await = None;
*worker.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
return CoordinatorTick { return CoordinatorTick {
sleep_ms: COMPLETED_RECHECK_MS, sleep_ms: COMPLETED_RECHECK_MS,
}; };
@@ -280,11 +288,13 @@ async fn coordinator_tick(
} }
if pending > 0 { if pending > 0 {
*worker.cursor.lock().await = None; *worker.cursor.lock().await = None;
*worker.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
return CoordinatorTick { return CoordinatorTick {
sleep_ms: EXHAUSTED_PENDING_RESCAN_MS, sleep_ms: EXHAUSTED_PENDING_RESCAN_MS,
}; };
} }
*worker.cursor.lock().await = None; *worker.cursor.lock().await = None;
*worker.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
return CoordinatorTick { return CoordinatorTick {
sleep_ms: EXHAUSTED_PAUSE_MS, sleep_ms: EXHAUSTED_PAUSE_MS,
}; };
@@ -321,6 +331,7 @@ fn on_sync_idle(app: &AppHandle, payload: SyncIdlePayload) {
return; return;
} }
*worker.cursor.lock().await = None; *worker.cursor.lock().await = None;
*worker.scan_phase.lock().await = AnalysisBackfillScanPhase::Candidates;
*worker.exhausted_streak.lock().await = 0; *worker.exhausted_streak.lock().await = 0;
}); });
} }
+1
View File
@@ -136,6 +136,7 @@ const CONTRIBUTOR_ENTRIES = [
'Artist detail: sort albums by year (newest/oldest) in the Albums section (PR #877)', 'Artist detail: sort albums by year (newest/oldest) in the Albums section (PR #877)',
'Cover art: Windows thumbnails, tier fallback, PNG decode, Subsonic coverArt id resolution (PR #878)', 'Cover art: Windows thumbnails, tier fallback, PNG decode, Subsonic coverArt id resolution (PR #878)',
'Analytics: native advanced library backfill coordinator — UI stays responsive on large libraries (PR #881)', 'Analytics: native advanced library backfill coordinator — UI stays responsive on large libraries (PR #881)',
'Analytics: library backfill scan phase/cursor persistence so advanced indexing can finish large libraries (PR #882)',
], ],
}, },
{ {