scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Activate atomic SQLite run history storage

Browse source
This commit is contained in:
Scott Werner 2026-08-28 15:18:25 -04:00
parent 9edcc780d6
commit f073684227
30 changed files with 2144 additions and 1789 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.

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

@ -96,6 +96,44 @@ 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. 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 backup represents the database
immediately before run-history import; once the activated server accepts new
run writes, recovery is forward-only unless an operator intentionally accepts
losing those newer writes by restoring that snapshot and the older binary.
## Submitting runs
Workflows are submitted via the REST API and executed in the background. The exact request body is documented in the API reference:

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

@ -0,0 +1,742 @@
//! 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 std::sync::Arc;
use chrono::{DateTime, Duration, Utc};
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, info};
const BACKUP_SUFFIX: &str = ".pre-run-history-activation.bak";
const STAGING_SUFFIX: &str = ".tmp";
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("identifying the legacy run-history source")]
SourceIdentity(#[source] fabro_store::LegacyRunHistorySourceIdentityError),
#[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("reading run-history activation backup metadata at {path}")]
BackupMetadata {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("run-history activation backup is not a regular file at {path}")]
BackupNotRegular { path: PathBuf },
#[error("run-history activation backup permissions are not private at {path}")]
BackupNotPrivate { path: PathBuf },
#[error(
"run-history activation backup is missing at {path} after SQLite import progress was recorded"
)]
MissingBackupAfterProgress { path: PathBuf },
#[error("opening or checking run-history activation backup integrity at {path}")]
BackupIntegrity {
path: PathBuf,
#[source]
source: sqlx::Error,
},
#[error("run-history activation backup integrity check failed at {path}")]
BackupIntegrityFailed { path: PathBuf },
#[error("staging the pre-activation SQLite backup")]
StageBackup(#[source] fabro_db::SnapshotStagingError),
#[error("joining the run-history backup publication task")]
JoinBackupPublication(#[source] JoinError),
#[error("publishing the run-history activation backup at {path} without overwriting")]
PublishBackup {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[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("the final SQLite WAL truncate checkpoint remained busy")]
FinalCheckpointBusy,
#[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: Arc<fabro_store::Database>,
) -> Result<Arc<fabro_store::Database>, 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 identity = store
.legacy_run_history_source_identity()
.await
.map_err(RunHistoryActivationError::SourceIdentity)?;
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_exists = backup_exists(&backup_path).await?;
if backup_exists {
validate_backup(&backup_path).await?;
}
let import_progress = target_events != 0 || marker.is_some();
if identity.events != 0 && import_progress && !backup_exists {
return Err(RunHistoryActivationError::MissingBackupAfterProgress { path: backup_path });
}
let backup_required = identity.events != 0 && !backup_exists;
if backup_required {
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,
);
let persisted = persist_activation_record(database.pool(), &identity, activated_at_ms).await?;
verify_marker(&persisted, &identity)?;
final_truncate_checkpoint(database.pool()).await?;
let activated_at = DateTime::<Utc>::from_timestamp_millis(persisted.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(store)
}
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<ActivationRecord, 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)?;
let mut transaction = pool
.begin()
.await
.map_err(RunHistoryActivationError::PersistMarker)?;
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(&mut *transaction)
.await
.map_err(RunHistoryActivationError::PersistMarker)?;
transaction
.commit()
.await
.map_err(RunHistoryActivationError::PersistMarker)?;
read_activation_record(pool)
.await?
.ok_or(RunHistoryActivationError::InvalidMarker)
}
async fn backup_exists(path: &Path) -> Result<bool, RunHistoryActivationError> {
match fs::metadata(path).await {
Ok(_) => Ok(true),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(RunHistoryActivationError::BackupMetadata {
path: path.to_path_buf(),
source,
}),
}
}
async fn create_backup(
pool: &sqlx::SqlitePool,
backup_path: &Path,
) -> Result<(), RunHistoryActivationError> {
let staging_path = fabro_db::append_to_path(backup_path, STAGING_SUFFIX);
fabro_db::write_snapshot_to_staging(pool, &staging_path)
.await
.map_err(RunHistoryActivationError::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(()) => {
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(RunHistoryActivationError::JoinBackupPublication)?
.map_err(|source| RunHistoryActivationError::PublishBackup {
path: backup_path.to_path_buf(),
source,
})?;
if already_exists {
debug!(
backup_path = %backup_path.display(),
"Reusing concurrently published SQLite run-history activation backup"
);
validate_backup(backup_path).await?;
}
Ok(())
}
async fn validate_backup(path: &Path) -> Result<(), RunHistoryActivationError> {
let metadata = fs::symlink_metadata(path).await.map_err(|source| {
RunHistoryActivationError::BackupMetadata {
path: path.to_path_buf(),
source,
}
})?;
if !metadata.is_file() {
return Err(RunHistoryActivationError::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| RunHistoryActivationError::BackupIntegrity {
path: path.to_path_buf(),
source,
})?;
let ok = integrity_check_is_ok(&mut connection)
.await
.map_err(|source| RunHistoryActivationError::BackupIntegrity {
path: path.to_path_buf(),
source,
})?;
if !ok {
return Err(RunHistoryActivationError::BackupIntegrityFailed {
path: path.to_path_buf(),
});
}
Ok(())
}
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<(), RunHistoryActivationError> {
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(RunHistoryActivationError::BackupNotPrivate {
path: path.to_path_buf(),
});
}
Ok(())
}
#[cfg(not(unix))]
fn validate_private_permissions(
_path: &Path,
_metadata: &std::fs::Metadata,
) -> Result<(), RunHistoryActivationError> {
Ok(())
}
async fn validate_live_integrity(pool: &sqlx::SqlitePool) -> Result<(), RunHistoryActivationError> {
let ok = 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 {
return Err(RunHistoryActivationError::FinalCheckpointBusy);
}
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 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
}
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?;
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.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,
Arc::clone(&context.store),
)
.await?;
assert_eq!(
read_activation_record(context.database.pool()).await?,
Some(first_marker)
);
Ok(())
}
#[tokio::test]
async fn busy_final_checkpoint_fails_activation_closed() -> 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?;
let error = final_truncate_checkpoint(&checkpoint_pool)
.await
.expect_err("a busy final truncate must fail startup closed");
assert!(matches!(
error,
RunHistoryActivationError::FinalCheckpointBusy
));
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?;
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.await?;
context
.put_event(&run_id, 2, "run.submitted", serde_json::json!({}))
.await?;
let error = activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.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?;
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.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 error = activate_run_history(&context.database, &context.sqlite_path, empty_store)
.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 error = activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.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?;
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.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,
Arc::clone(&context.store),
)
.await?;
assert!(!context.backup_path().exists());
Ok(())
}
}

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

