Cut redundant blob scans and hashing from server startup

Startup previously scanned the legacy SlateDB keyspace three times and
SHA-256-hashed every value in each pass (inventory, import,
verification), then read and rehashed every row of the live SQLite blobs
table — on every boot, even a warm restart with nothing to import. With
a large object-store-backed legacy keyspace that makes restart time
proportional to total blob bytes for the whole retention window.

The inventory pass now only validates key shapes and sizes the keyspace;
digests are still validated by the import pass before any row persists.
The independent verification sweep now runs only on boots whose import
actually inserted rows: the import pass itself byte-compares every
already-present legacy row each boot, so a no-op restart is already
fully cross-checked without a third scan or a full-table rehash.

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

lib/apps/fabro-server/src/blob_activation.rs

@@ -184,10 +184,19 @@ pub(crate) async fn activate_blob_storage(
         .import_legacy_blobs_into(database.pool())
         .await
         .map_err(|source| BlobActivationError::Import(Box::new(source)))?;
-    let verification = store
-        .verify_legacy_blobs_in(database.pool())
-        .await
-        .map_err(|source| BlobActivationError::Verification(Box::new(source)))?;
+    // The import pass already validates every legacy digest and byte-compares
+    // every already-present row on each boot, so the independent verification
+    // sweep only needs to double-check boots that actually inserted rows.
+    let verification = if import.imported_rows > 0 {
+        Some(
+            store
+                .verify_legacy_blobs_in(database.pool())
+                .await
+                .map_err(|source| BlobActivationError::Verification(Box::new(source)))?,
+        )
+    } else {
+        None
+    };
     validate_live_integrity(database.pool()).await?;
     final_truncate_checkpoint(database.pool()).await?;
 
@@ -196,8 +205,8 @@ pub(crate) async fn activate_blob_storage(
         legacy_bytes = inventory.bytes,
         imported_rows = import.imported_rows,
         existing_rows = import.existing_rows,
-        matched_rows = verification.matched_rows,
-        target_rows = verification.target_rows,
+        matched_rows = verification.as_ref().map(|report| report.matched_rows),
+        target_rows = verification.as_ref().map(|report| report.target_rows),
         passive_checkpoints = import.passive_checkpoints,
         backup_required,
         backup_path = ?retained_backup,
@@ -624,7 +633,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn empty_inventory_skips_backup_but_validates_sqlite_rows() -> TestResult<()> {
+    async fn empty_inventory_skips_backup_and_serves_existing_sqlite_rows() -> TestResult<()> {
         let directory = tempfile::tempdir()?;
         let sqlite_path = directory.path().join("fabro.sqlite3");
         let database = fabro_db::Database::connect(&sqlite_path).await?;

lib/components/fabro-store/src/legacy_blob_import.rs

@@ -79,8 +79,6 @@ enum LegacyBlobInventoryFailure {
     ReadSourceScan(#[source] slatedb::Error),
     #[error("a legacy blob key is not canonical")]
     InvalidSourceKey,
-    #[error("legacy blob bytes do not match their key digest")]
-    SourceDigestMismatch,
     #[error("a legacy blob inventory counter overflowed")]
     CounterOverflow,
 }
@@ -520,7 +518,11 @@ struct BatchReport {
 }
 
 impl Database {
-    /// Strictly inventories the exact legacy SlateDB blob keyspace.
+    /// Inventories the exact legacy SlateDB blob keyspace.
+    ///
+    /// 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.
     pub async fn legacy_blob_inventory(
         &self,
     ) -> std::result::Result<LegacyBlobInventory, LegacyBlobInventoryError> {
@@ -555,14 +557,8 @@ impl Database {
                 &mut report.bytes,
                 inventory_usize_to_u64(entry.value.len())?,
             )?;
-            validate_source_entry_common(&entry.key, &entry.value, &prefix).map_err(|failure| {
-                match failure {
-                    SourceEntryFailure::InvalidKey => LegacyBlobInventoryFailure::InvalidSourceKey,
-                    SourceEntryFailure::DigestMismatch => {
-                        LegacyBlobInventoryFailure::SourceDigestMismatch
-                    }
-                }
-            })?;
+            parse_source_key(&entry.key, &prefix)
+                .ok_or(LegacyBlobInventoryFailure::InvalidSourceKey)?;
         }
         Ok(())
     }
@@ -856,16 +852,18 @@ fn parse_canonical_hash(value: &str) -> Option<BlobHash> {
     canonical.then(|| value.parse().ok()).flatten()
 }
 
+fn parse_source_key(key: &[u8], prefix: &[u8]) -> Option<BlobHash> {
+    let suffix = key.strip_prefix(prefix)?;
+    let hash_text = std::str::from_utf8(suffix).ok()?;
+    parse_canonical_hash(hash_text)
+}
+
 fn validate_source_entry_common(
     key: &[u8],
     value: &[u8],
     prefix: &[u8],
 ) -> Result<BlobHash, SourceEntryFailure> {
-    let suffix = key
-        .strip_prefix(prefix)
-        .ok_or(SourceEntryFailure::InvalidKey)?;
-    let hash_text = std::str::from_utf8(suffix).map_err(|_| SourceEntryFailure::InvalidKey)?;
-    let hash = parse_canonical_hash(hash_text).ok_or(SourceEntryFailure::InvalidKey)?;
+    let hash = parse_source_key(key, prefix).ok_or(SourceEntryFailure::InvalidKey)?;
     if BlobHash::new(value) != hash {
         return Err(SourceEntryFailure::DigestMismatch);
     }