From 67b1dc790b127f98c13d53c762c6026d9edcec05 Mon Sep 17 00:00:00 2001 From: cucadmuh <49571317+cucadmuh@users.noreply.github.com> Date: Sat, 23 May 2026 00:58:37 +0300 Subject: [PATCH] fix(library): sweep orphan tracks after successful full resync (#861) * fix(library): sweep orphan tracks after successful full resync Full resync only upserted server tracks and left server-deleted rows live in SQLite. Add IS-7 mark-and-sweep via track.resync_gen: stamp rows on ingest during a re-sync, soft-delete unstamped rows when IS-6 completes. Delta tombstone reconcile is unchanged. * docs(release): CHANGELOG and credits for PR #861 resync orphan sweep * fix(library): satisfy clippy unnecessary_min_or_max in orphan sweep execute() returns usize; drop redundant max(0) so CI clippy -D warnings passes. --- CHANGELOG.md | 8 + .../migrations/007_resync_gen.sql | 4 + .../psysonic-library/src/repos/track.rs | 251 +++++++++++++++--- .../crates/psysonic-library/src/store.rs | 4 +- .../psysonic-library/src/sync/cursor.rs | 7 + .../psysonic-library/src/sync/initial.rs | 122 ++++++++- src/config/settingsCredits.ts | 1 + 7 files changed, 349 insertions(+), 48 deletions(-) create mode 100644 src-tauri/crates/psysonic-library/migrations/007_resync_gen.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index e914b2ac..ae3c50a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -215,6 +215,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 +### Local library index — full resync removes server-deleted tracks + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#861](https://github.com/Psychotoxical/psysonic/pull/861)** + +* **Settings → Library → Full resync** now soft-deletes local rows that no longer exist on the server after a successful re-sync (mark-and-sweep via `resync_gen`), so **Ready (N tracks)** no longer stays inflated when tracks were removed on Navidrome/Subsonic. Delta tombstone reconcile is unchanged. + + + ### Playlists & Favorites — column picker on short lists **By [@Psychotoxical](https://github.com/Psychotoxical), PR [#853](https://github.com/Psychotoxical/psysonic/pull/853)** diff --git a/src-tauri/crates/psysonic-library/migrations/007_resync_gen.sql b/src-tauri/crates/psysonic-library/migrations/007_resync_gen.sql new file mode 100644 index 00000000..ecac46dc --- /dev/null +++ b/src-tauri/crates/psysonic-library/migrations/007_resync_gen.sql @@ -0,0 +1,4 @@ +-- Full-resync orphan sweep (mark-and-sweep via generation stamp). +-- Rows ingested during a resync pass carry the active `resync_gen`; after +-- IS-6 succeeds, live rows with a stale generation are soft-deleted. +ALTER TABLE track ADD COLUMN resync_gen INTEGER NOT NULL DEFAULT 0; diff --git a/src-tauri/crates/psysonic-library/src/repos/track.rs b/src-tauri/crates/psysonic-library/src/repos/track.rs index 65709168..8df84807 100644 --- a/src-tauri/crates/psysonic-library/src/repos/track.rs +++ b/src-tauri/crates/psysonic-library/src/repos/track.rs @@ -79,58 +79,108 @@ impl<'a> TrackRepository<'a> { /// IS-3 initial-sync fast path: upsert rows only. Skips §6.9 remap /// detection and inline canonical linking — both run on delta sync /// or in a post-ingest canonical pass so 500-row batches stay fast. + /// + /// When `resync_gen` is `Some`, each row is stamped with that + /// generation so IS-7 can soft-delete stale rows after a successful + /// full resync. pub fn upsert_batch_initial_ingest(&self, rows: &[TrackRow]) -> Result<(), String> { - self.upsert_batch_initial_ingest_timed(rows).map(|_| ()) + self.upsert_batch_initial_ingest_timed(rows, None).map(|_| ()) } pub fn upsert_batch_initial_ingest_timed( &self, rows: &[TrackRow], + resync_gen: Option, ) -> Result { if rows.is_empty() { return Ok(WriteOpTiming::default()); } + let sql = match resync_gen { + Some(_) => UPSERT_INITIAL_RESYNC_SQL, + None => UPSERT_SQL, + }; let (_, timing) = self.store.with_conn_mut_timed("track.upsert_initial_ingest", |conn| { let tx = conn.transaction()?; - let mut upsert = tx.prepare_cached(UPSERT_SQL)?; + let mut upsert = tx.prepare_cached(sql)?; for r in rows { - upsert.execute(params![ - r.server_id, - r.id, - r.title, - r.title_sort, - r.artist, - r.artist_id, - r.album, - r.album_id, - r.album_artist, - r.duration_sec, - r.track_number, - r.disc_number, - r.year, - r.genre, - r.suffix, - r.bit_rate, - r.size_bytes, - r.cover_art_id, - r.starred_at, - r.user_rating, - r.play_count, - r.played_at, - r.server_path, - r.library_id, - r.isrc, - r.mbid_recording, - r.bpm, - r.replay_gain_track_db, - r.replay_gain_album_db, - r.content_hash, - r.server_updated_at, - r.server_created_at, - if r.deleted { 1_i64 } else { 0 }, - r.synced_at, - r.raw_json, - ])?; + if let Some(gen) = resync_gen { + upsert.execute(params![ + r.server_id, + r.id, + r.title, + r.title_sort, + r.artist, + r.artist_id, + r.album, + r.album_id, + r.album_artist, + r.duration_sec, + r.track_number, + r.disc_number, + r.year, + r.genre, + r.suffix, + r.bit_rate, + r.size_bytes, + r.cover_art_id, + r.starred_at, + r.user_rating, + r.play_count, + r.played_at, + r.server_path, + r.library_id, + r.isrc, + r.mbid_recording, + r.bpm, + r.replay_gain_track_db, + r.replay_gain_album_db, + r.content_hash, + r.server_updated_at, + r.server_created_at, + if r.deleted { 1_i64 } else { 0 }, + r.synced_at, + r.raw_json, + gen, + ])?; + } else { + upsert.execute(params![ + r.server_id, + r.id, + r.title, + r.title_sort, + r.artist, + r.artist_id, + r.album, + r.album_id, + r.album_artist, + r.duration_sec, + r.track_number, + r.disc_number, + r.year, + r.genre, + r.suffix, + r.bit_rate, + r.size_bytes, + r.cover_art_id, + r.starred_at, + r.user_rating, + r.play_count, + r.played_at, + r.server_path, + r.library_id, + r.isrc, + r.mbid_recording, + r.bpm, + r.replay_gain_track_db, + r.replay_gain_album_db, + r.content_hash, + r.server_updated_at, + r.server_created_at, + if r.deleted { 1_i64 } else { 0 }, + r.synced_at, + r.raw_json, + ])?; + } } drop(upsert); tx.commit()?; @@ -139,6 +189,30 @@ impl<'a> TrackRepository<'a> { Ok(timing) } + /// Next generation stamp for a full-resync orphan sweep on this server. + pub fn next_resync_gen(&self, server_id: &str) -> Result { + self.store.with_conn("misc", |c| { + c.query_row( + "SELECT COALESCE(MAX(resync_gen), 0) + 1 FROM track WHERE server_id = ?1", + params![server_id], + |r| r.get(0), + ) + }) + } + + /// IS-7 — soft-delete live rows not re-stamped during the active resync. + pub fn sweep_resync_orphans(&self, server_id: &str, resync_gen: i64) -> Result { + let now = now_unix_ms(); + let changed = self.store.with_conn_mut("misc", |c| { + c.execute( + "UPDATE track SET deleted = 1, synced_at = ?3 \ + WHERE server_id = ?1 AND deleted = 0 AND resync_gen != ?2", + params![server_id, resync_gen, now], + ) + })?; + Ok(changed as u32) + } + /// SELECT a single track by `(server_id, id)`. Returns `None` /// when missing or deleted (`deleted = 1`). Used by /// `library_get_track` and the offline-path command. @@ -514,6 +588,64 @@ ON CONFLICT(server_id, id) DO UPDATE SET raw_json = excluded.raw_json "#; +const UPSERT_INITIAL_RESYNC_SQL: &str = r#" +INSERT INTO track ( + server_id, id, title, title_sort, artist, artist_id, album, album_id, + album_artist, duration_sec, track_number, disc_number, year, genre, suffix, + bit_rate, size_bytes, cover_art_id, starred_at, user_rating, play_count, + played_at, server_path, library_id, isrc, mbid_recording, bpm, + replay_gain_track_db, replay_gain_album_db, content_hash, server_updated_at, + server_created_at, deleted, synced_at, raw_json, resync_gen +) VALUES ( + ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, + ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29, ?30, ?31, ?32, + ?33, ?34, ?35, ?36 +) +ON CONFLICT(server_id, id) DO UPDATE SET + title = excluded.title, + title_sort = excluded.title_sort, + artist = excluded.artist, + artist_id = excluded.artist_id, + album = excluded.album, + album_id = excluded.album_id, + album_artist = excluded.album_artist, + duration_sec = excluded.duration_sec, + track_number = excluded.track_number, + disc_number = excluded.disc_number, + year = excluded.year, + genre = excluded.genre, + suffix = excluded.suffix, + bit_rate = excluded.bit_rate, + size_bytes = excluded.size_bytes, + cover_art_id = excluded.cover_art_id, + starred_at = excluded.starred_at, + user_rating = excluded.user_rating, + play_count = excluded.play_count, + played_at = excluded.played_at, + server_path = excluded.server_path, + library_id = excluded.library_id, + isrc = excluded.isrc, + mbid_recording = excluded.mbid_recording, + bpm = excluded.bpm, + replay_gain_track_db = excluded.replay_gain_track_db, + replay_gain_album_db = excluded.replay_gain_album_db, + content_hash = COALESCE(NULLIF(excluded.content_hash, ''), track.content_hash), + server_updated_at = excluded.server_updated_at, + server_created_at = excluded.server_created_at, + deleted = 0, + synced_at = excluded.synced_at, + raw_json = excluded.raw_json, + resync_gen = excluded.resync_gen +"#; + +fn now_unix_ms() -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis().min(i64::MAX as u128) as i64) + .unwrap_or(0) +} + #[cfg(test)] mod tests { use super::*; @@ -558,6 +690,47 @@ mod tests { } } + #[test] + fn resync_upsert_stamps_generation_and_sweep_deletes_stale_rows() { + let store = LibraryStore::open_in_memory(); + let repo = TrackRepository::new(&store); + repo.upsert_batch_initial_ingest_timed(&[row("s1", "seen", "Seen")], Some(2)) + .unwrap(); + store + .with_conn_mut("misc", |c| { + c.execute( + "INSERT INTO track (server_id, id, title, album, duration_sec, deleted, synced_at, raw_json, resync_gen) \ + VALUES ('s1', 'orphan', 'Orphan', 'Al', 1, 0, 1, '{}', 1)", + [], + ) + }) + .unwrap(); + + assert_eq!(repo.sweep_resync_orphans("s1", 2).unwrap(), 1); + + let live: i64 = store + .with_conn("misc", |c| { + c.query_row( + "SELECT COUNT(*) FROM track WHERE server_id = 's1' AND deleted = 0", + [], + |r| r.get(0), + ) + }) + .unwrap(); + assert_eq!(live, 1); + + let orphan_deleted: i64 = store + .with_conn("misc", |c| { + c.query_row( + "SELECT deleted FROM track WHERE id = 'orphan'", + [], + |r| r.get(0), + ) + }) + .unwrap(); + assert_eq!(orphan_deleted, 1); + } + #[test] fn resync_does_not_clobber_playback_content_hash() { // E2 safety property: a sync (which passes content_hash = None) must diff --git a/src-tauri/crates/psysonic-library/src/store.rs b/src-tauri/crates/psysonic-library/src/store.rs index b5aa7b4f..5cd2f964 100644 --- a/src-tauri/crates/psysonic-library/src/store.rs +++ b/src-tauri/crates/psysonic-library/src/store.rs @@ -8,7 +8,7 @@ use tauri::Manager; /// Current head of the embedded migrations. Bump each time a new /// `migrations/NNN_*.sql` is added. -pub const LIBRARY_DB_SCHEMA_VERSION: i64 = 6; +pub const LIBRARY_DB_SCHEMA_VERSION: i64 = 7; /// Lowest applied schema version the current code can advance from purely /// additively. If a DB carries a version below this, the breaking-bump hook @@ -30,6 +30,7 @@ const MIGRATION_004_TRACK_TITLE_INDEX: &str = const MIGRATION_005_TRACK_GENRE_YEAR_INDEXES: &str = include_str!("../migrations/005_track_genre_year_indexes.sql"); const MIGRATION_006_PLAY_SESSION: &str = include_str!("../migrations/006_play_session.sql"); +const MIGRATION_007_RESYNC_GEN: &str = include_str!("../migrations/007_resync_gen.sql"); /// Embedded migrations. Ordered ascending by `version`; the runner sorts /// defensively before applying so the source order can stay readable. @@ -40,6 +41,7 @@ const MIGRATIONS: &[(i64, &str)] = &[ (4, MIGRATION_004_TRACK_TITLE_INDEX), (5, MIGRATION_005_TRACK_GENRE_YEAR_INDEXES), (6, MIGRATION_006_PLAY_SESSION), + (7, MIGRATION_007_RESYNC_GEN), ]; #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src-tauri/crates/psysonic-library/src/sync/cursor.rs b/src-tauri/crates/psysonic-library/src/sync/cursor.rs index eebcedcc..d72f7a32 100644 --- a/src-tauri/crates/psysonic-library/src/sync/cursor.rs +++ b/src-tauri/crates/psysonic-library/src/sync/cursor.rs @@ -29,6 +29,11 @@ pub struct InitialSyncCursor { /// future field add doesn't break old cursors. #[serde(default)] pub strategy_state: StrategyState, + /// Active full-resync generation for mark-and-sweep orphan cleanup + /// (IS-7). `Some(n)` when re-syncing an already-indexed server; + /// persisted so resume keeps the same generation. + #[serde(default)] + pub resync_gen: Option, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -73,6 +78,7 @@ impl InitialSyncCursor { library_scope, ingested_count: 0, strategy_state, + resync_gen: None, } } @@ -120,6 +126,7 @@ mod tests { library_scope: Some("lib-1".into()), ingested_count: 2500, strategy_state: StrategyState::LinearOffset { offset: 2500 }, + resync_gen: Some(2), }; let json = serde_json::to_value(&c).unwrap(); let back: InitialSyncCursor = serde_json::from_value(json.clone()).unwrap(); diff --git a/src-tauri/crates/psysonic-library/src/sync/initial.rs b/src-tauri/crates/psysonic-library/src/sync/initial.rs index 461a0b70..cff5f592 100644 --- a/src-tauri/crates/psysonic-library/src/sync/initial.rs +++ b/src-tauri/crates/psysonic-library/src/sync/initial.rs @@ -1,4 +1,4 @@ -//! `InitialSyncRunner` — spec §6.3 IS-1 … IS-6. PR-3b lands the runner, +//! `InitialSyncRunner` — spec §6.3 IS-1 … IS-7. PR-3b lands the runner, //! cursor persistence, and the N1/S1/S2 ingest loops. S3 (file-tree) //! is enumerated but returns `StrategyUnsupported`. IS-4 artist pass + //! IS-5 watermarks run after the bulk loop completes. @@ -29,6 +29,7 @@ use super::mapping::{navidrome_song_to_track_row, subsonic_song_to_track_row}; use super::progress::{IngestBatchMetrics, NoopProgress, Progress, ProgressEvent}; use super::strategy::IngestStrategy; use crate::bulk_ingest::{restore_track_secondary_indexes, suspend_track_secondary_indexes}; +use crate::dto::track_index_nonempty; use crate::repos::{RemapStats, SyncStateRepository, TrackRepository, TrackRow}; use crate::store::LibraryStore; use crate::store::WriteOpTiming; @@ -207,6 +208,7 @@ impl<'a> InitialSyncRunner<'a> { }); let mut cursor = self.load_or_init_cursor(&sync_state)?; + self.ensure_resync_generation(&mut cursor, &sync_state)?; let mut report = InitialSyncReport { strategy: Some(cursor.strategy.clone()), ingested_count: cursor.ingested_count, @@ -265,8 +267,19 @@ impl<'a> InitialSyncRunner<'a> { self.persist_cursor(&sync_state, &cursor)?; } - // IS-6 — phase=ready, clear cursor, stamp watermarks. + // IS-6 — phase=ready, optional IS-7 orphan sweep, clear cursor, stamp watermarks. let finished_at = now_unix_ms(); + if let Some(gen) = cursor.resync_gen { + let swept = TrackRepository::new(self.store) + .sweep_resync_orphans(&self.server_id, gen) + .map_err(SyncError::Storage)?; + if swept > 0 { + self.progress.emit(ProgressEvent::Tombstoned { + deleted_count: swept, + checked_count: swept, + }); + } + } let local_count = crate::dto::count_local_tracks(self.store, &self.server_id) .map_err(SyncError::Storage)?; sync_state @@ -419,9 +432,36 @@ impl<'a> InitialSyncRunner<'a> { } } - fn write_batch_timed(&self, rows: &[TrackRow]) -> Result { + fn ensure_resync_generation( + &self, + cursor: &mut InitialSyncCursor, + sync_state: &SyncStateRepository<'_>, + ) -> Result<(), SyncError> { + if cursor.resync_gen.is_some() { + return Ok(()); + } + let is_resync = sync_state + .has_last_full_sync_at(&self.server_id, &self.library_scope) + .map_err(SyncError::Storage)? + || track_index_nonempty(self.store, &self.server_id).map_err(SyncError::Storage)?; + if !is_resync { + return Ok(()); + } + let gen = TrackRepository::new(self.store) + .next_resync_gen(&self.server_id) + .map_err(SyncError::Storage)?; + cursor.resync_gen = Some(gen); + self.persist_cursor(sync_state, cursor)?; + Ok(()) + } + + fn write_batch_timed( + &self, + rows: &[TrackRow], + resync_gen: Option, + ) -> Result { TrackRepository::new(self.store) - .upsert_batch_initial_ingest_timed(rows) + .upsert_batch_initial_ingest_timed(rows, resync_gen) .map_err(SyncError::Storage) } @@ -430,8 +470,9 @@ impl<'a> InitialSyncRunner<'a> { rows: &[TrackRow], label: &str, offset: u32, + resync_gen: Option, ) -> Result<(RemapStats, WriteOpTiming), SyncError> { - let timing = self.write_batch_timed(rows)?; + let timing = self.write_batch_timed(rows, resync_gen)?; let total_ms = timing.total_ms(); if total_ms >= 500 { crate::app_eprintln!( @@ -667,7 +708,8 @@ impl<'a> InitialSyncRunner<'a> { ) }) .collect(); - let (_stats, _timing) = self.write_batch_logged(&rows, "N1", offset)?; + let (_stats, _timing) = + self.write_batch_logged(&rows, "N1", offset, ctx.cursor.resync_gen)?; ctx.report.ingested_count = ctx.report.ingested_count.saturating_add(rows.len() as u32); let next_offset = offset.saturating_add(self.batch_size); @@ -923,7 +965,8 @@ impl<'a> InitialSyncRunner<'a> { )); } let row_count = rows.len() as u32; - let (_stats, write_timing) = self.write_batch_logged(&rows, "S1", offset)?; + let (_stats, write_timing) = + self.write_batch_logged(&rows, "S1", offset, ctx.cursor.resync_gen)?; ctx.report.ingested_count = ctx.report.ingested_count.saturating_add(row_count); let next_offset = offset.saturating_add(self.batch_size); @@ -1094,7 +1137,8 @@ impl<'a> InitialSyncRunner<'a> { )); } if !rows.is_empty() { - let (_stats, _timing) = self.write_batch_logged(&rows, "S2", album_offset)?; + let (_stats, _timing) = + self.write_batch_logged(&rows, "S2", album_offset, cursor.resync_gen)?; report.ingested_count = report .ingested_count .saturating_add(rows.len() as u32); @@ -1367,6 +1411,68 @@ mod tests { assert_eq!(count, 7); } + #[tokio::test(flavor = "multi_thread")] + async fn full_resync_sweeps_orphans_not_seen_in_ingest() { + let server = MockServer::start().await; + mount_search3_pages(&server, /*total*/ 3, /*batch*/ 10).await; + mount_minimal_artists(&server).await; + + let store = LibraryStore::open_in_memory(); + let sync_state = SyncStateRepository::new(&store); + sync_state.ensure("s1", "").unwrap(); + sync_state + .set_last_full_sync_at("s1", "", 1) + .unwrap(); + + store + .with_conn_mut("misc", |c| { + for id in ["tr_stale_a", "tr_stale_b"] { + c.execute( + "INSERT INTO track (server_id, id, title, album, duration_sec, deleted, synced_at, raw_json, resync_gen) \ + VALUES ('s1', ?1, 'stale', 'Al', 1, 0, 1, '{}', 1)", + rusqlite::params![id], + )?; + } + Ok(()) + }) + .unwrap(); + + let subsonic = test_subsonic(&server.uri()); + InitialSyncRunner::new( + &store, + &subsonic, + "s1", + "", + flags(CapabilityFlags::SUBSONIC_SEARCH3_BULK | CapabilityFlags::SCAN_STATUS_AVAILABLE), + ) + .with_sleep_disabled() + .run() + .await + .unwrap(); + + let live: i64 = store + .with_conn("misc", |c| { + c.query_row( + "SELECT COUNT(*) FROM track WHERE server_id = 's1' AND deleted = 0", + [], + |r| r.get(0), + ) + }) + .unwrap(); + assert_eq!(live, 3); + + let stale_deleted: i64 = store + .with_conn("misc", |c| { + c.query_row( + "SELECT COUNT(*) FROM track WHERE id IN ('tr_stale_a', 'tr_stale_b') AND deleted = 1", + [], + |r| r.get(0), + ) + }) + .unwrap(); + assert_eq!(stale_deleted, 2); + } + // ── Per-batch progress is emitted during ingest ─────────────────── #[tokio::test(flavor = "multi_thread")] diff --git a/src/config/settingsCredits.ts b/src/config/settingsCredits.ts index 0d862b6b..73bdbc1b 100644 --- a/src/config/settingsCredits.ts +++ b/src/config/settingsCredits.ts @@ -126,6 +126,7 @@ const CONTRIBUTOR_ENTRIES = [ 'Library browse: local-vs-network text search race, All Albums/Artists catalog from index, DevTools browse-race logging (PR #847)', 'Player stats: local listening history tab with heatmap, year summary, recent days, and day drill-down (PR #849)', 'Playback speed: global 0.5–2.0× with Speed / Varispeed / Pitch shift strategies, player bar popover, Orbit passthrough (PR #852)', + 'Local library index: full resync orphan sweep (IS-7) — remove server-deleted tracks after successful re-sync (PR #861)', ], }, {