diff --git a/docs/public/reference/server-operations.mdx b/docs/public/reference/server-operations.mdx index 4f6ae94b7..82066052f 100644 --- a/docs/public/reference/server-operations.mdx +++ b/docs/public/reference/server-operations.mdx @@ -58,10 +58,13 @@ See [Server Configuration](/administration/server-configuration) for the full `s On startup, Fabro activates SQLite as the only live content-addressed blob store before it opens routes, schedulers, workers, webhooks, reapers, or the ready callback. The activation inventories the exact legacy SlateDB blob -prefix, checks disk headroom, imports in bounded transactions, compares every -legacy blob byte-for-byte with SQLite, validates every SQLite blob row, runs a -live SQLite integrity check, and completes a final WAL checkpoint. Any failure -stops startup. Rows committed by an interrupted import are retained so the +prefix, checks disk headroom sized to the rows not yet imported (a warm +restart with nothing left to import only needs a small fixed headroom), +imports in bounded transactions, compares every legacy blob byte-for-byte +with SQLite, runs a live SQLite integrity check, and completes a final WAL +checkpoint. Boots that import new rows additionally re-verify every legacy +blob against SQLite and validate every SQLite blob row independently. Any +failure stops startup. Rows committed by an interrupted import are retained so the next startup can resume, but the legacy source is never modified and there is no fallback or dual read/write path. diff --git a/lib/apps/fabro-server/src/blob_activation.rs b/lib/apps/fabro-server/src/blob_activation.rs index 5917ed22d..b24188d4f 100644 --- a/lib/apps/fabro-server/src/blob_activation.rs +++ b/lib/apps/fabro-server/src/blob_activation.rs @@ -142,7 +142,7 @@ pub(crate) async fn activate_blob_storage( )); let inventory = store - .legacy_blob_inventory() + .legacy_blob_inventory(database.pool()) .await .map_err(BlobActivationError::Inventory)?; let backup_exists = backup_exists(&backup_path).await?; @@ -159,11 +159,18 @@ pub(crate) async fn activate_blob_storage( } else { 0 }; - let required_free_bytes = - compute_disk_preflight(inventory.bytes, backup_reserve, available_free_bytes)?; + // Only the rows the import still has to copy need new space; rows already + // present in SQLite cost nothing on a warm restart. + let required_free_bytes = compute_disk_preflight( + inventory.pending_bytes, + backup_reserve, + available_free_bytes, + )?; debug!( legacy_rows = inventory.rows, legacy_bytes = inventory.bytes, + pending_rows = inventory.pending_rows, + pending_bytes = inventory.pending_bytes, backup_required, backup_reserve, required_free_bytes, @@ -217,16 +224,16 @@ pub(crate) async fn activate_blob_storage( /// Fail-closed disk capacity check; returns the required free bytes. fn compute_disk_preflight( - legacy_bytes: u64, + pending_bytes: u64, backup_reserve: u64, available_free_bytes: u64, ) -> Result { - let half = legacy_bytes + let half = pending_bytes .checked_add(1) .ok_or(BlobActivationError::DiskRequirementOverflow)? / 2; let required_free_bytes = backup_reserve - .checked_add(legacy_bytes) + .checked_add(pending_bytes) .and_then(|value| value.checked_add(half)) .and_then(|value| value.checked_add(DISK_HEADROOM_BYTES)) .ok_or(BlobActivationError::DiskRequirementOverflow)?; @@ -459,15 +466,15 @@ mod tests { #[test] fn disk_preflight_passes_at_equality_and_fails_one_byte_below() { - let legacy_bytes = 3; + let pending_bytes = 3; let backup_reserve = 10; - let required = backup_reserve + legacy_bytes + 2 + DISK_HEADROOM_BYTES; + let required = backup_reserve + pending_bytes + 2 + DISK_HEADROOM_BYTES; - let required_free_bytes = compute_disk_preflight(legacy_bytes, backup_reserve, required) + let required_free_bytes = compute_disk_preflight(pending_bytes, backup_reserve, required) .expect("exact equality must pass"); assert_eq!(required_free_bytes, required); - let error = compute_disk_preflight(legacy_bytes, backup_reserve, required - 1) + let error = compute_disk_preflight(pending_bytes, backup_reserve, required - 1) .expect_err("one byte below must fail"); assert!(matches!( error, diff --git a/lib/components/fabro-store/src/legacy_blob_import.rs b/lib/components/fabro-store/src/legacy_blob_import.rs index c781d9c4d..580fd5e48 100644 --- a/lib/components/fabro-store/src/legacy_blob_import.rs +++ b/lib/components/fabro-store/src/legacy_blob_import.rs @@ -25,8 +25,12 @@ const PASSIVE_CHECKPOINT_BYTES: u64 = 8 * 1024 * 1024; /// Aggregate size and row count of the exact legacy blob keyspace. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct LegacyBlobInventory { - pub rows: u64, - pub bytes: u64, + pub rows: u64, + pub bytes: u64, + /// Legacy rows whose hash is not yet present in the SQLite blobs table. + pub pending_rows: u64, + /// Bytes belonging to [`Self::pending_rows`]. + pub pending_bytes: u64, } /// A failed strict inventory and the aggregate progress observed before it. @@ -79,6 +83,8 @@ enum LegacyBlobInventoryFailure { ReadSourceScan(#[source] slatedb::Error), #[error("a legacy blob key is not canonical")] InvalidSourceKey, + #[error("reading a SQLite blob row for the legacy inventory")] + ReadDestination(#[source] sqlx::Error), #[error("a legacy blob inventory counter overflowed")] CounterOverflow, } @@ -518,16 +524,20 @@ struct BatchReport { } impl Database { - /// Inventories the exact legacy SlateDB blob keyspace. + /// Inventories the exact legacy SlateDB blob keyspace against the SQLite + /// blobs table in `pool`. /// /// Keys must be canonical, but value digests are not rehashed here: the /// inventory only sizes the keyspace, and the import pass validates every - /// digest before any row is persisted. + /// digest before any row is persisted. Rows whose hash the blobs table + /// does not contain yet are reported as pending so callers can size the + /// remaining import work. pub async fn legacy_blob_inventory( &self, + pool: &SqlitePool, ) -> std::result::Result { let mut report = LegacyBlobInventory::default(); - let result = self.run_legacy_blob_inventory(&mut report).await; + let result = self.run_legacy_blob_inventory(pool, &mut report).await; match result { Ok(()) => Ok(report), Err(failure) => Err(LegacyBlobInventoryError { report, failure }), @@ -536,6 +546,7 @@ impl Database { async fn run_legacy_blob_inventory( &self, + pool: &SqlitePool, report: &mut LegacyBlobInventory, ) -> Result<(), LegacyBlobInventoryFailure> { let source = self @@ -553,12 +564,20 @@ impl Database { .map_err(LegacyBlobInventoryFailure::ReadSourceScan)? { inventory_checked_add(&mut report.rows, 1)?; - inventory_checked_add( - &mut report.bytes, - inventory_usize_to_u64(entry.value.len())?, - )?; - parse_source_key(&entry.key, &prefix) + let value_bytes = inventory_usize_to_u64(entry.value.len())?; + inventory_checked_add(&mut report.bytes, value_bytes)?; + let hash = parse_source_key(&entry.key, &prefix) .ok_or(LegacyBlobInventoryFailure::InvalidSourceKey)?; + let imported: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blobs WHERE hash = ?)") + .bind(hash.to_string()) + .fetch_one(pool) + .await + .map_err(LegacyBlobInventoryFailure::ReadDestination)?; + if !imported { + inventory_checked_add(&mut report.pending_rows, 1)?; + inventory_checked_add(&mut report.pending_bytes, value_bytes)?; + } } Ok(()) } @@ -1305,10 +1324,36 @@ mod tests { ) .await?; - let inventory = context.source.legacy_blob_inventory().await?; + let inventory = context + .source + .legacy_blob_inventory(&context.sqlite) + .await?; assert_eq!(inventory.rows, 2); assert_eq!(inventory.bytes, 4); + assert_eq!(inventory.pending_rows, 2); + assert_eq!(inventory.pending_bytes, 4); + Ok(()) + } + + #[tokio::test] + async fn inventory_reports_already_imported_rows_as_not_pending() -> TestResult<()> { + let context = TestContext::new().await?; + context.put_blob(b"imported-before-inventory").await?; + context.import().await?; + context.put_blob(b"still-pending").await?; + + let inventory = context + .source + .legacy_blob_inventory(&context.sqlite) + .await?; + + assert_eq!(inventory.rows, 2); + assert_eq!(inventory.pending_rows, 1); + assert_eq!( + inventory.pending_bytes, + u64::try_from(b"still-pending".len())? + ); Ok(()) }