diff --git a/lib/components/fabro-store/src/keys.rs b/lib/components/fabro-store/src/keys.rs index da76f982f..6e0feb7c1 100644 --- a/lib/components/fabro-store/src/keys.rs +++ b/lib/components/fabro-store/src/keys.rs @@ -120,6 +120,11 @@ pub(crate) fn sessions_by_id_prefix() -> SlateKey { SlateKey::new("sessions").with("by-id").into_prefix() } +#[cfg(test)] +pub(crate) fn session_by_id_key(session_id: &fabro_types::SessionId) -> SlateKey { + SlateKey::new("sessions").with("by-id").with(session_id) +} + // --- Parsing --- #[cfg(test)] @@ -186,9 +191,7 @@ mod tests { // Sibling namespaces of the same run sort outside the range. assert!(!contains(&SlateKey::new("runs").with(run_id).with("state"))); assert!(!contains( - &SlateKey::new("sessions") - .with("by-id") - .with(fabro_types::SessionId::new()) + &session_by_id_key(&fabro_types::SessionId::new()) )); } 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 deb882e56..dfad82b55 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -1574,7 +1574,7 @@ mod tests { .await?; context .put_raw( - SlateKey::new("sessions").with("by-id").with(session_id), + keys::session_by_id_key(&session_id), b"opaque legacy reverse row", ) .await?; diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index d4623a652..b9907de18 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -3,8 +3,8 @@ use std::sync::LazyLock; use chrono::{DateTime, Utc}; use fabro_types::{ - BilledTokenCounts, EventBody, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, - RunTiming, SessionId, StageId, timing, + BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming, + SessionId, StageId, timing, }; use sqlx::pool::PoolConnection; use sqlx::query::Query; @@ -436,17 +436,12 @@ ON CONFLICT(singleton) DO NOTHING } pub(crate) async fn find_session_owner(&self, session_id: &SessionId) -> Result> { - let row = sqlx::query( - r" -SELECT run_id, seq, event_name, node_id, stage_id, session_id, event_json -FROM run_events -WHERE session_id = ? - AND event_name = 'run.session.created' -", - ) - .bind(session_id.to_string()) - .fetch_optional(&self.pool) - .await?; + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE session_id = ") + .push_bind(session_id.to_string()) + .push(" AND event_name = 'run.session.created'"); + let row = query.build().fetch_optional(&self.pool).await?; let Some(row) = row else { return Ok(None); }; @@ -455,21 +450,14 @@ WHERE session_id = ? let run_id = stored_run_id .parse::() .map_err(|_| Error::RunEventMismatch { - run_id: "".to_string(), + run_id: stored_run_id.clone(), seq: 0, field: "run_id", })?; - let envelope = decode_event_row(&row, &run_id, &stored_run_id)?; - if envelope.event.run_id != run_id { - return Err(run_event_mismatch(&run_id, envelope.seq, "run_id")); - } - let requested_session_id = session_id.to_string(); - if envelope.event.session_id.as_deref() != Some(requested_session_id.as_str()) { - return Err(run_event_mismatch(&run_id, envelope.seq, "session_id")); - } - if !matches!(envelope.event.body, EventBody::RunSessionCreated(_)) { - return Err(run_event_mismatch(&run_id, envelope.seq, "event_name")); - } + // The WHERE clause pins the row's session_id and event_name columns + // to the requested values, and decoding verifies the envelope against + // every stored column, so a successful decode proves ownership. + decode_event_row(&row, &run_id, &stored_run_id)?; Ok(Some(run_id)) } diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 747eab940..9e92eb894 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -888,11 +888,7 @@ mod tests { Some(first_id) ); - let legacy_key = keys::SlateKey::new("sessions") - .with("by-id") - .with(session_id) - .as_ref() - .to_vec(); + let legacy_key = keys::session_by_id_key(&session_id).as_ref().to_vec(); let legacy = store.open_db().await.unwrap(); assert!(legacy.get(&legacy_key).await.unwrap().is_none()); legacy diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index c3a8f0ab3..a921ec7eb 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -73,10 +73,11 @@ impl Database { } pub async fn migrate(&self) -> anyhow::Result<()> { - self.preflight_session_owner_index() + let applied = applied_migration_versions(&self.pool).await?; + self.preflight_session_owner_index(&applied) .await .context("checking session ownership before SQLite migrations")?; - self.snapshot_before_new_migrations() + self.snapshot_before_new_migrations(&applied) .await .context("snapshotting SQLite database before migrations")?; MIGRATOR @@ -88,7 +89,16 @@ impl Database { /// Refuse the unique owner index when old event history contains /// collisions. The diagnostic is deliberately count-only because session /// identifiers and event contents are not safe startup-log fields. - async fn preflight_session_owner_index(&self) -> anyhow::Result<()> { + /// + /// Temporary compatibility guard: once every supported database has + /// applied the session-owner index migration the version check below + /// always short-circuits, and this preflight can be deleted along with + /// the run-history compatibility window. + async fn preflight_session_owner_index(&self, applied: &HashSet) -> anyhow::Result<()> { + if applied.contains(&SESSION_OWNER_INDEX_MIGRATION_VERSION) { + return Ok(()); + } + let run_events_exists: bool = sqlx::query_scalar( "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'run_events')", ) @@ -99,11 +109,6 @@ impl Database { return Ok(()); } - let applied = applied_migration_versions(&self.pool).await?; - if applied.contains(&SESSION_OWNER_INDEX_MIGRATION_VERSION) { - return Ok(()); - } - let collision_groups: i64 = sqlx::query_scalar( r" SELECT COUNT(*) @@ -145,8 +150,7 @@ FROM ( /// from immediately before the most recent schema change. Failing to /// write the snapshot fails the migration: no rollback artifact, no /// schema change. - async fn snapshot_before_new_migrations(&self) -> anyhow::Result<()> { - let applied = applied_migration_versions(&self.pool).await?; + async fn snapshot_before_new_migrations(&self, applied: &HashSet) -> anyhow::Result<()> { let has_pending = MIGRATOR .iter() .any(|migration| !applied.contains(&migration.version));