From 7689fb0df6156ec92201fa283fbb20aa9c46322a Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Sun, 23 Aug 2026 07:53:20 -0600 Subject: [PATCH] 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 | 2 +- crates/qbz-local-catalog/src/bootstrap.rs | 56 ++- crates/qbz-local-catalog/src/catalog.rs | 235 ++++++++++- crates/qbz-local-catalog/src/legacy.rs | 157 ++++++- crates/qbz-local-catalog/src/lib.rs | 8 +- crates/qbz-local-catalog/src/projection.rs | 235 +++++++++++ .../qbz-local-catalog/src/projection_tests.rs | 395 ++++++++++++++++++ crates/qbz-qt/src/local_bridge_ops.rs | 1 + crates/qbz-qt/src/local_catalog_qt.rs | 179 +++++++- crates/qbz-qt/src/media_sync_qt.rs | 1 + crates/qbz-qt/src/settings_qt/library.rs | 6 + 11 files changed, 1241 insertions(+), 34 deletions(-) create mode 100644 crates/qbz-local-catalog/src/projection.rs create mode 100644 crates/qbz-local-catalog/src/projection_tests.rs diff --git a/crates/qbz-local-catalog/Cargo.toml b/crates/qbz-local-catalog/Cargo.toml index 66ec6e42a..9327c4482 100644 --- a/crates/qbz-local-catalog/Cargo.toml +++ b/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 diff --git a/crates/qbz-local-catalog/src/bootstrap.rs b/crates/qbz-local-catalog/src/bootstrap.rs index 1a889165c..124f5dc38 100644 --- a/crates/qbz-local-catalog/src/bootstrap.rs +++ b/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 { + 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) -> Result { + pub(crate) fn next_building_generation(&self, active_generation: Option) -> Result { 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 { +pub(crate) fn acquire_lock(path: &Path) -> Result { let file = OpenOptions::new() .create(true) .read(true) @@ -441,7 +475,7 @@ fn acquire_lock(path: &Path) -> Result { 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 { } #[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(()) } diff --git a/crates/qbz-local-catalog/src/catalog.rs b/crates/qbz-local-catalog/src/catalog.rs index 19e0842e8..e58bc1889 100644 --- a/crates/qbz-local-catalog/src/catalog.rs +++ b/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 { + 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 { + 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> { + 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);")?; diff --git a/crates/qbz-local-catalog/src/legacy.rs b/crates/qbz-local-catalog/src/legacy.rs index c089e3863..89cbe72b6 100644 --- a/crates/qbz-local-catalog/src/legacy.rs +++ b/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 { + 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 { + 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::>>()?; + let probes = readers + .iter() + .map(|reader| reader.probe.clone()) + .collect::>(); + 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::>(); + 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::>(); + 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) -> 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 { (false, false) => "''".to_string(), } } + +fn sqlite_file_epoch(path: &Path) -> Result { + 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(":")) +} diff --git a/crates/qbz-local-catalog/src/lib.rs b/crates/qbz-local-catalog/src/lib.rs index 2c08cbb39..b40674dcf 100644 --- a/crates/qbz-local-catalog/src/lib.rs +++ b/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 = std::result::Result; #[cfg(test)] mod bootstrap_tests; #[cfg(test)] +mod projection_tests; +#[cfg(test)] mod tests; diff --git a/crates/qbz-local-catalog/src/projection.rs b/crates/qbz-local-catalog/src/projection.rs new file mode 100644 index 000000000..422dc0bb7 --- /dev/null +++ b/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, + 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, + ) -> 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> { + self.catalog.source_checkpoint(source) + } + + pub fn source_watermark(&self, source: &SourceKey) -> Result> { + 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 { + self.catalog.apply_reconciliation_batch(batch) + } + + pub fn stats(&self) -> Result { + self.catalog.stats() + } + + pub fn activate(mut self, changed: &[SourceProbe]) -> Result { + 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::>(); + 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 { + 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) +} diff --git a/crates/qbz-local-catalog/src/projection_tests.rs b/crates/qbz-local-catalog/src/projection_tests.rs new file mode 100644 index 000000000..c367d303d --- /dev/null +++ b/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) -> 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::>(); + let original_jellyfin = (1..=50) + .map(|id| row(&jellyfin, id, format!("Jellyfin {id}"))) + .collect::>(); + 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::>(); + 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", ¤t_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", + ¤t_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::>(); + 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(); +} diff --git a/crates/qbz-qt/src/local_bridge_ops.rs b/crates/qbz-qt/src/local_bridge_ops.rs index e3a332ae0..8da8fc1c8 100644 --- a/crates/qbz-qt/src/local_bridge_ops.rs +++ b/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); diff --git a/crates/qbz-qt/src/local_catalog_qt.rs b/crates/qbz-qt/src/local_catalog_qt.rs index 746577dd2..f19c93f1f 100644 --- a/crates/qbz-qt/src/local_catalog_qt.rs +++ b/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); } diff --git a/crates/qbz-qt/src/media_sync_qt.rs b/crates/qbz-qt/src/media_sync_qt.rs index c67c9cf77..9a53eb15a 100644 --- a/crates/qbz-qt/src/media_sync_qt.rs +++ b/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, diff --git a/crates/qbz-qt/src/settings_qt/library.rs b/crates/qbz-qt/src/settings_qt/library.rs index 5006d2a6e..8d9c31dca 100644 --- a/crates/qbz-qt/src/settings_qt/library.rs +++ b/crates/qbz-qt/src/settings_qt/library.rs @@ -787,6 +787,7 @@ pub fn scan(folder_id: Option) { 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(); }