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

Harden SQLite activation and run history consistency

Browse source
This commit is contained in:
Scott Werner 2026-08-31 12:19:16 -04:00
parent 0574c818cf
commit c68e67c841
11 changed files with 324 additions and 135 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -58,13 +58,15 @@ See [Server Configuration](/administration/server-configuration) for the full `s
On startup, Fabro activates SQLite as the only live content-addressed blob
store before it opens routes, schedulers, workers, webhooks, reapers, or the
ready callback. The activation inventories the exact legacy SlateDB blob
prefix, checks disk headroom sized to the rows not yet imported (a warm
restart with nothing left to import only needs a small fixed headroom; on
filesystems whose free space cannot be determined the check is skipped with
a warning), imports in bounded transactions, compares every legacy blob
byte-for-byte with SQLite, runs a live SQLite integrity check, and attempts a
final WAL truncate checkpoint. A busy final truncate logs a warning and startup
continues so a later checkpoint can finish after the blocking reader exits.
prefix and run history, then checks disk headroom for the rows not yet
imported, any required blob backup, and the projected post-import database
snapshot required by run-history activation. A warm restart with no pending
imports or backups only needs a small fixed headroom; on filesystems whose free
space cannot be determined the check is skipped with a warning. Fabro then
imports in bounded transactions, compares every legacy blob byte-for-byte with
SQLite, runs a live SQLite integrity check, and attempts a final WAL truncate
checkpoint. A busy final truncate logs a warning and startup continues so a
later checkpoint can finish after the blocking reader exits.
Boots that import new rows additionally re-verify every
legacy blob against SQLite and validate every SQLite blob row independently.
Any failure stops startup. Warm boots that import no rows skip that full target
@ -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

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

@ -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<fabro_store::Database>,
pub(crate) run_history_identity: fabro_store::LegacyRunHistorySourceIdentity,
}
impl std::fmt::Debug for ActivatedBlobStorage {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ActivatedBlobStorage")
.field("run_history_identity", &self.run_history_identity)
.finish_non_exhaustive()
}
}
#[derive(Debug, thiserror::Error)]
pub(crate) enum BlobActivationError {
#[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<PathBuf>,
) -> Result<Arc<fabro_store::Database>, BlobActivationError> {
) -> Result<ActivatedBlobStorage, BlobActivationError> {
let canonical_path = fs::canonicalize(sqlite_path).await.map_err(|source| {
BlobActivationError::Canonicalize {
path: sqlite_path.to_path_buf(),
@ -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<u64, BlobActivationError> {
let half = pending_bytes
.checked_add(1)
.ok_or(BlobActivationError::DiskRequirementOverflow)?
/ 2;
let import_reserve = blob_import_reserve(pending_bytes)?;
let required_free_bytes = backup_reserve
.checked_add(pending_bytes)
.and_then(|value| value.checked_add(half))
.checked_add(import_reserve)
.and_then(|value| value.checked_add(run_history_backup_reserve))
.and_then(|value| value.checked_add(DISK_HEADROOM_BYTES))
.ok_or(BlobActivationError::DiskRequirementOverflow)?;
if available_free_bytes < required_free_bytes {
@ -227,6 +271,25 @@ fn compute_disk_preflight(
Ok(required_free_bytes)
}
fn projected_sqlite_bytes(
sqlite_bytes: u64,
pending_bytes: u64,
) -> Result<u64, BlobActivationError> {
sqlite_bytes
.checked_add(blob_import_reserve(pending_bytes)?)
.ok_or(BlobActivationError::DiskRequirementOverflow)
}
fn blob_import_reserve(pending_bytes: u64) -> Result<u64, BlobActivationError> {
let half = pending_bytes
.checked_add(1)
.ok_or(BlobActivationError::DiskRequirementOverflow)?
/ 2;
pending_bytes
.checked_add(half)
.ok_or(BlobActivationError::DiskRequirementOverflow)
}
async fn sqlite_file_set_bytes(path: &Path) -> Result<u64, BlobActivationError> {
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(())

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

@ -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::<Utc>::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<fabro_store::LegacyRunHistorySourceIdentity> {
Ok(self.store.legacy_run_history_source_identity().await?)
}
fn backup_path(&self) -> PathBuf {
fabro_db::append_to_path(&self.sqlite_path, BACKUP_SUFFIX)
}
@ -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(())
}

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

@ -773,7 +773,7 @@ where
} else {
None
};
let store = migrations::activate_blob_storage(
let blob_activation = migrations::activate_blob_storage(
&database,
&sqlite_path,
object_store,
@ -783,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

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

@ -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,
)

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

@ -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<Vec<(EventEnvelope, String)>> {
// The current-row head and event rows must come from one snapshot.
// Otherwise a concurrent append between the two SELECTs looks like
// durable corruption even though both versions are individually valid.
let mut transaction = connection.begin().await?;
let events = Self::list_events_with_json_in_transaction(&mut transaction, run_id).await?;
transaction.commit().await?;
Ok(events)
}
pub(crate) async fn list_events_with_json_in_transaction(
transaction: &mut Transaction<'_, Sqlite>,
run_id: &RunId,
) -> Result<Vec<(EventEnvelope, String)>> {
Self::list_events_with_json_in_snapshot(&mut *transaction, run_id).await
}
async fn list_events_with_json_in_snapshot(
connection: &mut SqliteConnection,
run_id: &RunId,
) -> Result<Vec<(EventEnvelope, String)>> {
let mut query = QueryBuilder::<Sqlite>::new(SELECT_EVENT_COLUMNS);
query

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

@ -150,7 +150,7 @@ impl Database {
) -> Result<RunDatabase> {
let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?;
let (envelope, cached) = run_store.commit_first_event(payload).await?;
run_store.install_in_memory_state(&envelope, &cached).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<Utc>,
) -> Result<Vec<CachedRunProjection>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.list(query, now).await)
Ok(self.projection_cache.list(query, now))
}
pub async fn list_unreadable_runs(&self) -> Result<Vec<UnreadableRun>> {
@ -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<Option<CachedRunProjection>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.get(run_id).await)
Ok(self.projection_cache.get(run_id))
}
pub async fn get_cached_projection(
@ -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<Utc>,
) -> Result<Option<Run>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.get_summary(run_id, now).await)
Ok(self.projection_cache.get_summary(run_id, now))
}
/// Run ids whose latest explicit pull request creation is still pending,
/// oldest request first.
pub async fn pending_pull_request_creation_run_ids(&self) -> Result<Vec<RunId>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.pending_pull_request_creations().await)
Ok(self.projection_cache.pending_pull_request_creations())
}
pub async fn put_session_run_index(
@ -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,

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

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

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

@ -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<BlobStore>,
pub(crate) state_lock: Mutex<()>,
projection_cache: Mutex<EventProjectionCache>,
pub(crate) state_lock: AsyncMutex<()>,
projection_cache: StdMutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
event_tx: broadcast::Sender<EventEnvelope>,
}
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<RunSummaryStore>,
) -> Result<Self> {
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<Arc<RunProjection>> {
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<Arc<RunProjection>> {
fn projected_state_locked(&self) -> Result<Arc<RunProjection>> {
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<EventEnvelope> {
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)?;

0
lib/foundation/fabro-db/migrations/2026082801_run_history_activation.sql → lib/foundation/fabro-db/migrations/2026082802_run_history_activation.sql
View file

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

@ -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 {