diff --git a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs index 494da96ac..c3c083502 100644 --- a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs +++ b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs @@ -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), #[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 { - 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 { 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 { } } -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 -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 = Result>; @@ -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()) diff --git a/lib/apps/fabro-server/migrations/2026082801_sqlite_run_history_activation.rs b/lib/apps/fabro-server/migrations/2026082801_sqlite_run_history_activation.rs index 1899e4dc1..727c7dd95 100644 --- a/lib/apps/fabro-server/migrations/2026082801_sqlite_run_history_activation.rs +++ b/lib/apps/fabro-server/migrations/2026082801_sqlite_run_history_activation.rs @@ -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), #[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, -) -> Result, 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::::from_timestamp_millis(persisted.activated_at_ms) + let activated_at = DateTime::::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 { +) -> 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 { - 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 -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(()) } diff --git a/lib/apps/fabro-server/migrations/sqlite_activation_backup.rs b/lib/apps/fabro-server/migrations/sqlite_activation_backup.rs new file mode 100644 index 000000000..d7ba2422a --- /dev/null +++ b/lib/apps/fabro-server/migrations/sqlite_activation_backup.rs @@ -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 { + 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 +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(()) +} diff --git a/lib/apps/fabro-server/src/migrations.rs b/lib/apps/fabro-server/src/migrations.rs index 6278607bf..5c89015d0 100644 --- a/lib/apps/fabro-server/src/migrations.rs +++ b/lib/apps/fabro-server/src/migrations.rs @@ -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"] diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index c2eb6bcd2..aa54cdba5 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -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 diff --git a/lib/components/fabro-store/src/keys.rs b/lib/components/fabro-store/src/keys.rs index 0360c2af2..1f5d93a76 100644 --- a/lib/components/fabro-store/src/keys.rs +++ b/lib/components/fabro-store/src/keys.rs @@ -60,6 +60,31 @@ pub(crate) fn run_events_prefix(run_id: &RunId) -> SlateKey { .into_prefix() } +/// Prefix of the retired `runs/_index/by-start/` 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/`. +pub(crate) fn parse_run_catalog_key(raw: &str) -> Option { + let segments = SlateKey::segments(raw).collect::>(); + 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 diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs index f4c438218..cebbaf85f 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -89,6 +89,12 @@ pub struct LegacyRunHistorySourceIdentityError { failure: LegacyRunHistorySourceFailure, } +impl From 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 { 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::>(); - let ["runs", "_index", "by-start", run_id] = segments.as_slice() else { - return Err(LegacyRunHistoryDiagnosticsFailure::InvalidCatalogKey); - }; catalog_ids.push( - run_id - .parse::() - .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, -) -> 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 { + 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 diff --git a/lib/components/fabro-store/src/record/mod.rs b/lib/components/fabro-store/src/record/mod.rs index d29e4d8d8..6482eb17c 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -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` 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; diff --git a/lib/components/fabro-store/src/record/repository.rs b/lib/components/fabro-store/src/record/repository.rs index c83b5de99..f851f6a64 100644 --- a/lib/components/fabro-store/src/record/repository.rs +++ b/lib/components/fabro-store/src/record/repository.rs @@ -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 { db: Arc, _record: PhantomData, diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 2deff15e5..ace5e57fb 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -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 { +pub(crate) fn next_event_seq_after(last_seq: u32) -> Result { last_seq .checked_add(1) .filter(|seq| *seq <= keys::MAX_EVENT_SEQ) @@ -1032,11 +1032,7 @@ fn decode_event_rows_with_json( ) -> Result> { 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 { + 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> { diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 776980d3b..181290053 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -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 { - 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 { + 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>>, + 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 { diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index 099bc102f..f8a57f669 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -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, shared_projection_cache: Arc, run_summary_store: Arc, - recent_events: Mutex>, - recent_event_limit: usize, event_tx: broadcast::Sender, } @@ -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, 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> { - 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 { - 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>, event: &EventEnvelope, diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index 803bf9167..b651ca096 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -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 { - 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 { /// production. #[must_use] pub fn test_run_summary_store_at(store_dir: &Path) -> Arc { + 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