diff --git a/Cargo.lock b/Cargo.lock index f7376d1bf..44a8a8e64 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2593,6 +2593,7 @@ dependencies = [ "chrono", "sqlx", "tempfile", + "thiserror 2.0.18", "tokio", "tracing", ] diff --git a/docs/public/reference/server-operations.mdx b/docs/public/reference/server-operations.mdx index 207f42525..320cd6330 100644 --- a/docs/public/reference/server-operations.mdx +++ b/docs/public/reference/server-operations.mdx @@ -53,6 +53,49 @@ Common flags: See [Server Configuration](/administration/server-configuration) for the full `settings.toml` reference. +### SQLite blob storage activation + +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 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 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 +SQLite-only blobs are hash-validated when read. 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. + +For a non-empty legacy inventory, the first activation also creates the +private sibling backup +`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. 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. + +Keep both the unchanged legacy `blobs/sha256` prefix and the private activation +backup for at least 30 consecutive calendar days after the first successful +production activation. Cleanup is eligible only after a successful cold +activation, a later warm restart that revalidates the backup and byte-compares +every retained legacy blob against SQLite, and 30 days of production observation +with no unresolved inventory, import, verification, integrity, backup, or +checkpoint failure. Scott must review that evidence and explicitly authorize a +separate cleanup change. Day 30 is only the earliest eligibility date; nothing +is deleted automatically, and incomplete evidence extends the support window. + ## Submitting runs Workflows are submitted via the REST API and executed in the background. The exact request body is documented in the API reference: diff --git a/lib/apps/fabro-server/Cargo.toml b/lib/apps/fabro-server/Cargo.toml index 5ad20cfbb..3e305823f 100644 --- a/lib/apps/fabro-server/Cargo.toml +++ b/lib/apps/fabro-server/Cargo.toml @@ -10,7 +10,7 @@ description = "HTTP server for Fabro pipelines" doctest = false [features] -test-support = [] +test-support = ["fabro-store/test-support"] [[test]] name = "it" @@ -74,6 +74,7 @@ tokio.workspace = true serde.workspace = true serde_json.workspace = true serde_yaml = "0.9" +sqlx.workspace = true anyhow.workspace = true async-trait.workspace = true async_zip.workspace = true @@ -117,7 +118,6 @@ tower = "0.5" http-body-util = "0.1" httpmock = "0.8" serde_yaml = "0.9" -sqlx.workspace = true tracing-subscriber.workspace = true tokio-util.workspace = true tokio-tungstenite.workspace = true diff --git a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs new file mode 100644 index 000000000..afd616500 --- /dev/null +++ b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs @@ -0,0 +1,816 @@ +//! Fail-closed activation of SQLite blob storage. +//! +//! This compatibility bridge remains until at least 30 calendar days after +//! the first successful production activation, and until the cold-start, +//! warm-restart, production-observation, and backup-integrity evidence is +//! complete and Scott explicitly approves its removal. The date is an +//! eligibility floor, never an automatic deletion trigger. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use futures_util::TryStreamExt as _; +use object_store::ObjectStore; +use sqlx::Connection as _; +use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection}; +use tokio::fs; +use tokio::task::{JoinError, spawn_blocking}; +use tracing::{debug, info, warn}; + +use crate::server::resource_sampler; + +/// Earliest date this bridge becomes eligible for removal, assuming the first +/// production activation happens no earlier than this change ships. Removal +/// additionally requires the evidence and explicit approval described in the +/// module docs; the date alone never triggers deletion. +pub(crate) const REMOVAL_DEADLINE: &str = "2026-09-22"; + +const DISK_HEADROOM_BYTES: u64 = 64 * 1024 * 1024; +const BACKUP_SUFFIX: &str = ".pre-blob-activation.bak"; +const STAGING_SUFFIX: &str = ".tmp"; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum BlobActivationError { + #[error("canonicalizing the SQLite database path {path}")] + Canonicalize { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("inventorying the legacy blob source")] + Inventory(#[source] fabro_store::LegacyBlobInventoryError), + #[error("reading activation backup metadata at {path}")] + BackupMetadata { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("activation backup is not a regular file at {path}")] + 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, + #[source] + source: sqlx::Error, + }, + #[error("activation backup integrity check did not return exactly one ok result at {path}")] + BackupIntegrityFailed { path: PathBuf }, + #[error("reading SQLite file metadata at {path}")] + SqliteMetadata { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("the blob activation disk requirement overflowed")] + DiskRequirementOverflow, + #[error( + "insufficient disk space for blob activation: {available_bytes} bytes available, {required_bytes} required" + )] + InsufficientDisk { + required_bytes: u64, + available_bytes: u64, + }, + #[error("staging the pre-activation SQLite backup")] + StageBackup(#[source] fabro_db::SnapshotStagingError), + #[error("joining the activation backup publication task")] + JoinBackupPublication(#[source] JoinError), + #[error("publishing the activation backup at {path} without overwriting")] + PublishBackup { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("importing legacy blobs into SQLite")] + Import(#[source] Box), + #[error("verifying legacy and SQLite blobs")] + Verification(#[source] Box), + #[error("running the live SQLite integrity check")] + LiveIntegrity(#[source] sqlx::Error), + #[error("the live SQLite integrity check did not return exactly one ok result")] + LiveIntegrityFailed, + #[error("running the final SQLite WAL truncate checkpoint")] + FinalCheckpoint(#[source] sqlx::Error), +} + +pub(crate) async fn activate_blob_storage( + database: &fabro_db::Database, + sqlite_path: &Path, + object_store: Arc, + slatedb_prefix: String, + flush_interval: Duration, + cache_path: Option, +) -> Result, BlobActivationError> { + let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| { + BlobActivationError::Canonicalize { + path: sqlite_path.to_path_buf(), + source, + } + })?; + let backup_path = fabro_db::append_to_path(&canonical_path, BACKUP_SUFFIX); + info!( + database_path = %canonical_path.display(), + backup_path = %backup_path.display(), + "Starting SQLite blob storage activation" + ); + + let blob_store = Arc::new(fabro_store::BlobStore::new(database.clone_pool())); + let store = Arc::new(fabro_store::Database::new( + object_store, + slatedb_prefix, + flush_interval, + cache_path, + Arc::clone(&blob_store), + )); + + let inventory = store + .legacy_blob_inventory(database.pool()) + .await + .map_err(BlobActivationError::Inventory)?; + let backup_exists = backup_exists(&backup_path).await?; + 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 + // filesystems, an unreadable mount table), so the preflight does too: + // skipping the capacity check must not block a boot the import itself + // could complete. + if let Some(available_free_bytes) = resource_sampler::available_space_for_path(&canonical_path) + { + let backup_reserve = if backup_required { + sqlite_file_set_bytes(&canonical_path).await? + } else { + 0 + }; + // 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, + available_free_bytes, + "Checked SQLite blob activation disk capacity" + ); + } else { + warn!( + database_path = %canonical_path.display(), + "No filesystem mount matched the SQLite database path; skipping the blob activation disk preflight" + ); + } + + let retained_backup = if backup_exists { + Some(backup_path) + } else if backup_required { + create_backup(database.pool(), &backup_path).await?; + Some(backup_path) + } else { + None + }; + + let import = store + .import_legacy_blobs_into(database.pool()) + .await + .map_err(|source| BlobActivationError::Import(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?; + + info!( + legacy_rows = inventory.rows, + legacy_bytes = inventory.bytes, + imported_rows = import.imported_rows, + existing_rows = import.existing_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, + removal_deadline = REMOVAL_DEADLINE, + "Activated SQLite blob storage" + ); + Ok(store) +} + +/// Fail-closed disk capacity check; returns the required free bytes. +fn compute_disk_preflight( + pending_bytes: u64, + backup_reserve: u64, + available_free_bytes: u64, +) -> Result { + let half = pending_bytes + .checked_add(1) + .ok_or(BlobActivationError::DiskRequirementOverflow)? + / 2; + let required_free_bytes = backup_reserve + .checked_add(pending_bytes) + .and_then(|value| value.checked_add(half)) + .and_then(|value| value.checked_add(DISK_HEADROOM_BYTES)) + .ok_or(BlobActivationError::DiskRequirementOverflow)?; + if available_free_bytes < required_free_bytes { + return Err(BlobActivationError::InsufficientDisk { + required_bytes: required_free_bytes, + available_bytes: available_free_bytes, + }); + } + Ok(required_free_bytes) +} + +async fn backup_exists(path: &Path) -> Result { + match fs::metadata(path).await { + Ok(_) => Ok(true), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(source) => Err(BlobActivationError::BackupMetadata { + path: path.to_path_buf(), + source, + }), + } +} + +async fn sqlite_file_set_bytes(path: &Path) -> Result { + let mut total = required_file_bytes(path).await?; + for suffix in ["-wal", "-shm"] { + let sibling = fabro_db::append_to_path(path, suffix); + let bytes = optional_file_bytes(&sibling).await?; + total = total + .checked_add(bytes) + .ok_or(BlobActivationError::DiskRequirementOverflow)?; + } + Ok(total) +} + +async fn required_file_bytes(path: &Path) -> Result { + fs::metadata(path) + .await + .map(|metadata| metadata.len()) + .map_err(|source| BlobActivationError::SqliteMetadata { + path: path.to_path_buf(), + source, + }) +} + +async fn optional_file_bytes(path: &Path) -> Result { + match fs::metadata(path).await { + Ok(metadata) => Ok(metadata.len()), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(0), + Err(source) => Err(BlobActivationError::SqliteMetadata { + path: path.to_path_buf(), + source, + }), + } +} + +async fn create_backup( + pool: &sqlx::SqlitePool, + backup_path: &Path, +) -> Result<(), BlobActivationError> { + let staging_path = fabro_db::append_to_path(backup_path, STAGING_SUFFIX); + fabro_db::write_snapshot_to_staging(pool, &staging_path) + .await + .map_err(BlobActivationError::StageBackup)?; + validate_backup(&staging_path).await?; + + let publish_staging = staging_path.clone(); + let publish_backup = backup_path.to_path_buf(); + let already_exists = spawn_blocking(move || { + let staging = tempfile::TempPath::from_path(publish_staging); + match staging.persist_noclobber(&publish_backup) { + Ok(()) => { + // Make the rename's directory entry durable: the retained + // backup is the documented rollback artifact, so it must not + // vanish in a crash after the import has already committed. + fabro_db::sync_parent_directory(&publish_backup)?; + Ok(false) + } + Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => Ok(true), + Err(error) => Err(error.error), + } + }) + .await + .map_err(BlobActivationError::JoinBackupPublication)? + .map_err(|source| BlobActivationError::PublishBackup { + path: backup_path.to_path_buf(), + source, + })?; + + // The staging copy was validated just before the atomic rename, so only a + // concurrently published file still needs its own validation. + if already_exists { + debug!( + backup_path = %backup_path.display(), + "Reusing concurrently published SQLite blob activation backup" + ); + validate_backup(backup_path).await?; + } + Ok(()) +} + +async fn validate_backup(path: &Path) -> Result<(), BlobActivationError> { + let metadata = + fs::symlink_metadata(path) + .await + .map_err(|source| BlobActivationError::BackupMetadata { + path: path.to_path_buf(), + source, + })?; + if !metadata.is_file() { + return Err(BlobActivationError::BackupNotRegular { + path: path.to_path_buf(), + }); + } + validate_private_permissions(path, &metadata)?; + + let options = SqliteConnectOptions::new() + .filename(path) + .read_only(true) + .immutable(true) + .create_if_missing(false); + let mut connection = SqliteConnection::connect_with(&options) + .await + .map_err(|source| BlobActivationError::BackupIntegrity { + path: path.to_path_buf(), + source, + })?; + let ok = integrity_check_is_ok(&mut connection) + .await + .map_err(|source| BlobActivationError::BackupIntegrity { + path: path.to_path_buf(), + source, + })?; + if !ok { + return Err(BlobActivationError::BackupIntegrityFailed { + path: path.to_path_buf(), + }); + } + Ok(()) +} + +/// Returns whether `PRAGMA integrity_check` reports exactly one `ok` row. +async fn integrity_check_is_ok<'a, E>(executor: E) -> Result +where + E: sqlx::Executor<'a, Database = sqlx::Sqlite>, +{ + let mut rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check").fetch(executor); + let first = rows.try_next().await?; + let second = rows.try_next().await?; + Ok(first.as_deref() == Some("ok") && second.is_none()) +} + +#[cfg(unix)] +fn validate_private_permissions( + path: &Path, + metadata: &std::fs::Metadata, +) -> Result<(), BlobActivationError> { + use std::os::unix::fs::PermissionsExt as _; + + if metadata.permissions().mode() & 0o077 != 0 { + return Err(BlobActivationError::BackupNotPrivate { + path: path.to_path_buf(), + }); + } + Ok(()) +} + +#[cfg(not(unix))] +fn validate_private_permissions( + _path: &Path, + _metadata: &std::fs::Metadata, +) -> Result<(), BlobActivationError> { + Ok(()) +} + +async fn validate_live_integrity(pool: &sqlx::SqlitePool) -> Result<(), BlobActivationError> { + let ok = integrity_check_is_ok(pool) + .await + .map_err(BlobActivationError::LiveIntegrity)?; + if !ok { + return Err(BlobActivationError::LiveIntegrityFailed); + } + Ok(()) +} + +async fn final_truncate_checkpoint(pool: &sqlx::SqlitePool) -> Result<(), BlobActivationError> { + let (busy, _, _): (i64, i64, i64) = sqlx::query_as("PRAGMA wal_checkpoint(TRUNCATE)") + .fetch_one(pool) + .await + .map_err(BlobActivationError::FinalCheckpoint)?; + if busy != 0 { + // A concurrent reader (a backup tool, a replication agent, an + // operator shell) can keep the WAL from truncating. An untruncated + // WAL threatens no data integrity, so it must not block startup; a + // later checkpoint truncates once the reader is gone. + warn!("The final SQLite WAL truncate checkpoint could not complete; continuing startup"); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use fabro_db::append_to_path; + use object_store::ObjectStore; + use object_store::memory::InMemory; + use tokio::fs; + + use super::{ + BACKUP_SUFFIX, BlobActivationError, DISK_HEADROOM_BYTES, activate_blob_storage, + compute_disk_preflight, create_backup, final_truncate_checkpoint, sqlite_file_set_bytes, + validate_backup, + }; + + type TestResult = Result>; + + #[test] + fn disk_preflight_passes_at_equality_and_fails_one_byte_below() { + let pending_bytes = 3; + let backup_reserve = 10; + let required = backup_reserve + pending_bytes + 2 + DISK_HEADROOM_BYTES; + + 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(pending_bytes, backup_reserve, required - 1) + .expect_err("one byte below must fail"); + assert!(matches!( + error, + BlobActivationError::InsufficientDisk { .. } + )); + } + + #[test] + fn disk_preflight_requires_only_headroom_without_a_backup_reserve() { + let required_free_bytes = + compute_disk_preflight(2, 0, u64::MAX).expect("available capacity should pass"); + assert_eq!(required_free_bytes, 3 + DISK_HEADROOM_BYTES); + } + + #[test] + fn disk_preflight_fails_closed_on_overflow() { + let error = + compute_disk_preflight(u64::MAX, 1, u64::MAX).expect_err("overflow must fail closed"); + assert!(matches!( + error, + BlobActivationError::DiskRequirementOverflow + )); + } + + #[tokio::test] + async fn disk_preflight_counts_the_sqlite_file_set_for_a_required_backup() -> TestResult<()> { + let directory = tempfile::tempdir()?; + let sqlite_path = directory.path().join("fabro.sqlite3"); + fs::write(&sqlite_path, [0_u8; 3]).await?; + fs::write(append_to_path(&sqlite_path, "-wal"), [0_u8; 5]).await?; + fs::write(append_to_path(&sqlite_path, "-shm"), [0_u8; 7]).await?; + + assert_eq!(sqlite_file_set_bytes(&sqlite_path).await?, 15); + Ok(()) + } + + #[tokio::test] + async fn backup_is_private_integrity_clean_and_does_not_create_journal_siblings() + -> 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 backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX); + + create_backup(database.pool(), &backup_path).await?; + validate_backup(&backup_path).await?; + + assert!(backup_path.is_file()); + assert!(!append_to_path(&backup_path, "-wal").exists()); + assert!(!append_to_path(&backup_path, "-shm").exists()); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + assert_eq!( + std::fs::metadata(&backup_path)?.permissions().mode() & 0o077, + 0 + ); + } + Ok(()) + } + + #[tokio::test] + async fn backup_publication_never_overwrites_an_existing_valid_backup() -> 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 backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX); + create_backup(database.pool(), &backup_path).await?; + let original = fs::read(&backup_path).await?; + + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(fabro_types::BlobHash::new(b"later").to_string()) + .bind(b"later".as_slice()) + .execute(database.pool()) + .await?; + create_backup(database.pool(), &backup_path).await?; + + assert_eq!(fs::read(&backup_path).await?, original); + Ok(()) + } + + #[tokio::test] + async fn failed_backup_copy_never_publishes_a_destination() -> 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 backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX); + database.pool().close().await; + + let error = create_backup(database.pool(), &backup_path) + .await + .expect_err("a closed pool must fail backup creation"); + + assert!(matches!( + error, + BlobActivationError::StageBackup(fabro_db::SnapshotStagingError::Write { .. }) + )); + assert!(!backup_path.exists()); + Ok(()) + } + + #[tokio::test] + async fn cold_activation_and_warm_restart_share_verified_sqlite_blobs() -> 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 = Arc::new(InMemory::new()); + let source = fabro_store::test_support::test_database( + Arc::clone(&object_store), + "activation-test", + Duration::from_millis(1), + None, + ); + let legacy_bytes = b"legacy-blob"; + let legacy_hash = fabro_store::test_support::put_legacy_blob(&source, legacy_bytes).await?; + drop(source); + + let store = activate_blob_storage( + &database, + &sqlite_path, + Arc::clone(&object_store), + "activation-test".to_string(), + Duration::from_millis(1), + None, + ) + .await?; + assert_eq!( + store.blobs().read(&legacy_hash).await?.as_deref(), + Some(legacy_bytes.as_slice()) + ); + + let backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX); + let original_backup = fs::read(&backup_path).await?; + let run_id = fabro_types::RunId::new(); + let writer = store.create_run(&run_id).await?; + let reader = store.open_run_reader(&run_id).await?; + let sqlite_only_bytes = b"written-after-activation"; + let sqlite_only_hash = writer.write_blob(sqlite_only_bytes).await?; + assert_eq!( + reader.read_blob(&sqlite_only_hash).await?.as_deref(), + Some(sqlite_only_bytes.as_slice()) + ); + drop(reader); + drop(writer); + drop(store); + + let warm = activate_blob_storage( + &database, + &sqlite_path, + object_store, + "activation-test".to_string(), + Duration::from_millis(1), + None, + ) + .await?; + assert_eq!(fs::read(&backup_path).await?, original_backup); + assert_eq!( + warm.blobs().read(&sqlite_only_hash).await?.as_deref(), + Some(sqlite_only_bytes.as_slice()) + ); + 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 = 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()?; + let sqlite_path = directory.path().join("fabro.sqlite3"); + let database = fabro_db::Database::connect(&sqlite_path).await?; + database.migrate().await?; + let bytes = b"sqlite-only"; + let hash = fabro_types::BlobHash::new(bytes); + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(hash.to_string()) + .bind(bytes.as_slice()) + .execute(database.pool()) + .await?; + + let activated = activate_blob_storage( + &database, + &sqlite_path, + Arc::new(InMemory::new()), + "empty-activation-test".to_string(), + Duration::from_millis(1), + None, + ) + .await?; + + assert!(!append_to_path(&sqlite_path, BACKUP_SUFFIX).exists()); + assert_eq!( + activated.blobs().read(&hash).await?.as_deref(), + Some(bytes.as_slice()) + ); + Ok(()) + } + + #[tokio::test] + async fn busy_final_checkpoint_warns_and_does_not_fail_startup() -> TestResult<()> { + use sqlx::Connection as _; + use sqlx::sqlite::{ + SqliteConnectOptions, SqliteConnection, SqliteJournalMode, SqlitePoolOptions, + }; + + let directory = tempfile::tempdir()?; + let sqlite_path = directory.path().join("fabro.sqlite3"); + let database = fabro_db::Database::connect(&sqlite_path).await?; + database.migrate().await?; + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(fabro_types::BlobHash::new(b"wal-content").to_string()) + .bind(b"wal-content".as_slice()) + .execute(database.pool()) + .await?; + + // A reader holding an open snapshot models a backup tool or operator + // shell that outlives the checkpoint's busy timeout. + let reader_options = SqliteConnectOptions::new() + .filename(&sqlite_path) + .read_only(true) + .create_if_missing(false); + let mut reader = SqliteConnection::connect_with(&reader_options).await?; + sqlx::query("BEGIN").execute(&mut reader).await?; + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM blobs") + .fetch_one(&mut reader) + .await?; + + // A short busy timeout keeps the blocked truncate from stalling the + // test for the production pool's full five seconds. + let checkpoint_options = SqliteConnectOptions::new() + .filename(&sqlite_path) + .journal_mode(SqliteJournalMode::Wal) + .busy_timeout(Duration::from_millis(50)) + .create_if_missing(false); + let checkpoint_pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(checkpoint_options) + .await?; + + final_truncate_checkpoint(&checkpoint_pool).await?; + + // The reader really did block the truncate: the WAL was not reset. + let wal_bytes = fs::metadata(append_to_path(&sqlite_path, "-wal")) + .await? + .len(); + assert!(wal_bytes > 0, "the WAL should remain untruncated"); + drop(reader); + Ok(()) + } + + #[tokio::test] + async fn invalid_retained_backup_fails_before_importing() -> 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 = Arc::new(InMemory::new()); + let source = fabro_store::test_support::test_database( + Arc::clone(&object_store), + "invalid-backup-test", + Duration::from_millis(1), + None, + ); + fabro_store::test_support::put_legacy_blob(&source, b"must-not-import").await?; + drop(source); + + let backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX); + fs::write(&backup_path, b"not a database").await?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(&backup_path, std::fs::Permissions::from_mode(0o600)).await?; + } + + let error = activate_blob_storage( + &database, + &sqlite_path, + object_store, + "invalid-backup-test".to_string(), + Duration::from_millis(1), + None, + ) + .await + .expect_err("an invalid retained backup must fail closed"); + assert!(matches!( + error, + BlobActivationError::BackupIntegrity { .. } + | BlobActivationError::BackupIntegrityFailed { .. } + )); + let destination_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs") + .fetch_one(database.pool()) + .await?; + assert_eq!(destination_rows, 0); + Ok(()) + } +} diff --git a/lib/apps/fabro-server/src/migrations.rs b/lib/apps/fabro-server/src/migrations.rs index 133dc885a..53d2cffb5 100644 --- a/lib/apps/fabro-server/src/migrations.rs +++ b/lib/apps/fabro-server/src/migrations.rs @@ -7,9 +7,12 @@ use fabro_vault::SecretStore; mod legacy_vault_entries; #[path = "../migrations/2026052501_optional_server_env_secrets_to_vault.rs"] mod optional_server_env_secrets_to_vault; +#[path = "../migrations/2026082301_sqlite_blob_activation.rs"] +mod sqlite_blob_activation; pub(crate) use legacy_vault_entries::REMOVAL_DEADLINE as LEGACY_VAULT_REMOVAL_DEADLINE; pub(crate) use optional_server_env_secrets_to_vault::REMOVAL_DEADLINE as OPTIONAL_SERVER_ENV_SECRETS_REMOVAL_DEADLINE; +pub(crate) use sqlite_blob_activation::activate_blob_storage; pub(crate) type LegacyVaultMigrationReport = legacy_vault_entries::LegacyVaultMigrationReport; pub(crate) type OptionalServerEnvSecretsMigrationReport = diff --git a/lib/apps/fabro-server/src/run_intent.rs b/lib/apps/fabro-server/src/run_intent.rs index 3614773d1..c122ceb69 100644 --- a/lib/apps/fabro-server/src/run_intent.rs +++ b/lib/apps/fabro-server/src/run_intent.rs @@ -16,11 +16,6 @@ use crate::run_compiler::{RunCompilerError, settings_layer_with_resolved_dockerf #[derive(Debug, Error)] pub(crate) enum RunIntentAdmissionError { - #[error("workflow-version storage could not be opened")] - StoreOpen { - #[source] - source: fabro_store::Error, - }, #[error("workflow-version closure could not be loaded")] VersionStore { #[source] @@ -385,7 +380,7 @@ mod tests { #[tokio::test] async fn lowers_nested_entrypoints_and_inlines_goal_files() { let (database, _) = crate::test_support::test_store_bundle(); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let store = WorkflowVersionStore::new(blobs); let grandchild = version( "deep/leaf.fabro", @@ -453,7 +448,7 @@ mod tests { #[tokio::test] async fn lowers_same_version_at_distinct_mount_paths() { let (database, _) = crate::test_support::test_store_bundle(); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let store = WorkflowVersionStore::new(blobs); let child = version( "pkg/child.fabro", @@ -494,7 +489,7 @@ mod tests { #[tokio::test] async fn rejects_closures_that_expand_past_the_mount_limit() { let (database, _) = crate::test_support::test_store_bundle(); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let store = WorkflowVersionStore::new(blobs); // A chain of tiny versions where each level mounts the next twice is // cheap to store and load (the closure dedupes by id) but expands to @@ -527,7 +522,7 @@ mod tests { #[tokio::test] async fn rejects_distinct_versions_that_converge_on_one_mount_path() { let (database, _) = crate::test_support::test_store_bundle(); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let store = WorkflowVersionStore::new(blobs); let first_leaf = version( "leaf/first.fabro", @@ -584,7 +579,7 @@ mod tests { #[tokio::test] async fn rejects_rebased_files_that_escape_the_runtime_root() { let (database, _) = crate::test_support::test_store_bundle(); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let store = WorkflowVersionStore::new(blobs); let child = version( "nested/child.fabro", diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index c263921d1..a9cfc51af 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -773,12 +773,16 @@ where } else { None }; - let store = Arc::new(fabro_store::Database::new( + let store = migrations::activate_blob_storage( + &database, + &sqlite_path, object_store, slatedb_prefix, flush_interval, cache_path, - )); + ) + .await + .context("activating SQLite blob storage")?; let auth_code_store = store.auth_codes().await?; // Refresh tokens now live in SQLite. Nothing reads the old records and no // reaper collects them any more, so clear them out once rather than diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 5c3e03d16..9500f578b 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -175,7 +175,7 @@ use crate::{ mod automation_scheduler; mod handler; mod pull_request_supervisor; -mod resource_sampler; +pub(crate) mod resource_sampler; mod session_runtime; pub(crate) use automation_scheduler::spawn_automation_scheduler; diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index 6369e7b24..ecf7100f5 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -624,12 +624,7 @@ async fn create_run_from_intent( Ok(id) => id, Err(error) => return run_intent_admission_error(error.into()), }; - let blobs = match state.store_ref().blobs().await { - Ok(blobs) => blobs, - Err(source) => { - return run_intent_admission_error(RunIntentAdmissionError::StoreOpen { source }); - } - }; + let blobs = state.store_ref().blobs(); let version_store = fabro_workflow_version::WorkflowVersionStore::new(blobs); let closure = match version_store.get_closure(&intent.workflow_version_id).await { Ok(Some(closure)) => closure, @@ -964,8 +959,7 @@ fn intent_error(status: StatusCode, detail: impl Into, code: &'static st fn run_intent_admission_error(error: RunIntentAdmissionError) -> Response { match &error { - RunIntentAdmissionError::StoreOpen { .. } - | RunIntentAdmissionError::VersionStore { .. } + RunIntentAdmissionError::VersionStore { .. } | RunIntentAdmissionError::VariableSnapshot { .. } | RunIntentAdmissionError::Environment(EnvironmentSelectionError::CredentialStore { .. @@ -987,8 +981,7 @@ fn run_intent_admission_error(error: RunIntentAdmissionError) -> Response { } match error { - RunIntentAdmissionError::StoreOpen { .. } - | RunIntentAdmissionError::VersionStore { .. } => intent_error( + RunIntentAdmissionError::VersionStore { .. } => intent_error( StatusCode::INTERNAL_SERVER_ERROR, "workflow version store operation failed", "workflow_version_store_error", diff --git a/lib/apps/fabro-server/src/server/handler/workflow_versions.rs b/lib/apps/fabro-server/src/server/handler/workflow_versions.rs index 1ded22adf..69845679b 100644 --- a/lib/apps/fabro-server/src/server/handler/workflow_versions.rs +++ b/lib/apps/fabro-server/src/server/handler/workflow_versions.rs @@ -38,14 +38,7 @@ async fn create_workflow_version( INVALID_VERSION_CODE, ) })?; - let blobs = state.store_ref().blobs().await.map_err(|err| { - tracing::error!( - error = %err, - error_chain = ?error::collect_chain(&err), - "Failed to open workflow version storage" - ); - internal_store_error() - })?; + let blobs = state.store_ref().blobs(); let store = WorkflowVersionStore::new(blobs); let workflow_version_id = store.put(&version).await.map_err(store_error)?; @@ -197,8 +190,6 @@ mod tests { state .store_ref() .blobs() - .await - .unwrap() .read(&id.into()) .await .unwrap() @@ -260,16 +251,7 @@ mod tests { .await; assert_eq!(error_code(&body), INVALID_VERSION_CODE); - assert!( - !state - .store_ref() - .blobs() - .await - .unwrap() - .exists(&id.into()) - .await - .unwrap() - ); + assert!(!state.store_ref().blobs().exists(&id.into()).await.unwrap()); } #[tokio::test] @@ -303,8 +285,6 @@ mod tests { state .store_ref() .blobs() - .await - .unwrap() .write(b"not a workflow version") .await .unwrap(), diff --git a/lib/apps/fabro-server/src/server/resource_sampler.rs b/lib/apps/fabro-server/src/server/resource_sampler.rs index d9353b938..e9e846ea6 100644 --- a/lib/apps/fabro-server/src/server/resource_sampler.rs +++ b/lib/apps/fabro-server/src/server/resource_sampler.rs @@ -254,12 +254,8 @@ fn compute_fabro_storage_usage( }) } -fn sample_disk_resources( - storage_path: &Path, - fabro_usage: FabroStorageUsage, -) -> SystemDiskResources { - let disks = Disks::new_with_refreshed_list(); - let candidates = disks +fn refreshed_disk_candidates() -> Vec { + Disks::new_with_refreshed_list() .list() .iter() .map(|disk| DiskCandidate { @@ -268,7 +264,14 @@ fn sample_disk_resources( total_bytes: disk.total_space(), available_bytes: disk.available_space(), }) - .collect::>(); + .collect() +} + +fn sample_disk_resources( + storage_path: &Path, + fabro_usage: FabroStorageUsage, +) -> SystemDiskResources { + let candidates = refreshed_disk_candidates(); let Some(disk) = select_storage_disk(storage_path, &candidates) else { return SystemDiskResources { @@ -339,6 +342,10 @@ fn select_storage_disk<'a>( .max_by_key(|disk| disk.mount_point.components().count()) } +pub(crate) fn available_space_for_path(storage_path: &Path) -> Option { + select_storage_disk(storage_path, &refreshed_disk_candidates()).map(|disk| disk.available_bytes) +} + fn percent(used: u64, total: u64) -> Option { if total == 0 { return None; diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 0c4c63515..43b1e434a 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -3574,7 +3574,7 @@ async fn store_workflow_version( ) .unwrap(); let version = fabro_workflow_version::ValidatedWorkflowVersion::new(version).unwrap(); - let blobs = state.store_ref().blobs().await.unwrap(); + let blobs = state.store_ref().blobs(); fabro_workflow_version::WorkflowVersionStore::new(blobs) .put(&version) .await diff --git a/lib/apps/fabro-server/src/test_support.rs b/lib/apps/fabro-server/src/test_support.rs index a28e38d56..a7a9f1432 100644 --- a/lib/apps/fabro-server/src/test_support.rs +++ b/lib/apps/fabro-server/src/test_support.rs @@ -22,7 +22,7 @@ use fabro_model::catalog::{LlmCatalogSettings, ProviderCatalogSettings}; use fabro_model::{Catalog, ProviderId}; use fabro_sandbox::SandboxProviderRegistry; use fabro_static::EnvVars; -use fabro_store::{ArtifactStore, Database}; +use fabro_store::{ArtifactStore, Database, test_support as store_test_support}; use fabro_types::settings::ServerAuthMethod; use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{AuthMethod, IdpIdentity, ServerSettings}; @@ -539,7 +539,7 @@ pub fn test_app_state_with_store( pub fn test_store_bundle() -> (Arc, ArtifactStore) { let object_store: Arc = Arc::new(MemoryObjectStore::new()); - let store = Arc::new(fabro_store::Database::new( + let store = Arc::new(store_test_support::test_database( Arc::clone(&object_store), "", Duration::from_millis(1), diff --git a/lib/apps/fabro-server/tests/it/api/auth_sessions.rs b/lib/apps/fabro-server/tests/it/api/auth_sessions.rs index be8300a23..d585aa949 100644 --- a/lib/apps/fabro-server/tests/it/api/auth_sessions.rs +++ b/lib/apps/fabro-server/tests/it/api/auth_sessions.rs @@ -9,8 +9,8 @@ use fabro_server::jwt_auth::resolve_auth_mode_with_lookup; use fabro_server::server::{AppState, RouterOptions, build_router_with_options}; use fabro_server::test_support::{TEST_SESSION_SECRET, TestAppStateBuilder}; use fabro_server::web_auth::{SESSION_COOKIE_NAME, SessionCookie}; +use fabro_store::ArtifactStore; use fabro_store::auth_session_store::{AuthSessionRecord, InitialRefreshToken}; -use fabro_store::{ArtifactStore, Database}; use hkdf::Hkdf; use object_store::memory::InMemory; use sha2::Sha256; @@ -22,7 +22,7 @@ use crate::helpers::{response_json, response_status, settings_from_toml}; fn test_app(source: &str) -> (axum::Router, Arc) { let settings = settings_from_toml(source); let object_store: Arc = Arc::new(InMemory::new()); - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::clone(&object_store), "", Duration::from_millis(1), diff --git a/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs b/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs index fc6ded632..5afd53fe3 100644 --- a/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs +++ b/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs @@ -19,7 +19,7 @@ use crate::helpers::{body_json, settings_from_toml}; fn test_app(source: &str) -> (axum::Router, Arc, Arc) { let settings = settings_from_toml(source); let object_store: Arc = Arc::new(InMemory::new()); - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::clone(&object_store), "", Duration::from_millis(1), diff --git a/lib/apps/fabro-server/tests/it/api/events.rs b/lib/apps/fabro-server/tests/it/api/events.rs index a5c8153c7..d0b448511 100644 --- a/lib/apps/fabro-server/tests/it/api/events.rs +++ b/lib/apps/fabro-server/tests/it/api/events.rs @@ -14,7 +14,7 @@ use crate::helpers::{MINIMAL_DOT, api, minimal_manifest_json, response_json, tes fn app_with_store(object_store: Arc) -> axum::Router { let settings = test_settings(); - let store = Arc::new(fabro_store::Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::clone(&object_store), "event-race", Duration::from_millis(1), diff --git a/lib/apps/fabro-server/tests/it/api/run_files.rs b/lib/apps/fabro-server/tests/it/api/run_files.rs index e30d09582..5e12786d5 100644 --- a/lib/apps/fabro-server/tests/it/api/run_files.rs +++ b/lib/apps/fabro-server/tests/it/api/run_files.rs @@ -39,7 +39,7 @@ fn files_url_with_scope(run_id: &str, scope: &str) -> String { fn store_bundle() -> (Arc, ArtifactStore) { let object_store: Arc = Arc::new(MemoryObjectStore::new()); - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::clone(&object_store), "", Duration::from_millis(1), diff --git a/lib/components/fabro-store/Cargo.toml b/lib/components/fabro-store/Cargo.toml index 588e285a0..0ad54c39a 100644 --- a/lib/components/fabro-store/Cargo.toml +++ b/lib/components/fabro-store/Cargo.toml @@ -12,9 +12,10 @@ doctest = false workspace = true [features] -test-support = [] +test-support = ["dep:fabro-db"] [dependencies] +fabro-db = { path = "../../foundation/fabro-db", optional = true } fabro-types = { path = "../../foundation/fabro-types" } fabro-util = { path = "../../foundation/fabro-util" } hex.workspace = true diff --git a/lib/components/fabro-store/src/blob_store.rs b/lib/components/fabro-store/src/blob_store.rs index 54b8124d3..e5d0dc128 100644 --- a/lib/components/fabro-store/src/blob_store.rs +++ b/lib/components/fabro-store/src/blob_store.rs @@ -1,10 +1,13 @@ +#[cfg(test)] use std::sync::Arc; use bytes::Bytes; use fabro_types::BlobHash; use sqlx::SqlitePool; -use crate::record::{RawBytesCodec, Record, Repository}; +#[cfg(test)] +use crate::record::Repository; +use crate::record::{RawBytesCodec, Record}; use crate::{Error, Result}; #[derive(Debug, Clone, PartialEq, Eq)] @@ -33,15 +36,15 @@ impl Record for Blob { } } -/// Which storage engine holds the blobs. +/// Temporary backend split for compatibility tests. /// -/// This enum is a transition vehicle, not a permanent abstraction: `Slate` -/// preserves current production behavior while the SQLite backend rolls out. -/// Once runtime blob storage switches to SQLite and legacy blobs are -/// imported, delete the `Slate` arm (and this enum) and inline the SQLite -/// implementation into [`BlobStore`]. The SQLite arm's semantics — verified -/// reads and loud failure on hash conflicts — are the intended end state. +/// Production compiles only the SQLite arm. The Slate arm remains test-only +/// while the startup import bridge is supported, so tests can exercise the +/// old-source boundary directly. Delete the Slate arm (and this enum) with the +/// separately authorized compatibility cleanup, then inline SQLite into +/// [`BlobStore`]. enum BlobBackend { + #[cfg(test)] Slate(Repository), Sqlite(SqlitePool), } @@ -53,6 +56,7 @@ pub struct BlobStore { impl std::fmt::Debug for BlobStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let backend = match &self.backend { + #[cfg(test)] BlobBackend::Slate(_) => "slate", BlobBackend::Sqlite(_) => "sqlite", }; @@ -71,21 +75,16 @@ impl BlobStore { } } + #[cfg(test)] pub(crate) fn from_slate(db: Arc) -> Self { Self { backend: BlobBackend::Slate(Repository::new(db)), } } - pub(crate) fn sqlite_pool_for_legacy_import(&self) -> Option<&SqlitePool> { - match &self.backend { - BlobBackend::Slate(_) => None, - BlobBackend::Sqlite(pool) => Some(pool), - } - } - pub async fn write(&self, bytes: &[u8]) -> Result { match &self.backend { + #[cfg(test)] BlobBackend::Slate(repo) => { let blob = Blob(Bytes::copy_from_slice(bytes)); let id = blob.id(); @@ -122,6 +121,7 @@ impl BlobStore { pub async fn read(&self, blob_hash: &BlobHash) -> Result> { match &self.backend { + #[cfg(test)] BlobBackend::Slate(repo) => Ok(repo.get(blob_hash).await?.map(|blob| blob.0)), BlobBackend::Sqlite(pool) => { let stored: Option> = @@ -144,6 +144,7 @@ impl BlobStore { pub async fn exists(&self, blob_hash: &BlobHash) -> Result { match &self.backend { + #[cfg(test)] BlobBackend::Slate(repo) => repo.exists(blob_hash).await, BlobBackend::Sqlite(pool) => { let exists: bool = @@ -160,26 +161,24 @@ impl BlobStore { #[cfg(test)] mod tests { use std::sync::Arc; - use std::time::Duration; use bytes::Bytes; use fabro_types::BlobHash; use object_store::memory::InMemory; use super::BlobStore; + use crate::Error; use crate::keys::SlateKey; - use crate::{Database, Error}; type TestResult = std::result::Result>; async fn slate_store() -> Arc { - let db = Database::new( - Arc::new(InMemory::new()), - "", - Duration::from_millis(1), - None, + let raw_db = Arc::new( + slatedb::Db::open("blob-store-tests", Arc::new(InMemory::new())) + .await + .unwrap(), ); - db.blobs().await.unwrap() + Arc::new(BlobStore::from_slate(raw_db)) } async fn raw_slate_store(name: &str) -> (Arc, BlobStore) { diff --git a/lib/components/fabro-store/src/legacy_blob_import.rs b/lib/components/fabro-store/src/legacy_blob_import.rs index d9ea4ba93..cc4739db7 100644 --- a/lib/components/fabro-store/src/legacy_blob_import.rs +++ b/lib/components/fabro-store/src/legacy_blob_import.rs @@ -8,19 +8,202 @@ 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}; +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; -use crate::{BlobStore, Database}; const MAX_BATCH_ROWS: usize = 100; const MAX_BATCH_BYTES: u64 = 1024 * 1024; 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, + /// 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. +pub struct LegacyBlobInventoryError { + report: LegacyBlobInventory, + failure: LegacyBlobInventoryFailure, +} + +impl LegacyBlobInventoryError { + #[must_use] + pub fn report(&self) -> &LegacyBlobInventory { + &self.report + } +} + +impl fmt::Debug for LegacyBlobInventoryError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyBlobInventoryError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyBlobInventoryError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy blob inventory failed after scanning {} rows: {}", + self.report.rows, self.failure + ) + } +} + +impl StdError for LegacyBlobInventoryError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.failure) + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyBlobInventoryFailure { + #[error("opening the legacy blob source")] + OpenSource(#[source] crate::Error), + #[error("opening the legacy blob scan")] + OpenSourceScan(#[source] slatedb::Error), + #[error("reading the legacy blob scan")] + 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, +} + +impl LegacyBlobInventoryFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyBlobInventoryFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyBlobInventoryFailure") + .field("kind", &self.kind()) + .finish() + } +} + +/// Aggregate proof produced by a complete legacy-source and SQLite-target +/// verification pass. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyBlobVerificationReport { + pub source_rows: u64, + pub source_bytes: u64, + pub matched_rows: u64, + pub matched_bytes: u64, + pub target_rows: u64, + pub target_bytes: u64, + pub missing_rows: u64, + pub invalid_source_rows: u64, + pub invalid_target_rows: u64, + pub conflicting_rows: u64, +} + +/// A failed complete verification and its aggregate partial report. +pub struct LegacyBlobVerificationError { + report: LegacyBlobVerificationReport, + failure: LegacyBlobVerificationFailure, +} + +impl LegacyBlobVerificationError { + #[must_use] + pub fn report(&self) -> &LegacyBlobVerificationReport { + &self.report + } +} + +impl fmt::Debug for LegacyBlobVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyBlobVerificationError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyBlobVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy blob verification failed after checking {} source rows and {} target rows: {}", + self.report.source_rows, self.report.target_rows, self.failure + ) + } +} + +impl StdError for LegacyBlobVerificationError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.failure) + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyBlobVerificationFailure { + #[error("opening the legacy blob source")] + OpenSource(#[source] crate::Error), + #[error("opening the legacy blob scan")] + OpenSourceScan(#[source] slatedb::Error), + #[error("reading the legacy blob scan")] + 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("reading a SQLite blob row for legacy verification")] + ReadDestination(#[source] sqlx::Error), + #[error("SQLite is missing a legacy blob row")] + MissingDestination, + #[error("SQLite contains different bytes for a legacy blob hash")] + DestinationConflict, + #[error("scanning SQLite blob rows")] + ScanTarget(#[source] sqlx::Error), + #[error("a SQLite blob hash is not canonical")] + InvalidTargetHash, + #[error("SQLite blob bytes do not match their hash")] + TargetDigestMismatch, + #[error("a legacy blob verification counter overflowed")] + CounterOverflow, +} + +impl LegacyBlobVerificationFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyBlobVerificationFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyBlobVerificationFailure") + .field("kind", &self.kind()) + .finish() + } +} + /// Aggregate progress from one legacy blob import attempt. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct LegacyBlobImportReport { @@ -101,9 +284,6 @@ impl StdError for LegacyBlobImportError { #[derive(strum::IntoStaticStr, thiserror::Error)] #[strum(serialize_all = "snake_case")] enum LegacyBlobImportFailure { - #[error("the legacy blob import target is not backed by SQLite")] - WrongTargetBackend, - #[error("opening the legacy blob source")] OpenSource(#[source] crate::Error), @@ -159,12 +339,6 @@ enum LegacyBlobImportFailure { #[error("the passive SQLite WAL checkpoint could not complete")] PassiveCheckpointBusy, - #[error("running the final SQLite WAL checkpoint")] - FinalCheckpoint(#[source] sqlx::Error), - - #[error("the final SQLite WAL checkpoint could not complete")] - FinalCheckpointBusy, - #[error("restoring the SQLite automatic checkpoint setting")] RestoreAutomaticCheckpoint { #[source] @@ -245,8 +419,6 @@ struct ImportControls { #[cfg(test)] passive_checkpoint: bool, #[cfg(test)] - final_checkpoint: bool, - #[cfg(test)] restore_automatic_checkpoint: bool, } @@ -281,17 +453,6 @@ impl ImportControls { None } - fn final_checkpoint_error(&self) -> Option { - #[cfg(test)] - if self.final_checkpoint { - return Some(sqlx::Error::Protocol( - "injected final checkpoint failure".to_owned(), - )); - } - let _ = self; - None - } - fn restore_automatic_checkpoint_error(&self) -> Option { #[cfg(test)] if self.restore_automatic_checkpoint { @@ -364,6 +525,158 @@ struct BatchReport { } impl Database { + /// 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. 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(pool, &mut report).await; + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyBlobInventoryError { report, failure }), + } + } + + async fn run_legacy_blob_inventory( + &self, + pool: &SqlitePool, + report: &mut LegacyBlobInventory, + ) -> Result<(), LegacyBlobInventoryFailure> { + let source = self + .open_db() + .await + .map_err(LegacyBlobInventoryFailure::OpenSource)?; + let prefix = legacy_blob_prefix(); + let mut entries = source + .scan_prefix(&prefix) + .await + .map_err(LegacyBlobInventoryFailure::OpenSourceScan)?; + while let Some(entry) = entries + .next() + .await + .map_err(LegacyBlobInventoryFailure::ReadSourceScan)? + { + inventory_checked_add(&mut report.rows, 1)?; + 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(()) + } + + /// Verifies every legacy blob against SQLite and validates every SQLite + /// blob row independently. + pub async fn verify_legacy_blobs_in( + &self, + pool: &SqlitePool, + ) -> std::result::Result { + let mut report = LegacyBlobVerificationReport::default(); + let result = self.run_legacy_blob_verification(pool, &mut report).await; + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyBlobVerificationError { report, failure }), + } + } + + async fn run_legacy_blob_verification( + &self, + pool: &SqlitePool, + report: &mut LegacyBlobVerificationReport, + ) -> Result<(), LegacyBlobVerificationFailure> { + let source = self + .open_db() + .await + .map_err(LegacyBlobVerificationFailure::OpenSource)?; + let prefix = legacy_blob_prefix(); + let mut entries = source + .scan_prefix(&prefix) + .await + .map_err(LegacyBlobVerificationFailure::OpenSourceScan)?; + while let Some(entry) = entries + .next() + .await + .map_err(LegacyBlobVerificationFailure::ReadSourceScan)? + { + verification_checked_add(&mut report.source_rows, 1)?; + let value_bytes = verification_usize_to_u64(entry.value.len())?; + verification_checked_add(&mut report.source_bytes, value_bytes)?; + let hash = match validate_source_entry_common(&entry.key, &entry.value, &prefix) { + Ok(hash) => hash, + Err(SourceEntryFailure::InvalidKey) => { + verification_checked_add(&mut report.invalid_source_rows, 1)?; + return Err(LegacyBlobVerificationFailure::InvalidSourceKey); + } + Err(SourceEntryFailure::DigestMismatch) => { + verification_checked_add(&mut report.invalid_source_rows, 1)?; + return Err(LegacyBlobVerificationFailure::SourceDigestMismatch); + } + }; + let equal: Option = + sqlx::query_scalar("SELECT data = ? FROM blobs WHERE hash = ?") + .bind(entry.value.as_ref()) + .bind(hash.to_string()) + .fetch_optional(pool) + .await + .map_err(LegacyBlobVerificationFailure::ReadDestination)?; + match equal { + Some(true) => { + verification_checked_add(&mut report.matched_rows, 1)?; + verification_checked_add(&mut report.matched_bytes, value_bytes)?; + } + Some(false) => { + verification_checked_add(&mut report.conflicting_rows, 1)?; + return Err(LegacyBlobVerificationFailure::DestinationConflict); + } + None => { + verification_checked_add(&mut report.missing_rows, 1)?; + return Err(LegacyBlobVerificationFailure::MissingDestination); + } + } + } + + let mut rows = + sqlx::query_as::<_, (String, Vec)>("SELECT hash, data FROM blobs").fetch(pool); + while let Some((hash_text, bytes)) = rows + .try_next() + .await + .map_err(LegacyBlobVerificationFailure::ScanTarget)? + { + verification_checked_add(&mut report.target_rows, 1)?; + verification_checked_add( + &mut report.target_bytes, + verification_usize_to_u64(bytes.len())?, + )?; + let Some(hash) = parse_canonical_hash(&hash_text) else { + verification_checked_add(&mut report.invalid_target_rows, 1)?; + return Err(LegacyBlobVerificationFailure::InvalidTargetHash); + }; + if BlobHash::new(&bytes) != hash { + verification_checked_add(&mut report.invalid_target_rows, 1)?; + return Err(LegacyBlobVerificationFailure::TargetDigestMismatch); + } + } + Ok(()) + } + /// Strictly imports the legacy SlateDB blob keyspace into a SQLite blob /// store. /// @@ -374,20 +687,20 @@ impl Database { /// cleanup fails. pub async fn import_legacy_blobs_into( &self, - target: &BlobStore, + pool: &SqlitePool, ) -> std::result::Result { - self.import_legacy_blobs_with_controls(target, &ImportControls::default()) + self.import_legacy_blobs_with_controls(pool, &ImportControls::default()) .await } async fn import_legacy_blobs_with_controls( &self, - target: &BlobStore, + pool: &SqlitePool, controls: &ImportControls, ) -> std::result::Result { let mut report = LegacyBlobImportReport::default(); let result = self - .run_legacy_blob_import(target, controls, &mut report) + .run_legacy_blob_import(pool, controls, &mut report) .await; match result { @@ -397,20 +710,25 @@ 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) } } } async fn run_legacy_blob_import( &self, - target: &BlobStore, + pool: &SqlitePool, controls: &ImportControls, report: &mut LegacyBlobImportReport, ) -> Result<(), LegacyBlobImportFailure> { - let pool = target - .sqlite_pool_for_legacy_import() - .ok_or(LegacyBlobImportFailure::WrongTargetBackend)?; let mut connection = pool .acquire() .await @@ -536,39 +854,65 @@ impl Database { report, ) .await?; - run_checkpoint(connection, CheckpointKind::Final, controls).await + Ok(()) } } +#[derive(Clone, Copy)] +enum SourceEntryFailure { + InvalidKey, + DigestMismatch, +} + +fn legacy_blob_prefix() -> Vec { + SlateKey::new("blobs") + .with("sha256") + .into_prefix() + .as_ref() + .to_vec() +} + +fn parse_canonical_hash(value: &str) -> Option { + let canonical = value.len() == 64 + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)); + canonical.then(|| value.parse().ok()).flatten() +} + +fn parse_source_key(key: &[u8], prefix: &[u8]) -> Option { + 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 { + let hash = parse_source_key(key, prefix).ok_or(SourceEntryFailure::InvalidKey)?; + if BlobHash::new(value) != hash { + return Err(SourceEntryFailure::DigestMismatch); + } + Ok(hash) +} + fn validate_source_entry( key: &[u8], value: &[u8], prefix: &[u8], report: &mut LegacyBlobImportReport, ) -> Result { - let Some(suffix) = key.strip_prefix(prefix) else { - return invalid_source_row(report, LegacyBlobImportFailure::InvalidSourceKey); - }; - let canonical = suffix.len() == 64 - && suffix - .iter() - .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)); - if !canonical { - return invalid_source_row(report, LegacyBlobImportFailure::InvalidSourceKey); + match validate_source_entry_common(key, value, prefix) { + Ok(hash) => Ok(hash), + Err(SourceEntryFailure::InvalidKey) => { + invalid_source_row(report, LegacyBlobImportFailure::InvalidSourceKey) + } + Err(SourceEntryFailure::DigestMismatch) => { + invalid_source_row(report, LegacyBlobImportFailure::SourceDigestMismatch) + } } - - let hash = std::str::from_utf8(suffix) - .ok() - .and_then(|value| value.parse().ok()) - .ok_or(LegacyBlobImportFailure::InvalidSourceKey); - let hash = match hash { - Ok(hash) => hash, - Err(failure) => return invalid_source_row(report, failure), - }; - if BlobHash::new(value) != hash { - return invalid_source_row(report, LegacyBlobImportFailure::SourceDigestMismatch); - } - Ok(hash) } fn invalid_source_row( @@ -605,7 +949,7 @@ async fn commit_pending_batch( checked_add(bytes_since_checkpoint, batch_report.imported_bytes)?; if *bytes_since_checkpoint >= PASSIVE_CHECKPOINT_BYTES { - run_checkpoint(connection, CheckpointKind::Passive, controls).await?; + run_checkpoint(connection, controls).await?; checked_add(&mut report.passive_checkpoints, 1)?; *bytes_since_checkpoint = 0; } @@ -675,50 +1019,49 @@ async fn commit_batch( } } -#[derive(Clone, Copy)] -enum CheckpointKind { - Passive, - Final, -} - async fn run_checkpoint( connection: &mut PoolConnection, - kind: CheckpointKind, controls: &ImportControls, ) -> Result<(), LegacyBlobImportFailure> { - let (statement, injected_error) = match kind { - CheckpointKind::Passive => ( - "PRAGMA wal_checkpoint(PASSIVE)", - controls.passive_checkpoint_error(), - ), - CheckpointKind::Final => ( - "PRAGMA wal_checkpoint(TRUNCATE)", - controls.final_checkpoint_error(), - ), - }; - if let Some(source) = injected_error { - return Err(match kind { - CheckpointKind::Passive => LegacyBlobImportFailure::PassiveCheckpoint(source), - CheckpointKind::Final => LegacyBlobImportFailure::FinalCheckpoint(source), - }); + if let Some(source) = controls.passive_checkpoint_error() { + return Err(LegacyBlobImportFailure::PassiveCheckpoint(source)); } - let result = sqlx::query_as::<_, (i64, i64, i64)>(statement) + let result = sqlx::query_as::<_, (i64, i64, i64)>("PRAGMA wal_checkpoint(PASSIVE)") .fetch_one(&mut **connection) .await; - let (busy, _, _) = result.map_err(|source| match kind { - CheckpointKind::Passive => LegacyBlobImportFailure::PassiveCheckpoint(source), - CheckpointKind::Final => LegacyBlobImportFailure::FinalCheckpoint(source), - })?; + let (busy, _, _) = result.map_err(LegacyBlobImportFailure::PassiveCheckpoint)?; if busy != 0 { - return Err(match kind { - CheckpointKind::Passive => LegacyBlobImportFailure::PassiveCheckpointBusy, - CheckpointKind::Final => LegacyBlobImportFailure::FinalCheckpointBusy, - }); + return Err(LegacyBlobImportFailure::PassiveCheckpointBusy); } Ok(()) } +fn inventory_usize_to_u64(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyBlobInventoryFailure::CounterOverflow) +} + +fn inventory_checked_add(value: &mut u64, amount: u64) -> Result<(), LegacyBlobInventoryFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyBlobInventoryFailure::CounterOverflow)?; + Ok(()) +} + +fn verification_usize_to_u64(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyBlobVerificationFailure::CounterOverflow) +} + +fn verification_checked_add( + value: &mut u64, + amount: u64, +) -> Result<(), LegacyBlobVerificationFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyBlobVerificationFailure::CounterOverflow)?; + Ok(()) +} + async fn set_automatic_checkpoint( connection: &mut PoolConnection, pages: i64, @@ -794,23 +1137,24 @@ mod tests { source: Database, source_db: slatedb::Db, sqlite: sqlx::SqlitePool, - target: BlobStore, + target: Arc, } impl TestContext { async fn new() -> TestResult { + let dir = tempfile::tempdir()?; + let sqlite = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + sqlite.migrate().await?; + let sqlite = sqlite.clone_pool(); + let target = Arc::new(BlobStore::new(sqlite.clone())); let source = Database::new( Arc::new(InMemory::new()), "legacy-blob-import-tests", Duration::from_millis(1), None, + Arc::clone(&target), ); let source_db = source.open_db().await?; - let dir = tempfile::tempdir()?; - let sqlite = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; - sqlite.migrate().await?; - let sqlite = sqlite.clone_pool(); - let target = BlobStore::new(sqlite.clone()); Ok(Self { _dir: dir, source, @@ -828,7 +1172,7 @@ mod tests { .max_connections(1) .connect_with(options) .await?; - context.target = BlobStore::new(sqlite.clone()); + context.target = Arc::new(BlobStore::new(sqlite.clone())); context.sqlite = sqlite; Ok(context) } @@ -892,7 +1236,7 @@ mod tests { } async fn import(&self) -> TestResult { - Ok(self.source.import_legacy_blobs_into(&self.target).await?) + Ok(self.source.import_legacy_blobs_into(&self.sqlite).await?) } } @@ -976,6 +1320,135 @@ mod tests { Ok(()) } + #[tokio::test] + async fn inventory_streams_the_exact_prefix_and_reports_logical_bytes() -> TestResult<()> { + let context = TestContext::new().await?; + context.put_blob(b"").await?; + context.put_blob(&[0, 0xff, 0x80, b'a']).await?; + context + .source_db + .put( + SlateKey::new("blobs").with("other").with("ignored"), + b"nearby", + ) + .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(()) + } + + #[tokio::test] + async fn verification_checks_every_source_and_allows_valid_sqlite_only_rows() -> TestResult<()> + { + let context = TestContext::new().await?; + context.put_blob(b"legacy").await?; + context.import().await?; + let sqlite_only = b"written-after-activation"; + context + .insert_destination(BlobHash::new(sqlite_only), sqlite_only) + .await?; + + let report = context + .source + .verify_legacy_blobs_in(&context.sqlite) + .await?; + + assert_eq!(report.source_rows, 1); + assert_eq!(report.matched_rows, 1); + assert_eq!(report.target_rows, 2); + assert_eq!(report.missing_rows, 0); + assert_eq!(report.invalid_source_rows, 0); + assert_eq!(report.invalid_target_rows, 0); + assert_eq!(report.conflicting_rows, 0); + Ok(()) + } + + #[tokio::test] + async fn verification_reports_a_missing_destination_without_exposing_its_hash() -> TestResult<()> + { + let context = TestContext::new().await?; + let bytes = b"missing-sensitive-content"; + let hash = context.put_blob(bytes).await?; + + let error = context + .source + .verify_legacy_blobs_in(&context.sqlite) + .await + .expect_err("a missing destination row must fail verification"); + + assert_eq!(error.report().source_rows, 1); + assert_eq!(error.report().missing_rows, 1); + let rendered = format!("{error} {error:?}"); + assert!(!rendered.contains(&hash.to_string())); + assert!(!rendered.contains(std::str::from_utf8(bytes)?)); + Ok(()) + } + + #[tokio::test] + async fn verification_rejects_invalid_sqlite_hashes_and_bytes() -> TestResult<()> { + let malformed_hash = TestContext::new().await?; + let mut connection = malformed_hash.sqlite.acquire().await?; + sqlx::query("PRAGMA ignore_check_constraints = ON") + .execute(&mut *connection) + .await?; + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind("not-a-canonical-hash") + .bind(b"bytes".as_slice()) + .execute(&mut *connection) + .await?; + sqlx::query("PRAGMA ignore_check_constraints = OFF") + .execute(&mut *connection) + .await?; + drop(connection); + let error = malformed_hash + .source + .verify_legacy_blobs_in(&malformed_hash.sqlite) + .await + .expect_err("malformed SQLite hashes must fail verification"); + assert_eq!(error.report().invalid_target_rows, 1); + + let mismatched_bytes = TestContext::new().await?; + mismatched_bytes + .insert_destination(BlobHash::new(b"expected"), b"different") + .await?; + let error = mismatched_bytes + .source + .verify_legacy_blobs_in(&mismatched_bytes.sqlite) + .await + .expect_err("SQLite bytes must match their hash"); + assert_eq!(error.report().invalid_target_rows, 1); + Ok(()) + } + #[tokio::test] async fn rejects_every_noncanonical_legacy_key_shape() -> TestResult<()> { let valid_hash = BlobHash::new(b"valid-shape").to_string(); @@ -1001,7 +1474,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .await .expect_err("noncanonical key should fail import"); @@ -1031,7 +1504,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .await .expect_err("digest mismatch should fail import"); @@ -1056,7 +1529,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .await .expect_err("invalid row should stop the import"); @@ -1097,7 +1570,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .await .expect_err("differing destination row should fail import"); @@ -1125,7 +1598,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .await .expect_err("last-row conflict should interrupt import"); @@ -1241,7 +1714,7 @@ mod tests { let error = context .source - .import_legacy_blobs_with_controls(&context.target, &controls) + .import_legacy_blobs_with_controls(&context.sqlite, &controls) .await .expect_err("injected passive checkpoint should fail import"); @@ -1260,33 +1733,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn final_checkpoint_failure_returns_committed_progress() -> TestResult<()> { - let context = TestContext::new().await?; - context - .put_blob(b"committed-before-final-checkpoint") - .await?; - let controls = ImportControls { - final_checkpoint: true, - ..ImportControls::default() - }; - - let error = context - .source - .import_legacy_blobs_with_controls(&context.target, &controls) - .await - .expect_err("injected final checkpoint should fail import"); - - assert_eq!(error.report().imported_rows, 1); - assert_eq!(error.report().committed_batches, 1); - assert!(matches!( - &error.failure, - LegacyBlobImportFailure::FinalCheckpoint(sqlx::Error::Protocol(_)) - )); - assert_eq!(context.destination_rows().await?, 1); - Ok(()) - } - #[tokio::test] async fn source_transport_failure_preserves_commits_and_typed_source() -> TestResult<()> { let context = TestContext::new().await?; @@ -1298,7 +1744,7 @@ mod tests { let error = context .source - .import_legacy_blobs_with_controls(&context.target, &controls) + .import_legacy_blobs_with_controls(&context.sqlite, &controls) .await .expect_err("injected source transport failure should stop import"); @@ -1329,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.target, &controls) + .import_legacy_blobs_with_controls(&context.sqlite, &controls) + .with_subscriber(capture.clone()) .await .expect_err("both injected failures should fail import"); @@ -1349,6 +1797,15 @@ mod tests { .any(|source| source.downcast_ref::().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(()) } @@ -1368,7 +1825,7 @@ mod tests { failure.put_raw(invalid_key, b"invalid").await?; failure .source - .import_legacy_blobs_into(&failure.target) + .import_legacy_blobs_into(&failure.sqlite) .await .expect_err("invalid source should fail import"); assert_eq!(failure.automatic_checkpoint().await?, 41); @@ -1377,12 +1834,6 @@ mod tests { #[tokio::test] async fn cancellation_retires_a_connection_with_disabled_checkpointing() -> TestResult<()> { - let source = Database::new( - Arc::new(InMemory::new()), - "legacy-blob-import-cancellation-test", - Duration::from_millis(1), - None, - ); let dir = tempfile::tempdir()?; let options = SqliteConnectOptions::new() .filename(dir.path().join("fabro.sqlite3")) @@ -1396,6 +1847,13 @@ mod tests { .execute(&pool) .await?; let target = Arc::new(BlobStore::new(pool.clone())); + let source = Database::new( + Arc::new(InMemory::new()), + "legacy-blob-import-cancellation-test", + Duration::from_millis(1), + None, + Arc::clone(&target), + ); let mut connection = pool.acquire().await?; set_automatic_checkpoint(&mut connection, 73).await?; @@ -1408,11 +1866,11 @@ mod tests { }; let task = tokio::spawn({ let source = source.clone(); - let target = Arc::clone(&target); + let pool = pool.clone(); async move { let mut report = LegacyBlobImportReport::default(); source - .run_legacy_blob_import(&target, &controls, &mut report) + .run_legacy_blob_import(&pool, &controls, &mut report) .await } }); @@ -1442,7 +1900,7 @@ mod tests { let error = context .source - .import_legacy_blobs_with_controls(&context.target, &controls) + .import_legacy_blobs_with_controls(&context.sqlite, &controls) .await .expect_err("injected restoration failure should fail import"); @@ -1458,27 +1916,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn slate_backed_target_is_rejected_before_scanning() -> TestResult<()> { - let context = TestContext::new().await?; - context.put_blob(b"never-scanned").await?; - let slate_target = context.source.blobs().await?; - - let error = context - .source - .import_legacy_blobs_into(&slate_target) - .await - .expect_err("Slate target should be rejected"); - - assert_eq!(*error.report(), LegacyBlobImportReport::default()); - assert!(matches!( - &error.failure, - LegacyBlobImportFailure::WrongTargetBackend - )); - assert_eq!(context.destination_rows().await?, 0); - Ok(()) - } - #[tokio::test] async fn logs_and_error_rendering_expose_counts_but_not_row_data() -> TestResult<()> { let context = TestContext::new().await?; @@ -1491,7 +1928,7 @@ mod tests { let error = context .source - .import_legacy_blobs_into(&context.target) + .import_legacy_blobs_into(&context.sqlite) .with_subscriber(capture.clone()) .await .expect_err("invalid key should fail import"); diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index b73695100..91363b5be 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -30,7 +30,10 @@ pub use fabro_types::{ BlobHash, EventEnvelope, PendingInterviewRecord, Run, RunProjection, StageId, StageProjection, }; pub use keyed_mutex::{KeyedMutex, KeyedMutexGuard}; -pub use legacy_blob_import::{LegacyBlobImportError, LegacyBlobImportReport}; +pub use legacy_blob_import::{ + LegacyBlobImportError, LegacyBlobImportReport, LegacyBlobInventory, LegacyBlobInventoryError, + LegacyBlobVerificationError, LegacyBlobVerificationReport, +}; pub use run_sessions::{ ProjectedRunSession, project_run_session, project_run_session_with_context, project_run_sessions, diff --git a/lib/components/fabro-store/src/record/repository.rs b/lib/components/fabro-store/src/record/repository.rs index 33196036c..60cbb2541 100644 --- a/lib/components/fabro-store/src/record/repository.rs +++ b/lib/components/fabro-store/src/record/repository.rs @@ -119,6 +119,7 @@ impl Repository { Ok(()) } + #[cfg(test)] pub(crate) async fn exists(&self, id: &R::Id) -> Result { Ok(self.db.get(key_for_id::(id)?).await?.is_some()) } diff --git a/lib/components/fabro-store/src/slate/auth_codes.rs b/lib/components/fabro-store/src/slate/auth_codes.rs index 65bcd5f16..0fbdb2ab9 100644 --- a/lib/components/fabro-store/src/slate/auth_codes.rs +++ b/lib/components/fabro-store/src/slate/auth_codes.rs @@ -91,10 +91,10 @@ mod tests { use tokio::task::JoinSet; use super::{AuthCode, AuthCodeStore}; - use crate::Database; + use crate::test_support; async fn store() -> Arc { - let db = Database::new( + let db = test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 7b4f206cc..77624e46a 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -43,7 +43,7 @@ pub struct Database { cache_path: Option, db: Arc>, active_runs: Arc>>>, - blobs: Arc>>, + blobs: Arc, catalog_index: Arc>>, auth_codes: Arc>>, projection_cache: Arc, @@ -67,6 +67,7 @@ impl Database { base_prefix: impl Into, flush_interval: Duration, cache_path: Option, + blobs: Arc, ) -> Self { Self { object_store, @@ -75,7 +76,7 @@ impl Database { cache_path, db: Arc::new(OnceCell::new()), active_runs: Arc::new(Mutex::new(HashMap::new())), - blobs: Arc::new(OnceCell::new()), + blobs, catalog_index: Arc::new(OnceCell::new()), auth_codes: Arc::new(OnceCell::new()), projection_cache: Arc::new(RunProjectionCache::default()), @@ -143,7 +144,7 @@ impl Database { *run_id, self.open_db().await?, read_only, - self.blobs().await?, + self.blobs(), Arc::clone(&self.projection_cache), Arc::clone(&self.run_summary_store), ) @@ -438,15 +439,9 @@ impl Database { Ok(Arc::clone(store)) } - pub async fn blobs(&self) -> Result> { - let store = self - .blobs - .get_or_try_init(|| async { - let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(BlobStore::from_slate(db))) - }) - .await?; - Ok(Arc::clone(store)) + #[must_use] + pub fn blobs(&self) -> Arc { + Arc::clone(&self.blobs) } /// Delete every record under the retired `auth/refresh` prefix. @@ -569,7 +564,7 @@ mod tests { fn make_store() -> (Arc, Database) { let object_store: Arc = Arc::new(InMemory::new()); - let store = Database::new( + let store = store_test_support::test_database( object_store.clone(), "runs/", Duration::from_millis(1), @@ -1050,7 +1045,12 @@ mod tests { .unwrap(); assert_ne!(stale.title, "Committed title"); - let reopened = Database::new(object_store, "runs/", Duration::from_millis(1), None); + let reopened = store_test_support::test_database( + object_store, + "runs/", + Duration::from_millis(1), + None, + ); reopened.attach_run_summary_store(Arc::clone(&repaired_summaries)); reopened.warm_projection_cache().await.unwrap(); let repaired = repaired_summaries @@ -1508,7 +1508,8 @@ mod tests { let run = store.create_run(&test_run_id("run-1")).await.unwrap(); append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); let summary = reopened .list_runs(&ListRunsQuery::default(), Utc::now()) .await @@ -1528,7 +1529,8 @@ mod tests { append_completed(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await; append_running(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await; - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); reopened.warm_projection_cache().await.unwrap(); let entries = reopened @@ -1592,7 +1594,8 @@ mod tests { .await .unwrap(); - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); reopened.warm_projection_cache().await.unwrap(); let entries = reopened @@ -1654,7 +1657,8 @@ mod tests { .await .unwrap(); - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); let unreadable = reopened.list_unreadable_runs().await.unwrap(); assert_eq!(unreadable.len(), 1); @@ -1850,7 +1854,8 @@ mod tests { let run = store.create_run(&test_run_id("run-1")).await.unwrap(); append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); let (_directory, summaries) = make_summary_store().await; reopened.attach_run_summary_store(Arc::clone(&summaries)); reopened.warm_projection_cache().await.unwrap(); @@ -1872,7 +1877,8 @@ mod tests { let run = store.create_run(&run_id).await.unwrap(); append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); reopened.warm_projection_cache().await.unwrap(); // If opening or projecting the run starts at the beginning, this @@ -1941,7 +1947,7 @@ mod tests { .await .unwrap(); - let reopened = Database::new( + let reopened = store_test_support::test_database( Arc::clone(&object_store), "runs/", Duration::from_millis(1), diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index 81450b44a..c01ab0904 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -890,7 +890,7 @@ mod tests { use object_store::memory::InMemory; use serde_json::json; - use crate::{Database, Error, EventPayload, keys}; + use crate::{Error, EventPayload, keys, test_support as store_test_support}; fn stage_prompt_payload(run_id: &RunId, idx: u32, node_id: Option<&str>) -> EventPayload { stage_prompt_payload_for_stage(run_id, idx, node_id, None) @@ -965,7 +965,8 @@ mod tests { async fn fresh_run() -> super::RunDatabase { let object_store = Arc::new(InMemory::new()); - let store = Database::new(object_store, "", Duration::from_millis(1), None); + let store = + store_test_support::test_database(object_store, "", Duration::from_millis(1), None); let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); let run = store.create_run(&run_id).await.unwrap(); run.append_event(&run_created_payload(&run_id)) @@ -1175,7 +1176,8 @@ mod tests { #[tokio::test] async fn recover_latest_seq_returns_zero_for_empty_history() { let object_store = Arc::new(InMemory::new()); - let store = Database::new(object_store, "", Duration::from_millis(1), None); + let store = + store_test_support::test_database(object_store, "", Duration::from_millis(1), None); let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); let run = store.create_run(&run_id).await.unwrap(); @@ -1226,7 +1228,12 @@ mod tests { #[tokio::test] async fn list_events_before_with_limit_serves_newest_page_from_cold_cache() { let object_store = Arc::new(InMemory::new()); - let store = Database::new(object_store.clone(), "", Duration::from_millis(1), None); + let store = store_test_support::test_database( + object_store.clone(), + "", + Duration::from_millis(1), + None, + ); let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); let run = store.create_run(&run_id).await.unwrap(); run.append_event(&run_created_payload(&run_id)) @@ -1238,7 +1245,8 @@ mod tests { .unwrap(); } - let reopened = Database::new(object_store, "", Duration::from_millis(1), None); + let reopened = + store_test_support::test_database(object_store, "", Duration::from_millis(1), None); let reader = reopened.open_run_reader(&run_id).await.unwrap(); let events = reader.list_events_before_with_limit(None, 2).await.unwrap(); diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index 758fe26a0..a77004142 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -1,11 +1,137 @@ -#[cfg(test)] -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; -use fabro_types::RunId; +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::{Database, Result}; +use crate::{BlobStore, Database, Result}; + +/// Returns an isolated SQLite blob authority backed by its own in-memory +/// database. +/// +/// Every call creates a fresh blob table, so tests never observe rows written +/// by other tests in the same process. Reopen-style tests that model one +/// process-wide blob authority across several store handles should call this +/// once and share the result through [`test_database_with_blobs`]. +/// +/// The pool connects lazily so synchronous fixture builders can remain +/// synchronous. Its single connection installs the production blob schema on +/// first use. +#[must_use] +pub fn test_blob_store() -> Arc { + let options = SqliteConnectOptions::new() + .filename(":memory:") + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + // A single in-memory test connection never needs reaping. Disabling + // both timers also keeps this lazy fixture constructible from sync + // tests, where SQLx has no Tokio runtime for maintenance tasks. + .max_lifetime(None) + .idle_timeout(None) + .after_connect(|connection, _metadata| { + Box::pin(async move { + sqlx::query(fabro_db::BLOBS_MIGRATION_SQL) + .execute(&mut *connection) + .await?; + Ok(()) + }) + }) + .connect_lazy_with(options); + Arc::new(BlobStore::new(pool)) +} + +/// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`. +#[must_use] +pub fn test_blob_store_path(store_dir: &Path) -> PathBuf { + fabro_db::append_to_path(store_dir, "-blobs.sqlite3") +} + +/// Returns a durable SQLite blob authority stored beside `store_dir`. +/// +/// Handles created for the same directory share one blob database file, so +/// reopen-style tests observe blobs across store handles the way production +/// handles share the process-wide blob authority. Tests that reuse a +/// directory must delete [`test_blob_store_path`] (and its `-wal`/`-shm` +/// siblings) when they reset the directory itself. +#[must_use] +pub fn test_blob_store_at(store_dir: &Path) -> Arc { + let options = SqliteConnectOptions::new() + .filename(test_blob_store_path(store_dir)) + .create_if_missing(true) + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .max_lifetime(None) + .idle_timeout(None) + .after_connect(|connection, _metadata| { + Box::pin(async move { + let installed: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM sqlite_master \ + WHERE type = 'table' AND name = 'blobs')", + ) + .fetch_one(&mut *connection) + .await?; + if !installed { + sqlx::query(fabro_db::BLOBS_MIGRATION_SQL) + .execute(&mut *connection) + .await?; + } + Ok(()) + }) + }) + .connect_lazy_with(options); + Arc::new(BlobStore::new(pool)) +} + +/// Builds a Slate-backed run database with its own isolated blob authority. +#[must_use] +pub fn test_database( + object_store: Arc, + base_prefix: impl Into, + flush_interval: Duration, + cache_path: Option, +) -> Database { + test_database_with_blobs( + object_store, + base_prefix, + flush_interval, + cache_path, + test_blob_store(), + ) +} + +/// Builds a Slate-backed run database sharing an explicit blob authority. +/// +/// Use this for reopen-style tests where two store handles must observe the +/// same signed SQLite blob table, mirroring the one blob authority a +/// production process shares across every run handle. +#[must_use] +pub fn test_database_with_blobs( + object_store: Arc, + base_prefix: impl Into, + flush_interval: Duration, + cache_path: Option, + blobs: Arc, +) -> Database { + Database::new(object_store, base_prefix, flush_interval, cache_path, blobs) +} + +/// Seeds one canonical row in the legacy SlateDB blob keyspace. +pub async fn put_legacy_blob(database: &Database, bytes: &[u8]) -> Result { + let hash = BlobHash::new(bytes); + let source = database.open_db().await?; + source + .put(SlateKey::new("blobs").with("sha256").with(hash), bytes) + .await?; + source.flush().await?; + Ok(hash) +} /// Writes an event without append validation to model a log corrupted by an /// older Fabro version. diff --git a/lib/components/fabro-workflow-version/Cargo.toml b/lib/components/fabro-workflow-version/Cargo.toml index bc91e3f89..f3eb60317 100644 --- a/lib/components/fabro-workflow-version/Cargo.toml +++ b/lib/components/fabro-workflow-version/Cargo.toml @@ -23,5 +23,6 @@ serde_json.workspace = true thiserror.workspace = true [dev-dependencies] +fabro-store = { path = "../fabro-store", features = ["test-support"] } object_store.workspace = true tokio = { workspace = true, features = ["full"] } diff --git a/lib/components/fabro-workflow-version/src/store.rs b/lib/components/fabro-workflow-version/src/store.rs index 72d462ec2..0c40dfedf 100644 --- a/lib/components/fabro-workflow-version/src/store.rs +++ b/lib/components/fabro-workflow-version/src/store.rs @@ -227,7 +227,7 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use fabro_store::{BlobStore, Database}; + use fabro_store::{BlobStore, test_support}; use fabro_types::{WorkflowPath, WorkflowVersion, WorkflowVersionId}; use object_store::memory::InMemory; @@ -259,21 +259,21 @@ mod tests { )) } - async fn stores() -> (Arc, WorkflowVersionStore) { - let database = Database::new( + fn stores() -> (Arc, WorkflowVersionStore) { + let database = test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), None, ); - let blobs = database.blobs().await.unwrap(); + let blobs = database.blobs(); let versions = WorkflowVersionStore::new(Arc::clone(&blobs)); (blobs, versions) } #[tokio::test] async fn put_get_reuses_exact_blob_digest() { - let (blobs, store) = stores().await; + let (blobs, store) = stores(); let version = version("digraph W {}", BTreeMap::new()); let expected_bytes = version.version().canonical_bytes().unwrap(); let expected_id = version_id(&version); @@ -290,7 +290,7 @@ mod tests { #[tokio::test] async fn identical_content_is_idempotent() { - let (_, store) = stores().await; + let (_, store) = stores(); let original = version("digraph W {}", BTreeMap::new()); assert_eq!( @@ -307,7 +307,7 @@ mod tests { #[tokio::test] async fn dependency_must_be_stored_first() { - let (blobs, store) = stores().await; + let (blobs, store) = stores(); let child = version("digraph Child {}", BTreeMap::new()); let child_id = version_id(&child); let root = version( @@ -328,7 +328,7 @@ mod tests { #[tokio::test] async fn dependency_closure_must_be_complete_before_root_write() { - let (blobs, store) = stores().await; + let (blobs, store) = stores(); let missing_grandchild_id = WorkflowVersionId::from(fabro_types::BlobHash::new(b"missing")); let child = version( r#"digraph Child { grandchild [stack.child_workflow="grandchild.fabro"] }"#, @@ -357,7 +357,7 @@ mod tests { #[tokio::test] async fn get_closure_returns_root_and_transitive_dependencies() { - let (_, store) = stores().await; + let (_, store) = stores(); let grandchild = version("digraph Grandchild {}", BTreeMap::new()); let grandchild_id = store.put(&grandchild).await.unwrap(); let child = version( @@ -392,7 +392,7 @@ mod tests { #[tokio::test] async fn get_closure_deduplicates_a_diamond() { - let (_, store) = stores().await; + let (_, store) = stores(); let leaf = version("digraph Leaf {}", BTreeMap::new()); let leaf_id = store.put(&leaf).await.unwrap(); let left = version( @@ -426,7 +426,7 @@ mod tests { #[tokio::test] async fn get_closure_preserves_noncanonical_dependency_errors() { - let (blobs, store) = stores().await; + let (blobs, store) = stores(); let dependency = version("digraph Dependency {}", BTreeMap::new()); let pretty = serde_json::to_vec_pretty(dependency.version()).unwrap(); let dependency_id = WorkflowVersionId::from(blobs.write(&pretty).await.unwrap()); @@ -457,7 +457,7 @@ mod tests { #[tokio::test] async fn get_projects_the_same_validated_root_as_get_closure() { - let (_, store) = stores().await; + let (_, store) = stores(); let child = version("digraph Child {}", BTreeMap::new()); let child_id = store.put(&child).await.unwrap(); let root = version( @@ -477,7 +477,7 @@ mod tests { #[tokio::test] async fn get_rejects_arbitrary_and_noncanonical_blobs() { - let (blobs, store) = stores().await; + let (blobs, store) = stores(); let arbitrary = WorkflowVersionId::from(blobs.write(b"not json").await.unwrap()); assert!(matches!( store.get(&arbitrary).await.unwrap_err(), diff --git a/lib/components/fabro-workflow/Cargo.toml b/lib/components/fabro-workflow/Cargo.toml index 0024c8b79..b326fb139 100644 --- a/lib/components/fabro-workflow/Cargo.toml +++ b/lib/components/fabro-workflow/Cargo.toml @@ -14,7 +14,7 @@ readme = "README.md" doctest = false [features] -test-support = ["fabro-auth/test-support"] +test-support = ["fabro-auth/test-support", "fabro-store/test-support"] [lints] workspace = true @@ -75,6 +75,7 @@ tempfile = "3" toml.workspace = true fabro-vault = { path = "../../foundation/fabro-vault" } [dev-dependencies] +fabro-store = { path = "../fabro-store", features = ["test-support"] } fabro-auth = { path = "../../foundation/fabro-auth", features = ["test-support"] } fabro-github = { path = "../fabro-github", features = ["test-support"] } base64.workspace = true diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index a96ff54d6..3fb27588f 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -831,7 +831,6 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use fabro_store::Database; use object_store::memory::InMemory; use ulid::Ulid; @@ -845,7 +844,12 @@ mod tests { async fn make_run_store(label: &str) -> fabro_store::RunDatabase { let object_store = Arc::new(InMemory::new()); - let store = Database::new(object_store, "runs/", Duration::from_millis(1), None); + let store = fabro_store::test_support::test_database( + object_store, + "runs/", + Duration::from_millis(1), + None, + ); store.create_run(&test_run_id(label)).await.unwrap() } diff --git a/lib/components/fabro-workflow/src/event/sink.rs b/lib/components/fabro-workflow/src/event/sink.rs index dfd8e3b4e..29d819718 100644 --- a/lib/components/fabro-workflow/src/event/sink.rs +++ b/lib/components/fabro-workflow/src/event/sink.rs @@ -293,7 +293,7 @@ mod tests { #[tokio::test] async fn append_event_writes_store_event_shape() { - let store = fabro_store::Database::new( + let store = fabro_store::test_support::test_database( std::sync::Arc::new(object_store::memory::InMemory::new()), "", std::time::Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/git.rs b/lib/components/fabro-workflow/src/git.rs index cb7b08a3f..e50561cd2 100644 --- a/lib/components/fabro-workflow/src/git.rs +++ b/lib/components/fabro-workflow/src/git.rs @@ -305,7 +305,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/handler/agent.rs b/lib/components/fabro-workflow/src/handler/agent.rs index 572de6429..e15e6d36f 100644 --- a/lib/components/fabro-workflow/src/handler/agent.rs +++ b/lib/components/fabro-workflow/src/handler/agent.rs @@ -458,7 +458,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/handler/command.rs b/lib/components/fabro-workflow/src/handler/command.rs index 8d08d17b1..0b58e52f8 100644 --- a/lib/components/fabro-workflow/src/handler/command.rs +++ b/lib/components/fabro-workflow/src/handler/command.rs @@ -435,7 +435,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/handler/manager_loop.rs b/lib/components/fabro-workflow/src/handler/manager_loop.rs index 8cc82f634..92969ddeb 100644 --- a/lib/components/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/components/fabro-workflow/src/handler/manager_loop.rs @@ -5,7 +5,7 @@ use std::time::Duration; use async_trait::async_trait; use fabro_graphviz::graph::{AttrValue, Graph, Node}; -use fabro_store::{ArtifactStore, Database}; +use fabro_store::ArtifactStore; use fabro_template::validate_static_reference; use fabro_types::WorkflowSettings; use fabro_types::graph::ReferenceKind; @@ -233,18 +233,8 @@ impl Handler for SubWorkflowHandler { let inputs = services.inputs.clone(); let dry_run = services.dry_run; let workflow_bundle = services.workflow_bundle.clone(); - let object_store = Arc::new(InMemory::new()); - let store = Arc::new(Database::new( - object_store.clone(), - "", - Duration::from_millis(1), - None, - )); - let run_store = store - .create_run(&child_run_options.run_id) - .await - .map_err(|err| Error::engine(err.to_string()))?; - let artifact_store = ArtifactStore::new(object_store, "artifacts"); + let run_store = services.run.run_store.clone(); + let artifact_store = ArtifactStore::new(Arc::new(InMemory::new()), "artifacts"); // Spawn child engine. Child runs receive a derived cancel token from // the parent run; parent cancellation propagates parent-to-child via @@ -252,7 +242,7 @@ impl Handler for SubWorkflowHandler { let child_run_token_for_services = child_run_token.clone(); let mut child_handle = tokio::spawn(async move { let child_run = parent_run - .with_run_store(run_store.into()) + .with_run_store(run_store) .with_cancel_token(child_run_token_for_services); let initialized = Initialized { graph: child_graph, @@ -549,6 +539,81 @@ mod tests { ); } + #[tokio::test] + async fn child_blob_writes_use_the_parent_run_store() { + const CHILD_BLOB: &[u8] = b"manager-child-shared-blob"; + + struct BlobWriter; + + #[async_trait] + impl Handler for BlobWriter { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + services: &EngineServices, + ) -> Result { + services + .run + .run_store + .write_blob(CHILD_BLOB) + .await + .map_err(|error| { + Error::handler_with_source("manager child blob write failed", error) + })?; + Ok(Outcome::success()) + } + } + + let mut registry = HandlerRegistry::new(Box::new(BlobWriter)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + let mut services = EngineServices::test_default(); + services.registry = Arc::new(registry); + + let handler = SubWorkflowHandler; + let mut node = Node::new("manager"); + node.attrs.insert( + "stack.child_dot_source".to_string(), + AttrValue::String( + "digraph Child { start [shape=Mdiamond]; work [shape=box]; exit [shape=Msquare]; start -> work -> exit }" + .to_string(), + ), + ); + node.attrs + .insert("manager.max_cycles".to_string(), AttrValue::Integer(100)); + node.attrs.insert( + "manager.poll_interval".to_string(), + AttrValue::Duration(Duration::from_millis(10)), + ); + + let outcome = handler + .execute( + &node, + &Context::new(), + &Graph::new("test"), + tempfile::tempdir().unwrap().path(), + &services, + ) + .await + .unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + let hash = fabro_types::BlobHash::new(CHILD_BLOB); + assert_eq!( + services + .run + .run_store + .read_blob(&hash) + .await + .unwrap() + .as_deref(), + Some(CHILD_BLOB) + ); + } + #[tokio::test] async fn child_workflow_reads_from_file() { let dir = tempfile::tempdir().unwrap(); diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 4e9d0d90f..edd09a56b 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -969,7 +969,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/handler/prompt.rs b/lib/components/fabro-workflow/src/handler/prompt.rs index 940f1711d..5e2058c3c 100644 --- a/lib/components/fabro-workflow/src/handler/prompt.rs +++ b/lib/components/fabro-workflow/src/handler/prompt.rs @@ -236,7 +236,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs index ccf017950..8d624bb76 100644 --- a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs @@ -411,7 +411,6 @@ mod tests { use fabro_core::graph::Graph as CoreGraph; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; - use fabro_store::Database; use fabro_types::fixtures; use object_store::memory::InMemory; @@ -462,7 +461,7 @@ mod tests { } async fn test_lifecycle(graph: &WorkflowGraph, run_dir: &Path) -> FidelityLifecycle { - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/lifecycle/git.rs b/lib/components/fabro-workflow/src/lifecycle/git.rs index 540c895fc..4f1b30602 100644 --- a/lib/components/fabro-workflow/src/lifecycle/git.rs +++ b/lib/components/fabro-workflow/src/lifecycle/git.rs @@ -646,7 +646,7 @@ mod tests { use fabro_core::state::ExecutionState; use fabro_graphviz::graph::types::{AttrValue, Edge, Graph, Node}; use fabro_model::Catalog; - use fabro_store::{Database, EventEnvelope, RunDatabase, RunProjection}; + use fabro_store::{EventEnvelope, RunDatabase, RunProjection}; use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase}; use fabro_types::{BlobHash, EventBody, RunEvent, WorkflowSettings, fixtures, test_support}; use object_store::memory::InMemory; @@ -766,7 +766,7 @@ mod tests { } async fn run_store(run_id: fabro_types::RunId) -> RunDatabase { - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/operations/archive.rs b/lib/components/fabro-workflow/src/operations/archive.rs index 188f625b2..1f0924bf0 100644 --- a/lib/components/fabro-workflow/src/operations/archive.rs +++ b/lib/components/fabro-workflow/src/operations/archive.rs @@ -144,7 +144,7 @@ mod tests { use super::*; fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/operations/create.rs b/lib/components/fabro-workflow/src/operations/create.rs index c14f48632..d29224204 100644 --- a/lib/components/fabro-workflow/src/operations/create.rs +++ b/lib/components/fabro-workflow/src/operations/create.rs @@ -712,7 +712,7 @@ mod tests { use crate::transforms::Transform; use crate::workflow_bundle::BundledWorkflow; fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), @@ -2345,7 +2345,7 @@ reasoning = false std::fs::create_dir_all(storage_dir.join("store")).unwrap(); let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap()); - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( object_store, "", Duration::from_millis(1), @@ -2405,7 +2405,7 @@ reasoning = false std::fs::create_dir_all(storage_dir.join("store")).unwrap(); let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap()); - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( object_store, "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/operations/fork.rs b/lib/components/fabro-workflow/src/operations/fork.rs index 5b827dbe8..bb5fee6cb 100644 --- a/lib/components/fabro-workflow/src/operations/fork.rs +++ b/lib/components/fabro-workflow/src/operations/fork.rs @@ -294,7 +294,7 @@ mod tests { use super::*; fn test_store() -> Database { - Database::new( + fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/operations/retry.rs b/lib/components/fabro-workflow/src/operations/retry.rs index 8fdd1dda4..5d58e9cc1 100644 --- a/lib/components/fabro-workflow/src/operations/retry.rs +++ b/lib/components/fabro-workflow/src/operations/retry.rs @@ -134,7 +134,7 @@ mod tests { use super::*; fn memory_store() -> Database { - Database::new( + fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/operations/start.rs b/lib/components/fabro-workflow/src/operations/start.rs index 980c5db42..d7a79aad7 100644 --- a/lib/components/fabro-workflow/src/operations/start.rs +++ b/lib/components/fabro-workflow/src/operations/start.rs @@ -1276,7 +1276,7 @@ mod tests { } fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 916b5883b..7e8647f6c 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -195,7 +195,7 @@ fn test_lifecycle(setup_commands: Vec<&str>) -> LifecycleOptions { } async fn test_run_store(run_id: &RunId) -> fabro_store::RunDatabase { - let store: Arc = Arc::new(Database::new( + let store: Arc = Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 5190bc4db..58f1b6930 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -776,7 +776,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 39f3629ef..d05d9acd9 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -785,7 +785,7 @@ mod tests { } fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/pipeline/persist.rs b/lib/components/fabro-workflow/src/pipeline/persist.rs index ef6bf4df4..c8162479c 100644 --- a/lib/components/fabro-workflow/src/pipeline/persist.rs +++ b/lib/components/fabro-workflow/src/pipeline/persist.rs @@ -103,7 +103,7 @@ mod tests { use crate::records::RunSpec; fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/pipeline/pull_request.rs b/lib/components/fabro-workflow/src/pipeline/pull_request.rs index 4de3f2ec4..10f324dd3 100644 --- a/lib/components/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/components/fabro-workflow/src/pipeline/pull_request.rs @@ -780,7 +780,7 @@ mod tests { } fn test_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/run_lookup.rs b/lib/components/fabro-workflow/src/run_lookup.rs index cff4f1657..d7ba5993f 100644 --- a/lib/components/fabro-workflow/src/run_lookup.rs +++ b/lib/components/fabro-workflow/src/run_lookup.rs @@ -458,7 +458,7 @@ mod tests { use crate::records::RunSpec; fn memory_store() -> Arc { - Arc::new(Database::new( + Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/runtime_store.rs b/lib/components/fabro-workflow/src/runtime_store.rs index 3217741d2..c80121d5a 100644 --- a/lib/components/fabro-workflow/src/runtime_store.rs +++ b/lib/components/fabro-workflow/src/runtime_store.rs @@ -116,7 +116,6 @@ mod tests { use std::time::Duration; use chrono::Utc; - use fabro_store::Database; use fabro_types::run_event::RunSubmittedProps; use fabro_types::{EventBody, RunEvent, fixtures, test_support}; use object_store::memory::InMemory; @@ -126,7 +125,7 @@ mod tests { use crate::records::RunSpec; async fn test_run_store() -> fabro_store::RunDatabase { - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/services.rs b/lib/components/fabro-workflow/src/services.rs index 64facc23e..9fa4c21e1 100644 --- a/lib/components/fabro-workflow/src/services.rs +++ b/lib/components/fabro-workflow/src/services.rs @@ -262,7 +262,6 @@ impl EngineServices { reason = "Test scaffolding must build a slate-backed run store from sync code." )] pub fn test_default() -> Self { - use fabro_store::Database; use object_store::memory::InMemory; use crate::handler::start; @@ -286,7 +285,7 @@ impl EngineServices { } } - let store = Arc::new(Database::new( + let store = Arc::new(fabro_store::test_support::test_database( Arc::new(InMemory::new()), "", Duration::from_millis(1), diff --git a/lib/components/fabro-workflow/src/test_support.rs b/lib/components/fabro-workflow/src/test_support.rs index a4b128c5d..08f657ff1 100644 --- a/lib/components/fabro-workflow/src/test_support.rs +++ b/lib/components/fabro-workflow/src/test_support.rs @@ -11,7 +11,7 @@ use fabro_interview::AutoApproveInterviewer; use fabro_model::Catalog; #[cfg(feature = "test-support")] use fabro_model::ProviderId; -use fabro_store::{ArtifactStore, Database, RunProjection}; +use fabro_store::{ArtifactStore, RunProjection, test_support as store_test_support}; use object_store::local::LocalFileSystem; use crate::artifact_upload::ArtifactSink; @@ -174,8 +174,14 @@ async fn initialized( std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir"); let store_dir = test_store_dir(&run_options.run_dir); let _ = std::fs::remove_dir_all(&store_dir); + let blob_store_path = store_test_support::test_blob_store_path(&store_dir); + for suffix in ["", "-wal", "-shm"] { + let mut sibling = blob_store_path.clone().into_os_string(); + sibling.push(suffix); + let _ = std::fs::remove_file(sibling); + } std::fs::create_dir_all(&store_dir).expect("failed to create local test run store dir"); - let store = Arc::new(Database::new( + let store = Arc::new(store_test_support::test_database_with_blobs( Arc::new( LocalFileSystem::new_with_prefix(&store_dir) .expect("failed to create local test run store"), @@ -183,6 +189,7 @@ async fn initialized( "", Duration::from_millis(1), None, + store_test_support::test_blob_store_at(&store_dir), )); let inner_store = store .create_run(&run_options.run_id) diff --git a/lib/components/fabro-workflow/tests/it/daytona_integration.rs b/lib/components/fabro-workflow/tests/it/daytona_integration.rs index fda3aabf7..313befccb 100644 --- a/lib/components/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/components/fabro-workflow/tests/it/daytona_integration.rs @@ -26,7 +26,7 @@ use fabro_agent::Sandbox; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox}; use fabro_static::EnvVars; -use fabro_store::{ArtifactKey, ArtifactStore, Database}; +use fabro_store::{ArtifactKey, ArtifactStore}; use fabro_types::{RunId, StageId, WorkflowSettings, parse_blob_ref}; use fabro_util::shell; use fabro_workflow::artifact; @@ -69,7 +69,7 @@ fn load_run_checkpoint(run_dir: &Path) -> Result Result PathBuf { append_to_path(database_path, ".pre-migration.bak") } -fn append_to_path(path: &Path, suffix: &str) -> PathBuf { +/// Returns `path` with `suffix` appended to its final component, preserving +/// any extension (`fabro.sqlite3` + `-wal` → `fabro.sqlite3-wal`). +pub fn append_to_path(path: &Path, suffix: &str) -> PathBuf { let mut path = path.as_os_str().to_os_string(); path.push(suffix); PathBuf::from(path) @@ -218,6 +218,169 @@ async fn applied_migration_versions(pool: &DbPool) -> anyhow::Result Result<(), SnapshotStagingError> { + write_snapshot_to_staging_inner(pool, staging_path, |_| {}).await +} + +async fn write_snapshot_to_staging_inner( + 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 = + private_staging_path + .to_str() + .ok_or_else(|| SnapshotStagingError::NonUtf8Path { + path: staging_path.to_path_buf(), + })?; + sqlx::query("VACUUM INTO ?") + .bind(staging_target) + .execute(pool) + .await + .map_err(|source| SnapshotStagingError::Write { + path: staging_path.to_path_buf(), + source, + })?; + after_write(&private_staging_path); + set_private_permissions(&private_staging_path) + .await + .map_err(|source| SnapshotStagingError::SetPermissions { + path: staging_path.to_path_buf(), + source, + })?; + // 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(&private_staging_path).await { + Ok(file) => file.sync_all().await, + Err(source) => Err(source), + }; + sync_result.map_err(|source| SnapshotStagingError::Sync { + 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 { + 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. +#[cfg(unix)] +#[expect( + clippy::disallowed_methods, + reason = "directory fds have no async open; callers run this on a blocking thread" +)] +pub fn sync_parent_directory(path: &Path) -> std::io::Result<()> { + std::fs::File::open(nonempty_parent(path))?.sync_all() +} + +/// 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. +#[cfg(not(unix))] +pub fn sync_parent_directory(_path: &Path) -> std::io::Result<()> { + Ok(()) +} + +/// Removes `path`, treating an already-missing file as success. async fn remove_file_if_exists(path: &Path) -> std::io::Result<()> { match fs::remove_file(path).await { Ok(()) => Ok(()), @@ -226,17 +389,17 @@ async fn remove_file_if_exists(path: &Path) -> std::io::Result<()> { } } +/// Restricts `path` to owner-only access (0o600). No-op off Unix. #[cfg(unix)] -async fn set_private_permissions(path: &Path) -> anyhow::Result<()> { +async fn set_private_permissions(path: &Path) -> std::io::Result<()> { use std::os::unix::fs::PermissionsExt; - fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) - .await - .with_context(|| format!("setting permissions on {}", path.display())) + fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).await } +/// Restricts `path` to owner-only access (0o600). No-op off Unix. #[cfg(not(unix))] -async fn set_private_permissions(_path: &Path) -> anyhow::Result<()> { +async fn set_private_permissions(_path: &Path) -> std::io::Result<()> { Ok(()) } @@ -274,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(()) + } +}