Size the blob activation disk preflight to the remaining import work

The preflight demanded ~1.5x the full legacy inventory bytes free on
every startup, with no credit for rows already imported. Because the
first activation itself consumes about twice the legacy bytes (the
SQLite copy plus the retained backup) and the legacy keyspace stays in
place for the whole retention window, a successfully activated server
could fall below the requirement and become unable to restart until an
operator freed space the server would never write.

The legacy inventory now checks each row's hash against the SQLite blobs
table and reports pending rows and bytes, and the preflight requires
1.5x only the pending bytes plus the backup reserve and fixed headroom.
A warm restart with nothing left to import needs only the headroom.
Also updates the server operations doc for this and for the
verification pass now running only on boots that import rows.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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.
 

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<u64, BlobActivationError> {
-    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,

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<LegacyBlobInventory, LegacyBlobInventoryError> {
         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(())
     }