scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #783 from fabro-sh/codex/sqlite-blob-startup-activation

Browse source 
Activate verified SQLite blob storage at server startup
This commit is contained in:
Scott Werner 2026-08-24 16:32:59 -04:00 • committed by GitHub
commit 2d292c28f8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
57 changed files with 2113 additions and 381 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -2593,6 +2593,7 @@ dependencies = [
"chrono",
"sqlx",
"tempfile",
"thiserror 2.0.18",
"tokio",
"tracing",
]

43
docs/public/reference/server-operations.mdx
View file

@ -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:

4
lib/apps/fabro-server/Cargo.toml
View file

@ -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

816
lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs Normal file
View file

@ -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<fabro_store::LegacyBlobImportError>),
#[error("verifying legacy and SQLite blobs")]
Verification(#[source] Box<fabro_store::LegacyBlobVerificationError>),
#[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<dyn ObjectStore>,
slatedb_prefix: String,
flush_interval: Duration,
cache_path: Option<PathBuf>,
) -> Result<Arc<fabro_store::Database>, 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<u64, BlobActivationError> {
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<bool, BlobActivationError> {
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<u64, BlobActivationError> {
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<u64, BlobActivationError> {
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<u64, BlobActivationError> {
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<bool, sqlx::Error>
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<T> = Result<T, Box<dyn std::error::Error>>;
#[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<dyn ObjectStore> = 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<dyn ObjectStore> = Arc::new(InMemory::new());
let source = fabro_store::test_support::test_database(
Arc::clone(&object_store),
"missing-backup-test",
Duration::from_millis(1),
None,
);
let bytes = b"already-imported";
let hash = fabro_store::test_support::put_legacy_blob(&source, bytes).await?;
sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
.bind(hash.to_string())
.bind(bytes.as_slice())
.execute(database.pool())
.await?;
drop(source);
let error = activate_blob_storage(
&database,
&sqlite_path,
object_store,
"missing-backup-test".to_string(),
Duration::from_millis(1),
None,
)
.await
.expect_err("startup must not move the pre-activation rollback boundary");
assert!(matches!(
error,
BlobActivationError::MissingBackupAfterImport {
legacy_rows: 1,
existing_rows: 1,
..
}
));
assert!(!append_to_path(&sqlite_path, BACKUP_SUFFIX).exists());
Ok(())
}
#[tokio::test]
async fn empty_inventory_skips_backup_and_serves_existing_sqlite_rows() -> TestResult<()> {
let directory = tempfile::tempdir()?;
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<dyn ObjectStore> = 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(())
}
}

3
lib/apps/fabro-server/src/migrations.rs
View file

@ -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 =

15
lib/apps/fabro-server/src/run_intent.rs
View file

@ -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",

8
lib/apps/fabro-server/src/serve.rs
View file

@ -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

2
lib/apps/fabro-server/src/server.rs
View file

@ -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;

13
lib/apps/fabro-server/src/server/handler/runs.rs
View file

@ -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<String>, 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",

24
lib/apps/fabro-server/src/server/handler/workflow_versions.rs
View file

@ -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(),

21
lib/apps/fabro-server/src/server/resource_sampler.rs
View file

@ -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<DiskCandidate> {
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::<Vec<_>>();
.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<u64> {
select_storage_disk(storage_path, &refreshed_disk_candidates()).map(|disk| disk.available_bytes)
}
fn percent(used: u64, total: u64) -> Option<f64> {
if total == 0 {
return None;

2
lib/apps/fabro-server/src/server/tests.rs
View file

@ -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

4
lib/apps/fabro-server/src/test_support.rs
View file

@ -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<Database>, ArtifactStore) {
let object_store: Arc<dyn object_store::ObjectStore> = 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),

4
lib/apps/fabro-server/tests/it/api/auth_sessions.rs
View file

@ -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<AppState>) {
let settings = settings_from_toml(source);
let object_store: Arc<dyn object_store::ObjectStore> = 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),

2
lib/apps/fabro-server/tests/it/api/cli_auth_token.rs
View file

@ -19,7 +19,7 @@ use crate::helpers::{body_json, settings_from_toml};
fn test_app(source: &str) -> (axum::Router, Arc<Database>, Arc<AppState>) {
let settings = settings_from_toml(source);
let object_store: Arc<dyn object_store::ObjectStore> = 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),

2
lib/apps/fabro-server/tests/it/api/events.rs
View file

@ -14,7 +14,7 @@ use crate::helpers::{MINIMAL_DOT, api, minimal_manifest_json, response_json, tes
fn app_with_store(object_store: Arc<dyn ObjectStore>) -> 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),

2
lib/apps/fabro-server/tests/it/api/run_files.rs
View file

@ -39,7 +39,7 @@ fn files_url_with_scope(run_id: &str, scope: &str) -> String {
fn store_bundle() -> (Arc<Database>, ArtifactStore) {
let object_store: Arc<dyn object_store::ObjectStore> = 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),

3
lib/components/fabro-store/Cargo.toml
View file

@ -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

45
lib/components/fabro-store/src/blob_store.rs
View file

@ -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<Blob>),
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<slatedb::Db>) -> 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<BlobHash> {
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<Option<Bytes>> {
match &self.backend {
#[cfg(test)]
BlobBackend::Slate(repo) => Ok(repo.get(blob_hash).await?.map(|blob| blob.0)),
BlobBackend::Sqlite(pool) => {
let stored: Option<Vec<u8>> =
@ -144,6 +144,7 @@ impl BlobStore {
pub async fn exists(&self, blob_hash: &BlobHash) -> Result<bool> {
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<T> = std::result::Result<T, Box<dyn std::error::Error>>;
async fn slate_store() -> Arc<BlobStore> {
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<slatedb::Db>, BlobStore) {

765
lib/components/fabro-store/src/legacy_blob_import.rs
View file

File diff suppressed because it is too large Load diff

5
lib/components/fabro-store/src/lib.rs
View file

@ -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,

1
lib/components/fabro-store/src/record/repository.rs
View file

@ -119,6 +119,7 @@ impl<R: Record> Repository<R> {
Ok(())
}
#[cfg(test)]
pub(crate) async fn exists(&self, id: &R::Id) -> Result<bool> {
Ok(self.db.get(key_for_id::<R>(id)?).await?.is_some())
}

4
lib/components/fabro-store/src/slate/auth_codes.rs
View file

@ -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<AuthCodeStore> {
let db = Database::new(
let db = test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

48
lib/components/fabro-store/src/slate/mod.rs
View file

@ -43,7 +43,7 @@ pub struct Database {
cache_path: Option<PathBuf>,
db: Arc<OnceCell<slatedb::Db>>,
active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
blobs: Arc<OnceCell<Arc<BlobStore>>>,
blobs: Arc<BlobStore>,
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
auth_codes: Arc<OnceCell<Arc<AuthCodeStore>>>,
projection_cache: Arc<RunProjectionCache>,
@ -67,6 +67,7 @@ impl Database {
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
) -> 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<Arc<BlobStore>> {
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<BlobStore> {
Arc::clone(&self.blobs)
}
/// Delete every record under the retired `auth/refresh` prefix.
@ -569,7 +564,7 @@ mod tests {
fn make_store() -> (Arc<dyn ObjectStore>, Database) {
let object_store: Arc<dyn ObjectStore> = 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),

18
lib/components/fabro-store/src/slate/run_store.rs
View file

@ -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();

134
lib/components/fabro-store/src/test_support/mod.rs
View file

@ -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<BlobStore> {
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<BlobStore> {
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<dyn ObjectStore>,
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
) -> 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<dyn ObjectStore>,
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
) -> 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<BlobHash> {
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.

1
lib/components/fabro-workflow-version/Cargo.toml
View file

@ -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"] }

26
lib/components/fabro-workflow-version/src/store.rs
View file

@ -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<BlobStore>, WorkflowVersionStore) {
let database = Database::new(
fn stores() -> (Arc<BlobStore>, 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(),

3
lib/components/fabro-workflow/Cargo.toml
View file

@ -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

8
lib/components/fabro-workflow/src/artifact.rs
View file

@ -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()
}

2
lib/components/fabro-workflow/src/event/sink.rs
View file

@ -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),

2
lib/components/fabro-workflow/src/git.rs
View file

@ -305,7 +305,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/handler/agent.rs
View file

@ -458,7 +458,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/handler/command.rs
View file

@ -435,7 +435,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

93
lib/components/fabro-workflow/src/handler/manager_loop.rs
View file

@ -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<Outcome, Error> {
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();

2
lib/components/fabro-workflow/src/handler/parallel.rs
View file

@ -969,7 +969,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/handler/prompt.rs
View file

@ -236,7 +236,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

3
lib/components/fabro-workflow/src/lifecycle/fidelity.rs
View file

@ -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),

4
lib/components/fabro-workflow/src/lifecycle/git.rs
View file

@ -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),

2
lib/components/fabro-workflow/src/operations/archive.rs
View file

@ -144,7 +144,7 @@ mod tests {
use super::*;
fn memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

6
lib/components/fabro-workflow/src/operations/create.rs
View file

@ -712,7 +712,7 @@ mod tests {
use crate::transforms::Transform;
use crate::workflow_bundle::BundledWorkflow;
fn memory_store() -> Arc<Database> {
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),

2
lib/components/fabro-workflow/src/operations/fork.rs
View file

@ -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),

2
lib/components/fabro-workflow/src/operations/retry.rs
View file

@ -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),

2
lib/components/fabro-workflow/src/operations/start.rs
View file

@ -1276,7 +1276,7 @@ mod tests {
}
fn memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -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<Database> = Arc::new(Database::new(
let store: Arc<Database> = Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/pipeline/finalize.rs
View file

@ -776,7 +776,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

@ -785,7 +785,7 @@ mod tests {
}
fn memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/pipeline/persist.rs
View file

@ -103,7 +103,7 @@ mod tests {
use crate::records::RunSpec;
fn memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -780,7 +780,7 @@ mod tests {
}
fn test_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

2
lib/components/fabro-workflow/src/run_lookup.rs
View file

@ -458,7 +458,7 @@ mod tests {
use crate::records::RunSpec;
fn memory_store() -> Arc<Database> {
Arc::new(Database::new(
Arc::new(fabro_store::test_support::test_database(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),

3
lib/components/fabro-workflow/src/runtime_store.rs
View file

@ -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),

3
lib/components/fabro-workflow/src/services.rs
View file

@ -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),

11
lib/components/fabro-workflow/src/test_support.rs
View file

@ -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)

11
lib/components/fabro-workflow/tests/it/daytona_integration.rs
View file

@ -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<Checkpoint, Box<dyn std::error:
test_store_dir(&run_dir)
};
let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?);
let store = Arc::new(Database::new(
let store = Arc::new(fabro_store::test_support::test_database(
object_store,
"",
std::time::Duration::from_millis(1),
@ -173,7 +173,12 @@ async fn resolve_checkpoint_text(
}
let object_store = Arc::new(LocalFileSystem::new_with_prefix(test_store_dir(run_dir))?);
let store = Database::new(object_store, "", std::time::Duration::from_millis(1), None);
let store = fabro_store::test_support::test_database(
object_store,
"",
std::time::Duration::from_millis(1),
None,
);
let run = store.open_run_reader(run_id).await?;
let run_store = RunStoreHandle::from(run);
Ok(artifact::resolve_text_or_blob_ref_str(current, &run_store).await?)

15
lib/components/fabro-workflow/tests/it/integration.rs
View file

@ -32,7 +32,7 @@ use fabro_interview::{
};
use fabro_model::catalog::{LlmCatalogSettings, ProviderCatalogSettings};
use fabro_model::{Catalog, ProviderId};
use fabro_store::{ArtifactKey, ArtifactStore, Database};
use fabro_store::{ArtifactKey, ArtifactStore};
use fabro_types::{EventBody, RunEvent, RunId, StageId, WorkflowSettings, parse_blob_ref};
use fabro_validate::{Severity, validate, validate_or_raise};
use fabro_workflow::artifact;
@ -117,12 +117,13 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
} else {
test_store_dir(&run_dir)
};
let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?);
let store = Arc::new(Database::new(
let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_dir)?);
let store = Arc::new(fabro_store::test_support::test_database_with_blobs(
object_store,
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
));
let state = if tokio::runtime::Handle::try_current().is_ok() {
std::thread::spawn(
@ -247,12 +248,13 @@ fn resolve_checkpoint_text(
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?);
let store = Arc::new(Database::new(
let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_dir)?);
let store = Arc::new(fabro_store::test_support::test_database_with_blobs(
object_store,
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
));
let run_id = if uses_shared_store {
run_dir
@ -7821,11 +7823,12 @@ async fn workflow_run_with_vault_only_openai_codex_builds_pr_body() {
assert_eq!(outcome.status, StageOutcome::Succeeded);
let store_dir = test_store_dir(&run_options.run_dir);
let store = Arc::new(Database::new(
let store = Arc::new(fabro_store::test_support::test_database_with_blobs(
Arc::new(LocalFileSystem::new_with_prefix(&store_dir).unwrap()),
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
));
let run_store = store.open_run_reader(&run_options.run_id).await.unwrap();
let run_store_handle: fabro_workflow::runtime_store::RunStoreHandle = run_store.into();

3
lib/foundation/fabro-db/Cargo.toml
View file

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

283
lib/foundation/fabro-db/src/lib.rs
View file

@ -16,6 +16,10 @@ pub type DbPool = sqlx::SqlitePool;
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
/// The blob-table migration, exposed so fixtures in other crates can install
/// the production blob schema without a filesystem path into this crate.
pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql");
#[derive(Clone)]
pub struct Database {
pool: DbPool,
@ -88,31 +92,12 @@ impl Database {
let database_path = connect_options.get_filename();
let snapshot_path = pre_migration_snapshot_path(database_path);
// VACUUM INTO produces a consistent single-file copy from the live
// pool, so the snapshot needs no -wal/-shm siblings to restore. It
// writes to a staging file that is renamed into place afterwards, so
// a failure mid-copy never leaves a partial file at the snapshot
// path.
// The snapshot is staged and then renamed into place, so a failure
// mid-copy never leaves a partial file at the snapshot path.
let staging_path = append_to_path(&snapshot_path, ".tmp");
remove_file_if_exists(&staging_path)
write_snapshot_to_staging(&self.pool, &staging_path)
.await
.with_context(|| {
format!(
"removing stale snapshot staging file {}",
staging_path.display()
)
})?;
let staging_target = staging_path
.to_str()
.context("snapshot staging path is not valid UTF-8")?;
sqlx::query("VACUUM INTO ?")
.bind(staging_target)
.execute(&self.pool)
.await
.with_context(|| {
format!("writing pre-migration snapshot {}", staging_path.display())
})?;
set_private_permissions(&staging_path).await?;
.context("staging the pre-migration snapshot")?;
remove_file_if_exists(&snapshot_path)
.await
.with_context(|| {
@ -129,6 +114,19 @@ impl Database {
snapshot_path.display()
)
})?;
#[cfg(unix)]
{
let published_path = snapshot_path.clone();
spawn_blocking(move || sync_parent_directory(&published_path))
.await
.context("joining the snapshot directory sync task")?
.with_context(|| {
format!(
"syncing the directory of pre-migration snapshot {}",
snapshot_path.display()
)
})?;
}
info!(
database = %database_path.display(),
@ -191,7 +189,9 @@ pub fn pre_migration_snapshot_path(database_path: &Path) -> 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<HashSet<i64
.collect())
}
/// Error writing a consistent single-file SQLite snapshot to a staging path.
#[derive(Debug, thiserror::Error)]
pub enum SnapshotStagingError {
#[error("removing stale snapshot staging file {path}")]
RemoveStale {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("creating private snapshot staging area for {path}")]
CreatePrivateStagingArea {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("snapshot staging path is not valid UTF-8 at {path}")]
NonUtf8Path { path: PathBuf },
#[error("writing SQLite snapshot {path}")]
Write {
path: PathBuf,
#[source]
source: sqlx::Error,
},
#[error("setting private permissions on snapshot staging file {path}")]
SetPermissions {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("flushing snapshot staging file {path} to disk")]
Sync {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("publishing private snapshot staging file {path}")]
Publish {
path: PathBuf,
#[source]
source: std::io::Error,
},
}
/// Writes a consistent single-file copy of the live pool to `staging_path`.
///
/// `VACUUM INTO` produces a snapshot that needs no `-wal`/`-shm` siblings to
/// restore. The copy is first written inside a private same-directory staging
/// area, then restricted to owner-only permissions and flushed before it is
/// exposed at `staging_path`. The caller publishes the staging file into its
/// final path and owns the durability of that rename.
pub async fn write_snapshot_to_staging(
pool: &DbPool,
staging_path: &Path,
) -> Result<(), SnapshotStagingError> {
write_snapshot_to_staging_inner(pool, staging_path, |_| {}).await
}
async fn write_snapshot_to_staging_inner<F>(
pool: &DbPool,
staging_path: &Path,
after_write: F,
) -> Result<(), SnapshotStagingError>
where
F: FnOnce(&Path),
{
remove_file_if_exists(staging_path)
.await
.map_err(|source| SnapshotStagingError::RemoveStale {
path: staging_path.to_path_buf(),
source,
})?;
let staging_parent = nonempty_parent(staging_path);
// SQLite's VACUUM INTO creates its destination with umask-derived
// permissions. Keep that file behind an owner-only directory until its
// own mode is restricted, so a traversable database directory never
// exposes a partially written snapshot.
let private_staging_area = create_private_staging_area(staging_parent).map_err(|source| {
SnapshotStagingError::CreatePrivateStagingArea {
path: staging_path.to_path_buf(),
source,
}
})?;
let private_staging_path = private_staging_area.path().join("snapshot.sqlite3");
let staging_target =
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<tempfile::TempDir> {
let mut builder = tempfile::Builder::new();
builder.prefix(".fabro-snapshot-");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
builder.permissions(std::fs::Permissions::from_mode(0o700));
}
builder.tempdir_in(parent)
}
fn nonempty_parent(path: &Path) -> &Path {
path.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."))
}
/// Flushes the directory entry metadata for `path`'s parent so a rename into
/// that directory survives power loss. No-op off Unix, where a directory
/// cannot be opened for syncing.
#[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(())
}
}