@ -9,10 +9,13 @@ mod legacy_vault_entries;
mod optional_server_env_secrets_to_vault;
#[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 =

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

@ -783,6 +783,9 @@ where
)
.await
.context("activating SQLite blob storage")?;
let store = migrations::activate_run_history(&database, &sqlite_path, store)
.await
.context("activating SQLite run history")?;
// 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 +832,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 {

17
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();
@ -7617,10 +7616,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}"

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/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}")]

9
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)
@ -73,6 +71,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 +82,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 +101,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"

366
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,80 @@ 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 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 +209,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 +227,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 +287,12 @@ impl fmt::Debug for LegacyRunHistoryImportFailure {
enum LegacyRunHistoryVerificationFailure {
#[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("acquiring a SQLite verification connection")]
AcquireConnection(#[source] sqlx::Error),
#[error("reading a destination run history")]
@ -306,7 +380,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 +448,7 @@ struct ValidatedLegacyRunEvent {
payload: EventPayload,
envelope: EventEnvelope,
event_json: String,
raw_key: Vec<u8>,
}
struct ValidatedLegacyRunHistory {
@ -500,6 +579,52 @@ 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
.map_err(|failure| LegacyRunHistorySourceIdentityError { failure })?;
let mut hasher = Sha256::new();
hasher.update(DOMAIN_SEPARATOR);
let mut runs = 0_u64;
let mut events = 0_u64;
loop {
let Some(history) = source
.next_run(None)
.await
.map_err(|failure| LegacyRunHistorySourceIdentityError { failure })?
else {
break;
};
runs = runs
.checked_add(1)
.ok_or_else(|| LegacyRunHistorySourceIdentityError {
failure: LegacyRunHistorySourceFailure::CounterOverflow,
})?;
for event in &history.events {
hash_source_part(&mut hasher, &event.raw_key)
.map_err(|failure| LegacyRunHistorySourceIdentityError { failure })?;
hash_source_part(&mut hasher, event.event_json.as_bytes())
.map_err(|failure| LegacyRunHistorySourceIdentityError { failure })?;
events =
events
.checked_add(1)
.ok_or_else(|| LegacyRunHistorySourceIdentityError {
failure: 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 +660,25 @@ 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)?;
require_tombstoned_destinations_absent(pool, &tombstones)
.await
.map_err(|error| match error {
TombstoneCheckError::Sqlite(source) => {
LegacyRunHistoryImportFailure::DeletionState(source)
}
TombstoneCheckError::DestinationPresent => {
LegacyRunHistoryImportFailure::TombstonedDestinationPresent
}
})?;
if !activated {
discard_projection_only_rows(pool, report).await?;
}
let mut source = LegacyRunHistorySource::open(self)
.await
.map_err(LegacyRunHistoryImportFailure::Source)?;
@ -553,7 +696,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 +739,22 @@ impl Database {
pool: &SqlitePool,
report: &mut LegacyRunHistoryVerificationReport,
) -> Result<(), LegacyRunHistoryVerificationFailure> {
legacy_run_history_is_activated(pool)
.await
.map_err(LegacyRunHistoryVerificationFailure::ActivationState)?;
let tombstones = legacy_run_history_tombstones(pool)
.await
.map_err(LegacyRunHistoryVerificationFailure::DeletionState)?;
require_tombstoned_destinations_absent(pool, &tombstones)
.await
.map_err(|error| match error {
TombstoneCheckError::Sqlite(source) => {
LegacyRunHistoryVerificationFailure::DeletionState(source)
}
TombstoneCheckError::DestinationPresent => {
LegacyRunHistoryVerificationFailure::TombstonedDestinationPresent
}
})?;
let mut source_ids = HashSet::new();
let mut source = LegacyRunHistorySource::open(self)
.await
@ -606,6 +773,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 +833,42 @@ 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(
SlateKey::new("runs")
.with("_index")
.with("by-start")
.into_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)?;
let segments = SlateKey::segments(key).collect::<Vec<_>>();
let ["runs", "_index", "by-start", run_id] = segments.as_slice() else {
return Err(LegacyRunHistoryDiagnosticsFailure::InvalidCatalogKey);
};
catalog_ids.push(
run_id
.parse::<RunId>()
.map_err(|_| 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 +965,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 +987,55 @@ 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(),
)
}
enum TombstoneCheckError {
Sqlite(sqlx::Error),
DestinationPresent,
}
async fn require_tombstoned_destinations_absent(
pool: &SqlitePool,
tombstones: &HashSet<String>,
) -> Result<(), TombstoneCheckError> {
for run_id in tombstones {
let destination_present: bool = sqlx::query_scalar(
r"
SELECT EXISTS(
SELECT 1 FROM runs WHERE id = ?
UNION ALL
SELECT 1 FROM run_events WHERE run_id = ?
)
",
)
.bind(run_id)
.bind(run_id)
.fetch_one(pool)
.await
.map_err(TombstoneCheckError::Sqlite)?;
if destination_present {
return Err(TombstoneCheckError::DestinationPresent);
}
}
Ok(())
}
async fn discard_projection_only_rows(
pool: &SqlitePool,
report: &mut LegacyRunHistoryImportReport,
@ -812,6 +1068,7 @@ async fn import_one_run(
pool: &SqlitePool,
controls: &ImportControls,
history: ValidatedLegacyRunHistory,
activated: bool,
report: &mut LegacyRunHistoryImportReport,
) -> Result<(), LegacyRunHistoryImportFailure> {
#[cfg(not(test))]
@ -849,6 +1106,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 +1271,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 +1291,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 +1445,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 +1596,23 @@ 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
.source
.catalog_index()
.await?
.add(&empty_marker)
.put_raw(
SlateKey::new("runs")
.with("_index")
.with("by-start")
.with(first.to_string()),
b"",
)
.await?;
context
.put_raw(
SlateKey::new("runs")
.with("_index")
.with("by-start")
.with(empty_marker.to_string()),
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)]

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

@ -7,8 +7,7 @@
//! - [`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.
//! rather than exposing `Repository<R>` directly.
mod codec;
mod record_id;

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

@ -77,7 +77,7 @@ use crate::{Error, Result, keys};
/// stores.
///
/// This type is intentionally `pub(crate)`: callers should interact through a
/// named store such as `RunCatalogIndex` or `BlobStore`, which can add
/// named store such as `BlobStore`, which can add
/// domain-specific behavior on top of the generic storage primitives here.
pub(crate) struct Repository<R: Record> {
db: Arc<Db>,

246
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::{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,132 @@ 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<()> {
let mut transaction = self.begin().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 +539,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,
@ -953,6 +1150,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,

428
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 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,50 @@ impl Database {
.await
}
pub async fn create_run_with_first_event(
&self,
run_id: &RunId,
payload: &EventPayload,
) -> Result<RunDatabase> {
self.warm_projection_cache().await?;
let mut 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(),
);
let (envelope, cached) = run_store.commit_first_event(payload).await?;
run_store.install_in_memory_state(&envelope, &cached).await;
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;
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(),
);
Self::cache_active_run(&mut active_runs, &run_store);
Ok(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 +199,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 +208,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 +216,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 +247,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,7 +264,6 @@ impl Database {
}
}
}
self.run_summary_store.reconcile(&entries).await?;
self.projection_cache.replace_all(entries).await;
Ok::<_, Error>(())
})
@ -256,15 +281,10 @@ impl Database {
}
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 +313,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).await;
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(
@ -364,28 +399,24 @@ impl Database {
}
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 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?;
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).await;
self.run_summary_store.delete(run_id).await?;
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"
);
}
Ok(())
}
@ -407,17 +438,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 +483,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 +901,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 +1084,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 +1098,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 +1126,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 +1158,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 +1553,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 +1580,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 +1635,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 +1812,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 +1863,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 +1896,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

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]
);
}
}

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

File diff suppressed because it is too large Load diff

79
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
@ -105,6 +113,65 @@ pub fn test_blob_store_at(store_dir: &Path) -> Arc<BlobStore> {
Arc::new(BlobStore::new(pool))
}
/// 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> {
let options = SqliteConnectOptions::new()
.filename(test_run_summary_store_path(store_dir))
.create_if_missing(true)
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
Box::pin(async move {
let installed: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM sqlite_master \
WHERE type = 'table' AND name = 'runs')",
)
.fetch_one(&mut *connection)
.await?;
if !installed {
for migration in [
fabro_db::RUNS_MIGRATION_SQL,
fabro_db::RUN_EVENTS_MIGRATION_SQL,
fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
] {
sqlx::raw_sql(migration).execute(&mut *connection).await?;
}
}
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(RunSummaryStore::new(pool))
}
/// 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.
#[must_use]
pub fn test_database(
@ -192,6 +259,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/2026082801_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/2026082801_run_history_activation.sql");
#[derive(Clone)]
pub struct Database {
pool: DbPool,