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

Merge pull request #823 from fabro-sh/codex/sqlite-run-history-activation

Browse source 
Activate atomic SQLite run history storage
This commit is contained in:
Scott Werner 2026-08-31 12:53:11 -04:00 • committed by GitHub
commit 05fb173767
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
34 changed files with 2441 additions and 2084 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
docs/public/administration/server-configuration.mdx
View file

@ -199,7 +199,10 @@ enabled = true
### `[server.slatedb]` section
Configure the embedded SlateDB key-value store used for run event storage.
Configure the embedded SlateDB key-value store used for the remaining
object-store-backed indexes and as the read-only source for temporary storage
migrations. Run history and content-addressed blobs live in SQLite; artifacts
use `[server.artifacts]`.
| Key | Description | Default |
|---|---|---|
@ -257,7 +260,11 @@ honors those hand-edited values even though the browser wizard does not manage t
### SQLite state and migration backups
Shared relational state, including vault entries, server-managed definitions, and CLI auth sessions, lives at `<storage_root>/db/fabro.sqlite3`. Run events continue to use the `[server.slatedb]` object store.
Shared relational state, including run events and current run rows,
content-addressed blobs, vault entries, server-managed definitions, and CLI
auth sessions, lives at `<storage_root>/db/fabro.sqlite3`. The
`[server.slatedb]` object store remains configured for compatibility imports
and session-to-run reverse indexes during the storage transition.
CLI auth sessions are stored as an `auth_sessions` row per signed-in CLI, with the rotating refresh tokens for that session in `refresh_tokens`. Pending browser-to-CLI handoffs live briefly in `oauth_authorization_codes`; the table contains a SHA-256 hash of each one-time code, never the raw bearer value. Revoking a session from **Settings → Sessions**, or with `DELETE /api/v1/auth/sessions/{id}`, deletes the session row and its tokens together.

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

