Harden SQLite blob activation safety

Keep VACUUM snapshots private until permissions and durability are established. Refuse to recreate a missing rollback backup after import has begun, and preserve secondary cleanup failures in startup logs.

docs/public/reference/server-operations.mdx

@@ -62,8 +62,10 @@ 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; on
 filesystems whose free space cannot be determined the check is skipped with
 a warning), 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
+byte-for-byte with SQLite, runs a live SQLite integrity check, and attempts a
+final WAL truncate checkpoint. A busy final truncate logs a warning and startup
+continues so a later checkpoint can finish after the blocking reader exits.
+Boots that import new rows additionally re-verify every
 legacy blob against SQLite and validate every SQLite blob row independently.
 Any failure stops startup. Warm boots that import no rows skip that full target
 scan: the import pass has already byte-compared every retained legacy row, and
@@ -73,10 +75,13 @@ never modified and there is no fallback or dual read/write path.
 
 For a non-empty legacy inventory, the first activation also creates the
 private sibling backup
-`fabro.sqlite3.pre-blob-activation.bak`. Fabro writes and validates a staging
-database before publishing the backup without overwriting an existing file.
+`fabro.sqlite3.pre-blob-activation.bak`. Fabro writes the staging database
+inside a private same-directory area, applies owner-only permissions, flushes
+and validates it, then publishes the backup without overwriting an existing file.
 A valid retained backup is revalidated on every warm restart and is preserved
-as the original pre-activation safety artifact. It is not a promise that an
+as the original pre-activation safety artifact. If any legacy row is already
+present in SQLite, a missing retained backup stops startup rather than silently
+moving that rollback boundary forward. It is not a promise that an
 older binary can safely resume after the activated server has accepted new
 work; recovery after that boundary is forward-only. Empty legacy inventories
 do not need this backup.

lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs

