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

Share activation backup helpers and trim run-history residue

Browse source 
Hoist the SQLite backup/integrity helpers duplicated between the blob and
run-history activation migrations into one shared module, drop the
write-only recent-events buffer and other dead state left behind by the
SQLite cutover, reuse existing helpers for projection bootstrap, sequence
allocation, catalog keys, and test pool construction, and collapse the
tombstone and activation-marker plumbing to single statements.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-08-28 17:20:05 -04:00
parent 3e6a88625c
commit 0574c818cf
13 changed files with 393 additions and 619 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -10,14 +10,11 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use futures_util::TryStreamExt as _;
use object_store::ObjectStore;
use sqlx::Connection as _;
use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection};
use tokio::fs;
use tokio::task::{JoinError, spawn_blocking};
use tracing::{debug, info, warn};
use crate::migrations::sqlite_activation_backup::{self, BackupError};
use crate::server::resource_sampler;
/// Earliest date this bridge becomes eligible for removal, assuming the first
@ -28,7 +25,6 @@ pub(crate) const REMOVAL_DEADLINE: &str = "2026-09-22";
const DISK_HEADROOM_BYTES: u64 = 64 * 1024 * 1024;
const BACKUP_SUFFIX: &str = ".pre-blob-activation.bak";
const STAGING_SUFFIX: &str = ".tmp";
#[derive(Debug, thiserror::Error)]
pub(crate) enum BlobActivationError {
@ -40,16 +36,6 @@ pub(crate) enum BlobActivationError {
},
#[error("inventorying the legacy blob source")]
Inventory(#[source] fabro_store::LegacyBlobInventoryError),
#[error("reading activation backup metadata at {path}")]
BackupMetadata {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("activation backup is not a regular file at {path}")]
BackupNotRegular { path: PathBuf },
#[error("activation backup permissions are not private at {path}")]
BackupNotPrivate { path: PathBuf },
#[error(
"activation backup is missing at {path} while {existing_rows} of {legacy_rows} legacy blob rows are already present in SQLite"
)]
@ -58,14 +44,6 @@ pub(crate) enum BlobActivationError {
legacy_rows: u64,
existing_rows: u64,
},
#[error("opening or checking activation backup integrity at {path}")]
BackupIntegrity {
path: PathBuf,
#[source]
source: sqlx::Error,
},
#[error("activation backup integrity check did not return exactly one ok result at {path}")]
BackupIntegrityFailed { path: PathBuf },
#[error("reading SQLite file metadata at {path}")]
SqliteMetadata {
path: PathBuf,
@ -81,16 +59,8 @@ pub(crate) enum BlobActivationError {
required_bytes: u64,
available_bytes: u64,
},
#[error("staging the pre-activation SQLite backup")]
StageBackup(#[source] fabro_db::SnapshotStagingError),
#[error("joining the activation backup publication task")]
JoinBackupPublication(#[source] JoinError),
#[error("publishing the activation backup at {path} without overwriting")]
PublishBackup {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error(transparent)]
Backup(#[from] BackupError),
#[error("importing legacy blobs into SQLite")]
Import(#[source] Box<fabro_store::LegacyBlobImportError>),
#[error("verifying legacy and SQLite blobs")]
@ -139,9 +109,9 @@ pub(crate) async fn activate_blob_storage(
.legacy_blob_inventory(database.pool())
.await
.map_err(BlobActivationError::Inventory)?;
let backup_exists = backup_exists(&backup_path).await?;
let backup_exists = sqlite_activation_backup::backup_exists(&backup_path).await?;
if backup_exists {
validate_backup(&backup_path).await?;
sqlite_activation_backup::validate_backup(&backup_path).await?;
}
if !backup_exists && inventory.pending_rows < inventory.rows {
return Err(BlobActivationError::MissingBackupAfterImport {
@ -191,7 +161,7 @@ pub(crate) async fn activate_blob_storage(
let retained_backup = if backup_exists {
Some(backup_path)
} else if backup_required {
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
Some(backup_path)
} else {
None
@ -257,17 +227,6 @@ fn compute_disk_preflight(
Ok(required_free_bytes)
}
async fn backup_exists(path: &Path) -> Result<bool, BlobActivationError> {
match fs::metadata(path).await {
Ok(_) => Ok(true),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(BlobActivationError::BackupMetadata {
path: path.to_path_buf(),
source,
}),
}
}
async fn sqlite_file_set_bytes(path: &Path) -> Result<u64, BlobActivationError> {
let mut total = required_file_bytes(path).await?;
for suffix in ["-wal", "-shm"] {
@ -301,127 +260,8 @@ async fn optional_file_bytes(path: &Path) -> Result<u64, BlobActivationError> {
}
}
async fn create_backup(
pool: &sqlx::SqlitePool,
backup_path: &Path,
) -> Result<(), BlobActivationError> {
let staging_path = fabro_db::append_to_path(backup_path, STAGING_SUFFIX);
fabro_db::write_snapshot_to_staging(pool, &staging_path)
.await
.map_err(BlobActivationError::StageBackup)?;
validate_backup(&staging_path).await?;
let publish_staging = staging_path.clone();
let publish_backup = backup_path.to_path_buf();
let already_exists = spawn_blocking(move || {
let staging = tempfile::TempPath::from_path(publish_staging);
match staging.persist_noclobber(&publish_backup) {
Ok(()) => {
// Make the rename's directory entry durable: the retained
// backup is the documented rollback artifact, so it must not
// vanish in a crash after the import has already committed.
fabro_db::sync_parent_directory(&publish_backup)?;
Ok(false)
}
Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => Ok(true),
Err(error) => Err(error.error),
}
})
.await
.map_err(BlobActivationError::JoinBackupPublication)?
.map_err(|source| BlobActivationError::PublishBackup {
path: backup_path.to_path_buf(),
source,
})?;
// The staging copy was validated just before the atomic rename, so only a
// concurrently published file still needs its own validation.
if already_exists {
debug!(
backup_path = %backup_path.display(),
"Reusing concurrently published SQLite blob activation backup"
);
validate_backup(backup_path).await?;
}
Ok(())
}
async fn validate_backup(path: &Path) -> Result<(), BlobActivationError> {
let metadata =
fs::symlink_metadata(path)
.await
.map_err(|source| BlobActivationError::BackupMetadata {
path: path.to_path_buf(),
source,
})?;
if !metadata.is_file() {
return Err(BlobActivationError::BackupNotRegular {
path: path.to_path_buf(),
});
}
validate_private_permissions(path, &metadata)?;
let options = SqliteConnectOptions::new()
.filename(path)
.read_only(true)
.immutable(true)
.create_if_missing(false);
let mut connection = SqliteConnection::connect_with(&options)
.await
.map_err(|source| BlobActivationError::BackupIntegrity {
path: path.to_path_buf(),
source,
})?;
let ok = integrity_check_is_ok(&mut connection)
.await
.map_err(|source| BlobActivationError::BackupIntegrity {
path: path.to_path_buf(),
source,
})?;
if !ok {
return Err(BlobActivationError::BackupIntegrityFailed {
path: path.to_path_buf(),
});
}
Ok(())
}
/// Returns whether `PRAGMA integrity_check` reports exactly one `ok` row.
async fn integrity_check_is_ok<'a, E>(executor: E) -> Result<bool, sqlx::Error>
where
E: sqlx::Executor<'a, Database = sqlx::Sqlite>,
{
let mut rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check").fetch(executor);
let first = rows.try_next().await?;
let second = rows.try_next().await?;
Ok(first.as_deref() == Some("ok") && second.is_none())
}
#[cfg(unix)]
fn validate_private_permissions(
path: &Path,
metadata: &std::fs::Metadata,
) -> Result<(), BlobActivationError> {
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(BlobActivationError::BackupNotPrivate {
path: path.to_path_buf(),
});
}
Ok(())
}
#[cfg(not(unix))]
fn validate_private_permissions(
_path: &Path,
_metadata: &std::fs::Metadata,
) -> Result<(), BlobActivationError> {
Ok(())
}
async fn validate_live_integrity(pool: &sqlx::SqlitePool) -> Result<(), BlobActivationError> {
let ok = integrity_check_is_ok(pool)
let ok = sqlite_activation_backup::integrity_check_is_ok(pool)
.await
.map_err(BlobActivationError::LiveIntegrity)?;
if !ok {
@ -457,9 +297,9 @@ mod tests {
use super::{
BACKUP_SUFFIX, BlobActivationError, DISK_HEADROOM_BYTES, activate_blob_storage,
compute_disk_preflight, create_backup, final_truncate_checkpoint, sqlite_file_set_bytes,
validate_backup,
compute_disk_preflight, final_truncate_checkpoint, sqlite_file_set_bytes,
};
use crate::migrations::sqlite_activation_backup::{self, BackupError, create_backup};
type TestResult<T> = Result<T, Box<dyn std::error::Error>>;
@ -519,8 +359,8 @@ mod tests {
database.migrate().await?;
let backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX);
create_backup(database.pool(), &backup_path).await?;
validate_backup(&backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::validate_backup(&backup_path).await?;
assert!(backup_path.is_file());
assert!(!append_to_path(&backup_path, "-wal").exists());
@ -543,7 +383,7 @@ mod tests {
let database = fabro_db::Database::connect(&sqlite_path).await?;
database.migrate().await?;
let backup_path = append_to_path(&sqlite_path, BACKUP_SUFFIX);
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
let original = fs::read(&backup_path).await?;
sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
@ -551,7 +391,7 @@ mod tests {
.bind(b"later".as_slice())
.execute(database.pool())
.await?;
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
assert_eq!(fs::read(&backup_path).await?, original);
Ok(())
@ -572,7 +412,7 @@ mod tests {
assert!(matches!(
error,
BlobActivationError::StageBackup(fabro_db::SnapshotStagingError::Write { .. })
BackupError::Stage(fabro_db::SnapshotStagingError::Write { .. })
));
assert!(!backup_path.exists());
Ok(())
@ -806,8 +646,9 @@ mod tests {
.expect_err("an invalid retained backup must fail closed");
assert!(matches!(
error,
BlobActivationError::BackupIntegrity { .. }
| BlobActivationError::BackupIntegrityFailed { .. }
BlobActivationError::Backup(
BackupError::Integrity { .. } | BackupError::IntegrityFailed { .. }
)
));
let destination_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs")
.fetch_one(database.pool())

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

@ -7,18 +7,14 @@
//! 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};
use tracing::info;
use crate::migrations::sqlite_activation_backup::{self, BackupError};
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)]
@ -52,38 +48,12 @@ pub(crate) enum RunHistoryActivationError {
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(transparent)]
Backup(#[from] BackupError),
#[error("importing legacy run history into SQLite")]
Import(#[source] Box<fabro_store::LegacyRunHistoryImportError>),
#[error("verifying legacy and SQLite run history")]
@ -107,8 +77,8 @@ pub(crate) enum RunHistoryActivationError {
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> {
store: &fabro_store::Database,
) -> Result<(), RunHistoryActivationError> {
let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| {
RunHistoryActivationError::Canonicalize {
path: sqlite_path.to_path_buf(),
@ -138,17 +108,17 @@ pub(crate) async fn activate_run_history(
});
}
let backup_exists = backup_exists(&backup_path).await?;
if backup_exists {
validate_backup(&backup_path).await?;
let backup_present = sqlite_activation_backup::backup_exists(&backup_path).await?;
if backup_present {
sqlite_activation_backup::validate_backup(&backup_path).await?;
}
let import_progress = target_events != 0 || marker.is_some();
if identity.events != 0 && import_progress && !backup_exists {
if identity.events != 0 && import_progress && !backup_present {
return Err(RunHistoryActivationError::MissingBackupAfterProgress { path: backup_path });
}
let backup_required = identity.events != 0 && !backup_exists;
let backup_required = identity.events != 0 && !backup_present;
if backup_required {
create_backup(database.pool(), &backup_path).await?;
sqlite_activation_backup::create_backup(database.pool(), &backup_path).await?;
}
let import = store
@ -165,11 +135,10 @@ pub(crate) async fn activate_run_history(
|| 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)?;
persist_activation_record(database.pool(), &identity, activated_at_ms).await?;
final_truncate_checkpoint(database.pool()).await?;
let activated_at = DateTime::<Utc>::from_timestamp_millis(persisted.activated_at_ms)
let activated_at = DateTime::<Utc>::from_timestamp_millis(activated_at_ms)
.ok_or(RunHistoryActivationError::InvalidActivationTimestamp)?;
let removal_eligible_at = activated_at + REMOVAL_WINDOW;
info!(
@ -191,7 +160,7 @@ pub(crate) async fn activate_run_history(
removal_eligible_at = %removal_eligible_at,
"Activated SQLite run history"
);
Ok(store)
Ok(())
}
async fn read_activation_record(
@ -254,15 +223,13 @@ async fn persist_activation_record(
pool: &sqlx::SqlitePool,
identity: &fabro_store::LegacyRunHistorySourceIdentity,
activated_at_ms: i64,
) -> Result<ActivationRecord, RunHistoryActivationError> {
) -> Result<(), RunHistoryActivationError> {
let source_runs =
i64::try_from(identity.runs).map_err(|_| RunHistoryActivationError::CountOverflow)?;
let source_events =
i64::try_from(identity.events).map_err(|_| RunHistoryActivationError::CountOverflow)?;
let mut transaction = pool
.begin()
.await
.map_err(RunHistoryActivationError::PersistMarker)?;
// A pre-existing marker was already verified against `identity` above, so
// leaving it untouched on conflict keeps the original activation time.
sqlx::query(
r"
INSERT INTO legacy_run_history_activation (
@ -275,142 +242,14 @@ ON CONFLICT(singleton) DO NOTHING
.bind(source_runs)
.bind(source_events)
.bind(activated_at_ms)
.execute(&mut *transaction)
.execute(pool)
.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)
let ok = sqlite_activation_backup::integrity_check_is_ok(pool)
.await
.map_err(RunHistoryActivationError::LiveIntegrity)?;
if !ok {
@ -538,12 +377,7 @@ mod tests {
.put_event(&run_id, 2, "run.submitted", serde_json::json!({}))
.await?;
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.await?;
activate_run_history(&context.database, &context.sqlite_path, &context.store).await?;
let first_marker = read_activation_record(context.database.pool())
.await?
.unwrap();
@ -569,12 +403,7 @@ mod tests {
2
);
activate_run_history(
&context.database,
&context.sqlite_path,
Arc::clone(&context.store),
)
.await?;
activate_run_history(&context.database, &context.sqlite_path, &context.store).await?;
assert_eq!(
read_activation_record(context.database.pool()).await?,
Some(first_marker)
@ -629,23 +458,14 @@ mod tests {
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?;
activate_run_history(&context.database, &context.sqlite_path, &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");
let error = activate_run_history(&context.database, &context.sqlite_path, &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")
@ -661,12 +481,7 @@ mod tests {
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?;
activate_run_history(&context.database, &context.sqlite_path, &context.store).await?;
sqlx::query("DELETE FROM legacy_run_history_activation")
.execute(context.database.pool())
.await?;
@ -681,7 +496,7 @@ mod tests {
context.database.clone_pool(),
)),
));
let error = activate_run_history(&context.database, &context.sqlite_path, empty_store)
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!(
@ -701,13 +516,9 @@ mod tests {
.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");
let error = activate_run_history(&context.database, &context.sqlite_path, &context.store)
.await
.expect_err("partial import progress requires the retained backup");
assert!(matches!(
error,
RunHistoryActivationError::MissingBackupAfterProgress { .. }
@ -718,24 +529,14 @@ mod tests {
#[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?;
activate_run_history(&context.database, &context.sqlite_path, &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?;
activate_run_history(&context.database, &context.sqlite_path, &context.store).await?;
assert!(!context.backup_path().exists());
Ok(())
}

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

@ -0,0 +1,186 @@
//! Pre-activation backup and integrity helpers shared by the SQLite
//! activation bridges.
//!
//! Every activation snapshots the live database to a private, integrity
//! checked backup file before importing legacy data, and re-validates any
//! backup it finds on a later start. This module owns that mechanism so the
//! blob and run-history bridges cannot drift apart.
use std::path::{Path, PathBuf};
use futures_util::TryStreamExt as _;
use sqlx::Connection as _;
use sqlx::sqlite::{SqliteConnectOptions, SqliteConnection};
use tokio::fs;
use tokio::task::{JoinError, spawn_blocking};
use tracing::debug;
const STAGING_SUFFIX: &str = ".tmp";
#[derive(Debug, thiserror::Error)]
pub(crate) enum BackupError {
#[error("reading activation backup metadata at {path}")]
Metadata {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("activation backup is not a regular file at {path}")]
NotRegular { path: PathBuf },
#[error("activation backup permissions are not private at {path}")]
NotPrivate { path: PathBuf },
#[error("opening or checking activation backup integrity at {path}")]
Integrity {
path: PathBuf,
#[source]
source: sqlx::Error,
},
#[error("activation backup integrity check did not return exactly one ok result at {path}")]
IntegrityFailed { path: PathBuf },
#[error("staging the pre-activation SQLite backup")]
Stage(#[source] fabro_db::SnapshotStagingError),
#[error("joining the activation backup publication task")]
JoinPublication(#[source] JoinError),
#[error("publishing the activation backup at {path} without overwriting")]
Publish {
path: PathBuf,
#[source]
source: std::io::Error,
},
}
pub(crate) async fn backup_exists(path: &Path) -> Result<bool, BackupError> {
match fs::metadata(path).await {
Ok(_) => Ok(true),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(BackupError::Metadata {
path: path.to_path_buf(),
source,
}),
}
}
/// Snapshots `pool` to `backup_path` without overwriting an existing file.
///
/// The snapshot is staged beside the target, validated, and then published
/// with an atomic no-clobber rename. A backup that another process published
/// concurrently is validated in place instead.
pub(crate) async fn create_backup(
pool: &sqlx::SqlitePool,
backup_path: &Path,
) -> Result<(), BackupError> {
let staging_path = fabro_db::append_to_path(backup_path, STAGING_SUFFIX);
fabro_db::write_snapshot_to_staging(pool, &staging_path)
.await
.map_err(BackupError::Stage)?;
validate_backup(&staging_path).await?;
let publish_staging = staging_path.clone();
let publish_backup = backup_path.to_path_buf();
let already_exists = spawn_blocking(move || {
let staging = tempfile::TempPath::from_path(publish_staging);
match staging.persist_noclobber(&publish_backup) {
Ok(()) => {
// Make the rename's directory entry durable: the retained
// backup is the documented rollback artifact, so it must not
// vanish in a crash after the import has already committed.
fabro_db::sync_parent_directory(&publish_backup)?;
Ok(false)
}
Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => Ok(true),
Err(error) => Err(error.error),
}
})
.await
.map_err(BackupError::JoinPublication)?
.map_err(|source| BackupError::Publish {
path: backup_path.to_path_buf(),
source,
})?;
// The staging copy was validated just before the atomic rename, so only a
// concurrently published file still needs its own validation.
if already_exists {
debug!(
backup_path = %backup_path.display(),
"Reusing concurrently published SQLite activation backup"
);
validate_backup(backup_path).await?;
}
Ok(())
}
/// Requires `path` to be a private regular file holding a SQLite database
/// whose `PRAGMA integrity_check` passes.
pub(crate) async fn validate_backup(path: &Path) -> Result<(), BackupError> {
let metadata = fs::symlink_metadata(path)
.await
.map_err(|source| BackupError::Metadata {
path: path.to_path_buf(),
source,
})?;
if !metadata.is_file() {
return Err(BackupError::NotRegular {
path: path.to_path_buf(),
});
}
validate_private_permissions(path, &metadata)?;
let options = SqliteConnectOptions::new()
.filename(path)
.read_only(true)
.immutable(true)
.create_if_missing(false);
let mut connection = SqliteConnection::connect_with(&options)
.await
.map_err(|source| BackupError::Integrity {
path: path.to_path_buf(),
source,
})?;
let ok = integrity_check_is_ok(&mut connection)
.await
.map_err(|source| BackupError::Integrity {
path: path.to_path_buf(),
source,
})?;
if !ok {
return Err(BackupError::IntegrityFailed {
path: path.to_path_buf(),
});
}
Ok(())
}
/// Returns whether `PRAGMA integrity_check` reports exactly one `ok` row.
pub(crate) async fn integrity_check_is_ok<'a, E>(executor: E) -> Result<bool, sqlx::Error>
where
E: sqlx::Executor<'a, Database = sqlx::Sqlite>,
{
let mut rows = sqlx::query_scalar::<_, String>("PRAGMA integrity_check").fetch(executor);
let first = rows.try_next().await?;
let second = rows.try_next().await?;
Ok(first.as_deref() == Some("ok") && second.is_none())
}
#[cfg(unix)]
fn validate_private_permissions(
path: &Path,
metadata: &std::fs::Metadata,
) -> Result<(), BackupError> {
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(BackupError::NotPrivate {
path: path.to_path_buf(),
});
}
Ok(())
}
#[cfg(not(unix))]
fn validate_private_permissions(
_path: &Path,
_metadata: &std::fs::Metadata,
) -> Result<(), BackupError> {
Ok(())
}

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

@ -7,6 +7,8 @@ use fabro_vault::SecretStore;
mod legacy_vault_entries;
#[path = "../migrations/2026052501_optional_server_env_secrets_to_vault.rs"]
mod optional_server_env_secrets_to_vault;
#[path = "../migrations/sqlite_activation_backup.rs"]
mod sqlite_activation_backup;
#[path = "../migrations/2026082301_sqlite_blob_activation.rs"]
mod sqlite_blob_activation;
#[path = "../migrations/2026082801_sqlite_run_history_activation.rs"]

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

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

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

@ -60,6 +60,31 @@ pub(crate) fn run_events_prefix(run_id: &RunId) -> SlateKey {
.into_prefix()
}
/// Prefix of the retired `runs/_index/by-start/<run_id>` catalog markers that
/// the legacy layout kept beside each run's events.
pub(crate) fn run_catalog_prefix() -> SlateKey {
run_catalog_root().into_prefix()
}
#[cfg(test)]
pub(crate) fn run_catalog_key(run_id: &RunId) -> SlateKey {
run_catalog_root().with(run_id)
}
/// Extracts the run id from a full catalog marker key, or `None` when the key
/// is not exactly `runs/_index/by-start/<run_id>`.
pub(crate) fn parse_run_catalog_key(raw: &str) -> Option<RunId> {
let segments = SlateKey::segments(raw).collect::<Vec<_>>();
let ["runs", "_index", "by-start", run_id] = segments.as_slice() else {
return None;
};
run_id.parse().ok()
}
fn run_catalog_root() -> SlateKey {
SlateKey::new("runs").with("_index").with("by-start")
}
// Sequence keys zero-pad `seq` to six digits so lexicographic key order
// matches numeric seq order through `MAX_EVENT_SEQ`. Seek-based event listing
// (`run_events_range`) depends on this invariant, so event allocation rejects

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

@ -89,6 +89,12 @@ pub struct LegacyRunHistorySourceIdentityError {
failure: LegacyRunHistorySourceFailure,
}
impl From<LegacyRunHistorySourceFailure> for LegacyRunHistorySourceIdentityError {
fn from(failure: LegacyRunHistorySourceFailure) -> Self {
Self { failure }
}
}
impl fmt::Debug for LegacyRunHistorySourceIdentityError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
@ -287,8 +293,6 @@ 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")]
@ -585,37 +589,21 @@ impl Database {
) -> 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 source = LegacyRunHistorySource::open(self).await?;
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;
};
while let Some(history) = source.next_run(None).await? {
runs = runs
.checked_add(1)
.ok_or_else(|| LegacyRunHistorySourceIdentityError {
failure: LegacyRunHistorySourceFailure::CounterOverflow,
})?;
.ok_or(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,
})?;
hash_source_part(&mut hasher, &event.raw_key)?;
hash_source_part(&mut hasher, event.event_json.as_bytes())?;
events = events
.checked_add(1)
.ok_or(LegacyRunHistorySourceFailure::CounterOverflow)?;
}
}
Ok(LegacyRunHistorySourceIdentity {
@ -666,16 +654,12 @@ impl Database {
let tombstones = legacy_run_history_tombstones(pool)
.await
.map_err(LegacyRunHistoryImportFailure::DeletionState)?;
require_tombstoned_destinations_absent(pool, &tombstones)
if tombstoned_destination_present(pool)
.await
.map_err(|error| match error {
TombstoneCheckError::Sqlite(source) => {
LegacyRunHistoryImportFailure::DeletionState(source)
}
TombstoneCheckError::DestinationPresent => {
LegacyRunHistoryImportFailure::TombstonedDestinationPresent
}
})?;
.map_err(LegacyRunHistoryImportFailure::DeletionState)?
{
return Err(LegacyRunHistoryImportFailure::TombstonedDestinationPresent);
}
if !activated {
discard_projection_only_rows(pool, report).await?;
}
@ -739,22 +723,15 @@ 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)
if tombstoned_destination_present(pool)
.await
.map_err(|error| match error {
TombstoneCheckError::Sqlite(source) => {
LegacyRunHistoryVerificationFailure::DeletionState(source)
}
TombstoneCheckError::DestinationPresent => {
LegacyRunHistoryVerificationFailure::TombstonedDestinationPresent
}
})?;
.map_err(LegacyRunHistoryVerificationFailure::DeletionState)?
{
return Err(LegacyRunHistoryVerificationFailure::TombstonedDestinationPresent);
}
let mut source_ids = HashSet::new();
let mut source = LegacyRunHistorySource::open(self)
.await
@ -839,12 +816,7 @@ impl Database {
.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(),
)
.scan_prefix(keys::run_catalog_prefix())
.await
.map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?;
while let Some(entry) = catalog
@ -854,14 +826,9 @@ impl Database {
{
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)?,
keys::parse_run_catalog_key(key)
.ok_or(LegacyRunHistoryDiagnosticsFailure::InvalidCatalogKey)?,
);
}
let mut diagnostics = LegacyRunHistoryDiagnostics {
@ -1005,35 +972,19 @@ async fn legacy_run_history_tombstones(pool: &SqlitePool) -> Result<HashSet<Stri
)
}
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"
/// Returns whether any tombstoned run still has canonical SQLite rows.
async fn tombstoned_destination_present(pool: &SqlitePool) -> Result<bool, sqlx::Error> {
sqlx::query_scalar(
r"
SELECT EXISTS(
SELECT 1 FROM runs WHERE id = ?
UNION ALL
SELECT 1 FROM run_events WHERE run_id = ?
SELECT 1 FROM legacy_run_history_deletions AS deletion
WHERE EXISTS(SELECT 1 FROM runs WHERE id = deletion.run_id)
OR EXISTS(SELECT 1 FROM run_events WHERE run_id = deletion.run_id)
)
",
)
.bind(run_id)
.bind(run_id)
.fetch_one(pool)
.await
.map_err(TombstoneCheckError::Sqlite)?;
if destination_present {
return Err(TombstoneCheckError::DestinationPresent);
}
}
Ok(())
)
.fetch_one(pool)
.await
}
async fn discard_projection_only_rows(
@ -1596,23 +1547,9 @@ mod tests {
context.put_event(&first, 1, 10, &first_created).await?;
context.put_event(&first, 4, 40, &first_submitted).await?;
context.put_event(&second, 1, 20, &second_created).await?;
context.put_raw(keys::run_catalog_key(&first), b"").await?;
context
.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"",
)
.put_raw(keys::run_catalog_key(&empty_marker), b"")
.await?;
context
.source

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

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

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

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

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

@ -1002,7 +1002,7 @@ fn sql_limit(limit: usize) -> i64 {
i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX)
}
fn next_event_seq_after(last_seq: u32) -> Result<u32> {
pub(crate) fn next_event_seq_after(last_seq: u32) -> Result<u32> {
last_seq
.checked_add(1)
.filter(|seq| *seq <= keys::MAX_EVENT_SEQ)
@ -1032,11 +1032,7 @@ fn decode_event_rows_with_json(
) -> Result<Vec<(EventEnvelope, String)>> {
let run_id_text = run_id.to_string();
rows.iter()
.map(|row| {
let event_json: String = row.try_get("event_json")?;
let envelope = decode_event_row(row, run_id, &run_id_text)?;
Ok((envelope, event_json))
})
.map(|row| decode_event_row_with_json(row, run_id, &run_id_text))
.collect()
}
@ -1072,6 +1068,17 @@ fn decode_event_row(
expected_run_id: &RunId,
expected_run_id_text: &str,
) -> Result<EventEnvelope> {
decode_event_row_with_json(row, expected_run_id, expected_run_id_text)
.map(|(envelope, _event_json)| envelope)
}
/// Decodes one event row and returns the raw `event_json` alongside it so
/// callers that need both do not fetch the column twice.
fn decode_event_row_with_json(
row: &SqliteRow,
expected_run_id: &RunId,
expected_run_id_text: &str,
) -> Result<(EventEnvelope, String)> {
let stored_run_id: String = row.try_get("run_id")?;
let raw_seq: i64 = row.try_get("seq")?;
let seq = stored_seq(raw_seq).ok_or_else(|| run_event_mismatch(expected_run_id, 0, "seq"))?;
@ -1113,7 +1120,7 @@ fn decode_event_row(
return Err(run_event_mismatch(expected_run_id, seq, "session_id"));
}
Ok(EventEnvelope { seq, event })
Ok((EventEnvelope { seq, event }, event_json))
}
async fn select_run_head(connection: &mut SqliteConnection, run_id: &RunId) -> Result<Option<u32>> {

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

@ -14,7 +14,7 @@ use projection_cache::RunProjectionCache;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, OnceCell};
use tokio::sync::{Mutex, MutexGuard, OnceCell};
use tracing::warn;
use crate::{
@ -148,17 +148,7 @@ impl Database {
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 (mut active_runs, run_store) = self.reserve_new_run(run_id).await?;
let (envelope, cached) = run_store.commit_first_event(payload).await?;
run_store.install_in_memory_state(&envelope, &cached).await;
Self::cache_active_run(&mut active_runs, &run_store);
@ -170,8 +160,23 @@ impl Database {
/// 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> {
let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?;
Self::cache_active_run(&mut active_runs, &run_store);
Ok(run_store)
}
/// Builds an empty handle for a run that exists neither in memory nor in
/// SQLite, returning the held `active_runs` guard so the caller can
/// register the handle before any concurrent creator observes the gap.
async fn reserve_new_run(
&self,
run_id: &RunId,
) -> Result<(
MutexGuard<'_, HashMap<RunId, Arc<RunDatabaseInner>>>,
RunDatabase,
)> {
self.warm_projection_cache().await?;
let mut active_runs = self.active_runs.lock().await;
let active_runs = self.active_runs.lock().await;
if active_runs.contains_key(run_id) || self.run_summary_store.contains(run_id).await? {
return Err(Error::RunAlreadyExists(run_id.to_string()));
}
@ -181,8 +186,7 @@ impl Database {
Arc::clone(&self.projection_cache),
self.run_summary_store(),
);
Self::cache_active_run(&mut active_runs, &run_store);
Ok(run_store)
Ok((active_runs, run_store))
}
pub async fn open_run(&self, run_id: &RunId) -> Result<RunDatabase> {

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

@ -1,4 +1,3 @@
use std::collections::VecDeque;
use std::sync::Arc;
use bytes::Bytes;
@ -11,10 +10,12 @@ use super::projection_cache::{CachedRunProjection, RunProjectionCache};
use crate::run_state::{EventProjectionCache, RunProjectionReducer};
use crate::{
BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId,
keys,
run_summary_store,
};
const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024;
/// Broadcast capacity for live event subscribers; a lagging subscriber refills
/// from SQLite.
const EVENT_BROADCAST_CAPACITY: usize = 1024;
#[derive(Clone)]
pub struct RunDatabase {
@ -39,8 +40,6 @@ pub(crate) struct RunDatabaseInner {
projection_cache: Mutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
recent_events: Mutex<VecDeque<EventEnvelope>>,
recent_event_limit: usize,
event_tx: broadcast::Sender<EventEnvelope>,
}
@ -60,13 +59,12 @@ impl RunDatabase {
state: Some(projection),
}
} else {
let events = run_summary_store.list_events_for_run(&run_id).await?;
let last_seq = events.last().map(|event| event.seq).ok_or_else(|| {
Error::InvalidEvent(format!("run {run_id} has no run.created event"))
})?;
let cached = Self::build_cached_projection(&run_summary_store, &run_id)
.await?
.ok_or_else(|| Error::RunNotFound(run_id.to_string()))?;
EventProjectionCache {
last_seq,
state: Some(Arc::new(RunProjection::apply_events(&events)?)),
last_seq: cached.last_seq,
state: Some(cached.projection),
}
};
Ok(Self::from_projection_cache(
@ -103,7 +101,7 @@ impl RunDatabase {
run_summary_store: Arc<RunSummaryStore>,
projection_cache: EventProjectionCache,
) -> Self {
let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16));
let (event_tx, _) = broadcast::channel(EVENT_BROADCAST_CAPACITY);
Self {
inner: Arc::new(RunDatabaseInner {
run_id,
@ -112,8 +110,6 @@ impl RunDatabase {
projection_cache: Mutex::new(projection_cache),
shared_projection_cache,
run_summary_store,
recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)),
recent_event_limit: DEFAULT_EVENT_TAIL_LIMIT,
event_tx,
}),
read_only,
@ -154,10 +150,11 @@ impl RunDatabase {
store: &RunSummaryStore,
run_id: &RunId,
) -> Result<Option<CachedRunProjection>> {
if !store.contains(run_id).await? {
return Ok(None);
}
let events = store.list_events_for_run(run_id).await?;
let events = match store.list_events_for_run(run_id).await {
Ok(events) => events,
Err(Error::RunNotFound(_)) => return Ok(None),
Err(error) => return Err(error),
};
let last_seq = events
.last()
.map(|event| event.seq)
@ -202,12 +199,6 @@ impl RunDatabase {
.shared_projection_cache
.replace(cached.clone())
.await;
let mut recent_events = self.inner.recent_events.lock().await;
recent_events.push_back(event.clone());
while recent_events.len() > self.inner.recent_event_limit {
recent_events.pop_front();
}
}
pub(crate) fn publish(&self, event: &EventEnvelope) {
@ -293,7 +284,7 @@ impl RunDatabase {
let cache = self.inner.projection_cache.lock().await;
(cache.last_seq, cache.state.clone())
};
let seq = next_event_seq(expected_last_seq)?;
let seq = run_summary_store::next_event_seq_after(expected_last_seq)?;
let prospective = EventEnvelope { seq, event };
apply_cached_projection_event(&mut next_state, &prospective).map_err(event_rejected)?;
let next_projection =
@ -459,9 +450,6 @@ async fn refill_from_sql(
}
};
for event in events {
if event.seq < *next_seq {
continue;
}
*next_seq = event.seq.saturating_add(1);
if sender.send(Ok(event)).is_err() {
return false;
@ -476,20 +464,6 @@ fn event_rejected(error: Error) -> Error {
}
}
fn next_event_seq(last_seq: u32) -> Result<u32> {
let seq = last_seq
.checked_add(1)
.ok_or(Error::EventSequenceExhausted {
max_seq: keys::MAX_EVENT_SEQ,
})?;
if seq > keys::MAX_EVENT_SEQ {
return Err(Error::EventSequenceExhausted {
max_seq: keys::MAX_EVENT_SEQ,
});
}
Ok(seq)
}
fn apply_cached_projection_event(
state: &mut Option<Arc<RunProjection>>,
event: &EventEnvelope,

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

@ -85,32 +85,11 @@ pub fn test_run_summary_store_path(store_dir: &Path) -> PathBuf {
/// siblings) when they reset the directory itself.
#[must_use]
pub fn test_blob_store_at(store_dir: &Path) -> Arc<BlobStore> {
let options = SqliteConnectOptions::new()
.filename(test_blob_store_path(store_dir))
.create_if_missing(true)
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
Box::pin(async move {
let installed: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM sqlite_master \
WHERE type = 'table' AND name = 'blobs')",
)
.fetch_one(&mut *connection)
.await?;
if !installed {
sqlx::query(fabro_db::BLOBS_MIGRATION_SQL)
.execute(&mut *connection)
.await?;
}
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(BlobStore::new(pool))
Arc::new(BlobStore::new(lazy_file_pool(
test_blob_store_path(store_dir),
"blobs",
&[fabro_db::BLOBS_MIGRATION_SQL],
)))
}
/// Returns a durable SQLite run-history authority stored beside `store_dir`.
@ -120,36 +99,54 @@ pub fn test_blob_store_at(store_dir: &Path) -> Arc<BlobStore> {
/// production.
#[must_use]
pub fn test_run_summary_store_at(store_dir: &Path) -> Arc<RunSummaryStore> {
Arc::new(RunSummaryStore::new(lazy_file_pool(
test_run_summary_store_path(store_dir),
"runs",
&[
fabro_db::RUNS_MIGRATION_SQL,
fabro_db::RUN_EVENTS_MIGRATION_SQL,
fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
],
)))
}
/// Builds a single-connection file-backed SQLite pool that installs
/// `migrations` the first time it opens a database without `probe_table`.
///
/// Like [`lazy_in_memory_pool`], the pool connects lazily so synchronous
/// fixture builders stay synchronous, and the file persists across handles so
/// reopen-style tests share one authority.
fn lazy_file_pool(
path: PathBuf,
probe_table: &'static str,
migrations: &'static [&'static str],
) -> sqlx::SqlitePool {
let options = SqliteConnectOptions::new()
.filename(test_run_summary_store_path(store_dir))
.filename(path)
.create_if_missing(true)
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
SqlitePoolOptions::new()
.max_connections(1)
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
.after_connect(move |connection, _metadata| {
Box::pin(async move {
let installed: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM sqlite_master \
WHERE type = 'table' AND name = 'runs')",
WHERE type = 'table' AND name = ?)",
)
.bind(probe_table)
.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?;
for migration in migrations {
sqlx::raw_sql(*migration).execute(&mut *connection).await?;
}
}
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(RunSummaryStore::new(pool))
.connect_lazy_with(options)
}
/// Builds a test database whose SQLite blob and run-history authorities are