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

Isolate the shared test blob store between tests

Browse source 
test_blob_store was a process-wide OnceLock singleton over one in-memory
SQLite connection, so content-addressed rows written by one test were
visible to every other test in the same process. nextest's
process-per-test model masked the bleed, but plain cargo test failed
(8/24 in fabro-workflow-version) because negative existence assertions
became order-dependent.

test_blob_store now builds a fresh isolated in-memory store per call,
and test_database gives every database its own blob authority.
Reopen-style tests that model one durable blob authority across several
store handles use the new test_blob_store_at, which keeps the blob table
in a SQLite file beside the store directory, plus
test_database_with_blobs to share it explicitly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-08-23 13:01:13 -04:00
parent eb54a8d0f8
commit 21421dce78
3 changed files with 100 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,7 +1,5 @@
#[cfg(test)]
use std::path::Path;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use fabro_types::{BlobHash, RunId};
@ -13,47 +11,85 @@ use crate::{AuthSessionStore, RunSummaryStore};
use crate::keys::SlateKey;
use crate::{BlobStore, Database, Result};
/// Returns the process-wide SQLite blob authority used by Slate-backed tests.
/// Returns an isolated SQLite blob authority backed by its own in-memory
/// database.
///
/// Production has one blob authority shared by every run handle. Keeping the
/// same shape in test processes also lets helpers reopen the Slate run store
/// without silently switching to an empty blob database. The CAS key is the
/// content hash, so sharing rows across otherwise isolated tests is safe.
/// 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> {
static STORE: OnceLock<Arc<BlobStore>> = OnceLock::new();
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))
}
Arc::clone(STORE.get_or_init(|| {
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 {
/// 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(())
})
}
Ok(())
})
.connect_lazy_with(options);
Arc::new(BlobStore::new(pool))
}))
})
.connect_lazy_with(options);
Arc::new(BlobStore::new(pool))
}
/// Builds a Slate-backed run database using the process-wide test blob
/// authority and its signed SQLite blob table.
/// Builds a Slate-backed run database with its own isolated blob authority.
#[must_use]
pub fn test_database(
object_store: Arc<dyn ObjectStore>,
@ -61,7 +97,7 @@ pub fn test_database(
flush_interval: Duration,
cache_path: Option<PathBuf>,
) -> Database {
Database::new(
test_database_with_blobs(
object_store,
base_prefix,
flush_interval,
@ -70,6 +106,22 @@ pub fn test_database(
)
}
/// 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);

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

@ -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(store_test_support::test_database(
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)

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

@ -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(fabro_store::test_support::test_database(
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(fabro_store::test_support::test_database(
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(fabro_store::test_support::test_database(
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();