feat(library): reconcile catalog by source

Clone the active generation for bounded source reconciliation, advance watermarks only after complete observation, and prune only the completed source.

Coalesce catch-up after authoritative local and media-server writes while retaining the last complete generation on failure.

cx

crates/qbz-local-catalog/Cargo.toml

@@ -6,7 +6,7 @@ license.workspace = true
 repository.workspace = true
 
 [dependencies]
-rusqlite = { version = "0.31", features = ["bundled"] }
+rusqlite = { version = "0.31", features = ["backup", "bundled"] }
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

crates/qbz-local-catalog/src/bootstrap.rs

@@ -12,7 +12,7 @@ pub const BOOTSTRAP_BATCH_ROWS: usize = 250;
 const CATALOG_BYTES_PER_TRACK: u64 = 1_280;
 const SAFETY_FLOOR_BYTES: u64 = 256 * 1024 * 1024;
 const MANIFEST_NAME: &str = "local_catalog-v1-manifest.json";
-const LOCK_NAME: &str = "local_catalog-v1-bootstrap.lock";
+pub(crate) const LOCK_NAME: &str = "local_catalog-v1-bootstrap.lock";
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SourceProbe {
@@ -77,6 +77,24 @@ impl PreflightReport {
         }
         Ok(report)
     }