@@ -50,6 +50,14 @@ pub(crate) enum BlobActivationError {
     BackupNotRegular { path: PathBuf },
     #[error("activation backup permissions are not private at {path}")]
     BackupNotPrivate { path: PathBuf },
+    #[error(
+        "activation backup is missing at {path} while {existing_rows} of {legacy_rows} legacy blob rows are already present in SQLite"
+    )]
+    MissingBackupAfterImport {
+        path:          PathBuf,
+        legacy_rows:   u64,
+        existing_rows: u64,
+    },
     #[error("opening or checking activation backup integrity at {path}")]
     BackupIntegrity {
         path:   PathBuf,
@@ -133,6 +141,13 @@ pub(crate) async fn activate_blob_storage(
     if backup_exists {
         validate_backup(&backup_path).await?;
     }
+    if !backup_exists && inventory.pending_rows < inventory.rows {
+        return Err(BlobActivationError::MissingBackupAfterImport {
+            path:          backup_path,
+            legacy_rows:   inventory.rows,
+            existing_rows: inventory.rows - inventory.pending_rows,
+        });
+    }
     let backup_required = inventory.rows > 0 && !backup_exists;
     // The resource sampler treats a path with no matching mount as an
     // unsupported-but-benign condition (tmpfs or squashfs roots, network
@@ -624,6 +639,51 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn missing_backup_after_prior_import_fails_closed() -> TestResult<()> {
+        let directory = tempfile::tempdir()?;
+        let sqlite_path = directory.path().join("fabro.sqlite3");
+        let database = fabro_db::Database::connect(&sqlite_path).await?;
+        database.migrate().await?;
+        let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let source = fabro_store::test_support::test_database(
+            Arc::clone(&object_store),
+            "missing-backup-test",
+            Duration::from_millis(1),
+            None,
+        );
+        let bytes = b"already-imported";
+        let hash = fabro_store::test_support::put_legacy_blob(&source, bytes).await?;
+        sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
+            .bind(hash.to_string())
+            .bind(bytes.as_slice())
+            .execute(database.pool())
+            .await?;
+        drop(source);
+
+        let error = activate_blob_storage(
+            &database,
+            &sqlite_path,
+            object_store,
+            "missing-backup-test".to_string(),
+            Duration::from_millis(1),
+            None,
+        )
+        .await
+        .expect_err("startup must not move the pre-activation rollback boundary");
+
+        assert!(matches!(
+            error,
+            BlobActivationError::MissingBackupAfterImport {
+                legacy_rows: 1,
+                existing_rows: 1,
+                ..
+            }
+        ));
+        assert!(!append_to_path(&sqlite_path, BACKUP_SUFFIX).exists());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn empty_inventory_skips_backup_and_serves_existing_sqlite_rows() -> TestResult<()> {
         let directory = tempfile::tempdir()?;

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

@@ -8,12 +8,13 @@ use std::fmt;
 
 use bytes::Bytes;
 use fabro_types::BlobHash;
+use fabro_util::error;
 use futures::TryStreamExt as _;
 use sqlx::pool::PoolConnection;
 use sqlx::{Acquire as _, Sqlite, SqlitePool};
 #[cfg(test)]
 use tokio::sync::Barrier;
-use tracing::debug;
+use tracing::{debug, error};
 
 use crate::Database;
 use crate::keys::SlateKey;
@@ -709,7 +710,15 @@ impl Database {
             }
             Err(failure) => {
                 debug_import_outcome("failed", &report, Some(failure.kind()));
-                Err(LegacyBlobImportError { report, failure })
+                let import_error = LegacyBlobImportError { report, failure };
+                for cleanup_error in import_error.cleanup_errors() {
+                    let rendered = error::collect_chain(cleanup_error).join(": ");
+                    error!(
+                        error = %rendered,
+                        "Legacy blob import cleanup failed"
+                    );
+                }
+                Err(import_error)
             }
         }
     }
@@ -1766,10 +1775,12 @@ mod tests {
             restore_automatic_checkpoint: true,
             ..ImportControls::default()
         };
+        let capture = CapturedEvents::default();
 
         let error = context
             .source
             .import_legacy_blobs_with_controls(&context.sqlite, &controls)
+            .with_subscriber(capture.clone())
             .await
             .expect_err("both injected failures should fail import");
 
@@ -1786,6 +1797,15 @@ mod tests {
                 .any(|source| source.downcast_ref::<sqlx::Error>().is_some()),
             "restoration source was absent from cleanup errors"
         );
+        let events = capture.events().join("\n");
+        assert!(
+            events.contains("Legacy blob import cleanup failed"),
+            "captured: {events}"
+        );
+        assert!(
+            events.contains("injected automatic checkpoint restoration failure"),
+            "captured: {events}"
+        );
         Ok(())
     }
 

lib/components/fabro-store/src/test_support/mod.rs

@@ -6,9 +6,9 @@ use fabro_types::{BlobHash, RunId};
 use object_store::ObjectStore;
 use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
 
+use crate::keys::SlateKey;
 #[cfg(test)]
 use crate::{AuthSessionStore, RunSummaryStore};
-use crate::keys::SlateKey;
 use crate::{BlobStore, Database, Result};
 
 /// Returns an isolated SQLite blob authority backed by its own in-memory

lib/foundation/fabro-db/Cargo.toml

@@ -16,10 +16,10 @@ workspace = true
 anyhow.workspace = true
 chrono.workspace = true
 sqlx.workspace = true
+tempfile = "3"
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
-tempfile = "3"
 tokio = { workspace = true, features = ["macros"] }

lib/foundation/fabro-db/src/lib.rs

@@ -227,6 +227,12 @@ pub enum SnapshotStagingError {
         #[source]
         source: std::io::Error,
     },
+    #[error("creating private snapshot staging area for {path}")]
+    CreatePrivateStagingArea {
+        path:   PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
     #[error("snapshot staging path is not valid UTF-8 at {path}")]
     NonUtf8Path { path: PathBuf },
     #[error("writing SQLite snapshot {path}")]
@@ -247,26 +253,57 @@ pub enum SnapshotStagingError {
         #[source]
         source: std::io::Error,
     },
+    #[error("publishing private snapshot staging file {path}")]
+    Publish {
+        path:   PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
 }
 
 /// Writes a consistent single-file copy of the live pool to `staging_path`.
 ///
 /// `VACUUM INTO` produces a snapshot that needs no `-wal`/`-shm` siblings to
-/// restore. Any stale staging file is removed first and the copy is
-/// restricted to private permissions. The caller publishes the staging file
-/// into its final path and owns the durability of that rename.
+/// restore. The copy is first written inside a private same-directory staging
+/// area, then restricted to owner-only permissions and flushed before it is
+/// exposed at `staging_path`. The caller publishes the staging file into its
+/// final path and owns the durability of that rename.
 pub async fn write_snapshot_to_staging(
     pool: &DbPool,
     staging_path: &Path,
 ) -> Result<(), SnapshotStagingError> {
+    write_snapshot_to_staging_inner(pool, staging_path, |_| {}).await
+}
+
+async fn write_snapshot_to_staging_inner<F>(
+    pool: &DbPool,
+    staging_path: &Path,
+    after_write: F,
+) -> Result<(), SnapshotStagingError>
+where
+    F: FnOnce(&Path),
+{
     remove_file_if_exists(staging_path)
         .await
         .map_err(|source| SnapshotStagingError::RemoveStale {
             path: staging_path.to_path_buf(),
             source,
         })?;
+
+    let staging_parent = nonempty_parent(staging_path);
+    // SQLite's VACUUM INTO creates its destination with umask-derived
+    // permissions. Keep that file behind an owner-only directory until its
+    // own mode is restricted, so a traversable database directory never
+    // exposes a partially written snapshot.
+    let private_staging_area = create_private_staging_area(staging_parent).map_err(|source| {
+        SnapshotStagingError::CreatePrivateStagingArea {
+            path: staging_path.to_path_buf(),
+            source,
+        }
+    })?;
+    let private_staging_path = private_staging_area.path().join("snapshot.sqlite3");
     let staging_target =
-        staging_path
+        private_staging_path
             .to_str()
             .ok_or_else(|| SnapshotStagingError::NonUtf8Path {
                 path: staging_path.to_path_buf(),
@@ -279,7 +316,8 @@ pub async fn write_snapshot_to_staging(
             path: staging_path.to_path_buf(),
             source,
         })?;
-    set_private_permissions(staging_path)
+    after_write(&private_staging_path);
+    set_private_permissions(&private_staging_path)
         .await
         .map_err(|source| SnapshotStagingError::SetPermissions {
             path: staging_path.to_path_buf(),
@@ -287,7 +325,7 @@ pub async fn write_snapshot_to_staging(
         })?;
     // The staging file must be durable before the caller renames it into a
     // path that later recovery logic treats as a complete snapshot.
-    let sync_result = match fs::File::open(staging_path).await {
+    let sync_result = match fs::File::open(&private_staging_path).await {
         Ok(file) => file.sync_all().await,
         Err(source) => Err(source),
     };
@@ -295,9 +333,33 @@ pub async fn write_snapshot_to_staging(
         path: staging_path.to_path_buf(),
         source,
     })?;
+    fs::rename(&private_staging_path, staging_path)
+        .await
+        .map_err(|source| SnapshotStagingError::Publish {
+            path: staging_path.to_path_buf(),
+            source,
+        })?;
     Ok(())
 }
 
+fn create_private_staging_area(parent: &Path) -> std::io::Result<tempfile::TempDir> {
+    let mut builder = tempfile::Builder::new();
+    builder.prefix(".fabro-snapshot-");
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt as _;
+
+        builder.permissions(std::fs::Permissions::from_mode(0o700));
+    }
+    builder.tempdir_in(parent)
+}
+
+fn nonempty_parent(path: &Path) -> &Path {
+    path.parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
+        .unwrap_or_else(|| Path::new("."))
+}
+
 /// Flushes the directory entry metadata for `path`'s parent so a rename into
 /// that directory survives power loss. No-op off Unix, where a directory
 /// cannot be opened for syncing.
@@ -307,10 +369,7 @@ pub async fn write_snapshot_to_staging(
     reason = "directory fds have no async open; callers run this on a blocking thread"
 )]
 pub fn sync_parent_directory(path: &Path) -> std::io::Result<()> {
-    let Some(parent) = path.parent() else {
-        return Ok(());
-    };
-    std::fs::File::open(parent)?.sync_all()
+    std::fs::File::open(nonempty_parent(path))?.sync_all()
 }
 
 /// Flushes the directory entry metadata for `path`'s parent so a rename into
@@ -378,3 +437,65 @@ async fn prepare_private_database_file(path: &Path) -> anyhow::Result<()> {
         .with_context(|| format!("creating SQLite database {}", path.display()))?;
     Ok(())
 }
+
+#[cfg(all(test, unix))]
+mod tests {
+    use std::os::unix::fs::PermissionsExt as _;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn snapshot_stays_hidden_until_it_has_private_permissions() -> anyhow::Result<()> {
+        let root = tempfile::tempdir()?;
+        let database_directory = root.path().join("traversable-db");
+        fs::create_dir(&database_directory).await?;
+        fs::set_permissions(&database_directory, std::fs::Permissions::from_mode(0o755)).await?;
+
+        let database = Database::connect(database_directory.join("fabro.sqlite3")).await?;
+        sqlx::query("CREATE TABLE snapshot_secret (value TEXT NOT NULL)")
+            .execute(database.pool())
+            .await?;
+        sqlx::query("INSERT INTO snapshot_secret (value) VALUES ('kept')")
+            .execute(database.pool())
+            .await?;
+
+        let staging_path = database_directory.join("snapshot.tmp");
+        let observed_private_stage = AtomicBool::new(false);
+        write_snapshot_to_staging_inner(database.pool(), &staging_path, |private_path| {
+            assert!(
+                !staging_path.exists(),
+                "the traversable parent must not expose the snapshot before chmod"
+            );
+            let private_directory = private_path
+                .parent()
+                .expect("private staging file should have a parent");
+            let mode = std::fs::metadata(private_directory)
+                .expect("private staging directory should exist")
+                .permissions()
+                .mode()
+                & 0o777;
+            assert_eq!(mode, 0o700, "temporary staging directory must be private");
+            observed_private_stage.store(true, Ordering::Relaxed);
+        })
+        .await?;
+        assert!(observed_private_stage.load(Ordering::Relaxed));
+
+        let mode = fs::metadata(&staging_path).await?.permissions().mode() & 0o777;
+        assert_eq!(mode, 0o600, "published staging file must be private");
+        let snapshot = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect_with(
+                SqliteConnectOptions::new()
+                    .filename(&staging_path)
+                    .read_only(true),
+            )
+            .await?;
+        let value: String = sqlx::query_scalar("SELECT value FROM snapshot_secret")
+            .fetch_one(&snapshot)
+            .await?;
+        assert_eq!(value, "kept");
+        snapshot.close().await;
+        Ok(())
+    }
+}