diff --git a/docs/public/reference/server-operations.mdx b/docs/public/reference/server-operations.mdx index ee86e37a3..812770802 100644 --- a/docs/public/reference/server-operations.mdx +++ b/docs/public/reference/server-operations.mdx @@ -58,13 +58,15 @@ See [Server Configuration](/administration/server-configuration) for the full `s On startup, Fabro activates SQLite as the only live content-addressed blob store before it opens routes, schedulers, workers, webhooks, reapers, or the ready callback. The activation inventories the exact legacy SlateDB blob -prefix, checks disk headroom sized to the rows not yet imported (a warm -restart with nothing left to import only needs a small fixed headroom; on -filesystems whose free space cannot be determined the check is skipped with -a warning), imports in bounded transactions, compares every legacy blob -byte-for-byte with SQLite, runs a live SQLite integrity check, and attempts a -final WAL truncate checkpoint. A busy final truncate logs a warning and startup -continues so a later checkpoint can finish after the blocking reader exits. +prefix and run history, then checks disk headroom for the rows not yet +imported, any required blob backup, and the projected post-import database +snapshot required by run-history activation. A warm restart with no pending +imports or backups only needs a small fixed headroom; on filesystems whose free +space cannot be determined the check is skipped with a warning. Fabro then +imports in bounded transactions, compares every legacy blob byte-for-byte with +SQLite, runs a live SQLite integrity check, and attempts a final WAL truncate +checkpoint. A busy final truncate logs a warning and startup continues so a +later checkpoint can finish after the blocking reader exits. Boots that import new rows additionally re-verify every legacy blob against SQLite and validate every SQLite blob row independently. Any failure stops startup. Warm boots that import no rows skip that full target @@ -105,8 +107,10 @@ projected row. The activation strictly validates and fingerprints the exact legacy SlateDB run-event key/value stream, imports each complete run in its own transaction, verifies every legacy history as an exact SQLite prefix, replays and verifies every SQLite run independently, and runs a full SQLite integrity -check. A source fingerprint or count change after activation stops startup. -There is no fallback or dual-read/write mode. +check. It attempts a final WAL truncate checkpoint, but a blocking reader only +produces a warning because committed activation data remains durable in the +WAL. A source fingerprint or count change after activation stops startup. There +is no fallback or dual-read/write mode. For a non-empty legacy run history, the first activation creates and validates the private sibling backup @@ -129,10 +133,17 @@ backup for at least 30 consecutive calendar days after the persisted first-success timestamp. Cleanup also requires successful cold and warm activation evidence, production observation, backup and restore validation, deletion/restart coverage, and explicit approval for a separate cleanup -change. Nothing is deleted automatically. The backup represents the database -immediately before run-history import; once the activated server accepts new -run writes, recovery is forward-only unless an operator intentionally accepts -losing those newer writes by restoring that snapshot and the older binary. +change. Nothing is deleted automatically. The run-history activation backup +represents the database immediately before run-history import and can be used +to retry or recover the activation with a binary that knows the activated +schema. It is not a binary-downgrade artifact because it already contains the +new SQL migrations. + +To return to the older binary, stop the server and restore the database's +`.pre-migration.bak` snapshot instead, then remove any `-wal` and `-shm` +siblings before starting the older binary. That snapshot was taken before the +new migrations were applied. Either recovery path loses writes accepted after +its snapshot, so make the rollback boundary explicit before restoring it. ## Submitting runs 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 c3c083502..5bc18a47b 100644 --- a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs +++ b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs @@ -15,6 +15,7 @@ use tokio::fs; use tracing::{debug, info, warn}; use crate::migrations::sqlite_activation_backup::{self, BackupError}; +use crate::migrations::sqlite_run_history_activation::BACKUP_SUFFIX as RUN_HISTORY_BACKUP_SUFFIX; use crate::server::resource_sampler; /// Earliest date this bridge becomes eligible for removal, assuming the first @@ -26,6 +27,20 @@ pub(crate) const REMOVAL_DEADLINE: &str = "2026-09-22"; const DISK_HEADROOM_BYTES: u64 = 64 * 1024 * 1024; const BACKUP_SUFFIX: &str = ".pre-blob-activation.bak"; +pub(crate) struct ActivatedBlobStorage { + pub(crate) store: Arc, + pub(crate) run_history_identity: fabro_store::LegacyRunHistorySourceIdentity, +} + +impl std::fmt::Debug for ActivatedBlobStorage { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ActivatedBlobStorage") + .field("run_history_identity", &self.run_history_identity) + .finish_non_exhaustive() + } +} + #[derive(Debug, thiserror::Error)] pub(crate) enum BlobActivationError { #[error("canonicalizing the SQLite database path {path}")] @@ -36,6 +51,8 @@ pub(crate) enum BlobActivationError { }, #[error("inventorying the legacy blob source")] Inventory(#[source] fabro_store::LegacyBlobInventoryError), + #[error("identifying legacy run history before importing blobs")] + RunHistorySourceIdentity(#[source] fabro_store::LegacyRunHistorySourceIdentityError), #[error( "activation backup is missing at {path} while {existing_rows} of {legacy_rows} legacy blob rows are already present in SQLite" )] @@ -80,7 +97,7 @@ pub(crate) async fn activate_blob_storage( slatedb_prefix: String, flush_interval: Duration, cache_path: Option, -) -> Result, BlobActivationError> { +) -> Result { let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| { BlobActivationError::Canonicalize { path: sqlite_path.to_path_buf(), @@ -109,6 +126,10 @@ pub(crate) async fn activate_blob_storage( .legacy_blob_inventory(database.pool()) .await .map_err(BlobActivationError::Inventory)?; + let run_history_identity = store + .legacy_run_history_source_identity() + .await + .map_err(BlobActivationError::RunHistorySourceIdentity)?; let backup_exists = sqlite_activation_backup::backup_exists(&backup_path).await?; if backup_exists { sqlite_activation_backup::validate_backup(&backup_path).await?; @@ -121,6 +142,15 @@ pub(crate) async fn activate_blob_storage( }); } let backup_required = inventory.rows > 0 && !backup_exists; + let run_history_backup_path = + fabro_db::append_to_path(&canonical_path, RUN_HISTORY_BACKUP_SUFFIX); + let run_history_backup_exists = + sqlite_activation_backup::backup_exists(&run_history_backup_path).await?; + if run_history_backup_exists { + sqlite_activation_backup::validate_backup(&run_history_backup_path).await?; + } + let run_history_backup_required = + run_history_identity.events != 0 && !run_history_backup_exists; // The resource sampler treats a path with no matching mount as an // unsupported-but-benign condition (tmpfs or squashfs roots, network // filesystems, an unreadable mount table), so the preflight does too: @@ -128,16 +158,25 @@ pub(crate) async fn activate_blob_storage( // could complete. if let Some(available_free_bytes) = resource_sampler::available_space_for_path(&canonical_path) { - let backup_reserve = if backup_required { + let sqlite_bytes = if backup_required || run_history_backup_required { sqlite_file_set_bytes(&canonical_path).await? } else { 0 }; + let backup_reserve = if backup_required { sqlite_bytes } else { 0 }; + let run_history_backup_reserve = if run_history_backup_required { + projected_sqlite_bytes(sqlite_bytes, inventory.pending_bytes)? + } else { + 0 + }; // Only the rows the import still has to copy need new space; rows - // already present in SQLite cost nothing on a warm restart. + // already present in SQLite cost nothing on a warm restart. Reserve + // the projected post-import database size as well when run-history + // activation will immediately take its own full SQLite snapshot. let required_free_bytes = compute_disk_preflight( inventory.pending_bytes, backup_reserve, + run_history_backup_reserve, available_free_bytes, )?; debug!( @@ -147,6 +186,9 @@ pub(crate) async fn activate_blob_storage( pending_bytes = inventory.pending_bytes, backup_required, backup_reserve, + run_history_events = run_history_identity.events, + run_history_backup_required, + run_history_backup_reserve, required_free_bytes, available_free_bytes, "Checked SQLite blob activation disk capacity" @@ -197,25 +239,27 @@ pub(crate) async fn activate_blob_storage( passive_checkpoints = import.passive_checkpoints, backup_required, backup_path = ?retained_backup, + run_history_backup_required, removal_deadline = REMOVAL_DEADLINE, "Activated SQLite blob storage" ); - Ok(store) + Ok(ActivatedBlobStorage { + store, + run_history_identity, + }) } /// Fail-closed disk capacity check; returns the required free bytes. fn compute_disk_preflight( pending_bytes: u64, backup_reserve: u64, + run_history_backup_reserve: u64, available_free_bytes: u64, ) -> Result { - let half = pending_bytes - .checked_add(1) - .ok_or(BlobActivationError::DiskRequirementOverflow)? - / 2; + let import_reserve = blob_import_reserve(pending_bytes)?; let required_free_bytes = backup_reserve - .checked_add(pending_bytes) - .and_then(|value| value.checked_add(half)) + .checked_add(import_reserve) + .and_then(|value| value.checked_add(run_history_backup_reserve)) .and_then(|value| value.checked_add(DISK_HEADROOM_BYTES)) .ok_or(BlobActivationError::DiskRequirementOverflow)?; if available_free_bytes < required_free_bytes { @@ -227,6 +271,25 @@ fn compute_disk_preflight( Ok(required_free_bytes) } +fn projected_sqlite_bytes( + sqlite_bytes: u64, + pending_bytes: u64, +) -> Result { + sqlite_bytes + .checked_add(blob_import_reserve(pending_bytes)?) + .ok_or(BlobActivationError::DiskRequirementOverflow) +} + +fn blob_import_reserve(pending_bytes: u64) -> Result { + let half = pending_bytes + .checked_add(1) + .ok_or(BlobActivationError::DiskRequirementOverflow)? + / 2; + pending_bytes + .checked_add(half) + .ok_or(BlobActivationError::DiskRequirementOverflow) +} + async fn sqlite_file_set_bytes(path: &Path) -> Result { let mut total = required_file_bytes(path).await?; for suffix in ["-wal", "-shm"] { @@ -297,7 +360,8 @@ mod tests { use super::{ BACKUP_SUFFIX, BlobActivationError, DISK_HEADROOM_BYTES, activate_blob_storage, - compute_disk_preflight, final_truncate_checkpoint, sqlite_file_set_bytes, + compute_disk_preflight, final_truncate_checkpoint, projected_sqlite_bytes, + sqlite_file_set_bytes, }; use crate::migrations::sqlite_activation_backup::{self, BackupError, create_backup}; @@ -307,14 +371,26 @@ mod tests { fn disk_preflight_passes_at_equality_and_fails_one_byte_below() { let pending_bytes = 3; let backup_reserve = 10; - let required = backup_reserve + pending_bytes + 2 + DISK_HEADROOM_BYTES; + let run_history_backup_reserve = 20; + let required = + backup_reserve + pending_bytes + 2 + run_history_backup_reserve + DISK_HEADROOM_BYTES; - let required_free_bytes = compute_disk_preflight(pending_bytes, backup_reserve, required) - .expect("exact equality must pass"); + let required_free_bytes = compute_disk_preflight( + pending_bytes, + backup_reserve, + run_history_backup_reserve, + required, + ) + .expect("exact equality must pass"); assert_eq!(required_free_bytes, required); - let error = compute_disk_preflight(pending_bytes, backup_reserve, required - 1) - .expect_err("one byte below must fail"); + let error = compute_disk_preflight( + pending_bytes, + backup_reserve, + run_history_backup_reserve, + required - 1, + ) + .expect_err("one byte below must fail"); assert!(matches!( error, BlobActivationError::InsufficientDisk { .. } @@ -324,14 +400,24 @@ mod tests { #[test] fn disk_preflight_requires_only_headroom_without_a_backup_reserve() { let required_free_bytes = - compute_disk_preflight(2, 0, u64::MAX).expect("available capacity should pass"); + compute_disk_preflight(2, 0, 0, u64::MAX).expect("available capacity should pass"); assert_eq!(required_free_bytes, 3 + DISK_HEADROOM_BYTES); } + #[test] + fn disk_preflight_reserves_the_projected_post_import_database() { + let projected = projected_sqlite_bytes(10, 3).expect("the projection should fit"); + assert_eq!(projected, 15); + + let required = compute_disk_preflight(3, 0, projected, u64::MAX) + .expect("available capacity should pass"); + assert_eq!(required, 20 + DISK_HEADROOM_BYTES); + } + #[test] fn disk_preflight_fails_closed_on_overflow() { - let error = - compute_disk_preflight(u64::MAX, 1, u64::MAX).expect_err("overflow must fail closed"); + let error = compute_disk_preflight(u64::MAX, 1, 0, u64::MAX) + .expect_err("overflow must fail closed"); assert!(matches!( error, BlobActivationError::DiskRequirementOverflow @@ -435,7 +521,7 @@ mod tests { let legacy_hash = fabro_store::test_support::put_legacy_blob(&source, legacy_bytes).await?; drop(source); - let store = activate_blob_storage( + let activation = activate_blob_storage( &database, &sqlite_path, Arc::clone(&object_store), @@ -444,6 +530,7 @@ mod tests { None, ) .await?; + let store = activation.store; assert_eq!( store.blobs().read(&legacy_hash).await?.as_deref(), Some(legacy_bytes.as_slice()) @@ -475,7 +562,7 @@ mod tests { .await?; assert_eq!(fs::read(&backup_path).await?, original_backup); assert_eq!( - warm.blobs().read(&sqlite_only_hash).await?.as_deref(), + warm.store.blobs().read(&sqlite_only_hash).await?.as_deref(), Some(sqlite_only_bytes.as_slice()) ); Ok(()) @@ -552,7 +639,7 @@ mod tests { assert!(!append_to_path(&sqlite_path, BACKUP_SUFFIX).exists()); assert_eq!( - activated.blobs().read(&hash).await?.as_deref(), + activated.store.blobs().read(&hash).await?.as_deref(), Some(bytes.as_slice()) ); Ok(()) 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 727c7dd95..faf49f52d 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 @@ -10,11 +10,11 @@ use std::path::{Path, PathBuf}; use chrono::{DateTime, Duration, Utc}; use tokio::fs; -use tracing::info; +use tracing::{info, warn}; use crate::migrations::sqlite_activation_backup::{self, BackupError}; -const BACKUP_SUFFIX: &str = ".pre-run-history-activation.bak"; +pub(crate) const BACKUP_SUFFIX: &str = ".pre-run-history-activation.bak"; const REMOVAL_WINDOW: Duration = Duration::days(30); #[derive(Clone, Debug, Eq, PartialEq)] @@ -33,8 +33,6 @@ pub(crate) enum RunHistoryActivationError { #[source] source: std::io::Error, }, - #[error("identifying the legacy run-history source")] - SourceIdentity(#[source] fabro_store::LegacyRunHistorySourceIdentityError), #[error("reading the SQLite run-history activation state")] ActivationState(#[source] sqlx::Error), #[error("the persisted run-history activation marker does not match the legacy source")] @@ -68,8 +66,6 @@ pub(crate) enum RunHistoryActivationError { InvalidActivationTimestamp, #[error("running the final SQLite WAL truncate checkpoint")] FinalCheckpoint(#[source] sqlx::Error), - #[error("the final SQLite WAL truncate checkpoint remained busy")] - FinalCheckpointBusy, #[error("a run-history activation count exceeds SQLite's integer range")] CountOverflow, } @@ -78,6 +74,7 @@ pub(crate) async fn activate_run_history( database: &fabro_db::Database, sqlite_path: &Path, store: &fabro_store::Database, + identity: &fabro_store::LegacyRunHistorySourceIdentity, ) -> Result<(), RunHistoryActivationError> { let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| { RunHistoryActivationError::Canonicalize { @@ -92,15 +89,11 @@ pub(crate) async fn activate_run_history( "Starting SQLite run-history activation" ); - let identity = store - .legacy_run_history_source_identity() - .await - .map_err(RunHistoryActivationError::SourceIdentity)?; let marker = read_activation_record(database.pool()).await?; let (target_runs, target_events) = target_counts(database.pool()).await?; if let Some(record) = &marker { - verify_marker(record, &identity)?; + verify_marker(record, identity)?; } else if identity.events == 0 && (target_runs != 0 || target_events != 0) { return Err(RunHistoryActivationError::EmptySourceWithTarget { target_runs, @@ -135,7 +128,7 @@ pub(crate) async fn activate_run_history( || Utc::now().timestamp_millis(), |record| record.activated_at_ms, ); - persist_activation_record(database.pool(), &identity, activated_at_ms).await?; + persist_activation_record(database.pool(), identity, activated_at_ms).await?; final_truncate_checkpoint(database.pool()).await?; let activated_at = DateTime::::from_timestamp_millis(activated_at_ms) @@ -266,7 +259,12 @@ async fn final_truncate_checkpoint( .await .map_err(RunHistoryActivationError::FinalCheckpoint)?; if busy != 0 { - return Err(RunHistoryActivationError::FinalCheckpointBusy); + // A concurrent reader can keep the WAL from truncating, but all + // activation data is already committed and remains durable in that + // WAL. A later checkpoint can truncate it after the reader exits. + warn!( + "The final SQLite run-history WAL truncate checkpoint could not complete; continuing startup" + ); } Ok(()) } @@ -281,6 +279,7 @@ mod tests { use fabro_types::{Graph, RunId, WorkflowSettings, test_support}; use object_store::memory::InMemory; use sqlx::Connection as _; + use tokio::fs; use ulid::Ulid; use super::{ @@ -359,6 +358,10 @@ mod tests { .await } + async fn source_identity(&self) -> TestResult { + Ok(self.store.legacy_run_history_source_identity().await?) + } + fn backup_path(&self) -> PathBuf { fabro_db::append_to_path(&self.sqlite_path, BACKUP_SUFFIX) } @@ -377,7 +380,14 @@ mod tests { .put_event(&run_id, 2, "run.submitted", serde_json::json!({})) .await?; - activate_run_history(&context.database, &context.sqlite_path, &context.store).await?; + let identity = context.source_identity().await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await?; let first_marker = read_activation_record(context.database.pool()) .await? .unwrap(); @@ -403,7 +413,13 @@ mod tests { 2 ); - activate_run_history(&context.database, &context.sqlite_path, &context.store).await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await?; assert_eq!( read_activation_record(context.database.pool()).await?, Some(first_marker) @@ -412,7 +428,7 @@ mod tests { } #[tokio::test] - async fn busy_final_checkpoint_fails_activation_closed() -> TestResult<()> { + async fn busy_final_checkpoint_warns_and_does_not_fail_activation() -> TestResult<()> { use sqlx::sqlite::{SqliteJournalMode, SqlitePoolOptions}; let context = TestContext::new("busy-final-run-checkpoint").await?; @@ -442,13 +458,12 @@ mod tests { .connect_with(checkpoint_options) .await?; - let error = final_truncate_checkpoint(&checkpoint_pool) - .await - .expect_err("a busy final truncate must fail startup closed"); - assert!(matches!( - error, - RunHistoryActivationError::FinalCheckpointBusy - )); + final_truncate_checkpoint(&checkpoint_pool).await?; + + let wal_bytes = fs::metadata(fabro_db::append_to_path(&context.sqlite_path, "-wal")) + .await? + .len(); + assert!(wal_bytes > 0, "the WAL should remain untruncated"); drop(reader); Ok(()) } @@ -458,14 +473,27 @@ 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, &context.store).await?; + let original_identity = context.source_identity().await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &original_identity, + ) + .await?; context .put_event(&run_id, 2, "run.submitted", serde_json::json!({})) .await?; - let error = activate_run_history(&context.database, &context.sqlite_path, &context.store) - .await - .expect_err("the source identity must remain stable after activation"); + let changed_identity = context.source_identity().await?; + let error = activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &changed_identity, + ) + .await + .expect_err("the source identity must remain stable after activation"); assert!(matches!(error, RunHistoryActivationError::MarkerMismatch)); assert_eq!( sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events") @@ -481,7 +509,14 @@ 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, &context.store).await?; + let identity = context.source_identity().await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await?; sqlx::query("DELETE FROM legacy_run_history_activation") .execute(context.database.pool()) .await?; @@ -496,9 +531,15 @@ mod tests { context.database.clone_pool(), )), )); - let error = activate_run_history(&context.database, &context.sqlite_path, &empty_store) - .await - .expect_err("unmarked SQLite rows cannot be adopted from an empty source"); + let empty_identity = empty_store.legacy_run_history_source_identity().await?; + let error = activate_run_history( + &context.database, + &context.sqlite_path, + &empty_store, + &empty_identity, + ) + .await + .expect_err("unmarked SQLite rows cannot be adopted from an empty source"); assert!(matches!( error, RunHistoryActivationError::EmptySourceWithTarget { .. } @@ -516,9 +557,15 @@ mod tests { .import_legacy_run_history_into(context.database.pool()) .await?; - let error = activate_run_history(&context.database, &context.sqlite_path, &context.store) - .await - .expect_err("partial import progress requires the retained backup"); + let identity = context.source_identity().await?; + let error = activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await + .expect_err("partial import progress requires the retained backup"); assert!(matches!( error, RunHistoryActivationError::MissingBackupAfterProgress { .. } @@ -529,14 +576,27 @@ 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, &context.store).await?; + let identity = context.source_identity().await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await?; let marker = read_activation_record(context.database.pool()) .await? .unwrap(); assert_eq!((marker.source_runs, marker.source_events), (0, 0)); assert!(!context.backup_path().exists()); - activate_run_history(&context.database, &context.sqlite_path, &context.store).await?; + activate_run_history( + &context.database, + &context.sqlite_path, + &context.store, + &identity, + ) + .await?; assert!(!context.backup_path().exists()); Ok(()) } diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index aa54cdba5..fe44e8a98 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -773,7 +773,7 @@ where } else { None }; - let store = migrations::activate_blob_storage( + let blob_activation = migrations::activate_blob_storage( &database, &sqlite_path, object_store, @@ -783,9 +783,15 @@ where ) .await .context("activating SQLite blob storage")?; - migrations::activate_run_history(&database, &sqlite_path, &store) - .await - .context("activating SQLite run history")?; + migrations::activate_run_history( + &database, + &sqlite_path, + &blob_activation.store, + &blob_activation.run_history_identity, + ) + .await + .context("activating SQLite run history")?; + let store = blob_activation.store; // Refresh tokens now live in SQLite. Nothing reads the old records and no // reaper collects them any more, so clear them out once rather than // leaving them in the object store forever. Pending authorization codes 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 cebbaf85f..0851bb1e2 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -1039,7 +1039,7 @@ async fn import_one_run( })?; let mut updated = *report; if has_destination { - let destination = RunSummaryStore::list_events_with_json_on_connection( + let destination = RunSummaryStore::list_events_with_json_in_transaction( &mut transaction, &history.run_id, ) diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index ace5e57fb..4750f1e21 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -9,7 +9,7 @@ use fabro_types::{ use sqlx::pool::PoolConnection; use sqlx::query::Query; use sqlx::sqlite::{SqliteArguments, SqliteConnection, SqliteRow}; -use sqlx::{QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction}; +use sqlx::{Connection as _, QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction}; use strum::VariantArray as _; use crate::run_state::projected_billing; @@ -598,6 +598,26 @@ impl RunSummaryStore { pub(crate) async fn list_events_with_json_on_connection( connection: &mut SqliteConnection, run_id: &RunId, + ) -> Result> { + // The current-row head and event rows must come from one snapshot. + // Otherwise a concurrent append between the two SELECTs looks like + // durable corruption even though both versions are individually valid. + let mut transaction = connection.begin().await?; + let events = Self::list_events_with_json_in_transaction(&mut transaction, run_id).await?; + transaction.commit().await?; + Ok(events) + } + + pub(crate) async fn list_events_with_json_in_transaction( + transaction: &mut Transaction<'_, Sqlite>, + run_id: &RunId, + ) -> Result> { + Self::list_events_with_json_in_snapshot(&mut *transaction, run_id).await + } + + async fn list_events_with_json_in_snapshot( + connection: &mut SqliteConnection, + run_id: &RunId, ) -> Result> { let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); query diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 181290053..360e06562 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -150,7 +150,7 @@ impl Database { ) -> Result { 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; + run_store.install_in_memory_state(&envelope, &cached); Self::cache_active_run(&mut active_runs, &run_store); run_store.publish(&envelope); Ok(run_store) @@ -268,7 +268,7 @@ impl Database { } } } - self.projection_cache.replace_all(entries).await; + self.projection_cache.replace_all(entries); Ok::<_, Error>(()) }) .await?; @@ -281,7 +281,7 @@ impl Database { now: DateTime, ) -> Result> { self.warm_projection_cache().await?; - Ok(self.projection_cache.list(query, now).await) + Ok(self.projection_cache.list(query, now)) } pub async fn list_unreadable_runs(&self) -> Result> { @@ -322,7 +322,7 @@ impl Database { .test_insert_unvalidated_event(run_id, seq, payload) .await?; self.active_runs.lock().await.remove(run_id); - self.projection_cache.remove(run_id).await; + self.projection_cache.remove(run_id); Ok(()) } @@ -344,7 +344,7 @@ impl Database { pub async fn get_cached_run(&self, run_id: &RunId) -> Result> { self.warm_projection_cache().await?; - Ok(self.projection_cache.get(run_id).await) + Ok(self.projection_cache.get(run_id)) } pub async fn get_cached_projection( @@ -355,7 +355,6 @@ impl Database { Ok(self .projection_cache .projection_snapshot(run_id) - .await .map(|(projection, _)| projection)) } @@ -365,14 +364,14 @@ impl Database { now: DateTime, ) -> Result> { self.warm_projection_cache().await?; - Ok(self.projection_cache.get_summary(run_id, now).await) + Ok(self.projection_cache.get_summary(run_id, now)) } /// Run ids whose latest explicit pull request creation is still pending, /// oldest request first. pub async fn pending_pull_request_creation_run_ids(&self) -> Result> { self.warm_projection_cache().await?; - Ok(self.projection_cache.pending_pull_request_creations().await) + Ok(self.projection_cache.pending_pull_request_creations()) } pub async fn put_session_run_index( @@ -398,8 +397,8 @@ impl Database { Ok(None) } - pub(crate) async fn remove_cached_run(&self, run_id: &RunId) { - self.projection_cache.remove(run_id).await; + pub(crate) fn remove_cached_run(&self, run_id: &RunId) { + self.projection_cache.remove(run_id); } pub async fn delete_run(&self, run_id: &RunId) -> Result<()> { @@ -413,7 +412,7 @@ impl Database { .delete_canonical(run_id, Utc::now().timestamp_millis()) .await?; active_runs.remove(run_id); - self.remove_cached_run(run_id).await; + self.remove_cached_run(run_id); if let Err(err) = self.delete_session_indexes_for_run(run_id).await { warn!( run_id = %run_id, diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index 6d53d2a6d..bd5171135 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -1,9 +1,8 @@ use std::collections::{BTreeSet, HashMap}; -use std::sync::Arc; +use std::sync::{Arc, Mutex, MutexGuard}; use chrono::{DateTime, Utc}; use fabro_types::{Run, RunId, RunProjection}; -use tokio::sync::Mutex; use crate::ListRunsQuery; use crate::run_state::build_summary; @@ -30,6 +29,9 @@ impl CachedRunProjection { #[derive(Debug, Default)] pub(crate) struct RunProjectionCache { + // Cache operations are bounded in-memory work and never await. Keeping + // this lock synchronous lets a committed event update both projection + // caches without introducing a cancellation point. state: Mutex, } @@ -109,21 +111,27 @@ fn apply_read_overlays(entry: &mut CachedRunProjection, now: DateTime) { } impl RunProjectionCache { - pub(crate) async fn replace_all(&self, entries: Vec) { - self.state.lock().await.replace_all(entries); + fn lock(&self) -> MutexGuard<'_, RunProjectionCacheState> { + self.state.lock().expect( + "run projection cache mutex is never poisoned: no code panics while holding this lock", + ) } - pub(crate) async fn replace(&self, entry: CachedRunProjection) { - self.state.lock().await.insert(entry); + pub(crate) fn replace_all(&self, entries: Vec) { + self.lock().replace_all(entries); } - pub(crate) async fn list( + pub(crate) fn replace(&self, entry: CachedRunProjection) { + self.lock().insert(entry); + } + + pub(crate) fn list( &self, query: &ListRunsQuery, now: DateTime, ) -> Vec { let entries = { - let state = self.state.lock().await; + let state = self.lock(); let raw = match query.parent_id { Some(parent_id) => state .children_by_parent @@ -166,8 +174,8 @@ impl RunProjectionCache { entries } - pub(crate) async fn get(&self, run_id: &RunId) -> Option { - let state = self.state.lock().await; + pub(crate) fn get(&self, run_id: &RunId) -> Option { + let state = self.lock(); state .entries .get(run_id) @@ -177,13 +185,8 @@ impl RunProjectionCache { /// Projection and last sequence for `run_id`, without the summary clone /// and children count that `get` computes under the cache mutex. - pub(crate) async fn projection_snapshot( - &self, - run_id: &RunId, - ) -> Option<(Arc, u32)> { - self.state - .lock() - .await + pub(crate) fn projection_snapshot(&self, run_id: &RunId) -> Option<(Arc, u32)> { + self.lock() .entries .get(run_id) .map(|entry| (Arc::clone(&entry.projection), entry.last_seq)) @@ -192,11 +195,9 @@ impl RunProjectionCache { /// Run ids whose latest explicit pull request creation is still pending, /// oldest request first. Clones only ids and timestamps, so callers can /// poll on an interval without materializing run summaries. - pub(crate) async fn pending_pull_request_creations(&self) -> Vec { + pub(crate) fn pending_pull_request_creations(&self) -> Vec { let mut pending = self - .state .lock() - .await .entries .values() .filter_map(|entry| { @@ -210,9 +211,9 @@ impl RunProjectionCache { pending.into_iter().map(|(_, run_id)| run_id).collect() } - pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { + pub(crate) fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { let mut entry = { - let state = self.state.lock().await; + let state = self.lock(); state .entries .get(run_id) @@ -223,7 +224,7 @@ impl RunProjectionCache { Some(entry.summary) } - pub(crate) async fn remove(&self, run_id: &RunId) { - self.state.lock().await.remove(run_id); + pub(crate) fn remove(&self, run_id: &RunId) { + self.lock().remove(run_id); } } diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index f8a57f669..cdbf92cc2 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -1,9 +1,9 @@ -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard}; use bytes::Bytes; use fabro_types::{BlobHash, RunEvent, RunId, SessionId}; use futures::Stream; -use tokio::sync::{Mutex, broadcast, mpsc}; +use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc}; use tokio_stream::wrappers::UnboundedReceiverStream; use super::projection_cache::{CachedRunProjection, RunProjectionCache}; @@ -36,13 +36,21 @@ impl std::fmt::Debug for RunDatabase { pub(crate) struct RunDatabaseInner { pub(crate) run_id: RunId, blob_store: Arc, - pub(crate) state_lock: Mutex<()>, - projection_cache: Mutex, + pub(crate) state_lock: AsyncMutex<()>, + projection_cache: StdMutex, shared_projection_cache: Arc, run_summary_store: Arc, event_tx: broadcast::Sender, } +impl RunDatabaseInner { + fn lock_projection_cache(&self) -> StdMutexGuard<'_, EventProjectionCache> { + self.projection_cache.lock().expect( + "event projection cache mutex is never poisoned: no code panics while holding this lock", + ) + } +} + impl RunDatabase { pub(crate) async fn build( run_id: RunId, @@ -52,7 +60,7 @@ impl RunDatabase { run_summary_store: Arc, ) -> Result { let projection_cache = if let Some((projection, last_seq)) = - shared_projection_cache.projection_snapshot(&run_id).await + shared_projection_cache.projection_snapshot(&run_id) { EventProjectionCache { last_seq, @@ -106,8 +114,8 @@ impl RunDatabase { inner: Arc::new(RunDatabaseInner { run_id, blob_store, - state_lock: Mutex::new(()), - projection_cache: Mutex::new(projection_cache), + state_lock: AsyncMutex::new(()), + projection_cache: StdMutex::new(projection_cache), shared_projection_cache, run_summary_store, event_tx, @@ -167,14 +175,12 @@ impl RunDatabase { async fn projected_state(&self) -> Result> { let _state_guard = self.inner.state_lock.lock().await; - self.projected_state_locked().await + self.projected_state_locked() } - async fn projected_state_locked(&self) -> Result> { + fn projected_state_locked(&self) -> Result> { self.inner - .projection_cache - .lock() - .await + .lock_projection_cache() .state .clone() .ok_or_else(|| { @@ -185,20 +191,17 @@ impl RunDatabase { }) } - pub(crate) async fn install_in_memory_state( + pub(crate) fn install_in_memory_state( &self, event: &EventEnvelope, cached: &CachedRunProjection, ) { { - let mut projection_cache = self.inner.projection_cache.lock().await; + let mut projection_cache = self.inner.lock_projection_cache(); projection_cache.state = Some(Arc::clone(&cached.projection)); projection_cache.last_seq = event.seq; } - self.inner - .shared_projection_cache - .replace(cached.clone()) - .await; + self.inner.shared_projection_cache.replace(cached.clone()); } pub(crate) fn publish(&self, event: &EventEnvelope) { @@ -212,7 +215,7 @@ impl RunDatabase { payload.validate(&self.inner.run_id)?; let event = RunEvent::try_from(payload)?; let _state_guard = self.inner.state_lock.lock().await; - if self.inner.projection_cache.lock().await.last_seq != 0 { + if self.inner.lock_projection_cache().last_seq != 0 { return Err(Error::RunAlreadyExists(self.inner.run_id.to_string())); } self.commit_event_locked(payload, event).await @@ -242,7 +245,7 @@ impl RunDatabase { payload.validate(&self.inner.run_id)?; let event = RunEvent::try_from(payload)?; let _state_guard = self.inner.state_lock.lock().await; - let projection = self.projected_state_locked().await?; + let projection = self.projected_state_locked()?; if !predicate(&projection) { return Ok(None); } @@ -270,7 +273,9 @@ impl RunDatabase { event: RunEvent, ) -> Result { let (envelope, cached) = self.commit_event_locked(payload, event).await?; - Box::pin(self.install_in_memory_state(&envelope, &cached)).await; + // Keep post-commit propagation await-free: cancellation after SQLite + // commits must not leave either cache stale or omit the broadcast. + self.install_in_memory_state(&envelope, &cached); self.publish(&envelope); Ok(envelope) } @@ -281,7 +286,7 @@ impl RunDatabase { event: RunEvent, ) -> Result<(EventEnvelope, CachedRunProjection)> { let (expected_last_seq, mut next_state) = { - let cache = self.inner.projection_cache.lock().await; + let cache = self.inner.lock_projection_cache(); (cache.last_seq, cache.state.clone()) }; let seq = run_summary_store::next_event_seq_after(expected_last_seq)?; diff --git a/lib/foundation/fabro-db/migrations/2026082801_run_history_activation.sql b/lib/foundation/fabro-db/migrations/2026082802_run_history_activation.sql similarity index 100% rename from lib/foundation/fabro-db/migrations/2026082801_run_history_activation.sql rename to lib/foundation/fabro-db/migrations/2026082802_run_history_activation.sql diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index c97a2278c..7d55de2ac 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -31,7 +31,7 @@ pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/202608270 /// The temporary run-history activation migration, exposed so fixtures in /// other crates can install the production compatibility schema. pub const RUN_HISTORY_ACTIVATION_MIGRATION_SQL: &str = - include_str!("../migrations/2026082801_run_history_activation.sql"); + include_str!("../migrations/2026082802_run_history_activation.sql"); #[derive(Clone)] pub struct Database {