@ -58,13 +58,15 @@ See [Server Configuration](/administration/server-configuration) for the full `s
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.
prefix and run history, then checks disk headroom for the rows not yet
imported, any required blob backup, and the projected post-import database
snapshot required by run-history activation. A warm restart with no pending
imports or backups only needs a small fixed headroom; on filesystems whose free
space cannot be determined the check is skipped with a warning. Fabro then
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
@ -96,6 +98,53 @@ 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.
### SQLite run-history activation
Immediately after blob activation, and still before routes, schedulers,
workers, webhooks, reapers, or readiness are exposed, Fabro activates SQLite
as the sole authority for run existence, run events, and each run's current
projected row. The activation strictly validates and fingerprints the exact
legacy SlateDB run-event key/value stream, imports each complete run in its own
transaction, verifies every legacy history as an exact SQLite prefix, replays
and verifies every SQLite run independently, and runs a full SQLite integrity
check. It attempts a final WAL truncate checkpoint, but a blocking reader only
produces a warning because committed activation data remains durable in the
WAL. A source fingerprint or count change after activation stops startup. There
is no fallback or dual-read/write mode.
For a non-empty legacy run history, the first activation creates and validates
the private sibling backup
`fabro.sqlite3.pre-run-history-activation.bak` before importing anything. The
backup is published without overwriting an existing file and is revalidated
on every restart. If import progress exists but that retained backup is
missing, startup stops. An empty legacy source is accepted without a backup
only when SQLite also has no unmarked run data. The activation marker stores
the source identity and first-success timestamp; retries preserve that
timestamp and repeat source, destination, backup, and integrity checks.
After activation, creating a run commits `run.created`, the run's current row,
and its existence atomically. Later appends update the event log and current
row in one transaction, and live streams advance only after commit. Deleting
a migrated run commits a tombstone with the SQL deletion so the retained
legacy source cannot resurrect it during a restart.
Keep the unchanged legacy `runs/*/events/*` data and the private activation
backup for at least 30 consecutive calendar days after the persisted
first-success timestamp. Cleanup also requires successful cold and warm
activation evidence, production observation, backup and restore validation,
deletion/restart coverage, and explicit approval for a separate cleanup
change. Nothing is deleted automatically. The run-history activation backup
represents the database immediately before run-history import and can be used
to retry or recover the activation with a binary that knows the activated
schema. It is not a binary-downgrade artifact because it already contains the
new SQL migrations.
To return to the older binary, stop the server and restore the database's
`.pre-migration.bak` snapshot instead, then remove any `-wal` and `-shm`
siblings before starting the older binary. That snapshot was taken before the
new migrations were applied. Either recovery path loses writes accepted after
its snapshot, so make the rollback boundary explicit before restoring it.
## Submitting runs
Workflows are submitted via the REST API and executed in the background. The exact request body is documented in the API reference:

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

@ -10,14 +10,12 @@ 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::migrations::sqlite_activation_backup::{self, BackupError};
use crate::migrations::sqlite_run_history_activation::BACKUP_SUFFIX as RUN_HISTORY_BACKUP_SUFFIX;
use crate::server::resource_sampler;
/// Earliest date this bridge becomes eligible for removal, assuming the first
@ -28,7 +26,20 @@ 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";
pub(crate) struct ActivatedBlobStorage {
pub(crate) store: Arc<fabro_store::Database>,
pub(crate) run_history_identity: fabro_store::LegacyRunHistorySourceIdentity,
}
impl std::fmt::Debug for ActivatedBlobStorage {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ActivatedBlobStorage")
.field("run_history_identity", &self.run_history_identity)
.finish_non_exhaustive()
}
}
#[derive(Debug, thiserror::Error)]
pub(crate) enum BlobActivationError {
@ -40,16 +51,8 @@ pub(crate) enum BlobActivationError {
},
#[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("identifying legacy run history before importing blobs")]
RunHistorySourceIdentity(#[source] fabro_store::LegacyRunHistorySourceIdentityError),
#[error(
"activation backup is missing at {path} while {existing_rows} of {legacy_rows} legacy blob rows are already present in SQLite"
)]
@ -58,14 +61,6 @@ pub(crate) enum BlobActivationError {
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,
@ -81,16 +76,8 @@ pub(crate) enum BlobActivationError {
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(transparent)]
Backup(#[from] BackupError),
#[error("importing legacy blobs into SQLite")]
Import(#[source] Box<fabro_store::LegacyBlobImportError>),
#[error("verifying legacy and SQLite blobs")]
@ -110,7 +97,7 @@ pub(crate) async fn activate_blob_storage(
slatedb_prefix: String,
flush_interval: Duration,
cache_path: Option<PathBuf>,
) -> Result<Arc<fabro_store::Database>, BlobActivationError> {
) -> Result<ActivatedBlobStorage, BlobActivationError> {
let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| {
BlobActivationError::Canonicalize {
path: sqlite_path.to_path_buf(),
@ -139,9 +126,13 @@ pub(crate) async fn activate_blob_storage(
.legacy_blob_inventory(database.pool())
.await
.map_err(BlobActivationError::Inventory)?;
let backup_exists = backup_exists(&backup_path).await?;
let run_history_identity = store
.legacy_run_history_source_identity()
.await
.map_err(BlobActivationError::RunHistorySourceIdentity)?;
let backup_exists = sqlite_activation_backup::backup_exists(&backup_path).await?;
if backup_exists {
validate_backup(&backup_path).await?;
sqlite_activation_backup::validate_backup(&backup_path).await?;
}
if !backup_exists && inventory.pending_rows < inventory.rows {
return Err(BlobActivationError::MissingBackupAfterImport {
@ -151,6 +142,15 @@ pub(crate) async fn activate_blob_storage(
});
}
let backup_required = inventory.rows > 0 && !backup_exists;
let run_history_backup_path =
fabro_db::append_to_path(&canonical_path, RUN_HISTORY_BACKUP_SUFFIX);
let run_history_backup_exists =
sqlite_activation_backup::backup_exists(&run_history_backup_path).await?;
if run_history_backup_exists {
sqlite_activation_backup::validate_backup(&run_history_backup_path).await?;
}
let run_history_backup_required =
run_history_identity.events != 0 && !run_history_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:
@ -158,16 +158,25 @@ pub(crate) async fn activate_blob_storage(
// could complete.
if let Some(available_free_bytes) = resource_sampler::available_space_for_path(&canonical_path)
{
let backup_reserve = if backup_required {
let sqlite_bytes = if backup_required || run_history_backup_required {
sqlite_file_set_bytes(&canonical_path).await?
} else {
0
};
let backup_reserve = if backup_required { sqlite_bytes } else { 0 };
let run_history_backup_reserve = if run_history_backup_required {
projected_sqlite_bytes(sqlite_bytes, inventory.pending_bytes)?
} 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.
// already present in SQLite cost nothing on a warm restart. Reserve
// the projected post-import database size as well when run-history
// activation will immediately take its own full SQLite snapshot.
let required_free_bytes = compute_disk_preflight(
inventory.pending_bytes,
backup_reserve,
run_history_backup_reserve,
available_free_bytes,
)?;
debug!(
@ -177,6 +186,9 @@ pub(crate) async fn activate_blob_storage(
pending_bytes = inventory.pending_bytes,
backup_required,
backup_reserve,
run_history_events = run_history_identity.events,
run_history_backup_required,
run_history_backup_reserve,
required_free_bytes,
available_free_bytes,
"Checked SQLite blob activation disk capacity"
@ -191,7 +203,7 @@ pub(crate) async fn activate_blob_storage(
let retained_backup = if backup_exists {
Some(backup_path)
} else if backup_required {
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
Some(backup_path)
} else {
None
@ -227,25 +239,27 @@ pub(crate) async fn activate_blob_storage(
passive_checkpoints = import.passive_checkpoints,
backup_required,
backup_path = ?retained_backup,
run_history_backup_required,
removal_deadline = REMOVAL_DEADLINE,
"Activated SQLite blob storage"
);
Ok(store)
Ok(ActivatedBlobStorage {
store,
run_history_identity,
})
}
/// Fail-closed disk capacity check; returns the required free bytes.
fn compute_disk_preflight(
pending_bytes: u64,
backup_reserve: u64,
run_history_backup_reserve: u64,
available_free_bytes: u64,
) -> Result<u64, BlobActivationError> {
let half = pending_bytes
.checked_add(1)
.ok_or(BlobActivationError::DiskRequirementOverflow)?
/ 2;
let import_reserve = blob_import_reserve(pending_bytes)?;
let required_free_bytes = backup_reserve
.checked_add(pending_bytes)
.and_then(|value| value.checked_add(half))
.checked_add(import_reserve)
.and_then(|value| value.checked_add(run_history_backup_reserve))
.and_then(|value| value.checked_add(DISK_HEADROOM_BYTES))
.ok_or(BlobActivationError::DiskRequirementOverflow)?;
if available_free_bytes < required_free_bytes {
@ -257,15 +271,23 @@ fn compute_disk_preflight(
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,
}),
}
fn projected_sqlite_bytes(
sqlite_bytes: u64,
pending_bytes: u64,
) -> Result<u64, BlobActivationError> {
sqlite_bytes
.checked_add(blob_import_reserve(pending_bytes)?)
.ok_or(BlobActivationError::DiskRequirementOverflow)
}
fn blob_import_reserve(pending_bytes: u64) -> Result<u64, BlobActivationError> {
let half = pending_bytes
.checked_add(1)
.ok_or(BlobActivationError::DiskRequirementOverflow)?
/ 2;
pending_bytes
.checked_add(half)
.ok_or(BlobActivationError::DiskRequirementOverflow)
}
async fn sqlite_file_set_bytes(path: &Path) -> Result<u64, BlobActivationError> {
@ -301,127 +323,8 @@ async fn optional_file_bytes(path: &Path) -> Result<u64, BlobActivationError> {
}
}
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)
let ok = sqlite_activation_backup::integrity_check_is_ok(pool)
.await
.map_err(BlobActivationError::LiveIntegrity)?;
if !ok {
@ -457,9 +360,10 @@ mod tests {
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,
compute_disk_preflight, final_truncate_checkpoint, projected_sqlite_bytes,
sqlite_file_set_bytes,
};
use crate::migrations::sqlite_activation_backup::{self, BackupError, create_backup};
type TestResult<T> = Result<T, Box<dyn std::error::Error>>;
@ -467,14 +371,26 @@ mod tests {
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 run_history_backup_reserve = 20;
let required =
backup_reserve + pending_bytes + 2 + run_history_backup_reserve + DISK_HEADROOM_BYTES;
let required_free_bytes = compute_disk_preflight(pending_bytes, backup_reserve, required)
.expect("exact equality must pass");
let required_free_bytes = compute_disk_preflight(
pending_bytes,
backup_reserve,
run_history_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");
let error = compute_disk_preflight(
pending_bytes,
backup_reserve,
run_history_backup_reserve,
required - 1,
)
.expect_err("one byte below must fail");
assert!(matches!(
error,
BlobActivationError::InsufficientDisk { .. }
@ -484,14 +400,24 @@ mod tests {
#[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");
compute_disk_preflight(2, 0, 0, u64::MAX).expect("available capacity should pass");
assert_eq!(required_free_bytes, 3 + DISK_HEADROOM_BYTES);
}
#[test]
fn disk_preflight_reserves_the_projected_post_import_database() {
let projected = projected_sqlite_bytes(10, 3).expect("the projection should fit");
assert_eq!(projected, 15);
let required = compute_disk_preflight(3, 0, projected, u64::MAX)
.expect("available capacity should pass");
assert_eq!(required, 20 + 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");
let error = compute_disk_preflight(u64::MAX, 1, 0, u64::MAX)
.expect_err("overflow must fail closed");
assert!(matches!(
error,
BlobActivationError::DiskRequirementOverflow
@ -519,8 +445,8 @@ mod tests {
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?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::validate_backup(&backup_path).await?;
assert!(backup_path.is_file());
assert!(!append_to_path(&backup_path, "-wal").exists());
@ -543,7 +469,7 @@ mod tests {
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?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
let original = fs::read(&backup_path).await?;
sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
@ -551,7 +477,7 @@ mod tests {
.bind(b"later".as_slice())
.execute(database.pool())
.await?;
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
assert_eq!(fs::read(&backup_path).await?, original);
Ok(())
@ -572,7 +498,7 @@ mod tests {
assert!(matches!(
error,
BlobActivationError::StageBackup(fabro_db::SnapshotStagingError::Write { .. })
BackupError::Stage(fabro_db::SnapshotStagingError::Write { .. })
));
assert!(!backup_path.exists());
Ok(())
@ -595,7 +521,7 @@ mod tests {
let legacy_hash = fabro_store::test_support::put_legacy_blob(&source, legacy_bytes).await?;
drop(source);
let store = activate_blob_storage(
let activation = activate_blob_storage(
&database,
&sqlite_path,
Arc::clone(&object_store),
@ -604,6 +530,7 @@ mod tests {
None,
)
.await?;
let store = activation.store;
assert_eq!(
store.blobs().read(&legacy_hash).await?.as_deref(),
Some(legacy_bytes.as_slice())
@ -635,7 +562,7 @@ mod tests {
.await?;
assert_eq!(fs::read(&backup_path).await?, original_backup);
assert_eq!(
warm.blobs().read(&sqlite_only_hash).await?.as_deref(),
warm.store.blobs().read(&sqlite_only_hash).await?.as_deref(),
Some(sqlite_only_bytes.as_slice())
);
Ok(())
@ -712,7 +639,7 @@ mod tests {
assert!(!append_to_path(&sqlite_path, BACKUP_SUFFIX).exists());
assert_eq!(
activated.blobs().read(&hash).await?.as_deref(),
activated.store.blobs().read(&hash).await?.as_deref(),
Some(bytes.as_slice())
);
Ok(())
@ -806,8 +733,9 @@ mod tests {
.expect_err("an invalid retained backup must fail closed");
assert!(matches!(
error,
BlobActivationError::BackupIntegrity { .. }
| BlobActivationError::BackupIntegrityFailed { .. }
BlobActivationError::Backup(
BackupError::Integrity { .. } | BackupError::IntegrityFailed { .. }
)
));
let destination_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs")
.fetch_one(database.pool())

603
lib/apps/fabro-server/migrations/2026082801_sqlite_run_history_activation.rs Normal file
View file

@ -0,0 +1,603 @@
//! Fail-closed activation of SQLite run history.
//!
//! This compatibility bridge remains for at least 30 days after the persisted
//! first-success timestamp, and until cold-start, warm-restart, production
//! observation, rollback-backup, deletion, and concurrent-reader evidence has
//! been accepted and Scott explicitly approves removal. The computed date is
//! an eligibility floor, never an automatic deletion trigger.
use std::path::{Path, PathBuf};
use chrono::{DateTime, Duration, Utc};
use tokio::fs;
use tracing::{info, warn};
use crate::migrations::sqlite_activation_backup::{self, BackupError};
pub(crate) const BACKUP_SUFFIX: &str = ".pre-run-history-activation.bak";
const REMOVAL_WINDOW: Duration = Duration::days(30);
#[derive(Clone, Debug, Eq, PartialEq)]
struct ActivationRecord {
source_fingerprint: Vec<u8>,
source_runs: u64,
source_events: u64,
activated_at_ms: i64,
}
#[derive(Debug, thiserror::Error)]
pub(crate) enum RunHistoryActivationError {
#[error("canonicalizing the SQLite database path {path}")]
Canonicalize {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("reading the SQLite run-history activation state")]
ActivationState(#[source] sqlx::Error),
#[error("the persisted run-history activation marker does not match the legacy source")]
MarkerMismatch,
#[error("the persisted run-history activation marker contains an invalid count or timestamp")]
InvalidMarker,
#[error(
"SQLite contains {target_runs} run rows and {target_events} run events, but the legacy run-history source is empty and no activation marker exists"
)]
EmptySourceWithTarget {
target_runs: u64,
target_events: u64,
},
#[error(
"run-history activation backup is missing at {path} after SQLite import progress was recorded"
)]
MissingBackupAfterProgress { path: PathBuf },
#[error(transparent)]
Backup(#[from] BackupError),
#[error("importing legacy run history into SQLite")]
Import(#[source] Box<fabro_store::LegacyRunHistoryImportError>),
#[error("verifying legacy and SQLite run history")]
Verification(#[source] Box<fabro_store::LegacyRunHistoryVerificationError>),
#[error("running the live SQLite integrity check")]
LiveIntegrity(#[source] sqlx::Error),
#[error("the live SQLite integrity check failed")]
LiveIntegrityFailed,
#[error("persisting the SQLite run-history activation marker")]
PersistMarker(#[source] sqlx::Error),
#[error("the run-history activation timestamp is outside the supported range")]
InvalidActivationTimestamp,
#[error("running the final SQLite WAL truncate checkpoint")]
FinalCheckpoint(#[source] sqlx::Error),
#[error("a run-history activation count exceeds SQLite's integer range")]
CountOverflow,
}
pub(crate) async fn activate_run_history(
database: &fabro_db::Database,
sqlite_path: &Path,
store: &fabro_store::Database,
identity: &fabro_store::LegacyRunHistorySourceIdentity,
) -> Result<(), RunHistoryActivationError> {
let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| {
RunHistoryActivationError::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 run-history activation"
);
let marker = read_activation_record(database.pool()).await?;
let (target_runs, target_events) = target_counts(database.pool()).await?;
if let Some(record) = &marker {
verify_marker(record, identity)?;
} else if identity.events == 0 && (target_runs != 0 || target_events != 0) {
return Err(RunHistoryActivationError::EmptySourceWithTarget {
target_runs,
target_events,
});
}
let backup_present = sqlite_activation_backup::backup_exists(&backup_path).await?;
if backup_present {
sqlite_activation_backup::validate_backup(&backup_path).await?;
}
let import_progress = target_events != 0 || marker.is_some();
if identity.events != 0 && import_progress && !backup_present {
return Err(RunHistoryActivationError::MissingBackupAfterProgress { path: backup_path });
}
let backup_required = identity.events != 0 && !backup_present;
if backup_required {
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
}
let import = store
.import_legacy_run_history_into(database.pool())
.await
.map_err(|source| RunHistoryActivationError::Import(Box::new(source)))?;
let verification = store
.verify_legacy_run_history_in(database.pool())
.await
.map_err(|source| RunHistoryActivationError::Verification(Box::new(source)))?;
validate_live_integrity(database.pool()).await?;
let activated_at_ms = marker.as_ref().map_or_else(
|| Utc::now().timestamp_millis(),
|record| record.activated_at_ms,
);
persist_activation_record(database.pool(), identity, activated_at_ms).await?;
final_truncate_checkpoint(database.pool()).await?;
let activated_at = DateTime::<Utc>::from_timestamp_millis(activated_at_ms)
.ok_or(RunHistoryActivationError::InvalidActivationTimestamp)?;
let removal_eligible_at = activated_at + REMOVAL_WINDOW;
info!(
source_runs = identity.runs,
source_events = identity.events,
imported_runs = import.imported_runs,
imported_events = import.imported_events,
existing_runs = import.verified_existing_runs,
existing_events = import.verified_existing_events,
tombstoned_source_runs = verification.tombstoned_source_runs,
tombstoned_source_events = verification.tombstoned_source_events,
target_runs = verification.target_runs,
target_events = verification.target_events,
sql_only_runs = verification.sql_only_runs,
sql_only_events = verification.sql_only_events,
backup_required,
backup_path = %backup_path.display(),
activated_at = %activated_at,
removal_eligible_at = %removal_eligible_at,
"Activated SQLite run history"
);
Ok(())
}
async fn read_activation_record(
pool: &sqlx::SqlitePool,
) -> Result<Option<ActivationRecord>, RunHistoryActivationError> {
let row = sqlx::query_as::<_, (Vec<u8>, i64, i64, i64)>(
r"
SELECT source_fingerprint, source_runs, source_events, activated_at_ms
FROM legacy_run_history_activation
WHERE singleton = 1
",
)
.fetch_optional(pool)
.await
.map_err(RunHistoryActivationError::ActivationState)?;
row.map(
|(source_fingerprint, source_runs, source_events, activated_at_ms)| {
Ok(ActivationRecord {
source_fingerprint,
source_runs: u64::try_from(source_runs)
.map_err(|_| RunHistoryActivationError::InvalidMarker)?,
source_events: u64::try_from(source_events)
.map_err(|_| RunHistoryActivationError::InvalidMarker)?,
activated_at_ms,
})
},
)
.transpose()
}
fn verify_marker(
marker: &ActivationRecord,
identity: &fabro_store::LegacyRunHistorySourceIdentity,
) -> Result<(), RunHistoryActivationError> {
if marker.activated_at_ms < 0 {
return Err(RunHistoryActivationError::InvalidMarker);
}
if marker.source_fingerprint.as_slice() != identity.fingerprint()
|| marker.source_runs != identity.runs
|| marker.source_events != identity.events
{
return Err(RunHistoryActivationError::MarkerMismatch);
}
Ok(())
}
async fn target_counts(pool: &sqlx::SqlitePool) -> Result<(u64, u64), RunHistoryActivationError> {
let (runs, events): (i64, i64) =
sqlx::query_as("SELECT (SELECT COUNT(*) FROM runs), (SELECT COUNT(*) FROM run_events)")
.fetch_one(pool)
.await
.map_err(RunHistoryActivationError::ActivationState)?;
Ok((
u64::try_from(runs).map_err(|_| RunHistoryActivationError::CountOverflow)?,
u64::try_from(events).map_err(|_| RunHistoryActivationError::CountOverflow)?,
))
}
async fn persist_activation_record(
pool: &sqlx::SqlitePool,
identity: &fabro_store::LegacyRunHistorySourceIdentity,
activated_at_ms: i64,
) -> Result<(), RunHistoryActivationError> {
let source_runs =
i64::try_from(identity.runs).map_err(|_| RunHistoryActivationError::CountOverflow)?;
let source_events =
i64::try_from(identity.events).map_err(|_| RunHistoryActivationError::CountOverflow)?;
// A pre-existing marker was already verified against `identity` above, so
// leaving it untouched on conflict keeps the original activation time.
sqlx::query(
r"
INSERT INTO legacy_run_history_activation (
singleton, source_fingerprint, source_runs, source_events, activated_at_ms
) VALUES (1, ?, ?, ?, ?)
ON CONFLICT(singleton) DO NOTHING
",
)
.bind(identity.fingerprint().as_slice())
.bind(source_runs)
.bind(source_events)
.bind(activated_at_ms)
.execute(pool)
.await
.map_err(RunHistoryActivationError::PersistMarker)?;
Ok(())
}
async fn validate_live_integrity(pool: &sqlx::SqlitePool) -> Result<(), RunHistoryActivationError> {
let ok = sqlite_activation_backup::integrity_check_is_ok(pool)
.await
.map_err(RunHistoryActivationError::LiveIntegrity)?;
if !ok {
return Err(RunHistoryActivationError::LiveIntegrityFailed);
}
Ok(())
}
async fn final_truncate_checkpoint(
pool: &sqlx::SqlitePool,
) -> Result<(), RunHistoryActivationError> {
let (busy, _, _): (i64, i64, i64) = sqlx::query_as("PRAGMA wal_checkpoint(TRUNCATE)")
.fetch_one(pool)
.await
.map_err(RunHistoryActivationError::FinalCheckpoint)?;
if busy != 0 {
// A concurrent reader can keep the WAL from truncating, but all
// activation data is already committed and remains durable in that
// WAL. A later checkpoint can truncate it after the reader exits.
warn!(
"The final SQLite run-history WAL truncate checkpoint could not complete; continuing startup"
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration as StdDuration;
use chrono::{TimeZone as _, Utc};
use fabro_types::{Graph, RunId, WorkflowSettings, test_support};
use object_store::memory::InMemory;
use sqlx::Connection as _;
use tokio::fs;
use ulid::Ulid;
use super::{
BACKUP_SUFFIX, RunHistoryActivationError, activate_run_history, final_truncate_checkpoint,
read_activation_record,
};
type TestResult<T> = Result<T, Box<dyn std::error::Error>>;
struct TestContext {
_directory: tempfile::TempDir,
sqlite_path: PathBuf,
database: fabro_db::Database,
store: Arc<fabro_store::Database>,
}
impl TestContext {
async fn new(prefix: &str) -> TestResult<Self> {
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 store = Arc::new(fabro_store::Database::new(
Arc::new(InMemory::new()),
prefix,
StdDuration::from_millis(1),
None,
Arc::new(fabro_store::BlobStore::new(database.clone_pool())),
Arc::new(fabro_store::RunSummaryStore::new(database.clone_pool())),
));
Ok(Self {
_directory: directory,
sqlite_path,
database,
store,
})
}
async fn put_event(
&self,
run_id: &RunId,
seq: u32,
event: &str,
properties: serde_json::Value,
) -> TestResult<()> {
let payload = serde_json::json!({
"id": format!("evt-{seq}-{event}"),
"ts": Utc
.timestamp_millis_opt(1_788_000_000_000 + i64::from(seq))
.single()
.unwrap()
.to_rfc3339(),
"run_id": run_id.to_string(),
"event": event,
"properties": properties,
});
fabro_store::test_support::put_legacy_run_event(&self.store, run_id, seq, &payload)
.await?;
Ok(())
}
async fn put_created(&self, run_id: &RunId) -> TestResult<()> {
self.put_event(
run_id,
1,
"run.created",
serde_json::json!({
"title": "Activation test",
"settings": WorkflowSettings::default(),
"graph": Graph::new("test"),
"workflow_slug": "test-workflow",
"labels": {},
"provenance": test_support::test_run_provenance(),
}),
)
.await
}
async fn source_identity(&self) -> TestResult<fabro_store::LegacyRunHistorySourceIdentity> {
Ok(self.store.legacy_run_history_source_identity().await?)
}
fn backup_path(&self) -> PathBuf {
fabro_db::append_to_path(&self.sqlite_path, BACKUP_SUFFIX)
}
}
fn run_id() -> RunId {
RunId::from(Ulid::from_parts(1_788_000_000_000, 1))
}
#[tokio::test]
async fn cold_activation_imports_and_warm_restart_preserves_marker() -> TestResult<()> {
let context = TestContext::new("cold-and-warm-run-activation").await?;
let run_id = run_id();
context.put_created(&run_id).await?;
context
.put_event(&run_id, 2, "run.submitted", serde_json::json!({}))
.await?;
let identity = context.source_identity().await?;
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await?;
let first_marker = read_activation_record(context.database.pool())
.await?
.unwrap();
assert_eq!(first_marker.source_runs, 1);
assert_eq!(first_marker.source_events, 2);
assert!(context.backup_path().is_file());
let backup_options = sqlx::sqlite::SqliteConnectOptions::new()
.filename(context.backup_path())
.read_only(true)
.create_if_missing(false);
let mut backup = sqlx::SqliteConnection::connect_with(&backup_options).await?;
assert_eq!(
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events")
.fetch_one(&mut backup)
.await?,
0,
"the retained backup must capture the exact pre-import boundary"
);
assert_eq!(
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events")
.fetch_one(context.database.pool())
.await?,
2
);
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await?;
assert_eq!(
read_activation_record(context.database.pool()).await?,
Some(first_marker)
);
Ok(())
}
#[tokio::test]
async fn busy_final_checkpoint_warns_and_does_not_fail_activation() -> TestResult<()> {
use sqlx::sqlite::{SqliteJournalMode, SqlitePoolOptions};
let context = TestContext::new("busy-final-run-checkpoint").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(context.database.pool())
.await?;
let reader_options = sqlx::sqlite::SqliteConnectOptions::new()
.filename(&context.sqlite_path)
.read_only(true)
.create_if_missing(false);
let mut reader = sqlx::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?;
let checkpoint_options = sqlx::sqlite::SqliteConnectOptions::new()
.filename(&context.sqlite_path)
.journal_mode(SqliteJournalMode::Wal)
.busy_timeout(StdDuration::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?;
let wal_bytes = fs::metadata(fabro_db::append_to_path(&context.sqlite_path, "-wal"))
.await?
.len();
assert!(wal_bytes > 0, "the WAL should remain untruncated");
drop(reader);
Ok(())
}
#[tokio::test]
async fn changed_legacy_source_fails_before_mutating_sqlite() -> TestResult<()> {
let context = TestContext::new("changed-run-activation-source").await?;
let run_id = run_id();
context.put_created(&run_id).await?;
let original_identity = context.source_identity().await?;
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&original_identity,
)
.await?;
context
.put_event(&run_id, 2, "run.submitted", serde_json::json!({}))
.await?;
let changed_identity = context.source_identity().await?;
let error = activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&changed_identity,
)
.await
.expect_err("the source identity must remain stable after activation");
assert!(matches!(error, RunHistoryActivationError::MarkerMismatch));
assert_eq!(
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events")
.fetch_one(context.database.pool())
.await?,
1
);
Ok(())
}
#[tokio::test]
async fn empty_source_with_unmarked_target_fails_closed() -> TestResult<()> {
let context = TestContext::new("empty-source-with-target").await?;
let run_id = run_id();
context.put_created(&run_id).await?;
let identity = context.source_identity().await?;
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await?;
sqlx::query("DELETE FROM legacy_run_history_activation")
.execute(context.database.pool())
.await?;
let empty_store = Arc::new(fabro_store::Database::new(
Arc::new(InMemory::new()),
"empty-source",
StdDuration::from_millis(1),
None,
Arc::new(fabro_store::BlobStore::new(context.database.clone_pool())),
Arc::new(fabro_store::RunSummaryStore::new(
context.database.clone_pool(),
)),
));
let empty_identity = empty_store.legacy_run_history_source_identity().await?;
let error = activate_run_history(
&context.database,
&context.sqlite_path,
&empty_store,
&empty_identity,
)
.await
.expect_err("unmarked SQLite rows cannot be adopted from an empty source");
assert!(matches!(
error,
RunHistoryActivationError::EmptySourceWithTarget { .. }
));
Ok(())
}
#[tokio::test]
async fn missing_backup_after_import_progress_fails_closed() -> TestResult<()> {
let context = TestContext::new("missing-run-activation-backup").await?;
let run_id = run_id();
context.put_created(&run_id).await?;
context
.store
.import_legacy_run_history_into(context.database.pool())
.await?;
let identity = context.source_identity().await?;
let error = activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await
.expect_err("partial import progress requires the retained backup");
assert!(matches!(
error,
RunHistoryActivationError::MissingBackupAfterProgress { .. }
));
Ok(())
}
#[tokio::test]
async fn empty_source_and_target_need_no_backup_on_cold_or_warm_start() -> TestResult<()> {
let context = TestContext::new("empty-run-activation").await?;
let identity = context.source_identity().await?;
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await?;
let marker = read_activation_record(context.database.pool())
.await?
.unwrap();
assert_eq!((marker.source_runs, marker.source_events), (0, 0));
assert!(!context.backup_path().exists());
activate_run_history(
&context.database,
&context.sqlite_path,
&context.store,
&identity,
)
.await?;
assert!(!context.backup_path().exists());
Ok(())
}
}

186
lib/apps/fabro-server/migrations/sqlite_activation_backup.rs Normal file
View file

@ -0,0 +1,186 @@
//! Pre-activation backup and integrity helpers shared by the SQLite
//! activation bridges.
//!
//! Every activation snapshots the live database to a private, integrity
//! checked backup file before importing legacy data, and re-validates any
//! backup it finds on a later start. This module owns that mechanism so the
//! blob and run-history bridges cannot drift apart.
use std::path::{Path, PathBuf};
use futures_util::TryStreamExt as _;
use sqlx::Connection as _;
use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection};
use tokio::fs;
use tokio::task::{JoinError, spawn_blocking};
use tracing::debug;
const STAGING_SUFFIX: &str = ".tmp";
#[derive(Debug, thiserror::Error)]
pub(crate) enum BackupError {
#[error("reading activation backup metadata at {path}")]
Metadata {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("activation backup is not a regular file at {path}")]
NotRegular { path: PathBuf },
#[error("activation backup permissions are not private at {path}")]
NotPrivate { path: PathBuf },
#[error("opening or checking activation backup integrity at {path}")]
Integrity {
path: PathBuf,
#[source]
source: sqlx::Error,
},
#[error("activation backup integrity check did not return exactly one ok result at {path}")]
IntegrityFailed { path: PathBuf },
#[error("staging the pre-activation SQLite backup")]
Stage(#[source] fabro_db::SnapshotStagingError),
#[error("joining the activation backup publication task")]
JoinPublication(#[source] JoinError),
#[error("publishing the activation backup at {path} without overwriting")]
Publish {
path: PathBuf,
#[source]
source: std::io::Error,
},
}
pub(crate) async fn backup_exists(path: &Path) -> Result<bool, BackupError> {
match fs::metadata(path).await {
Ok(_) => Ok(true),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(BackupError::Metadata {
path: path.to_path_buf(),
source,
}),
}
}
/// Snapshots `pool` to `backup_path` without overwriting an existing file.
///
/// The snapshot is staged beside the target, validated, and then published
/// with an atomic no-clobber rename. A backup that another process published
/// concurrently is validated in place instead.
pub(crate) async fn create_backup(
pool: &sqlx::SqlitePool,
backup_path: &Path,
) -> Result<(), BackupError> {
let staging_path = fabro_db::append_to_path(backup_path, STAGING_SUFFIX);
fabro_db::write_snapshot_to_staging(pool, &staging_path)
.await
.map_err(BackupError::Stage)?;
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(BackupError::JoinPublication)?
.map_err(|source| BackupError::Publish {
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 activation backup"
);
validate_backup(backup_path).await?;
}
Ok(())
}
/// Requires `path` to be a private regular file holding a SQLite database
/// whose `PRAGMA integrity_check` passes.
pub(crate) async fn validate_backup(path: &Path) -> Result<(), BackupError> {
let metadata = fs::symlink_metadata(path)
.await
.map_err(|source| BackupError::Metadata {
path: path.to_path_buf(),
source,
})?;
if !metadata.is_file() {
return Err(BackupError::NotRegular {
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| BackupError::Integrity {
path: path.to_path_buf(),
source,
})?;
let ok = integrity_check_is_ok(&mut connection)
.await
.map_err(|source| BackupError::Integrity {
path: path.to_path_buf(),
source,
})?;
if !ok {
return Err(BackupError::IntegrityFailed {
path: path.to_path_buf(),
});
}
Ok(())
}
/// Returns whether `PRAGMA integrity_check` reports exactly one `ok` row.
pub(crate) 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<(), BackupError> {
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(BackupError::NotPrivate {
path: path.to_path_buf(),
});
}
Ok(())
}
#[cfg(not(unix))]
fn validate_private_permissions(
_path: &Path,
_metadata: &std::fs::Metadata,
) -> Result<(), BackupError> {
Ok(())
}

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

@ -7,12 +7,17 @@ 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/sqlite_activation_backup.rs"]
mod sqlite_activation_backup;
#[path = "../migrations/2026082301_sqlite_blob_activation.rs"]
mod sqlite_blob_activation;
#[path = "../migrations/2026082801_sqlite_run_history_activation.rs"]
mod sqlite_run_history_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) use sqlite_run_history_activation::activate_run_history;
pub(crate) type LegacyVaultMigrationReport = legacy_vault_entries::LegacyVaultMigrationReport;
pub(crate) type OptionalServerEnvSecretsMigrationReport =

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

@ -773,7 +773,7 @@ where
} else {
None
};
let store = migrations::activate_blob_storage(
let blob_activation = migrations::activate_blob_storage(
&database,
&sqlite_path,
object_store,
@ -783,6 +783,15 @@ where
)
.await
.context("activating SQLite blob storage")?;
migrations::activate_run_history(
&database,
&sqlite_path,
&blob_activation.store,
&blob_activation.run_history_identity,
)
.await
.context("activating SQLite run history")?;
let store = blob_activation.store;
// 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
// leaving them in the object store forever. Pending authorization codes
@ -829,7 +838,7 @@ where
.runs
.warm_projection_cache()
.await
.context("warming run projection cache and reconciling run summaries")?;
.context("warming run projection cache from SQLite")?;
let reconciled = reconcile_incomplete_runs_on_startup(&state).await?;
if reconciled > 0 {
info!(

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

@ -89,7 +89,7 @@ async fn get_system_info(_auth: RequiredUser, State(state): State<Arc<AppState>>
profile: option_env!("FABRO_BUILD_PROFILE").map(str::to_string),
os: Some(std::env::consts::OS.to_string()),
arch: Some(std::env::consts::ARCH.to_string()),
storage_engine: Some("slatedb".to_string()),
storage_engine: Some("sqlite".to_string()),
storage_dir: Some(state.server_storage_dir().display().to_string()),
uptime_secs: Some(to_i64(state.started_at.elapsed().as_secs())),
runs: Some(SystemRunCounts {

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

@ -3265,17 +3265,16 @@ url = "http://127.0.0.1:32276"
}
#[tokio::test]
async fn system_repair_runs_lists_catalog_entries_without_projection() {
async fn system_repair_runs_lists_sql_rows_without_readable_history() {
let state = test_app_state();
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = RunId::new();
let run_store = state.stores.runs.create_run(&run_id).await.unwrap();
append_default_run_created(&run_store, run_id).await;
state
.stores
.runs
.catalog_index()
.await
.unwrap()
.add(&run_id)
.run_summaries
.test_delete_run_events(&run_id)
.await
.unwrap();
@ -3303,7 +3302,7 @@ async fn system_repair_runs_lists_catalog_entries_without_projection() {
body["runs"][0]["error"]
.as_str()
.unwrap()
.contains("no events"),
.contains("head mismatch"),
"got: {}",
body["runs"][0]["error"]
);
@ -7680,10 +7679,8 @@ async fn create_unreadable_durable_run(state: &Arc<AppState>, run_id: RunId) {
)
.await
.unwrap();
let err = run_store
.state()
.await
.expect_err("poison event should make the run projection unreadable");
let unreadable = state.stores.runs.open_run_reader(&run_id).await;
let err = unreadable.expect_err("poison event should make the run projection unreadable");
assert!(
err.to_string().contains("invalid completed stage status"),
"unexpected projection error: {err}"
@ -16551,7 +16548,6 @@ async fn cancel_run_requests_worker_runtime_stop_when_control_unavailable() {
.parse::<RunId>()
.unwrap();
let worker_ref = test_worker_ref(u32::MAX);
tokio::time::pause();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16571,6 +16567,7 @@ async fn cancel_run_requests_worker_runtime_stop_when_control_unavailable() {
assert_eq!(runtime.requested_refs(), vec![worker_ref.clone()]);
tokio::time::pause();
advance_past_worker_cancel_grace().await;
runtime.wait_for_forced_ref(&worker_ref).await;
@ -16592,7 +16589,6 @@ async fn cancel_run_force_stops_worker_when_delivered_control_does_not_converge(
.unwrap();
let worker_ref = test_worker_ref(u32::MAX);
let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
tokio::time::pause();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16613,6 +16609,7 @@ async fn cancel_run_force_stops_worker_when_delivered_control_does_not_converge(
assert!(runtime.requested_refs().is_empty());
assert!(runtime.forced_refs().is_empty());
tokio::time::pause();
advance_past_worker_cancel_grace().await;
runtime.wait_for_forced_ref(&worker_ref).await;
@ -16635,7 +16632,6 @@ async fn cancel_run_watchdog_does_not_stop_replacement_worker() {
let cancelled_worker_ref = test_worker_ref(u32::MAX - 1);
let replacement_worker_ref = test_worker_ref(u32::MAX);
let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
tokio::time::pause();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16653,6 +16649,7 @@ async fn cancel_run_watchdog_does_not_stop_replacement_worker() {
let response = app.oneshot(req).await.unwrap();
assert_status!(response, StatusCode::ACCEPTED).await;
tokio::time::pause();
tokio::task::yield_now().await;
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16679,7 +16676,6 @@ async fn cancel_run_watchdog_does_not_stop_worker_after_live_ref_clears() {
.unwrap();
let worker_ref = test_worker_ref(u32::MAX);
let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
tokio::time::pause();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16697,6 +16693,7 @@ async fn cancel_run_watchdog_does_not_stop_worker_after_live_ref_clears() {
let response = app.oneshot(req).await.unwrap();
assert_status!(response, StatusCode::ACCEPTED).await;
tokio::time::pause();
tokio::task::yield_now().await;
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16723,7 +16720,6 @@ async fn repeated_cancel_request_arms_one_watchdog_and_persists_one_intent() {
.unwrap();
let worker_ref = test_worker_ref(u32::MAX);
let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
tokio::time::pause();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -16760,6 +16756,7 @@ async fn repeated_cancel_request_arms_one_watchdog_and_persists_one_intent() {
.count();
assert_eq!(request_count, 1);
tokio::time::pause();
advance_past_worker_cancel_grace().await;
runtime.wait_for_forced_ref(&worker_ref).await;

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

@ -12,13 +12,19 @@ use tower::ServiceExt;
use crate::helpers::{MINIMAL_DOT, api, minimal_manifest_json, response_json, test_settings};
fn app_with_store(object_store: Arc<dyn ObjectStore>) -> axum::Router {
fn app_with_store(
object_store: Arc<dyn ObjectStore>,
blobs: Arc<fabro_store::BlobStore>,
run_summaries: Arc<fabro_store::RunSummaryStore>,
) -> axum::Router {
let settings = test_settings();
let store = Arc::new(fabro_store::test_support::test_database(
let store = Arc::new(fabro_store::test_support::test_database_with_stores(
Arc::clone(&object_store),
"event-race",
Duration::from_millis(1),
None,
blobs,
run_summaries,
));
let artifact_store = fabro_store::ArtifactStore::new(object_store, "artifacts");
let state = fabro_server::test_support::TestAppStateBuilder::new()
@ -103,7 +109,13 @@ async fn append_status_and_body(
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_event_appends_after_restart_keep_projection_cache_contiguous() {
let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
let first_app = app_with_store(Arc::clone(&object_store));
let blobs = fabro_store::test_support::test_blob_store();
let run_summaries = fabro_store::test_support::test_run_summary_store();
let first_app = app_with_store(
Arc::clone(&object_store),
Arc::clone(&blobs),
Arc::clone(&run_summaries),
);
let run_id = create_run(&first_app).await;
tokio::time::sleep(Duration::from_millis(25)).await;
@ -111,7 +123,7 @@ async fn concurrent_event_appends_after_restart_keep_projection_cache_contiguous
// Simulate a server restart: a fresh AppState opens the existing run with
// an empty active-run cache, so concurrent appends all race through the
// public event endpoint instead of sharing an already-open RunDatabase.
let restarted_app = app_with_store(object_store);
let restarted_app = app_with_store(object_store, blobs, run_summaries);
let appends = 64;
let barrier = Arc::new(Barrier::new(appends));
let mut tasks = Vec::with_capacity(appends);

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

@ -129,7 +129,7 @@ async fn get_system_info_returns_runtime_fields() {
let body = response_json(response, StatusCode::OK, "GET /api/v1/system/info").await;
assert!(body["version"].as_str().is_some());
assert_eq!(body["server_url"], configured_server_url);
assert_eq!(body["storage_engine"], "slatedb");
assert_eq!(body["storage_engine"], "sqlite");
assert_eq!(
body["storage_dir"],
expected_storage_dir.display().to_string()

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

@ -7,6 +7,7 @@ use sqlx::SqlitePool;
#[cfg(test)]
use crate::record::Repository;
#[cfg(test)]
use crate::record::{RawBytesCodec, Record};
use crate::{Error, Result};
@ -25,6 +26,7 @@ impl From<Bytes> for Blob {
}
}
#[cfg(test)]
impl Record for Blob {
type Id = BlobHash;
type Codec = RawBytesCodec;

2
lib/components/fabro-store/src/error.rs
View file

@ -39,6 +39,8 @@ pub enum Error {
},
#[error("Run not found: {0}")]
RunNotFound(String),
#[error("Run already exists: {0}")]
RunAlreadyExists(String),
#[error("Session not found: {0}")]
SessionNotFound(String),
#[error("Session already exists: {0}")]

34
lib/components/fabro-store/src/keys.rs
View file

@ -1,4 +1,5 @@
use std::fmt::{self, Write};
#[cfg(test)]
use std::ops::Range;
use fabro_types::{RunId, SessionId};
@ -28,6 +29,7 @@ impl SlateKey {
/// Exclusive end bound of this key's prefix keyspace: every key under
/// `self.into_prefix()` sorts below it and no other key sorts between.
#[cfg(test)]
fn into_prefix_end(mut self) -> Self {
self.0.push('\u{1}');
self
@ -51,10 +53,6 @@ impl AsRef<[u8]> for SlateKey {
// --- Construction ---
pub(crate) fn run_data_prefix(run_id: &RunId) -> SlateKey {
SlateKey::new("runs").with(run_id).into_prefix()
}
pub(crate) fn run_events_prefix(run_id: &RunId) -> SlateKey {
SlateKey::new("runs")
.with(run_id)
@ -62,6 +60,31 @@ pub(crate) fn run_events_prefix(run_id: &RunId) -> SlateKey {
.into_prefix()
}
/// Prefix of the retired `runs/_index/by-start/<run_id>` catalog markers that
/// the legacy layout kept beside each run's events.
pub(crate) fn run_catalog_prefix() -> SlateKey {
run_catalog_root().into_prefix()
}
#[cfg(test)]
pub(crate) fn run_catalog_key(run_id: &RunId) -> SlateKey {
run_catalog_root().with(run_id)
}
/// Extracts the run id from a full catalog marker key, or `None` when the key
/// is not exactly `runs/_index/by-start/<run_id>`.
pub(crate) fn parse_run_catalog_key(raw: &str) -> Option<RunId> {
let segments = SlateKey::segments(raw).collect::<Vec<_>>();
let ["runs", "_index", "by-start", run_id] = segments.as_slice() else {
return None;
};
run_id.parse().ok()
}
fn run_catalog_root() -> SlateKey {
SlateKey::new("runs").with("_index").with("by-start")
}
// Sequence keys zero-pad `seq` to six digits so lexicographic key order
// matches numeric seq order through `MAX_EVENT_SEQ`. Seek-based event listing
// (`run_events_range`) depends on this invariant, so event allocation rejects
@ -73,6 +96,7 @@ pub(crate) fn run_event_key(run_id: &RunId, seq: u32, epoch_ms: i64) -> SlateKey
.with(format!("{seq:06}-{epoch_ms}"))
}
#[cfg(test)]
pub(crate) fn run_event_seq_prefix(run_id: &RunId, seq: u32) -> SlateKey {
SlateKey::new("runs")
.with(run_id)
@ -83,6 +107,7 @@ pub(crate) fn run_event_seq_prefix(run_id: &RunId, seq: u32) -> SlateKey {
/// Scan range covering the run's event keys from `start_seq` to the end of
/// the run's event namespace, so seek-based listing never touches keys of
/// other runs or namespaces.
#[cfg(test)]
pub(crate) fn run_events_range(run_id: &RunId, start_seq: u32) -> Range<SlateKey> {
let end = SlateKey::new("runs")
.with(run_id)
@ -101,6 +126,7 @@ pub(crate) fn session_by_id_key(session_id: &SessionId) -> SlateKey {
// --- Parsing ---
#[cfg(test)]
pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
let mut segments = SlateKey::segments(key);
let _ = segments.next()?; // "runs"

305
lib/components/fabro-store/src/legacy_run_history_import.rs
View file

@ -9,6 +9,7 @@ use std::error::Error as StdError;
use std::fmt;
use fabro_types::{EventEnvelope, RunEvent, RunId, RunProjection};
use sha2::{Digest as _, Sha256};
use sqlx::SqlitePool;
#[cfg(test)]
use tokio::sync::Barrier;
@ -16,7 +17,7 @@ use tracing::debug;
use crate::keys::SlateKey;
use crate::slate::CachedRunProjection;
use crate::{Database, EventPayload, ListRunsQuery, RunProjectionReducer, RunSummaryStore, keys};
use crate::{Database, EventPayload, RunProjectionReducer, RunSummaryStore, keys};
/// Count-only observations about the legacy catalog and session indexes.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
@ -37,21 +38,86 @@ pub struct LegacyRunHistoryImportReport {
pub verified_existing_events: u64,
pub discarded_projection_only_rows: u64,
pub committed_run_transactions: u64,
pub tombstoned_source_runs: u64,
pub tombstoned_source_events: u64,
pub diagnostics: LegacyRunHistoryDiagnostics,
}
/// Aggregate proof from legacy-prefix and full-SQL-destination verification.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct LegacyRunHistoryVerificationReport {
pub source_runs: u64,
pub source_events: u64,
pub matched_prefix_runs: u64,
pub matched_prefix_events: u64,
pub target_runs: u64,
pub target_events: u64,
pub sql_only_runs: u64,
pub sql_only_events: u64,
pub diagnostics: LegacyRunHistoryDiagnostics,
pub source_runs: u64,
pub source_events: u64,
pub matched_prefix_runs: u64,
pub matched_prefix_events: u64,
pub target_runs: u64,
pub target_events: u64,
pub sql_only_runs: u64,
pub sql_only_events: u64,
pub tombstoned_source_runs: u64,
pub tombstoned_source_events: u64,
pub diagnostics: LegacyRunHistoryDiagnostics,
}
/// Stable, aggregate-only identity of the exact legacy run-event source.
#[derive(Clone, Copy, Eq, PartialEq)]
pub struct LegacyRunHistorySourceIdentity {
fingerprint: [u8; 32],
pub runs: u64,
pub events: u64,
}
impl LegacyRunHistorySourceIdentity {
#[must_use]
pub fn fingerprint(&self) -> &[u8; 32] {
&self.fingerprint
}
}
impl fmt::Debug for LegacyRunHistorySourceIdentity {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("LegacyRunHistorySourceIdentity")
.field("runs", &self.runs)
.field("events", &self.events)
.finish_non_exhaustive()
}
}
/// Failure while strictly identifying the legacy run-event source.
pub struct LegacyRunHistorySourceIdentityError {
failure: LegacyRunHistorySourceFailure,
}
impl From<LegacyRunHistorySourceFailure> for LegacyRunHistorySourceIdentityError {
fn from(failure: LegacyRunHistorySourceFailure) -> Self {
Self { failure }
}
}
impl fmt::Debug for LegacyRunHistorySourceIdentityError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("LegacyRunHistorySourceIdentityError")
.field("failure", &self.failure.kind())
.finish()
}
}
impl fmt::Display for LegacyRunHistorySourceIdentityError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
formatter,
"identifying the legacy run-history source: {}",
self.failure
)
}
}
impl StdError for LegacyRunHistorySourceIdentityError {
fn source(&self) -> Option<&(dyn StdError + 'static)> {
Some(&self.failure)
}
}
/// An import failure plus the durable progress completed before it.
@ -149,6 +215,12 @@ impl StdError for LegacyRunHistoryVerificationError {
enum LegacyRunHistoryImportFailure {
#[error("reading and validating the legacy run-history source")]
Source(#[source] LegacyRunHistorySourceFailure),
#[error("reading legacy run-history activation state")]
ActivationState(#[source] sqlx::Error),
#[error("reading legacy run-history deletion tombstones")]
DeletionState(#[source] sqlx::Error),
#[error("a legacy run-history deletion tombstone still has canonical SQLite data")]
TombstonedDestinationPresent,
#[error("starting the projection-only row cleanup transaction")]
BeginCleanup(#[source] sqlx::Error),
#[error("deleting projection-only run rows")]
@ -161,6 +233,8 @@ enum LegacyRunHistoryImportFailure {
ReadDestination(#[source] crate::Error),
#[error("the destination history is partial or conflicts with the legacy prefix")]
DestinationConflict,
#[error("SQLite is missing an activated legacy run-history prefix")]
MissingDestinationAfterActivation,
#[error("replaying an existing destination run history")]
ReplayDestination(#[source] crate::Error),
#[error("verifying an existing destination run row")]
@ -219,6 +293,10 @@ impl fmt::Debug for LegacyRunHistoryImportFailure {
enum LegacyRunHistoryVerificationFailure {
#[error("reading and validating the legacy run-history source")]
Source(#[source] LegacyRunHistorySourceFailure),
#[error("reading legacy run-history deletion tombstones")]
DeletionState(#[source] sqlx::Error),
#[error("a legacy run-history deletion tombstone still has canonical SQLite data")]
TombstonedDestinationPresent,
#[error("acquiring a SQLite verification connection")]
AcquireConnection(#[source] sqlx::Error),
#[error("reading a destination run history")]
@ -306,7 +384,11 @@ impl fmt::Debug for LegacyRunHistorySourceFailure {
#[strum(serialize_all = "snake_case")]
enum LegacyRunHistoryDiagnosticsFailure {
#[error("reading the legacy run catalog")]
ReadCatalog(#[source] crate::Error),
ReadCatalog(#[source] slatedb::Error),
#[error("a legacy run-catalog key is not UTF-8")]
CatalogKeyUtf8(#[source] std::str::Utf8Error),
#[error("a legacy run-catalog key is not canonical")]
InvalidCatalogKey,
#[error("opening the legacy run source for diagnostics")]
OpenSource(#[source] crate::Error),
#[error("opening a legacy run-event probe")]
@ -370,6 +452,7 @@ struct ValidatedLegacyRunEvent {
payload: EventPayload,
envelope: EventEnvelope,
event_json: String,
raw_key: Vec<u8>,
}
struct ValidatedLegacyRunHistory {
@ -500,6 +583,36 @@ impl LegacyRunHistorySource {
}
impl Database {
/// Strictly identifies the exact legacy run-event key/value stream.
pub async fn legacy_run_history_source_identity(
&self,
) -> Result<LegacyRunHistorySourceIdentity, LegacyRunHistorySourceIdentityError> {
const DOMAIN_SEPARATOR: &[u8] = b"fabro.legacy-run-history-source.v1\0";
let mut source = LegacyRunHistorySource::open(self).await?;
let mut hasher = Sha256::new();
hasher.update(DOMAIN_SEPARATOR);
let mut runs = 0_u64;
let mut events = 0_u64;
while let Some(history) = source.next_run(None).await? {
runs = runs
.checked_add(1)
.ok_or(LegacyRunHistorySourceFailure::CounterOverflow)?;
for event in &history.events {
hash_source_part(&mut hasher, &event.raw_key)?;
hash_source_part(&mut hasher, event.event_json.as_bytes())?;
events = events
.checked_add(1)
.ok_or(LegacyRunHistorySourceFailure::CounterOverflow)?;
}
}
Ok(LegacyRunHistorySourceIdentity {
fingerprint: hasher.finalize().into(),
runs,
events,
})
}
/// Strictly imports legacy SlateDB run history into the inactive SQLite
/// run store, committing one complete run at a time.
///
@ -535,7 +648,21 @@ impl Database {
controls: &ImportControls,
report: &mut LegacyRunHistoryImportReport,
) -> Result<(), LegacyRunHistoryImportFailure> {
discard_projection_only_rows(pool, report).await?;
let activated = legacy_run_history_is_activated(pool)
.await
.map_err(LegacyRunHistoryImportFailure::ActivationState)?;
let tombstones = legacy_run_history_tombstones(pool)
.await
.map_err(LegacyRunHistoryImportFailure::DeletionState)?;
if tombstoned_destination_present(pool)
.await
.map_err(LegacyRunHistoryImportFailure::DeletionState)?
{
return Err(LegacyRunHistoryImportFailure::TombstonedDestinationPresent);
}
if !activated {
discard_projection_only_rows(pool, report).await?;
}
let mut source = LegacyRunHistorySource::open(self)
.await
.map_err(LegacyRunHistoryImportFailure::Source)?;
@ -553,7 +680,15 @@ impl Database {
break;
};
import_checked_add(&mut report.scanned_source_runs, 1)?;
import_one_run(pool, controls, history, report).await?;
if tombstones.contains(&history.run_id.to_string()) {
import_checked_add(&mut report.tombstoned_source_runs, 1)?;
import_checked_add(
&mut report.tombstoned_source_events,
usize_to_import_count(history.events.len())?,
)?;
continue;
}
import_one_run(pool, controls, history, activated, report).await?;
}
report.diagnostics = self
@ -588,6 +723,15 @@ impl Database {
pool: &SqlitePool,
report: &mut LegacyRunHistoryVerificationReport,
) -> Result<(), LegacyRunHistoryVerificationFailure> {
let tombstones = legacy_run_history_tombstones(pool)
.await
.map_err(LegacyRunHistoryVerificationFailure::DeletionState)?;
if tombstoned_destination_present(pool)
.await
.map_err(LegacyRunHistoryVerificationFailure::DeletionState)?
{
return Err(LegacyRunHistoryVerificationFailure::TombstonedDestinationPresent);
}
let mut source_ids = HashSet::new();
let mut source = LegacyRunHistorySource::open(self)
.await
@ -606,6 +750,14 @@ impl Database {
};
verification_checked_add(&mut report.source_runs, 1)?;
source_ids.insert(history.run_id);
if tombstones.contains(&history.run_id.to_string()) {
verification_checked_add(&mut report.tombstoned_source_runs, 1)?;
verification_checked_add(
&mut report.tombstoned_source_events,
usize_to_verification_count(history.events.len())?,
)?;
continue;
}
verify_source_prefix(pool, &history).await?;
verification_checked_add(&mut report.matched_prefix_runs, 1)?;
verification_checked_add(
@ -658,22 +810,32 @@ impl Database {
async fn legacy_run_history_diagnostics(
&self,
) -> Result<LegacyRunHistoryDiagnostics, LegacyRunHistoryDiagnosticsFailure> {
let catalog_ids = self
.catalog_index()
let source = self
.open_db()
.await
.map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?
.list(&ListRunsQuery::default())
.map_err(LegacyRunHistoryDiagnosticsFailure::OpenSource)?;
let mut catalog_ids = Vec::new();
let mut catalog = source
.scan_prefix(keys::run_catalog_prefix())
.await
.map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?;
while let Some(entry) = catalog
.next()
.await
.map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?
{
let key = std::str::from_utf8(&entry.key)
.map_err(LegacyRunHistoryDiagnosticsFailure::CatalogKeyUtf8)?;
catalog_ids.push(
keys::parse_run_catalog_key(key)
.ok_or(LegacyRunHistoryDiagnosticsFailure::InvalidCatalogKey)?,
);
}
let mut diagnostics = LegacyRunHistoryDiagnostics {
catalog_markers: u64::try_from(catalog_ids.len())
.map_err(|_| LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?,
..LegacyRunHistoryDiagnostics::default()
};
let source = self
.open_db()
.await
.map_err(LegacyRunHistoryDiagnosticsFailure::OpenSource)?;
for run_id in catalog_ids {
let mut events = source
.scan_prefix(keys::run_events_prefix(&run_id))
@ -770,9 +932,21 @@ fn parse_source_event(
event,
},
event_json,
raw_key: key.to_vec(),
}))
}
fn hash_source_part(
hasher: &mut Sha256,
bytes: &[u8],
) -> Result<(), LegacyRunHistorySourceFailure> {
let length =
u64::try_from(bytes.len()).map_err(|_| LegacyRunHistorySourceFailure::CounterOverflow)?;
hasher.update(length.to_be_bytes());
hasher.update(bytes);
Ok(())
}
fn event_run_segment(key: &[u8]) -> Option<&[u8]> {
let mut segments = key.split(|byte| *byte == 0);
(segments.next()? == b"runs").then_some(())?;
@ -780,6 +954,39 @@ fn event_run_segment(key: &[u8]) -> Option<&[u8]> {
(segments.next()? == b"events").then_some(run_id)
}
async fn legacy_run_history_is_activated(pool: &SqlitePool) -> Result<bool, sqlx::Error> {
sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM legacy_run_history_activation WHERE singleton = 1)",
)
.fetch_one(pool)
.await
}
async fn legacy_run_history_tombstones(pool: &SqlitePool) -> Result<HashSet<String>, sqlx::Error> {
Ok(
sqlx::query_scalar("SELECT run_id FROM legacy_run_history_deletions")
.fetch_all(pool)
.await?
.into_iter()
.collect(),
)
}
/// Returns whether any tombstoned run still has canonical SQLite rows.
async fn tombstoned_destination_present(pool: &SqlitePool) -> Result<bool, sqlx::Error> {
sqlx::query_scalar(
r"
SELECT EXISTS(
SELECT 1 FROM legacy_run_history_deletions AS deletion
WHERE EXISTS(SELECT 1 FROM runs WHERE id = deletion.run_id)
OR EXISTS(SELECT 1 FROM run_events WHERE run_id = deletion.run_id)
)
",
)
.fetch_one(pool)
.await
}
async fn discard_projection_only_rows(
pool: &SqlitePool,
report: &mut LegacyRunHistoryImportReport,
@ -812,6 +1019,7 @@ async fn import_one_run(
pool: &SqlitePool,
controls: &ImportControls,
history: ValidatedLegacyRunHistory,
activated: bool,
report: &mut LegacyRunHistoryImportReport,
) -> Result<(), LegacyRunHistoryImportFailure> {
#[cfg(not(test))]
@ -831,7 +1039,7 @@ async fn import_one_run(
})?;
let mut updated = *report;
if has_destination {
let destination = RunSummaryStore::list_events_with_json_on_connection(
let destination = RunSummaryStore::list_events_with_json_in_transaction(
&mut transaction,
&history.run_id,
)
@ -849,6 +1057,8 @@ async fn import_one_run(
&mut updated.verified_existing_events,
usize_to_import_count(destination.len())?,
)?;
} else if activated {
return Err(LegacyRunHistoryImportFailure::MissingDestinationAfterActivation);
} else {
RunSummaryStore::insert_imported_run_on_connection(&mut transaction, &history.current)
.await
@ -1012,6 +1222,8 @@ fn debug_import_outcome(outcome: &'static str, report: &LegacyRunHistoryImportRe
verified_existing_events = report.verified_existing_events,
discarded_projection_only_rows = report.discarded_projection_only_rows,
committed_run_transactions = report.committed_run_transactions,
tombstoned_source_runs = report.tombstoned_source_runs,
tombstoned_source_events = report.tombstoned_source_events,
catalog_markers = report.diagnostics.catalog_markers,
empty_catalog_markers = report.diagnostics.empty_catalog_markers,
session_reverse_rows = report.diagnostics.session_reverse_rows,
@ -1030,6 +1242,8 @@ fn debug_verification_outcome(outcome: &'static str, report: &LegacyRunHistoryVe
target_events = report.target_events,
sql_only_runs = report.sql_only_runs,
sql_only_events = report.sql_only_events,
tombstoned_source_runs = report.tombstoned_source_runs,
tombstoned_source_events = report.tombstoned_source_events,
catalog_markers = report.diagnostics.catalog_markers,
empty_catalog_markers = report.diagnostics.empty_catalog_markers,
session_reverse_rows = report.diagnostics.session_reverse_rows,
@ -1182,6 +1396,46 @@ mod tests {
event_value(run_id, seq, "run.submitted", &serde_json::json!({}))
}
#[tokio::test]
async fn legacy_source_identity_covers_exact_keys_and_json_bytes() -> TestResult<()> {
let context = TestContext::new().await?;
let run_id = run_id(0);
let value = created_value(&run_id, "identity");
let compact = serde_json::to_string(&value)?;
context.put_event(&run_id, 1, 1, &compact).await?;
let compact_identity = context.source.legacy_run_history_source_identity().await?;
let pretty = serde_json::to_string_pretty(&value)?;
context.put_event(&run_id, 1, 1, &pretty).await?;
let pretty_identity = context.source.legacy_run_history_source_identity().await?;
assert_eq!(
(compact_identity.runs, compact_identity.events),
(pretty_identity.runs, pretty_identity.events)
);
assert_ne!(
compact_identity.fingerprint(),
pretty_identity.fingerprint(),
"semantically equivalent JSON bytes must still change the source identity"
);
context
.source_db
.delete(keys::run_event_key(&run_id, 1, 1))
.await?;
context.put_event(&run_id, 1, 2, &pretty).await?;
let moved_key_identity = context.source.legacy_run_history_source_identity().await?;
assert_eq!(
(pretty_identity.runs, pretty_identity.events),
(moved_key_identity.runs, moved_key_identity.events)
);
assert_ne!(
pretty_identity.fingerprint(),
moved_key_identity.fingerprint(),
"changing only the raw legacy key must change the source identity"
);
Ok(())
}
fn decode_event(
run_id: &RunId,
seq: u32,
@ -1293,12 +1547,9 @@ mod tests {
context.put_event(&first, 1, 10, &first_created).await?;
context.put_event(&first, 4, 40, &first_submitted).await?;
context.put_event(&second, 1, 20, &second_created).await?;
context.source.catalog_index().await?.add(&first).await?;
context.put_raw(keys::run_catalog_key(&first), b"").await?;
context
.source
.catalog_index()
.await?
.add(&empty_marker)
.put_raw(keys::run_catalog_key(&empty_marker), b"")
.await?;
context
.source

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

@ -9,6 +9,7 @@ mod keyed_mutex;
mod keys;
mod legacy_blob_import;
mod legacy_run_history_import;
#[cfg(test)]
mod record;
mod run_sessions;
mod run_state;
@ -40,6 +41,7 @@ pub use legacy_blob_import::{
};
pub use legacy_run_history_import::{
LegacyRunHistoryDiagnostics, LegacyRunHistoryImportError, LegacyRunHistoryImportReport,
LegacyRunHistorySourceIdentity, LegacyRunHistorySourceIdentityError,
LegacyRunHistoryVerificationError, LegacyRunHistoryVerificationReport,
};
pub use run_sessions::{
@ -52,7 +54,7 @@ pub use run_summary_store::{
RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility,
};
pub use serializable_projection::SerializableProjection;
pub use slate::{CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, UnreadableRun};
pub use slate::{CachedRunProjection, Database, RunDatabase, Runs, UnreadableRun};
pub use types::EventPayload;
#[derive(Debug, Default, Clone, PartialEq, Eq)]

6
lib/components/fabro-store/src/record/mod.rs
View file

@ -6,9 +6,9 @@
//! - [`RecordId`]: converts the typed id to and from key segments.
//! - [`Repository`]: performs the generic get/put/delete/scan operations.
//!
//! Production callers should add a named domain store on top of this layer
//! rather than exposing `Repository<R>` directly. See `slate/blob_store.rs`
//! and `slate/run_catalog_index.rs` for the intended pattern.
//! Production stores no longer read or write SlateDB records; this module is
//! compiled only for tests that model the retired Slate layout and goes away
//! with the remaining compatibility bridges.
mod codec;
mod record_id;

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

@ -76,9 +76,8 @@ use crate::{Error, Result, keys};
/// Generic typed key/value operations shared by the simple record-backed
/// stores.
///
/// This type is intentionally `pub(crate)`: callers should interact through a
/// named store such as `RunCatalogIndex` or `BlobStore`, which can add
/// domain-specific behavior on top of the generic storage primitives here.
/// Test-only: production stores are SQLite-backed, so this exists solely to
/// exercise the retired Slate layout from tests.
pub(crate) struct Repository<R: Record> {
db: Arc<Db>,
_record: PhantomData<R>,

326
lib/components/fabro-store/src/run_summary_store.rs
View file

@ -1,4 +1,3 @@
use std::collections::{HashMap, HashSet};
use std::fmt::Write as _;
use std::sync::LazyLock;
@ -7,9 +6,10 @@ use fabro_types::{
BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming,
SessionId, StageId, timing,
};
use sqlx::pool::PoolConnection;
use sqlx::query::Query;
use sqlx::sqlite::{SqliteArguments, SqliteConnection, SqliteRow};
use sqlx::{QueryBuilder, Row as _, Sqlite, SqlitePool};
use sqlx::{Connection as _, QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction};
use strum::VariantArray as _;
use crate::run_state::projected_billing;
@ -28,6 +28,7 @@ INSERT INTO runs (
)
";
#[cfg(test)]
const UPSERT_RUN_SQL: &str = r"
INSERT INTO runs (
id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms,
@ -199,19 +200,104 @@ impl RunSummaryStore {
Self { pool }
}
pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> {
let record = PreparedRunSummary::from_entry(entry);
let mut connection = self.pool.acquire().await?;
upsert_run_on_connection(&mut connection, &record).await?;
Ok(())
}
#[cfg(test)]
pub(crate) async fn close_pool(&self) {
self.pool.close().await;
}
/// Corrupts one fixture history while preserving its current row so
/// cross-crate repair-endpoint tests can exercise unreadable SQL runs.
#[cfg(any(test, feature = "test-support"))]
pub async fn test_delete_run_events(&self, run_id: &RunId) -> Result<()> {
sqlx::query("DELETE FROM run_events WHERE run_id = ?")
.bind(run_id.to_string())
.execute(&self.pool)
.await?;
Ok(())
}
/// Inserts a fixture event without reducing it into the current row.
///
/// This deliberately creates an unreadable history for cross-crate repair
/// endpoint tests. It is never linked into production builds.
#[cfg(any(test, feature = "test-support"))]
pub async fn test_insert_unvalidated_event(
&self,
run_id: &RunId,
seq: u32,
payload: &serde_json::Value,
) -> Result<()> {
let payload = EventPayload::new(payload.clone(), run_id)?;
let event = RunEvent::try_from(&payload)?;
let envelope = EventEnvelope { seq, event };
let event_json = serde_json::to_string(&payload)?;
let mut transaction = self.pool.begin().await?;
insert_event_json_on_connection(&mut transaction, run_id, &envelope, &event_json).await?;
let updated = sqlx::query("UPDATE runs SET source_last_seq = ? WHERE id = ?")
.bind(i64::from(seq))
.bind(run_id.to_string())
.execute(&mut *transaction)
.await?;
if updated.rows_affected() != 1 {
return Err(Error::RunNotFound(run_id.to_string()));
}
transaction.commit().await?;
Ok(())
}
#[cfg(test)]
pub(crate) async fn test_mark_run_history_activated(&self) -> Result<()> {
sqlx::query(
r"
INSERT INTO legacy_run_history_activation (
singleton, source_fingerprint, source_runs, source_events, activated_at_ms
) VALUES (1, zeroblob(32), 0, 0, 1)
ON CONFLICT(singleton) DO NOTHING
",
)
.execute(&self.pool)
.await?;
Ok(())
}
#[cfg(test)]
pub(crate) async fn test_is_run_history_tombstoned(&self, run_id: &RunId) -> Result<bool> {
Ok(sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM legacy_run_history_deletions WHERE run_id = ?)",
)
.bind(run_id.to_string())
.fetch_one(&self.pool)
.await?)
}
pub(crate) async fn acquire(&self) -> Result<PoolConnection<Sqlite>> {
Ok(self.pool.acquire().await?)
}
pub(crate) async fn begin(&self) -> Result<Transaction<'static, Sqlite>> {
Ok(self.pool.begin().await?)
}
pub(crate) async fn contains(&self, run_id: &RunId) -> Result<bool> {
Ok(
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM runs WHERE id = ?)")
.bind(run_id.to_string())
.fetch_one(&self.pool)
.await?,
)
}
#[cfg(test)]
pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> {
let record = PreparedRunSummary::from_entry(entry);
let mut connection = self.pool.acquire().await?;
upsert_run_on_connection(&mut connection, &record).await
}
#[cfg(test)]
pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> {
use std::collections::{HashMap, HashSet};
let mut transaction = self.pool.begin().await?;
let stored_seqs: HashMap<String, i64> =
sqlx::query_as::<_, (String, i64)>("SELECT id, source_last_seq FROM runs")
@ -251,6 +337,137 @@ impl RunSummaryStore {
Ok(())
}
pub(crate) async fn list_run_ids(&self) -> Result<Vec<RunId>> {
sqlx::query_scalar::<_, String>("SELECT id FROM runs ORDER BY id ASC")
.fetch_all(&self.pool)
.await?
.into_iter()
.map(|stored_id| {
stored_id
.parse::<RunId>()
.map_err(|_| Error::RunSummaryMismatch {
run_id: stored_id,
field: "id",
})
})
.collect()
}
pub(crate) async fn head(&self, run_id: &RunId) -> Result<Option<u32>> {
let mut connection = self.acquire().await?;
select_run_head(&mut connection, run_id).await
}
pub(crate) async fn list_events_for_run(&self, run_id: &RunId) -> Result<Vec<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::list_events_on_connection(&mut connection, run_id).await
}
pub(crate) async fn list_events_from_with_limit(
&self,
run_id: &RunId,
start_seq: u32,
limit: usize,
) -> Result<Vec<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::list_events_from_with_limit_on_connection(&mut connection, run_id, start_seq, limit)
.await
}
pub(crate) async fn list_events_before_with_limit(
&self,
run_id: &RunId,
before_seq: Option<u32>,
limit: usize,
) -> Result<Vec<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::list_events_before_with_limit_on_connection(
&mut connection,
run_id,
before_seq,
limit,
)
.await
}
pub(crate) async fn get_event_for_run(
&self,
run_id: &RunId,
seq: u32,
) -> Result<Option<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::get_event_on_connection(&mut connection, run_id, seq).await
}
pub(crate) async fn list_events_for_stage_from_with_limit(
&self,
run_id: &RunId,
stage_id: &StageId,
start_seq: u32,
limit: usize,
) -> Result<Vec<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::list_events_for_stage_from_with_limit_on_connection(
&mut connection,
run_id,
stage_id,
start_seq,
limit,
)
.await
}
pub(crate) async fn list_events_for_session_from_with_limit(
&self,
run_id: &RunId,
session_id: &SessionId,
start_seq: u32,
limit: usize,
) -> Result<Vec<EventEnvelope>> {
let mut connection = self.acquire().await?;
Self::list_events_for_session_from_with_limit_on_connection(
&mut connection,
run_id,
session_id,
start_seq,
limit,
)
.await
}
pub(crate) async fn delete_canonical(&self, run_id: &RunId, deleted_at_ms: i64) -> Result<()> {
// This transaction reads the activation marker before it writes the
// tombstone and run deletion. A deferred SQLite transaction can fail
// immediately when that read transaction is upgraded while another
// writer is active, bypassing the configured busy timeout. Reserve
// the write lock up front so concurrent deletes wait normally.
let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?;
let activated: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM legacy_run_history_activation WHERE singleton = 1)",
)
.fetch_one(&mut *transaction)
.await?;
if activated {
sqlx::query(
r"
INSERT INTO legacy_run_history_deletions (run_id, deleted_at_ms)
VALUES (?, ?)
ON CONFLICT(run_id) DO UPDATE SET deleted_at_ms = excluded.deleted_at_ms
",
)
.bind(run_id.to_string())
.bind(deleted_at_ms)
.execute(&mut *transaction)
.await?;
}
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind(run_id.to_string())
.execute(&mut *transaction)
.await?;
transaction.commit().await?;
Ok(())
}
pub async fn get(&self, run_id: &RunId, now: DateTime<Utc>) -> Result<Option<Run>> {
let mut query = QueryBuilder::<Sqlite>::new(SELECT_RUN_SUMMARIES_SQL);
query
@ -327,23 +544,8 @@ FROM runs",
has_more: consumed < total,
})
}
pub async fn delete(&self, run_id: &RunId) -> Result<()> {
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind(run_id.to_string())
.execute(&self.pool)
.await?;
Ok(())
}
}
#[cfg_attr(
not(test),
expect(
dead_code,
reason = "the atomic SQL run path stays inactive until the authority cutover"
)
)]
impl RunSummaryStore {
pub(crate) async fn insert_first_event_on_connection(
connection: &mut SqliteConnection,
@ -396,6 +598,26 @@ impl RunSummaryStore {
pub(crate) async fn list_events_with_json_on_connection(
connection: &mut SqliteConnection,
run_id: &RunId,
) -> Result<Vec<(EventEnvelope, String)>> {
// The current-row head and event rows must come from one snapshot.
// Otherwise a concurrent append between the two SELECTs looks like
// durable corruption even though both versions are individually valid.
let mut transaction = connection.begin().await?;
let events = Self::list_events_with_json_in_transaction(&mut transaction, run_id).await?;
transaction.commit().await?;
Ok(events)
}
pub(crate) async fn list_events_with_json_in_transaction(
transaction: &mut Transaction<'_, Sqlite>,
run_id: &RunId,
) -> Result<Vec<(EventEnvelope, String)>> {
Self::list_events_with_json_in_snapshot(&mut *transaction, run_id).await
}
async fn list_events_with_json_in_snapshot(
connection: &mut SqliteConnection,
run_id: &RunId,
) -> Result<Vec<(EventEnvelope, String)>> {
let mut query = QueryBuilder::<Sqlite>::new(SELECT_EVENT_COLUMNS);
query
@ -800,7 +1022,7 @@ fn sql_limit(limit: usize) -> i64 {
i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX)
}
fn next_event_seq_after(last_seq: u32) -> Result<u32> {
pub(crate) fn next_event_seq_after(last_seq: u32) -> Result<u32> {
last_seq
.checked_add(1)
.filter(|seq| *seq <= keys::MAX_EVENT_SEQ)
@ -830,11 +1052,7 @@ fn decode_event_rows_with_json(
) -> Result<Vec<(EventEnvelope, String)>> {
let run_id_text = run_id.to_string();
rows.iter()
.map(|row| {
let event_json: String = row.try_get("event_json")?;
let envelope = decode_event_row(row, run_id, &run_id_text)?;
Ok((envelope, event_json))
})
.map(|row| decode_event_row_with_json(row, run_id, &run_id_text))
.collect()
}
@ -870,6 +1088,17 @@ fn decode_event_row(
expected_run_id: &RunId,
expected_run_id_text: &str,
) -> Result<EventEnvelope> {
decode_event_row_with_json(row, expected_run_id, expected_run_id_text)
.map(|(envelope, _event_json)| envelope)
}
/// Decodes one event row and returns the raw `event_json` alongside it so
/// callers that need both do not fetch the column twice.
fn decode_event_row_with_json(
row: &SqliteRow,
expected_run_id: &RunId,
expected_run_id_text: &str,
) -> Result<(EventEnvelope, String)> {
let stored_run_id: String = row.try_get("run_id")?;
let raw_seq: i64 = row.try_get("seq")?;
let seq = stored_seq(raw_seq).ok_or_else(|| run_event_mismatch(expected_run_id, 0, "seq"))?;
@ -911,7 +1140,7 @@ fn decode_event_row(
return Err(run_event_mismatch(expected_run_id, seq, "session_id"));
}
Ok(EventEnvelope { seq, event })
Ok((EventEnvelope { seq, event }, event_json))
}
async fn select_run_head(connection: &mut SqliteConnection, run_id: &RunId) -> Result<Option<u32>> {
@ -953,6 +1182,7 @@ fn normalize_billing_for_read_model(mut billing: BilledTokenCounts) -> BilledTok
billing
}
#[cfg(test)]
async fn upsert_run_on_connection(
connection: &mut SqliteConnection,
record: &PreparedRunSummary,
@ -1195,6 +1425,7 @@ fn overlay_live_wall_time(run: &mut Run, now: DateTime<Utc>) {
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::time::Duration;
use chrono::{DateTime, Utc};
use fabro_types::{
@ -1204,6 +1435,7 @@ mod tests {
WorkflowSettings, test_support,
};
use strum::VariantArray as _;
use tokio::time;
use ulid::Ulid;
use super::{
@ -1575,6 +1807,38 @@ mod tests {
assert_eq!(sequences, vec![1, 2, 3]);
}
#[tokio::test]
async fn canonical_delete_waits_for_a_concurrent_writer() {
let (_directory, store) = store().await;
let created_at = dt("2026-08-27T12:00:00Z");
let id = run_id(created_at.timestamp_millis().cast_unsigned(), 5);
let first = entry(projection(id, "created", created_at), 1);
let mut transaction = store.pool.begin().await.unwrap();
RunSummaryStore::insert_first_event_on_connection(
&mut transaction,
&first,
&created_payload(&id),
)
.await
.unwrap();
transaction.commit().await.unwrap();
store.test_mark_run_history_activated().await.unwrap();
let blocker = store.pool.begin_with("BEGIN IMMEDIATE").await.unwrap();
let contender = store.clone();
let delete = tokio::spawn(async move { contender.delete_canonical(&id, 2).await });
time::sleep(Duration::from_millis(25)).await;
assert!(
!delete.is_finished(),
"delete should wait for the existing writer"
);
blocker.commit().await.unwrap();
delete.await.unwrap().unwrap();
assert!(!store.contains(&id).await.unwrap());
assert!(store.test_is_run_history_tombstoned(&id).await.unwrap());
}
#[tokio::test]
async fn sql_run_reads_preserve_paging_filters_json_and_legacy_gaps() {
let (_directory, store) = store().await;

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

@ -1,5 +1,4 @@
mod projection_cache;
mod run_catalog_index;
mod run_store;
use std::collections::HashMap;
@ -12,14 +11,15 @@ use fabro_types::{Run, RunId, SessionId};
use object_store::ObjectStore;
pub use projection_cache::CachedRunProjection;
use projection_cache::RunProjectionCache;
pub use run_catalog_index::RunCatalogIndex;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, OnceCell};
use tokio::sync::{Mutex, MutexGuard, OnceCell};
use tracing::warn;
use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys};
use crate::{
BlobStore, Error, EventPayload, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys,
};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnreadableRun {
@ -42,7 +42,6 @@ pub struct Database {
db: Arc<OnceCell<slatedb::Db>>,
active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
blobs: Arc<BlobStore>,
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
run_summary_store: Arc<RunSummaryStore>,
@ -75,7 +74,6 @@ impl Database {
db: Arc::new(OnceCell::new()),
active_runs: Arc::new(Mutex::new(HashMap::new())),
blobs,
catalog_index: Arc::new(OnceCell::new()),
projection_cache: Arc::new(RunProjectionCache::default()),
projection_cache_warmed: Arc::new(OnceCell::new()),
run_summary_store,
@ -117,13 +115,7 @@ impl Database {
active_run_from(&active_runs, run_id)
}
fn cache_active_run(
active_runs: &mut HashMap<RunId, Arc<RunDatabaseInner>>,
run_store: &RunDatabase,
) {
active_runs.insert(run_store.run_id(), run_store.inner_arc());
}
#[cfg(test)]
async fn remove_active_run(&self, run_id: &RunId) -> Option<RunDatabase> {
self.active_runs
.lock()
@ -132,11 +124,17 @@ impl Database {
.map(RunDatabase::from_inner)
}
fn cache_active_run(
active_runs: &mut HashMap<RunId, Arc<RunDatabaseInner>>,
run_store: &RunDatabase,
) {
active_runs.insert(run_store.run_id(), run_store.inner_arc());
}
/// Builds a run handle wired to the Database-owned shared stores.
async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result<RunDatabase> {
RunDatabase::build(
*run_id,
self.open_db().await?,
read_only,
self.blobs(),
Arc::clone(&self.projection_cache),
@ -145,18 +143,54 @@ impl Database {
.await
}
pub async fn create_run_with_first_event(
&self,
run_id: &RunId,
payload: &EventPayload,
) -> Result<RunDatabase> {
let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?;
let (envelope, cached) = run_store.commit_first_event(payload).await?;
run_store.install_in_memory_state(&envelope, &cached);
Self::cache_active_run(&mut active_runs, &run_store);
run_store.publish(&envelope);
Ok(run_store)
}
/// Creates an empty run handle for fixture setup. Production code must
/// create sequence-1 `run.created` atomically with the canonical row.
#[cfg(any(test, feature = "test-support"))]
pub async fn create_run(&self, run_id: &RunId) -> Result<RunDatabase> {
self.warm_projection_cache().await?;
self.catalog_index().await?.add(run_id).await?;
let run_store = self.open_run_database(run_id, false).await?;
let mut active_runs = self.active_runs.lock().await;
let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?;
Self::cache_active_run(&mut active_runs, &run_store);
Ok(run_store)
}
/// Builds an empty handle for a run that exists neither in memory nor in
/// SQLite, returning the held `active_runs` guard so the caller can
/// register the handle before any concurrent creator observes the gap.
async fn reserve_new_run(
&self,
run_id: &RunId,
) -> Result<(
MutexGuard<'_, HashMap<RunId, Arc<RunDatabaseInner>>>,
RunDatabase,
)> {
self.warm_projection_cache().await?;
let active_runs = self.active_runs.lock().await;
if active_runs.contains_key(run_id) || self.run_summary_store.contains(run_id).await? {
return Err(Error::RunAlreadyExists(run_id.to_string()));
}
let run_store = RunDatabase::build_empty(
*run_id,
self.blobs(),
Arc::clone(&self.projection_cache),
self.run_summary_store(),
);
Ok((active_runs, run_store))
}
pub async fn open_run(&self, run_id: &RunId) -> Result<RunDatabase> {
self.warm_projection_cache().await?;
let db = self.open_db().await?;
// Keep the active-writer miss and insert atomic. Otherwise concurrent
// callers can create independent writers with the same recovered seq.
let mut active_runs = self.active_runs.lock().await;
@ -169,7 +203,7 @@ impl Database {
}
return Ok(active);
}
if !RunDatabase::has_any_events(&db, run_id).await? {
if !self.run_summary_store.contains(run_id).await? {
return Err(Error::RunNotFound(run_id.to_string()));
}
let run_store = self.open_run_database(run_id, false).await?;
@ -178,7 +212,6 @@ impl Database {
}
pub async fn open_run_reader(&self, run_id: &RunId) -> Result<RunDatabase> {
let db = self.open_db().await?;
if let Some(active) = self.get_active_run(run_id).await {
if !active.matches_run(run_id) {
return Err(Error::Other(format!(
@ -187,7 +220,7 @@ impl Database {
}
return Ok(active.read_only_clone());
}
if !RunDatabase::has_any_events(&db, run_id).await? {
if !self.run_summary_store.contains(run_id).await? {
return Err(Error::RunNotFound(run_id.to_string()));
}
self.open_run_database(run_id, true).await
@ -218,15 +251,12 @@ impl Database {
pub async fn warm_projection_cache(&self) -> Result<()> {
self.projection_cache_warmed
.get_or_try_init(|| async {
let db = self.open_db().await?;
let run_ids = self
.catalog_index()
.await?
.list(&ListRunsQuery::default())
.await?;
let run_ids = self.run_summary_store.list_run_ids().await?;
let mut entries = Vec::new();
for run_id in run_ids {
match RunDatabase::build_cached_projection(&db, &run_id).await {
match RunDatabase::build_cached_projection(&self.run_summary_store, &run_id)
.await
{
Ok(Some(entry)) => entries.push(entry),
Ok(None) => {}
Err(err) => {
@ -238,8 +268,7 @@ impl Database {
}
}
}
self.run_summary_store.reconcile(&entries).await?;
self.projection_cache.replace_all(entries).await;
self.projection_cache.replace_all(entries);
Ok::<_, Error>(())
})
.await?;
@ -252,19 +281,14 @@ impl Database {
now: DateTime<Utc>,
) -> Result<Vec<CachedRunProjection>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.list(query, now).await)
Ok(self.projection_cache.list(query, now))
}
pub async fn list_unreadable_runs(&self) -> Result<Vec<UnreadableRun>> {
let db = self.open_db().await?;
let run_ids = self
.catalog_index()
.await?
.list(&ListRunsQuery::default())
.await?;
let run_ids = self.run_summary_store.list_run_ids().await?;
let mut unreadable = Vec::new();
for run_id in run_ids {
match RunDatabase::build_cached_projection(&db, &run_id).await {
match RunDatabase::build_cached_projection(&self.run_summary_store, &run_id).await {
Ok(Some(_)) => {}
Ok(None) => unreadable.push(UnreadableRun {
run_id,
@ -293,6 +317,21 @@ impl Database {
run_id: &RunId,
seq: u32,
payload: &serde_json::Value,
) -> Result<()> {
self.run_summary_store
.test_insert_unvalidated_event(run_id, seq, payload)
.await?;
self.active_runs.lock().await.remove(run_id);
self.projection_cache.remove(run_id);
Ok(())
}
#[cfg(any(test, feature = "test-support"))]
pub(crate) async fn put_unvalidated_legacy_run_event(
&self,
run_id: &RunId,
seq: u32,
payload: &serde_json::Value,
) -> Result<()> {
let db = self.open_db().await?;
db.put(
@ -305,7 +344,7 @@ impl Database {
pub async fn get_cached_run(&self, run_id: &RunId) -> Result<Option<CachedRunProjection>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.get(run_id).await)
Ok(self.projection_cache.get(run_id))
}
pub async fn get_cached_projection(
@ -316,7 +355,6 @@ impl Database {
Ok(self
.projection_cache
.projection_snapshot(run_id)
.await
.map(|(projection, _)| projection))
}
@ -326,14 +364,14 @@ impl Database {
now: DateTime<Utc>,
) -> Result<Option<Run>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.get_summary(run_id, now).await)
Ok(self.projection_cache.get_summary(run_id, now))
}
/// Run ids whose latest explicit pull request creation is still pending,
/// oldest request first.
pub async fn pending_pull_request_creation_run_ids(&self) -> Result<Vec<RunId>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.pending_pull_request_creations().await)
Ok(self.projection_cache.pending_pull_request_creations())
}
pub async fn put_session_run_index(
@ -359,33 +397,29 @@ impl Database {
Ok(None)
}
pub(crate) async fn remove_cached_run(&self, run_id: &RunId) {
self.projection_cache.remove(run_id).await;
pub(crate) fn remove_cached_run(&self, run_id: &RunId) {
self.projection_cache.remove(run_id);
}
pub async fn delete_run(&self, run_id: &RunId) -> Result<()> {
let active = self.remove_active_run(run_id).await;
if let Some(active) = &active {
active.close().await?;
let mut active_runs = self.active_runs.lock().await;
let active = active_runs.get(run_id).cloned();
let _state_guard = match &active {
Some(active) => Some(active.state_lock.lock().await),
None => None,
};
self.run_summary_store
.delete_canonical(run_id, Utc::now().timestamp_millis())
.await?;
active_runs.remove(run_id);
self.remove_cached_run(run_id);
if let Err(err) = self.delete_session_indexes_for_run(run_id).await {
warn!(
run_id = %run_id,
error = %err,
"Failed to remove retired session reverse indexes after deleting run"
);
}
let db = self.open_db().await?;
let mut keys_to_delete = Vec::new();
for prefix in [keys::run_data_prefix(run_id)] {
let mut iter = db.scan_prefix(&prefix).await?;
while let Some(entry) = iter.next().await? {
keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
Error::Other(format!("stored key is not valid UTF-8: {err}"))
})?);
}
}
for key in keys_to_delete {
db.delete(key).await?;
}
self.delete_session_indexes_for_run(run_id).await?;
self.catalog_index().await?.remove(run_id).await?;
self.remove_cached_run(run_id).await;
self.run_summary_store.delete(run_id).await?;
Ok(())
}
@ -407,17 +441,6 @@ impl Database {
Ok(())
}
pub async fn catalog_index(&self) -> Result<Arc<RunCatalogIndex>> {
let store = self
.catalog_index
.get_or_try_init(|| async {
let db = Arc::new(self.open_db().await?);
Ok::<_, Error>(Arc::new(RunCatalogIndex::new(db)))
})
.await?;
Ok(Arc::clone(store))
}
#[must_use]
pub fn blobs(&self) -> Arc<BlobStore> {
Arc::clone(&self.blobs)
@ -463,7 +486,7 @@ impl Runs {
}
pub async fn find(&self, run_id: &RunId) -> Result<Option<Run>> {
self.db.get_cached_summary(run_id, Utc::now()).await
self.db.run_summary_store.get(run_id, Utc::now()).await
}
pub async fn list(&self, query: &ListRunsQuery) -> Result<Vec<Run>> {
@ -881,6 +904,67 @@ mod tests {
assert_eq!(read.as_deref(), Some(shared_blob.as_slice()));
}
#[tokio::test]
async fn activated_delete_tombstone_prevents_legacy_history_resurrection() {
let (directory, summaries) = make_run_summary_store().await;
let (_object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let created = event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
&serde_json::json!({
"settings": sample_run_spec("run-1").settings,
"graph": sample_run_spec("run-1").graph,
"provenance": test_support::test_run_provenance(),
}),
);
store
.put_unvalidated_legacy_run_event(&run_id, 1, created.as_value())
.await
.unwrap();
let run = store.create_run(&run_id).await.unwrap();
run.append_event(&created).await.unwrap();
summaries.test_mark_run_history_activated().await.unwrap();
store.delete_run(&run_id).await.unwrap();
assert!(
summaries
.test_is_run_history_tombstoned(&run_id)
.await
.unwrap()
);
assert!(store.open_run(&run_id).await.is_err());
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
let report = store
.import_legacy_run_history_into(database.pool())
.await
.unwrap();
assert_eq!(report.tombstoned_source_runs, 1);
assert_eq!(report.tombstoned_source_events, 1);
assert!(store.open_run(&run_id).await.is_err());
let verification = store
.verify_legacy_run_history_in(database.pool())
.await
.unwrap();
assert_eq!(verification.tombstoned_source_runs, 1);
assert_eq!(verification.tombstoned_source_events, 1);
let recreated = store.create_run(&run_id).await.unwrap();
recreated.append_event(&created).await.unwrap();
assert!(
store
.verify_legacy_run_history_in(database.pool())
.await
.is_err(),
"a tombstone and live canonical data must fail verification"
);
}
#[tokio::test]
async fn open_run_reader_is_read_only() {
let (_object_store, store) = make_store();
@ -1003,12 +1087,13 @@ mod tests {
}
#[tokio::test]
async fn best_effort_run_summary_update_failure_keeps_slate_append_repairable() {
let (directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
async fn sql_failure_leaves_event_and_projection_unpublished() {
let (_directory, summaries) = make_run_summary_store().await;
let (_object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let cached_before = store.get_cached_run(&run_id).await.unwrap().unwrap();
summaries.close_pool().await;
let result = run
@ -1016,42 +1101,17 @@ mod tests {
"run-1",
"2026-03-27T12:00:01Z",
"run.title.updated",
&serde_json::json!({ "title": "Committed title" }),
&serde_json::json!({ "title": "Uncommitted title" }),
))
.await;
assert!(result.is_ok(), "committed append returned {result:?}");
assert_eq!(run.list_events().await.unwrap().len(), 2);
assert!(matches!(
result,
Err(Error::Sqlite(sqlx::Error::PoolClosed))
));
let cached = store.get_cached_run(&run_id).await.unwrap().unwrap();
assert_eq!(cached.last_seq, 2);
assert_eq!(cached.summary.title, "Committed title");
let stored = run.get_event(2).await.unwrap().unwrap();
assert_eq!(stored.event, result.unwrap().event);
let repaired_summaries =
Arc::new(store_test_support::sqlite_run_summary_store_at(directory.path()).await);
let stale = repaired_summaries
.get(&run_id, Utc::now())
.await
.unwrap()
.unwrap();
assert_ne!(stale.title, "Committed title");
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&repaired_summaries),
);
reopened.warm_projection_cache().await.unwrap();
let repaired = repaired_summaries
.get(&run_id, Utc::now())
.await
.unwrap()
.unwrap();
assert_eq!(repaired.title, "Committed title");
assert_eq!(cached.last_seq, cached_before.last_seq);
assert_eq!(cached.summary.title, cached_before.summary.title);
}
#[tokio::test]
@ -1069,7 +1129,7 @@ mod tests {
let err = run.append_event(&invalid_first).await.unwrap_err();
assert!(matches!(err, Error::EventRejected { .. }));
assert!(run.list_events().await.unwrap().is_empty());
assert_eq!(run.last_event_seq().await.unwrap(), None);
append_created(&run, "run-1", dt("2026-03-27T12:00:01Z")).await;
assert_eq!(run.list_events().await.unwrap().len(), 1);
@ -1101,7 +1161,7 @@ mod tests {
let err = run.append_event(&malformed).await.unwrap_err();
assert!(matches!(err, Error::InvalidEvent(_)));
assert!(run.list_events().await.unwrap().is_empty());
assert_eq!(run.last_event_seq().await.unwrap(), None);
}
#[tokio::test]
@ -1496,13 +1556,20 @@ mod tests {
}
#[tokio::test]
async fn reopening_store_rebuilds_from_shared_db() {
let (object_store, store) = make_store();
async fn reopening_store_rebuilds_from_shared_sqlite() {
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
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 =
store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let summary = reopened
.list_runs(&ListRunsQuery::default(), Utc::now())
.await
@ -1516,14 +1583,21 @@ mod tests {
#[tokio::test]
async fn projection_cache_warmup_lists_newest_first_and_applies_date_filters() {
let (object_store, store) = make_store();
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
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 =
store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
reopened.warm_projection_cache().await.unwrap();
let entries = reopened
@ -1564,100 +1638,6 @@ mod tests {
});
}
#[tokio::test]
async fn projection_cache_warmup_skips_unreplayable_catalog_run() {
let (object_store, store) = make_store();
let good_run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_completed(&good_run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let bad_run_id = test_run_id("run-2");
store
.catalog_index()
.await
.unwrap()
.add(&bad_run_id)
.await
.unwrap();
store
.put_unvalidated_run_event(
&bad_run_id,
1,
&serde_json::json!({ "not": "a valid run event" }),
)
.await
.unwrap();
let reopened =
store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
reopened.warm_projection_cache().await.unwrap();
let entries = reopened
.list_cached_runs(&ListRunsQuery::default(), Utc::now())
.await
.unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].run_id, test_run_id("run-1"));
assert!(
reopened
.get_cached_run(&bad_run_id)
.await
.unwrap()
.is_none()
);
assert!(reopened.runs().find(&bad_run_id).await.unwrap().is_none());
}
#[tokio::test]
async fn list_unreadable_runs_reports_catalog_entries_that_fail_projection() {
let (object_store, store) = make_store();
let good_run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_completed(&good_run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let bad_run_id = test_run_id("run-2");
store
.catalog_index()
.await
.unwrap()
.add(&bad_run_id)
.await
.unwrap();
let run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap();
store
.put_unvalidated_run_event(
&bad_run_id,
1,
&serde_json::json!({
"id": "evt-run-2-run.created",
"ts": "2026-03-27T12:00:10Z",
"run_id": bad_run_id,
"event": "run.created",
"properties": {
"settings": run_spec["settings"],
"graph": run_spec["graph"],
"workflow_slug": run_spec["workflow_slug"],
"source_directory": run_spec["source_directory"],
"git": run_spec["git"],
"labels": run_spec["labels"],
},
}),
)
.await
.unwrap();
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);
assert_eq!(unreadable[0].run_id, bad_run_id);
assert_eq!(unreadable[0].created_at, bad_run_id.created_at());
assert!(
unreadable[0].error.contains("missing field `provenance`"),
"expected missing provenance error, got: {}",
unreadable[0].error
);
}
#[tokio::test]
async fn required_run_summary_append_refreshes_cache_and_delete_removes_rows() {
let (_directory, summaries) = make_run_summary_store().await;
@ -1835,43 +1815,23 @@ mod tests {
}
#[tokio::test]
async fn required_run_summary_warmup_backfills_sqlite_run_summaries() {
let (object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
async fn opening_sql_backed_run_does_not_read_legacy_event_history() {
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let (_directory, summaries) = make_run_summary_store().await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&summaries),
summaries,
);
reopened.warm_projection_cache().await.unwrap();
let summary = summaries
.get(&test_run_id("run-1"), Utc::now())
.await
.unwrap()
.unwrap();
assert_eq!(summary.lifecycle.status, RunStatus::Succeeded {
reason: SuccessReason::Completed,
});
}
#[tokio::test]
async fn opening_cached_run_does_not_read_older_event_history() {
let (object_store, store) = make_store();
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
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
// unreadable old key makes the operation fail. A hydrated run starts
// after the shared projection's last sequence instead.
@ -1906,7 +1866,8 @@ mod tests {
#[tokio::test]
async fn append_event_hydrates_local_projection_cache_for_fresh_writer() {
let (object_store, store) = make_store();
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
@ -1938,11 +1899,13 @@ mod tests {
.await
.unwrap();
let reopened = store_test_support::test_database(
Arc::clone(&object_store),
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let fresh_writer = reopened.open_run(&run_id).await.unwrap();
fresh_writer

49
lib/components/fabro-store/src/slate/projection_cache.rs
View file

@ -1,9 +1,8 @@
use std::collections::{BTreeSet, HashMap};
use std::sync::Arc;
use std::sync::{Arc, Mutex, MutexGuard};
use chrono::{DateTime, Utc};
use fabro_types::{Run, RunId, RunProjection};
use tokio::sync::Mutex;
use crate::ListRunsQuery;
use crate::run_state::build_summary;
@ -30,6 +29,9 @@ impl CachedRunProjection {
#[derive(Debug, Default)]
pub(crate) struct RunProjectionCache {
// Cache operations are bounded in-memory work and never await. Keeping
// this lock synchronous lets a committed event update both projection
// caches without introducing a cancellation point.
state: Mutex<RunProjectionCacheState>,
}
@ -109,21 +111,27 @@ fn apply_read_overlays(entry: &mut CachedRunProjection, now: DateTime<Utc>) {
}
impl RunProjectionCache {
pub(crate) async fn replace_all(&self, entries: Vec<CachedRunProjection>) {
self.state.lock().await.replace_all(entries);
fn lock(&self) -> MutexGuard<'_, RunProjectionCacheState> {
self.state.lock().expect(
"run projection cache mutex is never poisoned: no code panics while holding this lock",
)
}
pub(crate) async fn replace(&self, entry: CachedRunProjection) {
self.state.lock().await.insert(entry);
pub(crate) fn replace_all(&self, entries: Vec<CachedRunProjection>) {
self.lock().replace_all(entries);
}
pub(crate) async fn list(
pub(crate) fn replace(&self, entry: CachedRunProjection) {
self.lock().insert(entry);
}
pub(crate) fn list(
&self,
query: &ListRunsQuery,
now: DateTime<Utc>,
) -> Vec<CachedRunProjection> {
let entries = {
let state = self.state.lock().await;
let state = self.lock();
let raw = match query.parent_id {
Some(parent_id) => state
.children_by_parent
@ -166,8 +174,8 @@ impl RunProjectionCache {
entries
}
pub(crate) async fn get(&self, run_id: &RunId) -> Option<CachedRunProjection> {
let state = self.state.lock().await;
pub(crate) fn get(&self, run_id: &RunId) -> Option<CachedRunProjection> {
let state = self.lock();
state
.entries
.get(run_id)
@ -177,13 +185,8 @@ impl RunProjectionCache {
/// Projection and last sequence for `run_id`, without the summary clone
/// and children count that `get` computes under the cache mutex.
pub(crate) async fn projection_snapshot(
&self,
run_id: &RunId,
) -> Option<(Arc<RunProjection>, u32)> {
self.state
.lock()
.await
pub(crate) fn projection_snapshot(&self, run_id: &RunId) -> Option<(Arc<RunProjection>, u32)> {
self.lock()
.entries
.get(run_id)
.map(|entry| (Arc::clone(&entry.projection), entry.last_seq))
@ -192,11 +195,9 @@ impl RunProjectionCache {
/// Run ids whose latest explicit pull request creation is still pending,
/// oldest request first. Clones only ids and timestamps, so callers can
/// poll on an interval without materializing run summaries.
pub(crate) async fn pending_pull_request_creations(&self) -> Vec<RunId> {
pub(crate) fn pending_pull_request_creations(&self) -> Vec<RunId> {
let mut pending = self
.state
.lock()
.await
.entries
.values()
.filter_map(|entry| {
@ -210,9 +211,9 @@ impl RunProjectionCache {
pending.into_iter().map(|(_, run_id)| run_id).collect()
}
pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
pub(crate) fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
let mut entry = {
let state = self.state.lock().await;
let state = self.lock();
state
.entries
.get(run_id)
@ -223,7 +224,7 @@ impl RunProjectionCache {
Some(entry.summary)
}
pub(crate) async fn remove(&self, run_id: &RunId) {
self.state.lock().await.remove(run_id);
pub(crate) fn remove(&self, run_id: &RunId) {
self.lock().remove(run_id);
}
}

152
lib/components/fabro-store/src/slate/run_catalog_index.rs
View file

@ -1,152 +0,0 @@
use std::sync::Arc;
use chrono::{Datelike, Timelike};
use fabro_types::RunId;
use futures::TryStreamExt;
use crate::record::{MarkerCodec, Record, Repository};
use crate::{ListRunsQuery, Result};
#[derive(Debug, Default)]
pub(crate) struct RunCatalogEntry;
impl Record for RunCatalogEntry {
type Id = RunId;
type Codec = MarkerCodec;
const PREFIX: &'static str = "runs/_index/by-start";
#[cfg(test)]
fn id(&self) -> Self::Id {
unreachable!("marker records must use put_at")
}
}
pub struct RunCatalogIndex {
repo: Repository<RunCatalogEntry>,
}
impl std::fmt::Debug for RunCatalogIndex {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RunCatalogIndex").finish_non_exhaustive()
}
}
impl RunCatalogIndex {
pub(crate) fn new(db: Arc<slatedb::Db>) -> Self {
Self {
repo: Repository::new(db),
}
}
pub async fn add(&self, run_id: &RunId) -> Result<()> {
self.repo.put_at(run_id, &RunCatalogEntry).await
}
pub async fn remove(&self, run_id: &RunId) -> Result<()> {
self.repo.delete(run_id).await
}
pub async fn list(&self, query: &ListRunsQuery) -> Result<Vec<RunId>> {
let mut run_ids = self.repo.scan_ids_stream().try_collect::<Vec<_>>().await?;
run_ids.retain(|run_id| {
let created_at = run_id.created_at();
if let Some(start) = query.start {
if created_at < start {
return false;
}
}
if let Some(end) = query.end {
if created_at > end {
return false;
}
}
true
});
run_ids.sort_by_key(|run_id| {
let created_at = run_id.created_at();
(
created_at.year(),
created_at.month(),
created_at.day(),
created_at.hour(),
created_at.minute(),
*run_id,
)
});
Ok(run_ids)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use chrono::{Duration as ChronoDuration, TimeZone, Utc};
use object_store::memory::InMemory;
use ulid::Ulid;
use super::RunCatalogIndex;
use crate::ListRunsQuery;
async fn index() -> RunCatalogIndex {
let db = Arc::new(
slatedb::Db::open("run-catalog-index-tests", Arc::new(InMemory::new()))
.await
.unwrap(),
);
RunCatalogIndex::new(db)
}
#[tokio::test]
async fn add_list_and_remove_round_trip() {
let index = index().await;
let early = fabro_types::RunId::from(Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 4, 20, 9, 0, 0).unwrap().into(),
));
let later = fabro_types::RunId::from(Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 4, 20, 9, 1, 0).unwrap().into(),
));
index.add(&later).await.unwrap();
index.add(&early).await.unwrap();
assert_eq!(index.list(&ListRunsQuery::default()).await.unwrap(), vec![
early, later
]);
index.remove(&early).await.unwrap();
assert_eq!(index.list(&ListRunsQuery::default()).await.unwrap(), vec![
later
]);
}
#[tokio::test]
async fn list_applies_start_and_end_filters() {
let index = index().await;
let first = fabro_types::RunId::from(Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 4, 20, 9, 0, 0).unwrap().into(),
));
let second = fabro_types::RunId::from(Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 4, 20, 9, 1, 0).unwrap().into(),
));
let third = fabro_types::RunId::from(Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 4, 20, 9, 2, 0).unwrap().into(),
));
for run_id in [first, second, third] {
index.add(&run_id).await.unwrap();
}
assert_eq!(
index
.list(&ListRunsQuery {
start: Some(second.created_at()),
end: Some(second.created_at() + ChronoDuration::seconds(1)),
parent_id: None,
})
.await
.unwrap(),
vec![second]
);
}
}

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

File diff suppressed because it is too large Load diff

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

@ -32,6 +32,7 @@ pub fn test_run_summary_store() -> Arc<RunSummaryStore> {
Arc::new(RunSummaryStore::new(lazy_in_memory_pool(&[
fabro_db::RUNS_MIGRATION_SQL,
fabro_db::RUN_EVENTS_MIGRATION_SQL,
fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
])))
}
@ -68,6 +69,13 @@ pub fn test_blob_store_path(store_dir: &Path) -> PathBuf {
fabro_db::append_to_path(store_dir, "-blobs.sqlite3")
}
/// Returns the SQLite file backing [`test_run_summary_store_at`] for
/// `store_dir`.
#[must_use]
pub fn test_run_summary_store_path(store_dir: &Path) -> PathBuf {
fabro_db::append_to_path(store_dir, "-runs.sqlite3")
}
/// Returns a durable SQLite blob authority stored beside `store_dir`.
///
/// Handles created for the same directory share one blob database file, so
@ -77,32 +85,88 @@ pub fn test_blob_store_path(store_dir: &Path) -> PathBuf {
/// siblings) when they reset the directory itself.
#[must_use]
pub fn test_blob_store_at(store_dir: &Path) -> Arc<BlobStore> {
Arc::new(BlobStore::new(lazy_file_pool(
test_blob_store_path(store_dir),
"blobs",
&[fabro_db::BLOBS_MIGRATION_SQL],
)))
}
/// Returns a durable SQLite run-history authority stored beside `store_dir`.
///
/// Handles created for the same directory share one database file, which lets
/// reopen-style tests model the process-wide SQLite authority used in
/// production.
#[must_use]
pub fn test_run_summary_store_at(store_dir: &Path) -> Arc<RunSummaryStore> {
Arc::new(RunSummaryStore::new(lazy_file_pool(
test_run_summary_store_path(store_dir),
"runs",
&[
fabro_db::RUNS_MIGRATION_SQL,
fabro_db::RUN_EVENTS_MIGRATION_SQL,
fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
],
)))
}
/// Builds a single-connection file-backed SQLite pool that installs
/// `migrations` the first time it opens a database without `probe_table`.
///
/// Like [`lazy_in_memory_pool`], the pool connects lazily so synchronous
/// fixture builders stay synchronous, and the file persists across handles so
/// reopen-style tests share one authority.
fn lazy_file_pool(
path: PathBuf,
probe_table: &'static str,
migrations: &'static [&'static str],
) -> sqlx::SqlitePool {
let options = SqliteConnectOptions::new()
.filename(test_blob_store_path(store_dir))
.filename(path)
.create_if_missing(true)
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
SqlitePoolOptions::new()
.max_connections(1)
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
.after_connect(move |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')",
WHERE type = 'table' AND name = ?)",
)
.bind(probe_table)
.fetch_one(&mut *connection)
.await?;
if !installed {
sqlx::query(fabro_db::BLOBS_MIGRATION_SQL)
.execute(&mut *connection)
.await?;
for migration in migrations {
sqlx::raw_sql(*migration).execute(&mut *connection).await?;
}
}
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(BlobStore::new(pool))
.connect_lazy_with(options)
}
/// Builds a test database whose SQLite blob and run-history authorities are
/// durable beside `store_dir` and shared by reopen-style handles.
#[must_use]
pub fn test_database_at(
object_store: Arc<dyn ObjectStore>,
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
store_dir: &Path,
) -> Database {
test_database_with_stores(
object_store,
base_prefix,
flush_interval,
cache_path,
test_blob_store_at(store_dir),
test_run_summary_store_at(store_dir),
)
}
/// Builds a Slate-backed run database with its own isolated blob authority.
@ -192,6 +256,18 @@ pub async fn put_unvalidated_run_event(
.await
}
/// Seeds one event in the retired Slate run-history keyspace.
pub async fn put_legacy_run_event(
database: &Database,
run_id: &RunId,
seq: u32,
payload: &serde_json::Value,
) -> Result<()> {
database
.put_unvalidated_legacy_run_event(run_id, seq, payload)
.await
}
/// Connects to a migrated `fabro.sqlite3` in `directory` and returns its pool.
#[cfg(test)]
async fn sqlite_test_pool(directory: &Path) -> sqlx::SqlitePool {

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

@ -19,6 +19,6 @@ pub use self::redaction::{
};
pub use self::sink::{
RunEventLogger, RunEventPersistenceError, RunEventSink, StoreProgressLogger, append_event,
append_event_if, append_event_to_sink,
append_event_if, append_event_to_sink, create_run,
};
pub use crate::stage_scope::StageScope;

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

@ -4,14 +4,15 @@ use std::sync::Arc;
use ::fabro_types::{RunEvent, RunId, RunProjection};
use anyhow::Result;
use fabro_store::RunDatabase;
use chrono::{DateTime, Utc};
use fabro_store::{Database, RunDatabase};
use fabro_util::error::{SharedError, collect_chain};
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot, watch};
use super::emitter::Emitter;
use super::redaction::{build_redacted_event_payload, redacted_event_json};
use super::{Event, to_run_event};
use super::{Event, to_run_event, to_run_event_at};
use crate::runtime_store::RunStoreHandle;
pub async fn append_event(run_store: &RunDatabase, run_id: &RunId, event: &Event) -> Result<()> {
@ -24,6 +25,21 @@ pub async fn append_event(run_store: &RunDatabase, run_id: &RunId, event: &Event
.map_err(anyhow::Error::from)
}
/// Creates a run by committing its redacted `run.created` event and canonical
/// current row in one SQLite transaction.
pub async fn create_run(
store: &Database,
run_id: &RunId,
event: &Event,
timestamp: DateTime<Utc>,
) -> Result<RunDatabase> {
let stored = to_run_event_at(run_id, event, timestamp, None);
let payload = build_redacted_event_payload(&stored, run_id)?;
Box::pin(store.create_run_with_first_event(run_id, &payload))
.await
.map_err(anyhow::Error::from)
}
pub async fn append_event_if(
run_store: &RunDatabase,
run_id: &RunId,

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

@ -13,7 +13,7 @@ use std::sync::Arc;
use fabro_config::Storage;
use fabro_graphviz::graph::{AttrValue, Graph};
use fabro_model::{Catalog, ProviderId};
use fabro_store::{Database, RunDatabase};
use fabro_store::{BlobStore, Database};
use fabro_template::TemplateContext;
use fabro_types::{
AutomationRef, BlobHash, ForkSourceRef, GitContext, ManifestPath, RunId, RunProvenance,
@ -24,7 +24,7 @@ use tokio::task::spawn_blocking;
use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow};
use crate::error::Error;
use crate::event::{Event, append_event, to_run_event_at};
use crate::event::{self, Event, append_event};
use crate::pipeline::types::PersistOptions;
use crate::pipeline::{self, Persisted, TransformOptions, Validated};
use crate::records::RunSpec;
@ -539,69 +539,57 @@ async fn persist_created_run(
web_url: Option<String>,
) -> Result<(), Error> {
let record = persisted.run_spec();
let run_store = store
.create_run(&record.run_id)
.await
.map_err(|err| Error::engine_with_source("failed to create run store", err))?;
let definition_bytes = accepted_definition
.map(serde_json::to_vec)
.transpose()
.map_err(|err| Error::engine_with_source("failed to serialize run definition", err))?;
let spec_bytes = serde_json::to_vec(record)
.map_err(|err| Error::engine_with_source("failed to serialize run spec", err))?;
let blob_store = store.blobs();
let (manifest_blob, definition_blob, spec_blob) = tokio::try_join!(
write_optional_blob(&run_store, submitted_manifest_bytes),
write_optional_blob(&run_store, definition_bytes.as_deref()),
async { run_store.write_blob(&spec_bytes).await.map_err(store_error) },
write_optional_blob(&blob_store, submitted_manifest_bytes),
write_optional_blob(&blob_store, definition_bytes.as_deref()),
async { blob_store.write(&spec_bytes).await.map_err(store_error) },
)?;
let title = explicit_title.unwrap_or_else(|| fabro_types::infer_run_title(record.graph.goal()));
let stored = to_run_event_at(
let first_event = Event::RunCreated {
run_id: record.run_id,
title: Some(title),
settings: normalize_json_value(
serde_json::to_value(&record.settings).map_err(|err| Error::engine(err.to_string()))?,
),
graph: normalize_json_value(
serde_json::to_value(&record.graph).map_err(|err| Error::engine(err.to_string()))?,
),
workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()),
labels: record
.labels
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>(),
source_directory: record.source_directory.clone(),
workflow_slug: record.workflow_slug.clone(),
workflow_version_id: record.workflow_version_id,
target: record.target.clone(),
automation: record.automation.clone(),
provenance: record.provenance.clone(),
manifest_blob,
spec_blob: Some(spec_blob),
git: record.git.clone(),
fork_source_ref: record.fork_source_ref.clone(),
retried_from: None,
parent_id,
web_url,
};
let run_store = event::create_run(
store,
&record.run_id,
&Event::RunCreated {
run_id: record.run_id,
title: Some(title),
settings: normalize_json_value(
serde_json::to_value(&record.settings)
.map_err(|err| Error::engine(err.to_string()))?,
),
graph: normalize_json_value(
serde_json::to_value(&record.graph)
.map_err(|err| Error::engine(err.to_string()))?,
),
workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()),
labels: record
.labels
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>(),
source_directory: record.source_directory.clone(),
workflow_slug: record.workflow_slug.clone(),
workflow_version_id: record.workflow_version_id,
target: record.target.clone(),
automation: record.automation.clone(),
provenance: record.provenance.clone(),
manifest_blob,
spec_blob: Some(spec_blob),
git: record.git.clone(),
fork_source_ref: record.fork_source_ref.clone(),
retried_from: None,
parent_id,
web_url,
},
&first_event,
record.run_id.created_at(),
None,
);
let payload = fabro_store::EventPayload::new(
serde_json::to_value(&stored).map_err(|err| Error::engine(err.to_string()))?,
&record.run_id,
)
.map_err(store_error)?;
run_store
.append_event(&payload)
.await
.map(|_| ())
.map_err(store_error)?;
.await
.map_err(|err| Error::engine_with_source("failed to create run store", err))?;
append_event(&run_store, &record.run_id, &Event::RunSubmitted {
definition_blob,
})
@ -610,15 +598,11 @@ async fn persist_created_run(
}
async fn write_optional_blob(
run_store: &RunDatabase,
blob_store: &BlobStore,
bytes: Option<&[u8]>,
) -> Result<Option<BlobHash>, Error> {
match bytes {
Some(bytes) => run_store
.write_blob(bytes)
.await
.map(Some)
.map_err(store_error),
Some(bytes) => blob_store.write(bytes).await.map(Some).map_err(store_error),
None => Ok(None),
}
}

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

@ -1,4 +1,5 @@
use anyhow::Result as AnyResult;
use chrono::Utc;
use fabro_store::{Database, RunProjection, RunProjectionReducer};
use fabro_types::{EventBody, EventEnvelope, ForkSourceRef, RunId, RunTarget};
@ -154,12 +155,7 @@ async fn persist_forked_run(
.current_checkpoint()
.ok_or_else(|| Error::engine("forked run projection has no checkpoint"))?;
let run_store = store
.create_run(&spec.run_id)
.await
.map_err(|err| Error::engine(err.to_string()))?;
event::append_event(&run_store, &spec.run_id, &Event::RunCreated {
let first_event = Event::RunCreated {
run_id: spec.run_id,
title: None,
settings: serde_json::to_value(&spec.settings)
@ -183,9 +179,10 @@ async fn persist_forked_run(
retried_from: None,
parent_id: None,
web_url: None,
})
.await
.map_err(|err| Error::engine(err.to_string()))?;
};
let run_store = event::create_run(store, &spec.run_id, &first_event, Utc::now())
.await
.map_err(|err| Error::engine(err.to_string()))?;
let replayed_checkpoint =
replay_historical_projection_events(&run_store, spec.run_id, historical_events).await?;

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

@ -1,5 +1,6 @@
use std::collections::BTreeMap;
use chrono::Utc;
use fabro_store::Database;
use fabro_types::{RunId, RunProvenance, RunSpec, RunStatus};
@ -64,12 +65,7 @@ pub async fn retry_run(
let settings = serde_json::to_value(&settings).map_err(|err| Error::engine(err.to_string()))?;
let graph = serde_json::to_value(&graph).map_err(|err| Error::engine(err.to_string()))?;
let retry_store = store
.create_run(&new_run_id)
.await
.map_err(|err| Error::engine(err.to_string()))?;
event::append_event(&retry_store, &new_run_id, &Event::RunCreated {
let first_event = Event::RunCreated {
run_id: new_run_id,
title: Some(title),
settings,
@ -91,9 +87,10 @@ pub async fn retry_run(
retried_from: Some(source_run_id),
parent_id,
web_url: input.web_url.clone(),
})
.await
.map_err(|err| Error::engine(err.to_string()))?;
};
let retry_store = event::create_run(store, &new_run_id, &first_event, Utc::now())
.await
.map_err(|err| Error::engine(err.to_string()))?;
event::append_event(&retry_store, &new_run_id, &Event::RunSubmitted {
definition_blob,

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

@ -174,14 +174,18 @@ 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);
for database_path in [
store_test_support::test_blob_store_path(&store_dir),
store_test_support::test_run_summary_store_path(&store_dir),
] {
for suffix in ["", "-wal", "-shm"] {
let mut sibling = database_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_with_blobs(
let store = Arc::new(store_test_support::test_database_at(
Arc::new(
LocalFileSystem::new_with_prefix(&store_dir)
.expect("failed to create local test run store"),
@ -189,7 +193,7 @@ async fn initialized(
"",
Duration::from_millis(1),
None,
store_test_support::test_blob_store_at(&store_dir),
&store_dir,
));
let inner_store = store
.create_run(&run_options.run_id)

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

@ -68,12 +68,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_at(
object_store,
"",
std::time::Duration::from_millis(1),
None,
&store_dir,
));
let state = if tokio::runtime::Handle::try_current().is_ok() {
std::thread::spawn(
@ -172,12 +173,14 @@ async fn resolve_checkpoint_text(
return Ok(current.to_string());
}
let object_store = Arc::new(LocalFileSystem::new_with_prefix(test_store_dir(run_dir))?);
let store = fabro_store::test_support::test_database(
let store_dir = test_store_dir(run_dir);
let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_dir)?);
let store = fabro_store::test_support::test_database_at(
object_store,
"",
std::time::Duration::from_millis(1),
None,
&store_dir,
);
let run = store.open_run_reader(run_id).await?;
let run_store = RunStoreHandle::from(run);

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

@ -118,12 +118,12 @@ 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(fabro_store::test_support::test_database_with_blobs(
let store = Arc::new(fabro_store::test_support::test_database_at(
object_store,
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
&store_dir,
));
let state = if tokio::runtime::Handle::try_current().is_ok() {
std::thread::spawn(
@ -249,12 +249,12 @@ fn resolve_checkpoint_text(
.enable_all()
.build()?;
let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_dir)?);
let store = Arc::new(fabro_store::test_support::test_database_with_blobs(
let store = Arc::new(fabro_store::test_support::test_database_at(
object_store,
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
&store_dir,
));
let run_id = if uses_shared_store {
run_dir
@ -8200,12 +8200,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_with_blobs(
let store = Arc::new(fabro_store::test_support::test_database_at(
Arc::new(LocalFileSystem::new_with_prefix(&store_dir).unwrap()),
"",
Duration::from_millis(1),
None,
fabro_store::test_support::test_blob_store_at(&store_dir),
&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();

20
lib/foundation/fabro-db/migrations/2026082802_run_history_activation.sql Normal file
View file

@ -0,0 +1,20 @@
CREATE TABLE legacy_run_history_activation (
singleton INTEGER PRIMARY KEY NOT NULL,
source_fingerprint BLOB NOT NULL,
source_runs INTEGER NOT NULL,
source_events INTEGER NOT NULL,
activated_at_ms INTEGER NOT NULL,
CHECK (singleton = 1),
CHECK (length(source_fingerprint) = 32),
CHECK (source_runs >= 0),
CHECK (source_events >= 0),
CHECK (activated_at_ms >= 0)
);
CREATE TABLE legacy_run_history_deletions (
run_id TEXT PRIMARY KEY NOT NULL,
deleted_at_ms INTEGER NOT NULL,
CHECK (deleted_at_ms >= 0)
);

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

@ -28,6 +28,11 @@ pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs
/// the production schema without a filesystem path into this crate.
pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026082701_run_events.sql");
/// The temporary run-history activation migration, exposed so fixtures in
/// other crates can install the production compatibility schema.
pub const RUN_HISTORY_ACTIVATION_MIGRATION_SQL: &str =
include_str!("../migrations/2026082802_run_history_activation.sql");
#[derive(Clone)]
pub struct Database {
pool: DbPool,