+
+    pub(crate) fn with_catalog_floor(mut self, catalog_bytes: u64) -> Result<Self> {
+        if catalog_bytes <= self.estimated_catalog_bytes {
+            return Ok(self);
+        }
+        self.estimated_catalog_bytes = catalog_bytes;
+        let margin = catalog_bytes.saturating_add(3) / 4;
+        self.required_available_bytes = catalog_bytes
+            .saturating_add(margin)
+            .saturating_add(SAFETY_FLOOR_BYTES);
+        if self.available_bytes < self.required_available_bytes {
+            return Err(CatalogError::InsufficientSpace {
+                required_bytes: self.required_available_bytes,
+                available_bytes: self.available_bytes,
+            });
+        }
+        Ok(self)
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -111,6 +129,7 @@ pub struct SourceCheckpoint {
     pub checkpoint_cursor: String,
     pub checkpoint_rows: u64,
     pub checkpoint_version: String,
+    pub complete_generation: u64,
     pub complete: bool,
 }
 
@@ -247,18 +266,33 @@ impl BootstrapLayout {
             .map(|manifest| manifest.active_generation);
         let generation = self.next_building_generation(active_generation)?;
         let building_path = self.building_path(generation);
-        let catalog = match Catalog::open(&building_path, generation) {
+        let base_generation = active_generation.unwrap_or(0);
+        let catalog = match Catalog::open(&building_path, generation).and_then(|mut catalog| {
+            let (phase, base) = catalog.build_phase()?;
+            if phase.is_empty() {
+                catalog.set_build_phase("bootstrap", base_generation)?;
+                Ok(catalog)
+            } else if phase == "bootstrap" && base == base_generation {
+                Ok(catalog)
+            } else {
+                Err(CatalogError::InvalidInput(format!(
+                    "building generation belongs to {phase:?} base {base}"
+                )))
+            }
+        }) {
             Ok(catalog) => catalog,
             Err(error) if building_path.is_file() => {
                 // Only the exact incomplete derived sidecar is recoverable here.
                 // Authoritative databases and active generations are never targets.
                 fs::remove_file(&building_path)?;
                 remove_sqlite_sidecars(&building_path)?;
-                Catalog::open(&building_path, generation).map_err(|retry| {
+                let mut catalog = Catalog::open(&building_path, generation).map_err(|retry| {
                     CatalogError::InvalidSource(format!(
                         "rebuild after rejected sidecar ({error}) also failed: {retry}"
                     ))
-                })?
+                })?;
+                catalog.set_build_phase("bootstrap", base_generation)?;
+                catalog
             }
             Err(error) => return Err(error),
         };
@@ -275,7 +309,7 @@ impl BootstrapLayout {
         ))
     }
 
-    fn write_manifest(&self, manifest: &BootstrapManifest) -> Result<()> {
+    pub(crate) fn write_manifest(&self, manifest: &BootstrapManifest) -> Result<()> {
         let path = self.manifest_path();
         let temporary = self.data_dir.join(format!("{MANIFEST_NAME}.tmp"));
         let bytes = serde_json::to_vec_pretty(manifest)?;
@@ -294,7 +328,7 @@ impl BootstrapLayout {
         Ok(())
     }
 
-    fn next_building_generation(&self, active_generation: Option<u64>) -> Result<u64> {
+    pub(crate) fn next_building_generation(&self, active_generation: Option<u64>) -> Result<u64> {
         let mut maximum_final = active_generation.unwrap_or(0);
         let mut building = Vec::new();
         for entry in fs::read_dir(&self.data_dir)? {
@@ -421,7 +455,7 @@ impl BootstrapSession {
     }
 }
 
-fn acquire_lock(path: &Path) -> Result<File> {
+pub(crate) fn acquire_lock(path: &Path) -> Result<File> {
     let file = OpenOptions::new()
         .create(true)
         .read(true)
@@ -441,7 +475,7 @@ fn acquire_lock(path: &Path) -> Result<File> {
     Ok(file)
 }
 
-fn remove_sqlite_sidecars(database: &Path) -> Result<()> {
+pub(crate) fn remove_sqlite_sidecars(database: &Path) -> Result<()> {
     for suffix in ["-wal", "-shm"] {
         let path = PathBuf::from(format!("{}{suffix}", database.display()));
         match fs::remove_file(path) {
@@ -453,7 +487,7 @@ fn remove_sqlite_sidecars(database: &Path) -> Result<()> {
     Ok(())
 }
 
-fn now_unix_ms() -> u64 {
+pub(crate) fn now_unix_ms() -> u64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
@@ -486,12 +520,12 @@ fn filesystem_available_bytes(_path: &Path) -> Result<u64> {
 }
 
 #[cfg(unix)]
-fn sync_directory(path: &Path) -> Result<()> {
+pub(crate) fn sync_directory(path: &Path) -> Result<()> {
     File::open(path)?.sync_all()?;
     Ok(())
 }
 
 #[cfg(not(unix))]
-fn sync_directory(_path: &Path) -> Result<()> {
+pub(crate) fn sync_directory(_path: &Path) -> Result<()> {
     Ok(())
 }

crates/qbz-local-catalog/src/catalog.rs

@@ -4,7 +4,8 @@ use std::time::{Duration, Instant};
 
 use rusqlite::types::Value;
 use rusqlite::{
-    params, params_from_iter, Connection, OpenFlags, OptionalExtension, Row, Statement,
+    params, params_from_iter, Connection, DatabaseName, OpenFlags, OptionalExtension, Row,
+    Statement,
 };
 
 use crate::bootstrap::{BootstrapBatch, SourceCheckpoint, BOOTSTRAP_BATCH_ROWS};
@@ -12,6 +13,7 @@ use crate::model::{
     ProjectedTrack, QueryDescriptor, SourceKey, SourceKind, TrackCursor, TrackGroup, TrackPage,
     TrackRecord, TrackRef, TrackSort,
 };
+use crate::projection::ReconciliationBatch;
 use crate::schema;
 use crate::{CatalogError, Result, SCHEMA_VERSION};
 
@@ -73,6 +75,67 @@ impl Catalog {
         Ok(Self { conn, generation })
     }
 
+    pub(crate) fn backup_to(&self, path: &Path) -> Result<()> {
+        self.conn.backup(DatabaseName::Main, path, None)?;
+        Ok(())
+    }
+
+    pub(crate) fn adopt_generation(
+        path: &Path,
+        previous_generation: u64,
+        generation: u64,
+    ) -> Result<Self> {
+        let conn = Connection::open(path)?;
+        schema::configure(&conn)?;
+        schema::verify(&conn, previous_generation)?;
+        conn.execute(
+            "UPDATE catalog_meta SET value=?1 WHERE key='generation'",
+            [generation.to_string()],
+        )?;
+        let mut catalog = Self { conn, generation };
+        catalog.set_build_phase("projection", previous_generation)?;
+        Ok(catalog)
+    }
+
+    pub(crate) fn set_build_phase(&mut self, phase: &str, base_generation: u64) -> Result<()> {
+        let tx = self.conn.transaction()?;
+        tx.execute(
+            "INSERT INTO catalog_meta(key,value) VALUES ('build_phase',?1)
+             ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+            [phase],
+        )?;
+        tx.execute(
+            "INSERT INTO catalog_meta(key,value) VALUES ('base_generation',?1)
+             ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+            [base_generation.to_string()],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn build_phase(&self) -> Result<(String, u64)> {
+        let phase = self
+            .conn
+            .query_row(
+                "SELECT value FROM catalog_meta WHERE key='build_phase'",
+                [],
+                |row| row.get::<_, String>(0),
+            )
+            .optional()?
+            .unwrap_or_default();
+        let base = self
+            .conn
+            .query_row(
+                "SELECT value FROM catalog_meta WHERE key='base_generation'",
+                [],
+                |row| row.get::<_, String>(0),
+            )
+            .optional()?
+            .and_then(|value| value.parse().ok())
+            .unwrap_or(0);
+        Ok((phase, base))
+    }
+
     pub fn generation(&self) -> u64 {
         self.generation
     }
@@ -127,6 +190,7 @@ impl Catalog {
                         checkpoint_cursor: row.get(1)?,
                         checkpoint_rows: row.get::<_, i64>(2)?.max(0) as u64,
                         checkpoint_version: row.get(3)?,
+                        complete_generation: row.get::<_, i64>(4)?.max(0) as u64,
                         complete: row.get::<_, i64>(4)? == self.generation as i64,
                     })
                 },
@@ -294,10 +358,179 @@ impl Catalog {
             checkpoint_cursor: batch.next_cursor.clone(),
             checkpoint_rows: next_rows.max(0) as u64,
             checkpoint_version: batch.snapshot_version.clone(),
+            complete_generation: if batch.complete { self.generation } else { 0 },
             complete: batch.complete,
         })
     }
 
+    pub(crate) fn begin_reconciliation(
+        &mut self,
+        source: &SourceKey,
+        snapshot_version: &str,
+    ) -> Result<()> {
+        if source.source_instance.trim().is_empty() || snapshot_version.trim().is_empty() {
+            return Err(CatalogError::InvalidInput(
+                "source instance and snapshot version must not be empty".to_string(),
+            ));
+        }
+        let tx = self.conn.transaction()?;
+        // A restarted attempt in the same derived generation must not inherit
+        // observation marks from a superseded snapshot.
+        tx.execute(
+            "UPDATE tracks SET last_observed_generation=0
+              WHERE source_kind=?1 AND source_instance=?2",
+            params![source.source.as_str(), source.source_instance],
+        )?;
+        tx.execute(
+            "INSERT INTO source_state(
+                 source_kind,source_instance,available,last_observed_at,watermark,
+                 complete_generation,checkpoint_cursor,checkpoint_rows,checkpoint_version
+             ) VALUES (?1,?2,1,0,'',0,'',0,?3)
+             ON CONFLICT(source_kind,source_instance) DO UPDATE SET
+                 available=1,
+                 complete_generation=0,
+                 checkpoint_cursor='',
+                 checkpoint_rows=0,
+                 checkpoint_version=excluded.checkpoint_version",
+            params![
+                source.source.as_str(),
+                source.source_instance,
+                snapshot_version
+            ],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn apply_reconciliation_batch(
+        &mut self,
+        batch: &ReconciliationBatch,
+    ) -> Result<SourceCheckpoint> {
+        if batch.tracks.len() > BOOTSTRAP_BATCH_ROWS {
+            return Err(CatalogError::BatchTooLarge {
+                found: batch.tracks.len(),
+                maximum: BOOTSTRAP_BATCH_ROWS,
+            });
+        }
+        for track in &batch.tracks {
+            if track.track_ref.source != batch.source.source
+                || track.track_ref.source_instance != batch.source.source_instance
+            {
+                return Err(CatalogError::InvalidInput(format!(
+                    "reconciliation {:?} contains a row from {:?}/{}",
+                    batch.source, track.track_ref.source, track.track_ref.source_instance
+                )));
+            }
+        }
+        let tx = self.conn.transaction()?;
+        let (committed_cursor, committed_rows, committed_version): (String, i64, String) = tx
+            .query_row(
+                "SELECT checkpoint_cursor,checkpoint_rows,checkpoint_version
+                   FROM source_state
+                  WHERE source_kind=?1 AND source_instance=?2",
+                params![batch.source.source.as_str(), batch.source.source_instance],
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+            )?;
+        if committed_version != batch.snapshot_version {
+            return Err(CatalogError::SourceSnapshotChanged(batch.source.clone()));
+        }
+        if committed_cursor != batch.expected_cursor {
+            return Err(CatalogError::CheckpointMismatch(batch.source.clone()));
+        }
+        {
+            let mut upsert = tx.prepare_cached(UPSERT_TRACK_SQL)?;
+            let mut clear_credits =
+                tx.prepare_cached("DELETE FROM artist_credits WHERE catalog_id=?1")?;
+            let mut insert_credit = tx.prepare_cached(
+                "INSERT OR IGNORE INTO artist_credits(
+                     catalog_id,artist_key,display_name,role,ordinal
+                 ) VALUES (?1,?2,?3,?4,?5)",
+            )?;
+            for track in &batch.tracks {
+                let mut track = track.clone();
+                track.observed_generation = self.generation as i64;
+                track.source_copy_id = Some(ensure_source_copy(&tx, &track, self.generation)?);
+                validate_track(&track)?;
+                upsert_track(&mut upsert, &mut clear_credits, &mut insert_credit, &track)?;
+            }
+        }
+        let next_rows = committed_rows.saturating_add(batch.tracks.len() as i64);
+        if batch.complete {
+            tx.execute(
+                "DELETE FROM tracks
+                  WHERE source_kind=?1 AND source_instance=?2
+                    AND last_observed_generation != ?3",
+                params![
+                    batch.source.source.as_str(),
+                    batch.source.source_instance,
+                    self.generation as i64
+                ],
+            )?;
+            tx.execute(
+                "DELETE FROM source_copies
+                  WHERE source_kind=?1 AND source_instance=?2
+                    AND NOT EXISTS (
+                        SELECT 1 FROM tracks t
+                         WHERE t.source_copy_id=source_copies.source_copy_id
+                    )",
+                params![batch.source.source.as_str(), batch.source.source_instance],
+            )?;
+            tx.execute(
+                "DELETE FROM editions
+                  WHERE NOT EXISTS (
+                        SELECT 1 FROM source_copies sc
+                         WHERE sc.edition_id=editions.edition_id
+                  );
+                 DELETE FROM logical_albums
+                  WHERE NOT EXISTS (
+                        SELECT 1 FROM editions e
+                         WHERE e.logical_album_id=logical_albums.logical_album_id
+                  );",
+                [],
+            )?;
+        }
+        tx.execute(
+            "UPDATE source_state
+                SET available=1,last_observed_at=?3,
+                    watermark=CASE WHEN ?4 THEN ?5 ELSE watermark END,
+                    complete_generation=CASE WHEN ?4 THEN ?6 ELSE 0 END,
+                    checkpoint_cursor=?7,checkpoint_rows=?8
+              WHERE source_kind=?1 AND source_instance=?2",
+            params![
+                batch.source.source.as_str(),
+                batch.source.source_instance,
+                now_unix_seconds(),
+                batch.complete,
+                batch.snapshot_version,
+                self.generation as i64,
+                batch.next_cursor,
+                next_rows,
+            ],
+        )?;
+        tx.commit()?;
+        Ok(SourceCheckpoint {
+            source: batch.source.clone(),
+            available: true,
+            checkpoint_cursor: batch.next_cursor.clone(),
+            checkpoint_rows: next_rows.max(0) as u64,
+            checkpoint_version: batch.snapshot_version.clone(),
+            complete_generation: if batch.complete { self.generation } else { 0 },
+            complete: batch.complete,
+        })
+    }
+
+    pub(crate) fn source_watermark(&self, source: &SourceKey) -> Result<Option<String>> {
+        self.conn
+            .query_row(
+                "SELECT watermark FROM source_state
+                  WHERE source_kind=?1 AND source_instance=?2",
+                params![source.source.as_str(), source.source_instance],
+                |row| row.get(0),
+            )
+            .optional()
+            .map_err(Into::into)
+    }
+
     pub(crate) fn checkpoint_for_activation(&mut self) -> Result<()> {
         self.conn
             .execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;

crates/qbz-local-catalog/src/legacy.rs

@@ -11,6 +11,7 @@ use crate::{
     BootstrapProgress, CatalogError, CreditRole, ProjectedTrack, Result, SourceKey, SourceKind,
     SourceProbe, TrackRef, BOOTSTRAP_BATCH_ROWS,
 };
+use crate::{ProjectionOutcome, ProjectionProgress, ReconciliationBatch};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum LegacyTable {
@@ -136,6 +137,134 @@ pub fn bootstrap_legacy_caches_with_progress(
     })
 }
 
+pub fn reconcile_legacy_caches(
+    data_dir: &Path,
+    cancelled: &AtomicBool,
+) -> Result<ProjectionOutcome> {
+    reconcile_legacy_caches_with_progress(data_dir, cancelled, |_| {})
+}
+
+pub fn reconcile_legacy_caches_with_progress(
+    data_dir: &Path,
+    cancelled: &AtomicBool,
+    mut publish: impl FnMut(&ProjectionProgress),
+) -> Result<ProjectionOutcome> {
+    let layout = BootstrapLayout::new(data_dir);
+    let (active, active_generation) = match layout.open_active() {
+        ActiveCatalog::Ready { catalog, manifest } => (catalog, manifest.active_generation),
+        ActiveCatalog::Fallback(reason) => {
+            return Err(CatalogError::ActivationNotReady(format!(
+                "reconciliation fallback: {reason:?}"
+            )))
+        }
+    };
+    let specs = discover_legacy_sources(data_dir)?;
+    let mut readers = specs
+        .into_iter()
+        .map(LegacyReader::open)
+        .collect::<Result<Vec<_>>>()?;
+    let probes = readers
+        .iter()
+        .map(|reader| reader.probe.clone())
+        .collect::<Vec<_>>();
+    let changed_keys = readers
+        .iter()
+        .filter_map(|reader| {
+            let checkpoint = active.source_checkpoint(&reader.spec.source).ok().flatten();
+            (!checkpoint.is_some_and(|checkpoint| {
+                checkpoint.complete_generation > 0
+                    && checkpoint.checkpoint_version == reader.probe.snapshot_version
+                    && checkpoint.checkpoint_rows == reader.probe.row_count
+            }))
+            .then(|| reader.spec.source.clone())
+        })
+        .collect::<HashSet<_>>();
+    if changed_keys.is_empty() {
+        return Ok(ProjectionOutcome::UpToDate {
+            generation: active_generation,
+            track_count: active.stats()?.track_count,
+        });
+    }
+    drop(active);
+
+    let changed_probes = probes
+        .iter()
+        .filter(|probe| changed_keys.contains(&probe.source))
+        .cloned()
+        .collect::<Vec<_>>();
+    let (mut session, _) = layout.prepare_projection(&probes, None)?;
+    let mut resumed_rows = 0_u64;
+    let mut committed_rows = 0_u64;
+    for reader in readers
+        .iter_mut()
+        .filter(|reader| changed_keys.contains(&reader.spec.source))
+    {
+        let checkpoint = session.checkpoint(&reader.spec.source)?;
+        let mut cursor = if let Some(checkpoint) = checkpoint {
+            if checkpoint.complete
+                && checkpoint.checkpoint_version == reader.probe.snapshot_version
+                && checkpoint.checkpoint_rows == reader.probe.row_count
+            {
+                continue;
+            } else if checkpoint.checkpoint_version == reader.probe.snapshot_version
+                && !checkpoint.complete
+                && checkpoint.checkpoint_rows <= reader.probe.row_count
+            {
+                resumed_rows = resumed_rows.saturating_add(checkpoint.checkpoint_rows);
+                checkpoint.checkpoint_cursor
+            } else {
+                session.begin_source(&reader.spec.source, &reader.probe.snapshot_version)?;
+                String::new()
+            }
+        } else {
+            session.begin_source(&reader.spec.source, &reader.probe.snapshot_version)?;
+            String::new()
+        };
+        loop {
+            if cancelled.load(Ordering::Acquire) {
+                return Ok(ProjectionOutcome::Paused {
+                    generation: session.generation(),
+                    source: reader.spec.source.clone(),
+                    committed_rows,
+                });
+            }
+            let batch = reader.read_batch(&cursor)?;
+            let projection_batch = ReconciliationBatch {
+                source: batch.source,
+                snapshot_version: batch.snapshot_version,
+                expected_cursor: batch.expected_cursor,
+                next_cursor: batch.next_cursor,
+                tracks: batch.tracks,
+                complete: batch.complete,
+            };
+            let saved = session.apply_batch(&projection_batch)?;
+            committed_rows = committed_rows.saturating_add(projection_batch.tracks.len() as u64);
+            cursor = saved.checkpoint_cursor;
+            publish(&ProjectionProgress {
+                generation: session.generation(),
+                source: reader.spec.source.clone(),
+                rows_written: saved.checkpoint_rows,
+                checkpoint_cursor: cursor.clone(),
+                source_complete: saved.complete,
+                prune_authorized: saved.complete,
+            });
+            if saved.complete {
+                break;
+            }
+            std::thread::yield_now();
+            std::thread::sleep(Duration::from_millis(1));
+        }
+    }
+    let stats = session.stats()?;
+    let manifest = session.activate(&changed_probes)?;
+    Ok(ProjectionOutcome::Activated {
+        generation: manifest.active_generation,
+        track_count: stats.track_count,
+        changed_sources: changed_probes.len(),
+        resumed_rows,
+    })
+}
+
 fn discover_local(path: &Path, specs: &mut Vec<LegacySourceSpec>) -> Result<()> {
     let Some(conn) = open_if_present(path)? else {
         return Ok(());
@@ -255,8 +384,9 @@ impl LegacyReader {
         let page_count: i64 = conn.query_row("PRAGMA page_count", [], |row| row.get(0))?;
         let schema_version: i64 = conn.query_row("PRAGMA schema_version", [], |row| row.get(0))?;
         let (row_count, maximum_id, maximum_updated) = source_summary(&conn, &spec)?;
+        let file_epoch = sqlite_file_epoch(&spec.database_path)?;
         let snapshot_version =
-            format!("v1:{schema_version}:{page_count}:{row_count}:{maximum_id}:{maximum_updated}");
+            format!("v1:{schema_version}:{page_count}:{row_count}:{maximum_id}:{maximum_updated}:{file_epoch}");
         let probe = SourceProbe {
             source: spec.source.clone(),
             source_path: spec.database_path.clone(),
@@ -633,3 +763,28 @@ fn format_expression(columns: &HashSet<String>) -> String {
         (false, false) => "''".to_string(),
     }
 }
+
+fn sqlite_file_epoch(path: &Path) -> Result<String> {
+    let mut parts = Vec::new();
+    for candidate in [
+        path.to_path_buf(),
+        PathBuf::from(format!("{}-wal", path.display())),
+    ] {
+        match std::fs::metadata(candidate) {
+            Ok(metadata) => {
+                let modified = metadata
+                    .modified()
+                    .ok()
+                    .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
+                    .map(|duration| duration.as_nanos())
+                    .unwrap_or(0);
+                parts.push(format!("{}:{modified}", metadata.len()));
+            }
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+                parts.push("0:0".to_string())
+            }
+            Err(error) => return Err(error.into()),
+        }
+    }
+    Ok(parts.join(":"))
+}

crates/qbz-local-catalog/src/lib.rs

@@ -8,6 +8,7 @@ mod bootstrap;
 mod catalog;
 mod legacy;
 mod model;
+mod projection;
 mod schema;
 
 pub use bootstrap::{
@@ -20,12 +21,15 @@ pub use catalog::{
 };
 pub use legacy::{
     bootstrap_legacy_caches, bootstrap_legacy_caches_with_progress, discover_legacy_sources,
-    LegacySourceSpec,
+    reconcile_legacy_caches, reconcile_legacy_caches_with_progress, LegacySourceSpec,
 };
 pub use model::{
     ArtistCredit, CreditRole, ProjectedTrack, QueryDescriptor, QuerySurface, SourceKey, SourceKind,
     TrackCursor, TrackGroup, TrackPage, TrackRecord, TrackRef, TrackSort,
 };
+pub use projection::{
+    ProjectionOutcome, ProjectionProgress, ProjectionSession, ReconciliationBatch,
+};
 pub use schema::SCHEMA_VERSION;
 
 #[derive(Debug, thiserror::Error)]
@@ -70,4 +74,6 @@ pub type Result<T> = std::result::Result<T, CatalogError>;
 #[cfg(test)]
 mod bootstrap_tests;
 #[cfg(test)]
+mod projection_tests;
+#[cfg(test)]
 mod tests;

crates/qbz-local-catalog/src/projection.rs

@@ -0,0 +1,235 @@
+use std::collections::BTreeMap;
+use std::fs::{self, File};
+use std::path::{Path, PathBuf};
+
+use crate::bootstrap::{
+    acquire_lock, now_unix_ms, remove_sqlite_sidecars, sync_directory, ActiveCatalog,
+    BootstrapLayout, BootstrapManifest, PreflightReport, SourceCheckpoint, SourceProbe, LOCK_NAME,
+};
+use crate::{Catalog, CatalogError, ProjectedTrack, Result, SourceKey, SCHEMA_VERSION};
+
+#[derive(Debug, Clone)]
+pub struct ReconciliationBatch {
+    pub source: SourceKey,
+    pub snapshot_version: String,
+    pub expected_cursor: String,
+    pub next_cursor: String,
+    pub tracks: Vec<ProjectedTrack>,
+    pub complete: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ProjectionProgress {
+    pub generation: u64,
+    pub source: SourceKey,
+    pub rows_written: u64,
+    pub checkpoint_cursor: String,
+    pub source_complete: bool,
+    pub prune_authorized: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ProjectionOutcome {
+    UpToDate {
+        generation: u64,
+        track_count: u64,
+    },
+    Activated {
+        generation: u64,
+        track_count: u64,
+        changed_sources: usize,
+        resumed_rows: u64,
+    },
+    Paused {
+        generation: u64,
+        source: SourceKey,
+        committed_rows: u64,
+    },
+}
+
+pub struct ProjectionSession {
+    layout: BootstrapLayout,
+    catalog: Catalog,
+    generation: u64,
+    building_path: PathBuf,
+    active_generation: u64,
+    _lock: File,
+}
+
+impl BootstrapLayout {
+    pub fn prepare_projection(
+        &self,
+        probes: &[SourceProbe],
+        available_bytes: Option<u64>,
+    ) -> Result<(ProjectionSession, PreflightReport)> {
+        let available = match available_bytes {
+            Some(bytes) => bytes,
+            None => self.available_bytes()?,
+        };
+        let active_hint = self.read_manifest().map_err(|reason| {
+            CatalogError::ActivationNotReady(format!(
+                "cannot preflight projection without a manifest: {reason:?}"
+            ))
+        })?;
+        let active_bytes = fs::metadata(self.generation_path(active_hint.active_generation))?.len();
+        let preflight =
+            PreflightReport::evaluate(probes, available)?.with_catalog_floor(active_bytes)?;
+        fs::create_dir_all(self.data_dir())?;
+        let lock = acquire_lock(&self.data_dir().join(LOCK_NAME))?;
+        let (active, active_manifest) = match self.open_active() {
+            ActiveCatalog::Ready { catalog, manifest } => (catalog, manifest),
+            ActiveCatalog::Fallback(reason) => {
+                return Err(CatalogError::ActivationNotReady(format!(
+                    "cannot project without an active catalog: {reason:?}"
+                )))
+            }
+        };
+        let active_generation = active_manifest.active_generation;
+        let generation = self.next_building_generation(Some(active_generation))?;
+        let building_path = self.building_path(generation);
+        let catalog = if building_path.is_file() {
+            match Catalog::open(&building_path, generation).and_then(|catalog| {
+                let (phase, base) = catalog.build_phase()?;
+                if phase == "projection" && base == active_generation {
+                    Ok(catalog)
+                } else {
+                    Err(CatalogError::InvalidInput(format!(
+                        "building generation belongs to {phase:?} base {base}"
+                    )))
+                }
+            }) {
+                Ok(catalog) => catalog,
+                Err(_) => {
+                    discard_building(&building_path)?;
+                    clone_generation(&active, &building_path, active_generation, generation)?
+                }
+            }
+        } else {
+            clone_generation(&active, &building_path, active_generation, generation)?
+        };
+        Ok((
+            ProjectionSession {
+                layout: self.clone(),
+                catalog,
+                generation,
+                building_path,
+                active_generation,
+                _lock: lock,
+            },
+            preflight,
+        ))
+    }
+}
+
+impl ProjectionSession {
+    pub fn generation(&self) -> u64 {
+        self.generation
+    }
+
+    pub fn checkpoint(&self, source: &SourceKey) -> Result<Option<SourceCheckpoint>> {
+        self.catalog.source_checkpoint(source)
+    }
+
+    pub fn source_watermark(&self, source: &SourceKey) -> Result<Option<String>> {
+        self.catalog.source_watermark(source)
+    }
+
+    pub fn begin_source(&mut self, source: &SourceKey, snapshot_version: &str) -> Result<()> {
+        self.catalog.begin_reconciliation(source, snapshot_version)
+    }
+
+    pub fn apply_batch(&mut self, batch: &ReconciliationBatch) -> Result<SourceCheckpoint> {
+        self.catalog.apply_reconciliation_batch(batch)
+    }
+
+    pub fn stats(&self) -> Result<crate::CatalogStats> {
+        self.catalog.stats()
+    }
+
+    pub fn activate(mut self, changed: &[SourceProbe]) -> Result<BootstrapManifest> {
+        self.catalog.rebuild_materialized_views()?;
+        if !self.catalog.materialized_views_valid()? {
+            return Err(CatalogError::ActivationNotReady(
+                "projection materializations do not match tracks".to_string(),
+            ));
+        }
+        let actual = self
+            .catalog
+            .stats()?
+            .source_counts
+            .into_iter()
+            .collect::<BTreeMap<_, _>>();
+        for probe in changed {
+            let count = actual.get(&probe.source).copied().unwrap_or(0);
+            if count != probe.row_count {
+                return Err(CatalogError::ActivationNotReady(format!(
+                    "reconciled count for {:?} is {count}, expected {}",
+                    probe.source, probe.row_count
+                )));
+            }
+            let checkpoint = self
+                .catalog
+                .source_checkpoint(&probe.source)?
+                .ok_or_else(|| {
+                    CatalogError::ActivationNotReady(format!(
+                        "missing reconciled checkpoint for {:?}",
+                        probe.source
+                    ))
+                })?;
+            let watermark = self
+                .catalog
+                .source_watermark(&probe.source)?
+                .unwrap_or_default();
+            if !checkpoint.complete
+                || checkpoint.checkpoint_rows != probe.row_count
+                || checkpoint.checkpoint_version != probe.snapshot_version
+                || watermark != probe.snapshot_version
+            {
+                return Err(CatalogError::ActivationNotReady(format!(
+                    "watermark/checkpoint is incomplete for {:?}",
+                    probe.source
+                )));
+            }
+        }
+        let integrity = self.catalog.integrity_check()?;
+        if !integrity.sqlite_ok || integrity.foreign_key_violations != 0 || !integrity.fts_ok {
+            return Err(CatalogError::ActivationNotReady(format!(
+                "projection integrity failed: {integrity:?}"
+            )));
+        }
+        self.catalog.checkpoint_for_activation()?;
+        let final_path = self.layout.generation_path(self.generation);
+        drop(self.catalog);
+        fs::rename(&self.building_path, &final_path)?;
+        sync_directory(self.layout.data_dir())?;
+        let manifest = BootstrapManifest {
+            manifest_version: 1,
+            schema_version: SCHEMA_VERSION,
+            active_generation: self.generation,
+            previous_generation: Some(self.active_generation),
+            activated_at_unix_ms: now_unix_ms(),
+        };
+        self.layout.write_manifest(&manifest)?;
+        Ok(manifest)
+    }
+}
+
+fn clone_generation(
+    active: &Catalog,
+    building_path: &Path,
+    active_generation: u64,
+    generation: u64,
+) -> Result<Catalog> {
+    discard_building(building_path)?;
+    active.backup_to(building_path)?;
+    Catalog::adopt_generation(building_path, active_generation, generation)
+}
+
+fn discard_building(path: &Path) -> Result<()> {
+    match fs::remove_file(path) {
+        Ok(()) => {}
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+        Err(error) => return Err(error.into()),
+    }
+    remove_sqlite_sidecars(path)
+}

crates/qbz-local-catalog/src/projection_tests.rs

@@ -0,0 +1,395 @@
+use std::collections::HashSet;
+use std::fs;
+use std::sync::atomic::AtomicBool;
+use std::time::Instant;
+
+use rusqlite::Connection;
+use tempfile::tempdir;
+
+use crate::{
+    bootstrap_legacy_caches, reconcile_legacy_caches, ActiveCatalog, BootstrapBatch,
+    BootstrapLayout, BootstrapOutcome, ProjectedTrack, ProjectionOutcome, QueryDescriptor,
+    ReconciliationBatch, SourceKey, SourceKind, SourceProbe, TrackRef, BOOTSTRAP_BATCH_ROWS,
+};
+
+fn key(kind: SourceKind, instance: &str) -> SourceKey {
+    SourceKey {
+        source: kind,
+        source_instance: instance.to_string(),
+    }
+}
+
+fn probe(source: &SourceKey, version: &str, rows: u64) -> SourceProbe {
+    SourceProbe {
+        source: source.clone(),
+        source_path: std::path::PathBuf::from(format!("{}.db", source.source.as_str())),
+        snapshot_version: version.to_string(),
+        row_count: rows,
+        page_bytes: rows * 512,
+        integrity_ok: true,
+    }
+}
+
+fn row(source: &SourceKey, id: u64, title: impl Into<String>) -> ProjectedTrack {
+    ProjectedTrack {
+        track_ref: TrackRef {
+            source: source.source,
+            source_instance: source.source_instance.clone(),
+            native_id: id.to_string(),
+        },
+        local_track_id: (source.source == SourceKind::Local).then_some(id as i64),
+        local_path: (source.source == SourceKind::Local).then(|| format!("/music/{id}.flac")),
+        native_album_id: Some(format!("album-{}", id / 10)),
+        source_copy_id: None,
+        title: title.into(),
+        artist: format!("Artist {}", id % 17),
+        album_artist: format!("Artist {}", id % 17),
+        album: format!("Album {}", id / 10),
+        duration_ms: 200_000 + id,
+        year: Some(2020),
+        disc_number: Some(1),
+        track_number: Some((id % 20 + 1) as u32),
+        format: "flac".to_string(),
+        bit_depth: Some(24),
+        sample_rate_hz: Some(96_000),
+        artwork_token: None,
+        isrc: None,
+        musicbrainz_recording_id: None,
+        added_at: id as i64,
+        available: true,
+        observed_generation: 0,
+        credits: Vec::new(),
+    }
+}
+
+fn bootstrap_source(
+    session: &mut crate::BootstrapSession,
+    source: &SourceKey,
+    version: &str,
+    rows: &[ProjectedTrack],
+) {
+    let mut cursor = String::new();
+    for (batch_index, chunk) in rows.chunks(BOOTSTRAP_BATCH_ROWS).enumerate() {
+        let end = (batch_index * BOOTSTRAP_BATCH_ROWS + chunk.len()).to_string();
+        let complete = batch_index * BOOTSTRAP_BATCH_ROWS + chunk.len() == rows.len();
+        let saved = session
+            .apply_batch(&BootstrapBatch {
+                source: source.clone(),
+                snapshot_version: version.to_string(),
+                expected_cursor: cursor,
+                next_cursor: end,
+                tracks: chunk.to_vec(),
+                complete,
+            })
+            .unwrap();
+        cursor = saved.checkpoint_cursor;
+    }
+}
+
+fn reconcile_range(
+    session: &mut crate::ProjectionSession,
+    source: &SourceKey,
+    version: &str,
+    rows: &[ProjectedTrack],
+    start: usize,
+    stop: usize,
+) {
+    let mut cursor = if start == 0 {
+        String::new()
+    } else {
+        start.to_string()
+    };
+    let mut offset = start;
+    while offset < stop {
+        let end = (offset + BOOTSTRAP_BATCH_ROWS).min(stop);
+        let saved = session
+            .apply_batch(&ReconciliationBatch {
+                source: source.clone(),
+                snapshot_version: version.to_string(),
+                expected_cursor: cursor,
+                next_cursor: end.to_string(),
+                tracks: rows[offset..end].to_vec(),
+                complete: end == rows.len(),
+            })
+            .unwrap();
+        cursor = saved.checkpoint_cursor;
+        offset = end;
+    }
+}
+
+#[test]
+fn full_source_reconciliation_resumes_and_prunes_only_on_completion() {
+    let temp = tempdir().unwrap();
+    let layout = BootstrapLayout::new(temp.path());
+    let plex = key(SourceKind::Plex, "plex-a");
+    let jellyfin = key(SourceKind::Jellyfin, "jf-a");
+    let original_plex = (1..=600)
+        .map(|id| row(&plex, id, format!("Old {id}")))
+        .collect::<Vec<_>>();
+    let original_jellyfin = (1..=50)
+        .map(|id| row(&jellyfin, id, format!("Jellyfin {id}")))
+        .collect::<Vec<_>>();
+    let initial_probes = [probe(&plex, "plex-v1", 600), probe(&jellyfin, "jf-v1", 50)];
+    let (mut bootstrap, _) = layout.prepare(&initial_probes, Some(u64::MAX)).unwrap();
+    bootstrap_source(&mut bootstrap, &plex, "plex-v1", &original_plex);
+    bootstrap_source(&mut bootstrap, &jellyfin, "jf-v1", &original_jellyfin);
+    bootstrap.activate(&initial_probes).unwrap();
+
+    let mut current_plex = (101..=600)
+        .map(|id| {
+            row(
+                &plex,
+                id,
+                if id == 200 {
+                    "Updated 200".to_string()
+                } else {
+                    format!("Old {id}")
+                },
+            )
+        })
+        .collect::<Vec<_>>();
+    current_plex.extend((1_000..=1_100).map(|id| row(&plex, id, format!("New {id}"))));
+    let current_probe = probe(&plex, "plex-v2", current_plex.len() as u64);
+    let all_probes = [current_probe.clone(), initial_probes[1].clone()];
+    let projection_started = Instant::now();
+    let (mut interrupted, _) = layout
+        .prepare_projection(&all_probes, Some(u64::MAX))
+        .unwrap();
+    interrupted.begin_source(&plex, "plex-v2").unwrap();
+    reconcile_range(&mut interrupted, &plex, "plex-v2", &current_plex, 0, 250);
+    assert_eq!(interrupted.source_watermark(&plex).unwrap().unwrap(), "");
+
+    // The active generation remains wholly readable and unpruned while g2 is
+    // incomplete, even with a reader held across the side-by-side write.
+    let ActiveCatalog::Ready {
+        catalog: old_reader,
+        ..
+    } = layout.open_active()
+    else {
+        panic!("generation one must remain active")
+    };
+    assert_eq!(old_reader.stats().unwrap().track_count, 650);
+    assert!(old_reader
+        .resolve(&TrackRef {
+            source: SourceKind::Plex,
+            source_instance: "plex-a".to_string(),
+            native_id: "1".to_string(),
+        })
+        .unwrap()
+        .is_some());
+    drop(interrupted);
+
+    let (mut resumed, _) = layout
+        .prepare_projection(&all_probes, Some(u64::MAX))
+        .unwrap();
+    let checkpoint = resumed.checkpoint(&plex).unwrap().unwrap();
+    assert_eq!(checkpoint.checkpoint_rows, 250);
+    reconcile_range(
+        &mut resumed,
+        &plex,
+        "plex-v2",
+        &current_plex,
+        250,
+        current_plex.len(),
+    );
+    assert_eq!(resumed.source_watermark(&plex).unwrap().unwrap(), "plex-v2");
+    let manifest = resumed.activate(&[current_probe]).unwrap();
+    assert_eq!(manifest.active_generation, 2);
+    assert_eq!(manifest.previous_generation, Some(1));
+    assert!(layout.generation_path(1).is_file());
+
+    let ActiveCatalog::Ready { catalog, .. } = layout.open_active() else {
+        panic!("generation two must be active")
+    };
+    assert_eq!(catalog.stats().unwrap().track_count, 651);
+    assert!(catalog
+        .resolve(&TrackRef {
+            source: SourceKind::Plex,
+            source_instance: "plex-a".to_string(),
+            native_id: "1".to_string(),
+        })
+        .unwrap()
+        .is_none());
+    assert_eq!(
+        catalog
+            .resolve(&TrackRef {
+                source: SourceKind::Plex,
+                source_instance: "plex-a".to_string(),
+                native_id: "200".to_string(),
+            })
+            .unwrap()
+            .unwrap()
+            .title,
+        "Updated 200"
+    );
+    assert_eq!(
+        catalog.count_tracks(&QueryDescriptor::tracks()).unwrap(),
+        651
+    );
+    let ids = catalog
+        .query_tracks(&QueryDescriptor::tracks(), None, 500)
+        .unwrap()
+        .rows
+        .into_iter()
+        .map(|track| track.track_ref)
+        .collect::<HashSet<_>>();
+    assert_eq!(ids.len(), 500);
+    println!(
+        "D_PROJECTION_METRIC observed_rows={} pruned_rows={} retained_other_source_rows={} elapsed_ms={:.3} catalog_bytes={}",
+        current_plex.len(),
+        100,
+        original_jellyfin.len(),
+        projection_started.elapsed().as_secs_f64() * 1_000.0,
+        fs::metadata(layout.generation_path(2)).unwrap().len()
+    );
+    drop(old_reader);
+}
+
+#[test]
+fn projection_preflight_accounts_for_copying_the_active_generation() {
+    let temp = tempdir().unwrap();
+    let layout = BootstrapLayout::new(temp.path());
+    let local = key(SourceKind::Local, "library");
+    let initial_probe = probe(&local, "v1", 0);
+    let (mut bootstrap, _) = layout
+        .prepare(&[initial_probe.clone()], Some(u64::MAX))
+        .unwrap();
+    bootstrap
+        .apply_batch(&BootstrapBatch {
+            source: local,
+            snapshot_version: "v1".to_string(),
+            expected_cursor: String::new(),
+            next_cursor: String::new(),
+            tracks: Vec::new(),
+            complete: true,
+        })
+        .unwrap();
+    bootstrap.activate(&[initial_probe.clone()]).unwrap();
+
+    let error = match layout.prepare_projection(&[initial_probe], Some(256 * 1024 * 1024)) {
+        Ok(_) => panic!("copying an active catalog needs room above the fixed floor"),
+        Err(error) => error,
+    };
+    assert!(matches!(
+        error,
+        crate::CatalogError::InsufficientSpace { .. }
+    ));
+    assert!(!layout.building_path(2).exists());
+}
+
+#[test]
+fn legacy_catch_up_detects_authoritative_updates_and_then_is_a_no_op() {
+    let temp = tempdir().unwrap();
+    let path = temp.path().join("library.db");
+    create_local(&path);
+    assert!(matches!(
+        bootstrap_legacy_caches(temp.path(), &AtomicBool::new(false)).unwrap(),
+        BootstrapOutcome::Activated {
+            generation: 1,
+            track_count: 3,
+            ..
+        }
+    ));
+    {
+        let conn = Connection::open(&path).unwrap();
+        conn.execute("UPDATE local_tracks SET title='Changed' WHERE id=2", [])
+            .unwrap();
+        conn.execute("DELETE FROM local_tracks WHERE id=1", [])
+            .unwrap();
+        conn.execute(
+            "INSERT INTO local_tracks VALUES
+             (4,'/music/d.flac','Added','Artist','Artist','Album',180,2024,1,4,
+              'flac',24,96000,NULL,40,'album')",
+            [],
+        )
+        .unwrap();
+    }
+    let outcome = reconcile_legacy_caches(temp.path(), &AtomicBool::new(false)).unwrap();
+    assert!(matches!(
+        outcome,
+        ProjectionOutcome::Activated {
+            generation: 2,
+            track_count: 3,
+            changed_sources: 1,
+            ..
+        }
+    ));
+    let layout = BootstrapLayout::new(temp.path());
+    let ActiveCatalog::Ready { catalog, .. } = layout.open_active() else {
+        panic!("catch-up did not activate")
+    };
+    let local_ref = |id: &str| TrackRef {
+        source: SourceKind::Local,
+        source_instance: "library".to_string(),
+        native_id: id.to_string(),
+    };
+    assert!(catalog.resolve(&local_ref("1")).unwrap().is_none());
+    assert_eq!(
+        catalog.resolve(&local_ref("2")).unwrap().unwrap().title,
+        "Changed"
+    );
+    assert!(catalog.resolve(&local_ref("4")).unwrap().is_some());
+    drop(catalog);
+
+    // A metadata-only edit keeps count, max id and max indexed_at unchanged;
+    // the SQLite file epoch still makes it visible to catch-up.
+    Connection::open(&path)
+        .unwrap()
+        .execute(
+            "UPDATE local_tracks SET title='Metadata only' WHERE id=3",
+            [],
+        )
+        .unwrap();
+    assert!(matches!(
+        reconcile_legacy_caches(temp.path(), &AtomicBool::new(false)).unwrap(),
+        ProjectionOutcome::Activated {
+            generation: 3,
+            track_count: 3,
+            changed_sources: 1,
+            ..
+        }
+    ));
+    let ActiveCatalog::Ready { catalog, .. } = layout.open_active() else {
+        panic!("metadata catch-up did not activate")
+    };
+    assert_eq!(
+        catalog.resolve(&local_ref("3")).unwrap().unwrap().title,
+        "Metadata only"
+    );
+    drop(catalog);
+
+    assert!(matches!(
+        reconcile_legacy_caches(temp.path(), &AtomicBool::new(false)).unwrap(),
+        ProjectionOutcome::UpToDate {
+            generation: 3,
+            track_count: 3,
+        }
+    ));
+    assert!(!layout.building_path(4).exists());
+
+    fs::write(&path, b"corrupt authoritative fixture").unwrap();
+    assert!(reconcile_legacy_caches(temp.path(), &AtomicBool::new(false)).is_err());
+    let ActiveCatalog::Ready { catalog, manifest } = layout.open_active() else {
+        panic!("source failure must preserve the last complete generation")
+    };
+    assert_eq!(manifest.active_generation, 3);
+    assert_eq!(catalog.stats().unwrap().track_count, 3);
+}
+
+fn create_local(path: &std::path::Path) {
+    let conn = Connection::open(path).unwrap();
+    conn.execute_batch(
+        "CREATE TABLE local_tracks (
+             id INTEGER PRIMARY KEY,file_path TEXT NOT NULL,title TEXT NOT NULL,
+             artist TEXT NOT NULL,album_artist TEXT,album TEXT NOT NULL,
+             duration_secs INTEGER,year INTEGER,disc_number INTEGER,
+             track_number INTEGER,format TEXT,bit_depth INTEGER,sample_rate REAL,
+             artwork_path TEXT,indexed_at INTEGER,album_group_key TEXT
+         );
+         INSERT INTO local_tracks VALUES
+             (1,'/music/a.flac','A','Artist','Artist','Album',180,2024,1,1,'flac',24,96000,NULL,10,'album'),
+             (2,'/music/b.flac','B','Artist','Artist','Album',180,2024,1,2,'flac',24,96000,NULL,20,'album'),
+             (3,'/music/c.flac','C','Artist','Artist','Album',180,2024,1,3,'flac',24,96000,NULL,30,'album');",
+    )
+    .unwrap();
+}

crates/qbz-qt/src/local_bridge_ops.rs

@@ -165,6 +165,7 @@ pub(crate) fn run_sync() {
         match result {
             Ok(total) => {
                 log::info!("[qbz-qt] plex sync finished: {total} tracks");
+                crate::local_catalog_qt::request_catch_up();
                 ui(move |mut b| {
                     b.as_mut().set_plex_last_sync_tracks(total as i32);
                     b.as_mut().set_plex_syncing(false);

crates/qbz-qt/src/local_catalog_qt.rs

@@ -8,31 +8,36 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
-use qbz_local_catalog::{BootstrapOutcome, CatalogError};
+use qbz_local_catalog::{BootstrapOutcome, CatalogError, ProjectionOutcome};
 
 static STARTED: AtomicBool = AtomicBool::new(false);
 static CANCELLED: AtomicBool = AtomicBool::new(false);
+static CATCHUP_RUNNING: AtomicBool = AtomicBool::new(false);
+static CATCHUP_REQUESTED: AtomicBool = AtomicBool::new(false);
 
 pub(crate) fn start() {
     if STARTED.swap(true, Ordering::AcqRel) {
         return;
     }
     CANCELLED.store(false, Ordering::Release);
+    CATCHUP_RUNNING.store(true, Ordering::Release);
     crate::spawn(async {
         // Component.onCompleted triggers `on_boot` before the first paint.
         // This delay keeps source probes and SQLite open off that critical path.
         tokio::time::sleep(Duration::from_secs(1)).await;
         if CANCELLED.load(Ordering::Acquire) {
+            worker_finished();
             return;
         }
         let Some(data_dir) = dirs::data_dir().map(|root| root.join("qbz")) else {
             log::warn!("[local-catalog] fallback=missing-data-directory");
+            worker_finished();
             return;
         };
         let result = tokio::task::spawn_blocking(move || {
             let mut last_publish = std::time::Instant::now();
             let mut last_source = None;
-            qbz_local_catalog::bootstrap_legacy_caches_with_progress(
+            let bootstrap = qbz_local_catalog::bootstrap_legacy_caches_with_progress(
                 &data_dir,
                 &CANCELLED,
                 |progress| {
@@ -53,26 +58,66 @@ pub(crate) fn start() {
                         last_source = Some(progress.source.clone());
                     }
                 },
-            )
+            )?;
+            let projection = if matches!(bootstrap, BootstrapOutcome::Activated { .. })
+                && !CANCELLED.load(Ordering::Acquire)
+            {
+                let mut last_projection = std::time::Instant::now();
+                let mut last_projection_source = None;
+                Some(qbz_local_catalog::reconcile_legacy_caches_with_progress(
+                    &data_dir,
+                    &CANCELLED,
+                    |progress| {
+                        let source_changed =
+                            last_projection_source.as_ref() != Some(&progress.source);
+                        if source_changed
+                            || progress.source_complete
+                            || last_projection.elapsed() >= Duration::from_secs(2)
+                        {
+                            log::info!(
+                                "[local-catalog] phase=catch-up generation={} source={:?} rows={} checkpoint={} complete={} prune_authorized={}",
+                                progress.generation,
+                                progress.source,
+                                progress.rows_written,
+                                progress.checkpoint_cursor,
+                                progress.source_complete,
+                                progress.prune_authorized
+                            );
+                            last_projection = std::time::Instant::now();
+                            last_projection_source = Some(progress.source.clone());
+                        }
+                    },
+                )?)
+            } else {
+                None
+            };
+            Ok::<_, CatalogError>((bootstrap, projection))
         })
         .await;
         match result {
-            Ok(Ok(BootstrapOutcome::Activated {
-                generation,
-                track_count,
-                resumed_rows,
-            })) => log::info!(
-                "[local-catalog] phase=active generation={generation} tracks={track_count} resumed_rows={resumed_rows}"
-            ),
-            Ok(Ok(BootstrapOutcome::Paused {
-                generation,
-                source,
-                committed_rows,
-            })) => log::info!(
-                "[local-catalog] phase=paused generation={generation} source={source:?} committed_rows={committed_rows}"
-            ),
-            Ok(Ok(BootstrapOutcome::Fallback(reason))) => {
-                log::warn!("[local-catalog] fallback={reason:?}")
+            Ok(Ok((bootstrap, projection))) => {
+                match bootstrap {
+                    BootstrapOutcome::Activated {
+                        generation,
+                        track_count,
+                        resumed_rows,
+                    } => log::info!(
+                        "[local-catalog] phase=active generation={generation} tracks={track_count} resumed_rows={resumed_rows}"
+                    ),
+                    BootstrapOutcome::Paused {
+                        generation,
+                        source,
+                        committed_rows,
+                    } => log::info!(
+                        "[local-catalog] phase=paused generation={generation} source={source:?} committed_rows={committed_rows}"
+                    ),
+                    BootstrapOutcome::Fallback(reason) => {
+                        log::warn!("[local-catalog] fallback={reason:?}")
+                    }
+                }
+                if let Some(projection) = projection {
+                    log_projection(projection);
+                }
             }
             Ok(Err(CatalogError::InsufficientSpace {
                 required_bytes,
@@ -83,9 +128,105 @@ pub(crate) fn start() {
             Ok(Err(error)) => log::warn!("[local-catalog] fallback=bootstrap-error error={error}"),
             Err(error) => log::warn!("[local-catalog] fallback=worker-join error={error}"),
         }
+        worker_finished();
     });
 }
 
+/// Coalesced post-write hook for local scan and media-server sync completion.
+/// If the boot worker is still running, one follow-up pass is remembered.
+pub(crate) fn request_catch_up() {
+    CATCHUP_REQUESTED.store(true, Ordering::Release);
+    if CATCHUP_RUNNING.swap(true, Ordering::AcqRel) {
+        return;
+    }
+    crate::spawn(async {
+        tokio::time::sleep(Duration::from_millis(500)).await;
+        CATCHUP_REQUESTED.store(false, Ordering::Release);
+        if CANCELLED.load(Ordering::Acquire) {
+            worker_finished();
+            return;
+        }
+        let Some(data_dir) = dirs::data_dir().map(|root| root.join("qbz")) else {
+            log::warn!("[local-catalog] fallback=missing-data-directory");
+            worker_finished();
+            return;
+        };
+        let result = tokio::task::spawn_blocking(move || {
+            let mut last_publish = std::time::Instant::now();
+            let mut last_source = None;
+            qbz_local_catalog::reconcile_legacy_caches_with_progress(
+                &data_dir,
+                &CANCELLED,
+                |progress| {
+                    let source_changed = last_source.as_ref() != Some(&progress.source);
+                    if source_changed
+                        || progress.source_complete
+                        || last_publish.elapsed() >= Duration::from_secs(2)
+                    {
+                        log::info!(
+                            "[local-catalog] phase=catch-up generation={} source={:?} rows={} checkpoint={} complete={} prune_authorized={}",
+                            progress.generation,
+                            progress.source,
+                            progress.rows_written,
+                            progress.checkpoint_cursor,
+                            progress.source_complete,
+                            progress.prune_authorized
+                        );
+                        last_publish = std::time::Instant::now();
+                        last_source = Some(progress.source.clone());
+                    }
+                },
+            )
+        })
+        .await;
+        match result {
+            Ok(Ok(outcome)) => log_projection(outcome),
+            Ok(Err(CatalogError::InsufficientSpace {
+                required_bytes,
+                available_bytes,
+            })) => log::warn!(
+                "[local-catalog] phase=catch-up-paused reason=low-disk required_bytes={required_bytes} available_bytes={available_bytes}"
+            ),
+            Ok(Err(error)) => log::warn!("[local-catalog] fallback=catch-up-error error={error}"),
+            Err(error) => log::warn!("[local-catalog] fallback=catch-up-join error={error}"),
+        }
+        worker_finished();
+    });
+}
+
+fn log_projection(outcome: ProjectionOutcome) {
+    match outcome {
+        ProjectionOutcome::UpToDate {
+            generation,
+            track_count,
+        } => log::info!(
+            "[local-catalog] phase=fresh generation={generation} tracks={track_count}"
+        ),
+        ProjectionOutcome::Activated {
+            generation,
+            track_count,
+            changed_sources,
+            resumed_rows,
+        } => log::info!(
+            "[local-catalog] phase=catch-up-active generation={generation} tracks={track_count} changed_sources={changed_sources} resumed_rows={resumed_rows}"
+        ),
+        ProjectionOutcome::Paused {
+            generation,
+            source,
+            committed_rows,
+        } => log::info!(
+            "[local-catalog] phase=catch-up-paused generation={generation} source={source:?} committed_rows={committed_rows}"
+        ),
+    }
+}
+
+fn worker_finished() {
+    CATCHUP_RUNNING.store(false, Ordering::Release);
+    if CATCHUP_REQUESTED.swap(false, Ordering::AcqRel) && !CANCELLED.load(Ordering::Acquire) {
+        request_catch_up();
+    }
+}
+
 pub(crate) fn cancel() {
     CANCELLED.store(true, Ordering::Release);
 }

crates/qbz-qt/src/media_sync_qt.rs

@@ -476,6 +476,7 @@ fn finish(
         "[qbz-qt] {} sync finished: {saved} saved, {pruned} pruned, {total} cached",
         kind.as_str()
     );
+    crate::local_catalog_qt::request_catch_up();
     Ok(SyncReport {
         saved,
         pruned,

crates/qbz-qt/src/settings_qt/library.rs

@@ -787,6 +787,7 @@ pub fn scan(folder_id: Option<i64>) {
 
         SCANNING.store(false, Ordering::SeqCst);
         *CURRENT_FILE.lock().unwrap_or_else(|e| e.into_inner()) = String::new();
+        crate::local_catalog_qt::request_catch_up();
         refresh_browse();
         super::publish_snapshot().await;
     });
@@ -852,6 +853,9 @@ pub async fn cleanup_missing() {
         None => qbz_i18n::t("Cleanup failed."),
     };
     CLEANING.store(false, Ordering::SeqCst);
+    if result.is_some() {
+        crate::local_catalog_qt::request_catch_up();
+    }
     refresh_browse();
     super::publish_snapshot().await;
 }
@@ -870,6 +874,7 @@ pub async fn clear_library() {
     })
     .await;
     CLEARING.store(false, Ordering::SeqCst);
+    crate::local_catalog_qt::request_catch_up();
     refresh_browse();
     super::publish_snapshot().await;
 }
@@ -891,6 +896,7 @@ pub async fn plex_clear_cache() {
         }
     })
     .await;
+    crate::local_catalog_qt::request_catch_up();
     crate::local_bridge_ops::publish_plex_state();
     refresh_browse();
 }