mirror of
https://github.com/Psychotoxical/psysonic.git
synced 2026-07-22 07:15:47 +00:00
fix(cover-backfill): kill idle CPU spin and offline-cache menu re-walks (#943)
* fix(cover-backfill): snapshot-diff worklist and live-tunable parallelism Aggressive cover backfill pegged one tokio worker at ~100% on large, fully-synced libraries while the download queues stayed empty. - Take two snapshots once per pass — the DB catalog (single GROUP BY) and the on-disk cover bucket (one directory walk) — and download the set-difference. No per-row `stat` syscalls and no re-scan loop; the empty cache case (heavy backfill) costs zero per-item disk hits. - Replace the front-loaded enumeration with a producer/consumer pipeline: the producer streams the catalog in chunks and feeds misses into a bounded channel; a fixed consumer pool keeps the download/encode pools saturated. - Make cover backfill parallelism runtime-tunable from the Performance Probe (threads slider + "Run full pass now"); HTTP download and CPU encode semaphores resize live. Not surfaced in app settings. - Add a "nothing changed" idle gate (catalog signature) so a settled pass is not re-run on every library:sync-idle, mirroring the analysis worker. - Cancel promptly on switch to lazy: consumers bail on enabled/focus change and the producer feeds via try_send so a full channel cannot deadlock. - Drop the per-item recursive disk walk from the ensure hot path. * fix(cover-backfill): cheap idle gate, settle on 404s, transient retries Follow-up to the snapshot-diff backfill: stop the periodic CPU spikes and the 89%-plateau wake storm on libraries whose covers can never reach 100%. - Idle gate is now disk-free: compare only the catalog COUNT(DISTINCT) instead of walking ~all cover dirs on every sync-idle. "Did the server change?" never touches the filesystem. Clear-cache commands re-arm the gate (rearm_idle_gate) since a clear leaves the catalog total unchanged, and the settings UI wakes the active server after a clear. - Settle the gate on any completed pass regardless of pending: remaining items are unfetchable-for-now (404), so the wake/sync-idle storm stops once the fetchable set is exhausted. - Stop auto-clearing .fetch-failed markers every pass (it defeated the 30-min backoff and re-attempted 404s forever). The manual "Run full pass now" sends force=true to clear them and retry; wake/sync-idle/configure stay opportunistic. - Rate-limit sync-idle passes (60s cooldown) as defence against chatty syncs. - Retry cover downloads up to 3x with backoff on transient failures (5xx / 429 / network), but never on a real 4xx so missing covers don't hammer the server. * fix(cover-cache): stop re-walking cover dirs from offline & cache menu The settings cover-cache section polled disk usage + progress every 15s for every server, each call doing a full recursive walk of the per-server cover directory. On a fully populated cache this caused periodic CPU spikes whenever that menu was open. - mod.rs: add a 10s TTL memo around the per-server cover dir walk (cached_dir_usage_for_server), shared by cover_cache_stats_server and library_cover_progress; invalidate on clear (per-server and clear-all). - CoverCacheStrategySection: recompute on entry only; rely on the cover:library-progress and cover:cache-cleared events for live updates; drop the per-cover cover:tier-ready refresh storm; turn the 15s loop into a 5-minute safety net. * fix(cover-backfill): keep emitting progress during the whole pass The producer finishes enumerating the worklist long before the consumer pool finishes downloading it, so progress was only emitted while feeding the channel — the "offline & cache" menu and overlay then froze through the entire drain phase. Replace the per-chunk emit with a 3s progress ticker that runs for the lifetime of the pass and is aborted once the consumers drain (final accurate emit still happens at settle). * docs(changelog): record cover-backfill idle-CPU fix (PR #943) Add [1.47.0] Fixed + Changed entries and a settingsCredits line for the cover-backfill idle CPU / offline & cache menu work.
This commit is contained in:
cucadmuh
GitHub
@@ -2,27 +2,50 @@
|
||||
|
||||
use super::{state, CoverCacheEnsureArgs, CoverCacheState};
|
||||
use psysonic_library::cover_backfill::{
|
||||
clear_cover_fetch_failures, collect_cover_backfill_batch, collect_cover_progress,
|
||||
LibraryCoverBackfillBatchDto, LIBRARY_COVER_CANONICAL_TIER,
|
||||
clear_cover_fetch_failures, collect_cover_progress, count_distinct_cover_ids,
|
||||
diff_missing_against_snapshot, fetch_all_catalog_rows, snapshot_cover_disk,
|
||||
LIBRARY_COVER_CANONICAL_TIER,
|
||||
};
|
||||
use psysonic_library::payload::LibrarySyncProgressPayload;
|
||||
use psysonic_library::repos::sync_state::SyncStateRepository;
|
||||
use psysonic_library::LibraryRuntime;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tauri::{AppHandle, Emitter, Listener, Manager};
|
||||
use tokio::sync::{Mutex, Semaphore};
|
||||
|
||||
use super::{count_cached_cover_ids, dir_usage_for_server};
|
||||
|
||||
/// Concurrent library downloads + encodes (hard cap — avoids saturating all CPU cores).
|
||||
const LIBRARY_BACKFILL_PARALLEL: usize = 2;
|
||||
const BATCH_SIZE: u32 = 24;
|
||||
const PENDING_RESTART_THRESHOLD: i64 = 32;
|
||||
/// Default concurrent library downloads + encodes. Runtime-tunable via the
|
||||
/// perf probe (`set_parallel`); the constant is only the startup value.
|
||||
const LIBRARY_BACKFILL_PARALLEL_DEFAULT: usize = 2;
|
||||
/// Bounds for the runtime knob — keep it sane so a stray value cannot DoS the
|
||||
/// host or starve the audio path.
|
||||
pub const LIBRARY_BACKFILL_PARALLEL_MIN: usize = 1;
|
||||
pub const LIBRARY_BACKFILL_PARALLEL_MAX: usize = 16;
|
||||
/// Raw catalog rows diffed per streaming chunk. Small enough that downloads
|
||||
/// start almost immediately after the one-shot enumeration, large enough to
|
||||
/// amortize the per-chunk `spawn_blocking` hop.
|
||||
const SCAN_CHUNK_ROWS: usize = 512;
|
||||
const SYNC_WAIT_MS: u64 = 5000;
|
||||
const PROGRESS_EVERY_BATCHES: u32 = 8;
|
||||
/// Cadence of the in-pass progress ticker (drives the "offline & cache" menu and
|
||||
/// the perf-probe overlay while a pass downloads). Only runs for the duration of
|
||||
/// an active pass.
|
||||
const PROGRESS_TICK_SECS: u64 = 3;
|
||||
/// Minimum gap between `library:sync-idle`-driven passes. Each such pass runs the
|
||||
/// idle-gate signature (a full cover-dir walk + DB count), so a chatty sync (e.g.
|
||||
/// periodic delta syncs) must not make that walk fire every few seconds. Manual
|
||||
/// runs and strategy-toggle wakes bypass this.
|
||||
const SYNC_IDLE_COOLDOWN_MS: u64 = 60_000;
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct CoverBackfillSession {
|
||||
@@ -33,6 +56,19 @@ pub struct CoverBackfillSession {
|
||||
pub password: String,
|
||||
}
|
||||
|
||||
/// Catalog signature captured when a full pass completes.
|
||||
///
|
||||
/// While it still matches, `library:sync-idle` must NOT re-trigger a rescan —
|
||||
/// mirrors the analysis coordinator's `completed_total` gate. Deliberately the
|
||||
/// **cheap** `COUNT(DISTINCT)` over the catalog only: a server change (track
|
||||
/// add/remove shifts `total`) re-arms the next pass, but checking it never
|
||||
/// touches the filesystem. A cover-cache clear leaves `total` unchanged, so the
|
||||
/// clear commands re-arm the gate explicitly (`rearm_idle_gate`).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
struct CoverIdleSignature {
|
||||
total: i64,
|
||||
}
|
||||
|
||||
pub struct CoverBackfillWorker {
|
||||
pub enabled: AtomicBool,
|
||||
/// When true, the active pass yields so visible-route cover IPC is not starved.
|
||||
@@ -41,6 +77,15 @@ pub struct CoverBackfillWorker {
|
||||
cursor: Mutex<String>,
|
||||
pass_running: AtomicBool,
|
||||
backfill_http: Arc<Semaphore>,
|
||||
/// Live download/encode concurrency for backfill passes. Mirrors the
|
||||
/// `backfill_http` permit count and gates per-batch `ensure_one` tasks.
|
||||
parallel: AtomicUsize,
|
||||
/// Set when a pass found nothing pending; suppresses idle-driven rescans
|
||||
/// until the catalog signature changes. `None` means "re-armed".
|
||||
settled: Mutex<Option<CoverIdleSignature>>,
|
||||
/// Epoch-ms of the last `sync-idle`-driven pass, to rate-limit the idle-gate
|
||||
/// disk walk against chatty syncs. 0 = never.
|
||||
last_sync_idle_ms: AtomicU64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
@@ -75,7 +120,10 @@ impl CoverBackfillWorker {
|
||||
session: Mutex::new(None),
|
||||
cursor: Mutex::new(String::new()),
|
||||
pass_running: AtomicBool::new(false),
|
||||
backfill_http: Arc::new(Semaphore::new(LIBRARY_BACKFILL_PARALLEL)),
|
||||
backfill_http: Arc::new(Semaphore::new(LIBRARY_BACKFILL_PARALLEL_DEFAULT)),
|
||||
parallel: AtomicUsize::new(LIBRARY_BACKFILL_PARALLEL_DEFAULT),
|
||||
settled: Mutex::new(None),
|
||||
last_sync_idle_ms: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,9 +131,49 @@ impl CoverBackfillWorker {
|
||||
self.ui_priority_hold.store(hold, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Re-arm the idle gate so the next opportunistic pass runs even though the
|
||||
/// catalog `total` is unchanged — used after a cover-cache clear, which
|
||||
/// drops files the cheap signature cannot see.
|
||||
pub async fn rearm_idle_gate(&self) {
|
||||
*self.settled.lock().await = None;
|
||||
self.last_sync_idle_ms.store(0, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Current backfill download/encode concurrency.
|
||||
pub fn parallel(&self) -> usize {
|
||||
self.parallel.load(Ordering::Relaxed).max(LIBRARY_BACKFILL_PARALLEL_MIN)
|
||||
}
|
||||
|
||||
/// Retune backfill concurrency at runtime. Resizes the shared HTTP permit
|
||||
/// pool to match (next batch picks up the new per-batch slot count). Returns
|
||||
/// the clamped value actually applied.
|
||||
pub fn set_parallel(&self, threads: usize) -> usize {
|
||||
let next = threads.clamp(LIBRARY_BACKFILL_PARALLEL_MIN, LIBRARY_BACKFILL_PARALLEL_MAX);
|
||||
let prev = self.parallel.swap(next, Ordering::SeqCst);
|
||||
if next > prev {
|
||||
self.backfill_http.add_permits(next - prev);
|
||||
} else if next < prev {
|
||||
// Shrinking: drain surplus permits as they free up so in-flight
|
||||
// fetches finish but no new ones start beyond the new cap.
|
||||
let sem = self.backfill_http.clone();
|
||||
let surplus = prev - next;
|
||||
tauri::async_runtime::spawn(async move {
|
||||
for _ in 0..surplus {
|
||||
if let Ok(permit) = sem.acquire().await {
|
||||
permit.forget();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
next
|
||||
}
|
||||
|
||||
pub async fn set_session(&self, enabled: bool, session: Option<CoverBackfillSession>) {
|
||||
self.enabled.store(enabled, Ordering::Relaxed);
|
||||
*self.session.lock().await = session;
|
||||
// Server switch or enable/disable invalidates any settled state: re-arm
|
||||
// so the next idle event runs a real pass for the new focus.
|
||||
*self.settled.lock().await = None;
|
||||
if !enabled {
|
||||
*self.cursor.lock().await = String::new();
|
||||
}
|
||||
@@ -97,7 +185,7 @@ impl CoverBackfillWorker {
|
||||
|
||||
/// Semaphore-backed library backfill HTTP slots (perf probe).
|
||||
pub fn pipeline_http_stats(&self) -> (u32, u32, bool) {
|
||||
let max = LIBRARY_BACKFILL_PARALLEL as u32;
|
||||
let max = self.parallel() as u32;
|
||||
let active = max.saturating_sub(self.backfill_http.available_permits() as u32);
|
||||
let pass_running = self.pass_running.load(Ordering::Relaxed);
|
||||
(max, active, pass_running)
|
||||
@@ -187,7 +275,7 @@ async fn ensure_one(
|
||||
let _ = CoverCacheState::ensure_inner(&st, &app, &args, Some(http_sem)).await;
|
||||
}
|
||||
|
||||
async fn run_full_pass(app: AppHandle, worker: Arc<CoverBackfillWorker>) {
|
||||
async fn run_full_pass(app: AppHandle, worker: Arc<CoverBackfillWorker>, force: bool) {
|
||||
if !worker.enabled.load(Ordering::Relaxed) {
|
||||
return;
|
||||
}
|
||||
@@ -201,6 +289,14 @@ async fn run_full_pass(app: AppHandle, worker: Arc<CoverBackfillWorker>) {
|
||||
None => return,
|
||||
};
|
||||
|
||||
// Opportunistic triggers (wake on track change, sync-idle) skip the whole
|
||||
// scan when a prior pass already settled and nothing changed — otherwise a
|
||||
// library with permanently-unfetchable covers (404s) would re-scan on every
|
||||
// wake forever. The manual "Run full pass now" sets `force` to bypass this.
|
||||
if !force && cover_idle_gate_should_skip(&app, &worker, &session).await {
|
||||
return;
|
||||
}
|
||||
|
||||
while !sync_allows_cover_backfill(&runtime.store, &session.library_server_id) {
|
||||
if !worker.enabled.load(Ordering::Relaxed) {
|
||||
return;
|
||||
@@ -220,125 +316,208 @@ async fn run_full_pass(app: AppHandle, worker: Arc<CoverBackfillWorker>) {
|
||||
|
||||
worker.reset_cursor().await;
|
||||
let http_sem = worker.backfill_http.clone();
|
||||
let mut batch_count = 0u32;
|
||||
|
||||
// A forced pass is an explicit user retry: drop the `.fetch-failed` backoff
|
||||
// markers and the settled gate so previously-404'd covers are attempted
|
||||
// again. Opportunistic passes leave the markers in place (30-min TTL).
|
||||
if force {
|
||||
*worker.settled.lock().await = None;
|
||||
let root2 = root.clone();
|
||||
let index_key2 = session.server_index_key.clone();
|
||||
let _ = tauri::async_runtime::spawn_blocking(move || {
|
||||
clear_cover_fetch_failures(&root2, &index_key2)
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
// Two snapshots, taken ONCE per pass: the DB catalog (single GROUP BY) and
|
||||
// the on-disk cover bucket (one directory walk). The delta = catalog minus
|
||||
// disk, streamed in chunks below. No per-row `stat` on the filesystem and no
|
||||
// re-scan loop — pure set math against the captured disk snapshot.
|
||||
let (raw_rows, snapshot) = {
|
||||
let store = runtime.store.clone();
|
||||
let lib_id = session.library_server_id.clone();
|
||||
let root_for_scan = root.clone();
|
||||
let index_key = session.server_index_key.clone();
|
||||
match tauri::async_runtime::spawn_blocking(move || {
|
||||
let rows = fetch_all_catalog_rows(&store, &lib_id)?;
|
||||
let snap = snapshot_cover_disk(&root_for_scan, &index_key);
|
||||
Ok::<_, String>((rows, snap))
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(Ok(pair)) => pair,
|
||||
_ => (Vec::new(), Default::default()),
|
||||
}
|
||||
};
|
||||
let snapshot = Arc::new(snapshot);
|
||||
|
||||
// Producer/consumer: a fixed pool of consumer tasks pulls misses off a
|
||||
// bounded channel and downloads them continuously, while the producer scans
|
||||
// the catalog in chunks and feeds misses in. This keeps the pool saturated
|
||||
// even when misses are sparse across chunks — no per-chunk drain barrier.
|
||||
// True concurrency stays governed by the resizable `http_sem` / encode
|
||||
// semaphores inside `ensure_one`, so the threads slider still applies live.
|
||||
let (tx, rx) =
|
||||
tokio::sync::mpsc::channel::<psysonic_library::cover_backfill::CoverBackfillItem>(256);
|
||||
let rx = Arc::new(Mutex::new(rx));
|
||||
let mut consumers = tokio::task::JoinSet::new();
|
||||
for _ in 0..LIBRARY_BACKFILL_PARALLEL_MAX {
|
||||
let rx = rx.clone();
|
||||
let st = st_arc.clone();
|
||||
let http_sem = http_sem.clone();
|
||||
let app = app.clone();
|
||||
let session = session.clone();
|
||||
let worker_arc = worker.clone();
|
||||
consumers.spawn(async move {
|
||||
loop {
|
||||
// Bail the moment the strategy flips to lazy / focus changes, so a
|
||||
// switch to "lazy" abandons the buffered backlog instead of
|
||||
// draining the whole channel (mirrors the producer's check).
|
||||
if !session_still_focused(&worker_arc, &session).await {
|
||||
break;
|
||||
}
|
||||
let item = {
|
||||
let mut guard = rx.lock().await;
|
||||
guard.recv().await
|
||||
};
|
||||
let Some(item) = item else { break };
|
||||
ensure_one(
|
||||
worker_arc.as_ref(),
|
||||
st.clone(),
|
||||
http_sem.clone(),
|
||||
app.clone(),
|
||||
session.clone(),
|
||||
item,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Progress ticker: the producer finishes enumerating the worklist long before
|
||||
// the consumers finish downloading it, so emitting only while feeding would
|
||||
// freeze the "offline & cache" menu through the whole drain phase. Tick a
|
||||
// periodic snapshot for the lifetime of the pass instead; aborted once the
|
||||
// consumers drain (a final accurate emit happens at settle below).
|
||||
let progress_ticker = {
|
||||
let app = app.clone();
|
||||
let store = runtime.store.clone();
|
||||
let root = root.clone();
|
||||
let session = session.clone();
|
||||
let lib_id = session.library_server_id.clone();
|
||||
let index_key = session.server_index_key.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(PROGRESS_TICK_SECS)).await;
|
||||
if let Ok((done, total, pending)) =
|
||||
progress_snapshot(&store, &root, &lib_id, &index_key).await
|
||||
{
|
||||
emit_library_progress(&app, &session, done, total, pending, &root).await;
|
||||
}
|
||||
}
|
||||
})
|
||||
};
|
||||
|
||||
let mut rows_iter = raw_rows.into_iter();
|
||||
let mut completed = false;
|
||||
loop {
|
||||
if !session_still_focused(&worker, &session).await {
|
||||
break;
|
||||
}
|
||||
|
||||
if worker.ui_priority_hold.load(Ordering::Relaxed) {
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
continue;
|
||||
}
|
||||
|
||||
let cursor = worker.cursor.lock().await.clone();
|
||||
let cursor_opt = if cursor.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(cursor)
|
||||
};
|
||||
let store = runtime.store.clone();
|
||||
let lib_id = session.library_server_id.clone();
|
||||
let index_key = session.server_index_key.clone();
|
||||
let root_for_batch = root.clone();
|
||||
let scan_chunk: Vec<_> = rows_iter.by_ref().take(SCAN_CHUNK_ROWS).collect();
|
||||
if scan_chunk.is_empty() {
|
||||
completed = true;
|
||||
break;
|
||||
}
|
||||
|
||||
let batch: Option<LibraryCoverBackfillBatchDto> =
|
||||
// Diff this chunk against the captured snapshot off-thread (in-memory set
|
||||
// math + DB expand only for rows not already cached) → misses to download.
|
||||
let missing: Vec<_> = {
|
||||
let store = runtime.store.clone();
|
||||
let lib_id = session.library_server_id.clone();
|
||||
let snapshot = snapshot.clone();
|
||||
match tauri::async_runtime::spawn_blocking(move || {
|
||||
collect_cover_backfill_batch(
|
||||
&store,
|
||||
&lib_id,
|
||||
&root_for_batch,
|
||||
&index_key,
|
||||
cursor_opt.as_deref(),
|
||||
Some(BATCH_SIZE),
|
||||
)
|
||||
diff_missing_against_snapshot(&store, &lib_id, &snapshot, scan_chunk)
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(Ok(b)) => Some(b),
|
||||
_ => None,
|
||||
};
|
||||
|
||||
let Some(batch) = batch else {
|
||||
break;
|
||||
Ok(Ok(missing)) => missing,
|
||||
_ => Vec::new(),
|
||||
}
|
||||
};
|
||||
|
||||
batch_count += 1;
|
||||
if !session_still_focused(&worker, &session).await {
|
||||
break;
|
||||
}
|
||||
let items = batch.items.clone();
|
||||
let mut paused_for_ui_priority = false;
|
||||
let batch_slots = Arc::new(Semaphore::new(LIBRARY_BACKFILL_PARALLEL));
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for item in items {
|
||||
if worker.ui_priority_hold.load(Ordering::Relaxed) {
|
||||
paused_for_ui_priority = true;
|
||||
break;
|
||||
}
|
||||
let st = st_arc.clone();
|
||||
let http_sem = http_sem.clone();
|
||||
let app = app.clone();
|
||||
let session = session.clone();
|
||||
let worker_arc = worker.clone();
|
||||
let batch_slots = batch_slots.clone();
|
||||
set.spawn(async move {
|
||||
let Ok(_slot) = batch_slots.acquire().await else {
|
||||
return;
|
||||
};
|
||||
ensure_one(worker_arc.as_ref(), st, http_sem, app, session, item).await;
|
||||
});
|
||||
}
|
||||
while set.join_next().await.is_some() {}
|
||||
if paused_for_ui_priority || worker.ui_priority_hold.load(Ordering::Relaxed) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if batch_count.is_multiple_of(PROGRESS_EVERY_BATCHES) {
|
||||
if let Ok((done, total, pending)) = progress_snapshot(
|
||||
&runtime.store,
|
||||
&root,
|
||||
&session.library_server_id,
|
||||
&session.server_index_key,
|
||||
)
|
||||
.await
|
||||
{
|
||||
emit_library_progress(&app, &session, done, total, pending, &root).await;
|
||||
}
|
||||
}
|
||||
|
||||
if batch.exhausted {
|
||||
worker.cursor.lock().await.clear();
|
||||
if let Ok((done, total, pending)) = progress_snapshot(
|
||||
&runtime.store,
|
||||
&root,
|
||||
&session.library_server_id,
|
||||
&session.server_index_key,
|
||||
)
|
||||
.await
|
||||
{
|
||||
if pending > PENDING_RESTART_THRESHOLD {
|
||||
let root3 = root.clone();
|
||||
let index_key3 = session.server_index_key.clone();
|
||||
let _ = tauri::async_runtime::spawn_blocking(move || {
|
||||
clear_cover_fetch_failures(&root3, &index_key3)
|
||||
})
|
||||
.await;
|
||||
// Focus-aware feed: never park indefinitely on a full channel, or a
|
||||
// switch to lazy (which stops the consumers) would deadlock the producer
|
||||
// here. `try_send` + a short retry lets us re-check focus and bail.
|
||||
let mut feed_closed = false;
|
||||
'feed: for mut item in missing {
|
||||
loop {
|
||||
match tx.try_send(item) {
|
||||
Ok(()) => break,
|
||||
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
|
||||
feed_closed = true;
|
||||
break 'feed;
|
||||
}
|
||||
Err(tokio::sync::mpsc::error::TrySendError::Full(returned)) => {
|
||||
if !session_still_focused(&worker, &session).await {
|
||||
feed_closed = true;
|
||||
break 'feed;
|
||||
}
|
||||
item = returned;
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
}
|
||||
emit_library_progress(&app, &session, done, total, pending, &root).await;
|
||||
}
|
||||
}
|
||||
if feed_closed {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(next) = batch.next_cursor {
|
||||
*worker.cursor.lock().await = next;
|
||||
// Close the channel so consumers drain the remaining backlog and exit.
|
||||
drop(tx);
|
||||
while consumers.join_next().await.is_some() {}
|
||||
progress_ticker.abort();
|
||||
|
||||
// Only settle the idle gate on a natural finish (worklist drained), never on
|
||||
// a session-switch break — that belongs to the previous focus.
|
||||
if completed {
|
||||
worker.cursor.lock().await.clear();
|
||||
match progress_snapshot(
|
||||
&runtime.store,
|
||||
&root,
|
||||
&session.library_server_id,
|
||||
&session.server_index_key,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok((done, total, pending)) => {
|
||||
// Settle on a full scan regardless of `pending`: whatever is left
|
||||
// is unfetchable for now (404s with a fresh `.fetch-failed`
|
||||
// marker). The cheap `total` signature re-triggers a pass only if
|
||||
// the server catalog changes; a cache clear re-arms via the clear
|
||||
// command. This stops the wake storm on libraries whose covers
|
||||
// can never reach 100%.
|
||||
*worker.settled.lock().await = Some(CoverIdleSignature { total });
|
||||
emit_library_progress(&app, &session, done, total, pending, &root).await;
|
||||
}
|
||||
Err(_) => {
|
||||
*worker.settled.lock().await = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Start one full-catalog pass on the Tokio runtime (survives inactive webview).
|
||||
pub async fn try_schedule_full_pass(app: &AppHandle) -> bool {
|
||||
/// `force` bypasses the idle gate and clears fetch-failed backoff (explicit user
|
||||
/// retry); opportunistic callers (wake / sync-idle) pass `false`.
|
||||
pub async fn try_schedule_full_pass(app: &AppHandle, force: bool) -> bool {
|
||||
let worker = match app.try_state::<Arc<CoverBackfillWorker>>() {
|
||||
Some(w) => w.inner().clone(),
|
||||
None => return false,
|
||||
@@ -356,12 +535,44 @@ pub async fn try_schedule_full_pass(app: &AppHandle) -> bool {
|
||||
|
||||
let app = app.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
run_full_pass(app, worker.clone()).await;
|
||||
run_full_pass(app, worker.clone(), force).await;
|
||||
worker.pass_running.store(false, Ordering::SeqCst);
|
||||
});
|
||||
true
|
||||
}
|
||||
|
||||
/// Cheap catalog signature for the idle gate: a single `COUNT(DISTINCT)` over
|
||||
/// the cover catalog. No filesystem access — checking "did anything change on
|
||||
/// the server?" must never walk the on-disk cover cache.
|
||||
async fn current_cover_signature(
|
||||
app: &AppHandle,
|
||||
session: &CoverBackfillSession,
|
||||
) -> Option<CoverIdleSignature> {
|
||||
let runtime = app.try_state::<LibraryRuntime>()?;
|
||||
let store = runtime.store.clone();
|
||||
let lib_id = session.library_server_id.clone();
|
||||
tauri::async_runtime::spawn_blocking(move || {
|
||||
count_distinct_cover_ids(&store, &lib_id)
|
||||
.ok()
|
||||
.map(|total| CoverIdleSignature { total })
|
||||
})
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
}
|
||||
|
||||
/// True when the previous pass settled with nothing pending and the catalog
|
||||
/// still matches that signature — so an idle event need not rescan.
|
||||
async fn cover_idle_gate_should_skip(app: &AppHandle, worker: &CoverBackfillWorker, session: &CoverBackfillSession) -> bool {
|
||||
let Some(settled) = *worker.settled.lock().await else {
|
||||
return false;
|
||||
};
|
||||
match current_cover_signature(app, session).await {
|
||||
Some(current) => current == settled,
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn on_sync_idle(app: &AppHandle, payload: SyncIdlePayload) {
|
||||
if !payload.ok {
|
||||
return;
|
||||
@@ -382,7 +593,18 @@ fn on_sync_idle(app: &AppHandle, payload: SyncIdlePayload) {
|
||||
if !session_matches_server(&session, &payload.server_id) {
|
||||
return;
|
||||
}
|
||||
let _ = try_schedule_full_pass(&app).await;
|
||||
// Rate-limit sync-idle passes: each runs the idle-gate disk walk, so a
|
||||
// chatty sync must not trigger it every few seconds. The gate inside the
|
||||
// pass still skips the actual rescan when nothing changed.
|
||||
let now = now_ms();
|
||||
let last = worker.last_sync_idle_ms.load(Ordering::Relaxed);
|
||||
if last != 0 && now.saturating_sub(last) < SYNC_IDLE_COOLDOWN_MS {
|
||||
return;
|
||||
}
|
||||
worker.last_sync_idle_ms.store(now, Ordering::Relaxed);
|
||||
// Opportunistic: the gate (checked inside the pass) skips the rescan when
|
||||
// a prior pass settled and nothing changed (mirrors the analysis gate).
|
||||
let _ = try_schedule_full_pass(&app, false).await;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -399,7 +621,7 @@ pub fn setup_library_sync_idle_listener(app: &AppHandle) {
|
||||
|
||||
/// Legacy single-step API (optional diagnostics).
|
||||
pub async fn pulse_backfill(app: &AppHandle, _worker: &Arc<CoverBackfillWorker>) -> CoverBackfillPulseDto {
|
||||
if try_schedule_full_pass(app).await {
|
||||
if try_schedule_full_pass(app, false).await {
|
||||
return CoverBackfillPulseDto {
|
||||
scheduled: 0,
|
||||
exhausted: false,
|
||||
|
||||
Reference in New Issue
Block a user