From a98c0db7134d6543b0fa395db8c7a2e1048201ac Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Thu, 27 Aug 2026 17:27:47 -0400 Subject: [PATCH] Import and verify legacy run history --- .../src/legacy_run_history_import.rs | 1817 +++++++++++++++++ lib/components/fabro-store/src/lib.rs | 5 + .../fabro-store/src/run_summary_store.rs | 190 +- 3 files changed, 2010 insertions(+), 2 deletions(-) create mode 100644 lib/components/fabro-store/src/legacy_run_history_import.rs diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs new file mode 100644 index 000000000..2756068f4 --- /dev/null +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -0,0 +1,1817 @@ +//! Temporary compatibility importer for legacy SlateDB run history. +//! +//! Keep this source reader, its reports, and its verification path through the +//! 30-day run-history compatibility window. Remove them only after the +//! production evidence gate for that window has been accepted. + +use std::collections::HashSet; +use std::error::Error as StdError; +use std::fmt; + +use fabro_types::{EventEnvelope, RunEvent, RunId, RunProjection}; +use sqlx::SqlitePool; +#[cfg(test)] +use tokio::sync::Barrier; +use tracing::debug; + +use crate::keys::SlateKey; +use crate::slate::CachedRunProjection; +use crate::{Database, EventPayload, ListRunsQuery, RunProjectionReducer, RunSummaryStore, keys}; + +/// Count-only observations about the legacy catalog and session indexes. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryDiagnostics { + pub catalog_markers: u64, + pub empty_catalog_markers: u64, + pub session_reverse_rows: u64, +} + +/// Durable progress from one legacy run-history import attempt. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryImportReport { + pub scanned_source_runs: u64, + pub scanned_source_events: u64, + pub imported_runs: u64, + pub imported_events: u64, + pub verified_existing_runs: u64, + pub verified_existing_events: u64, + pub discarded_projection_only_rows: u64, + pub committed_run_transactions: u64, + pub diagnostics: LegacyRunHistoryDiagnostics, +} + +/// Aggregate proof from legacy-prefix and full-SQL-destination verification. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct LegacyRunHistoryVerificationReport { + pub source_runs: u64, + pub source_events: u64, + pub matched_prefix_runs: u64, + pub matched_prefix_events: u64, + pub target_runs: u64, + pub target_events: u64, + pub sql_only_runs: u64, + pub sql_only_events: u64, + pub diagnostics: LegacyRunHistoryDiagnostics, +} + +/// An import failure plus the durable progress completed before it. +pub struct LegacyRunHistoryImportError { + report: LegacyRunHistoryImportReport, + failure: LegacyRunHistoryImportFailure, +} + +impl LegacyRunHistoryImportError { + #[must_use] + pub fn report(&self) -> &LegacyRunHistoryImportReport { + &self.report + } +} + +impl fmt::Debug for LegacyRunHistoryImportError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryImportError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyRunHistoryImportError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy run-history import failed after scanning {} runs and committing {} run transactions: {}", + self.report.scanned_source_runs, self.report.committed_run_transactions, self.failure + ) + } +} + +impl StdError for LegacyRunHistoryImportError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(self.failure.primary_failure()) + } +} + +/// A verification failure plus the aggregate proof completed before it. +pub struct LegacyRunHistoryVerificationError { + report: LegacyRunHistoryVerificationReport, + failure: LegacyRunHistoryVerificationFailure, +} + +impl LegacyRunHistoryVerificationError { + #[must_use] + pub fn report(&self) -> &LegacyRunHistoryVerificationReport { + &self.report + } +} + +impl fmt::Debug for LegacyRunHistoryVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryVerificationError") + .field("report", &self.report) + .field("failure", &self.failure.kind()) + .finish() + } +} + +impl fmt::Display for LegacyRunHistoryVerificationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "legacy run-history verification failed after checking {} source runs and {} target runs: {}", + self.report.source_runs, self.report.target_runs, self.failure + ) + } +} + +impl StdError for LegacyRunHistoryVerificationError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.failure) + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryImportFailure { + #[error("reading and validating the legacy run-history source")] + Source(#[source] LegacyRunHistorySourceFailure), + #[error("starting the projection-only row cleanup transaction")] + BeginCleanup(#[source] sqlx::Error), + #[error("deleting projection-only run rows")] + DeleteProjectionOnlyRows(#[source] sqlx::Error), + #[error("committing the projection-only row cleanup")] + CommitCleanup(#[source] sqlx::Error), + #[error("starting a per-run import transaction")] + BeginRunTransaction(#[source] sqlx::Error), + #[error("reading an existing destination run history")] + ReadDestination(#[source] crate::Error), + #[error("the destination history is partial or conflicts with the legacy prefix")] + DestinationConflict, + #[error("replaying an existing destination run history")] + ReplayDestination(#[source] crate::Error), + #[error("verifying an existing destination run row")] + VerifyDestination(#[source] crate::Error), + #[error("inserting the imported final run row")] + InsertRun(#[source] crate::Error), + #[error("inserting an imported run event")] + InsertEvent(#[source] crate::Error), + #[error("committing a per-run import transaction")] + CommitRunTransaction(#[source] sqlx::Error), + #[error("rolling back a failed per-run import transaction")] + RollbackRunTransaction { + #[source] + source: sqlx::Error, + prior: Box, + }, + #[error("collecting count-only legacy index diagnostics")] + Diagnostics(#[source] LegacyRunHistoryDiagnosticsFailure), + #[error("a legacy run-history import counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryImportFailure { + fn kind(&self) -> &'static str { + self.into() + } + + fn primary_failure(&self) -> &Self { + match self { + Self::RollbackRunTransaction { prior, .. } => prior.primary_failure(), + _ => self, + } + } +} + +impl fmt::Debug for LegacyRunHistoryImportFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut debug = formatter.debug_struct("LegacyRunHistoryImportFailure"); + debug.field("kind", &self.kind()); + if let Self::RollbackRunTransaction { prior, .. } = self { + debug.field("prior_failure", &prior.kind()); + } + debug.finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryVerificationFailure { + #[error("reading and validating the legacy run-history source")] + Source(#[source] LegacyRunHistorySourceFailure), + #[error("acquiring a SQLite verification connection")] + AcquireConnection(#[source] sqlx::Error), + #[error("reading a destination run history")] + ReadDestination(#[source] crate::Error), + #[error("SQLite is missing all or part of a legacy run-history prefix")] + MissingDestinationPrefix, + #[error("a destination run-history prefix conflicts with legacy JSON or sequence identity")] + DestinationPrefixConflict, + #[error("enumerating destination run rows")] + ListDestinationRuns(#[source] sqlx::Error), + #[error("a destination run row has an invalid identity")] + InvalidDestinationRunId, + #[error("a destination run has no event history")] + EmptyDestinationHistory, + #[error("replaying a destination run history")] + ReplayDestination(#[source] crate::Error), + #[error("verifying a destination run row")] + VerifyDestination(#[source] crate::Error), + #[error("collecting count-only legacy index diagnostics")] + Diagnostics(#[source] LegacyRunHistoryDiagnosticsFailure), + #[error("a legacy run-history verification counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryVerificationFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistoryVerificationFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryVerificationFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistorySourceFailure { + #[error("opening the legacy run-history source")] + OpenSource(#[source] crate::Error), + #[error("opening the legacy run-history scan")] + OpenScan(#[source] slatedb::Error), + #[error("reading the legacy run-history scan")] + ReadScan(#[source] slatedb::Error), + #[error("a legacy run-event key is not UTF-8")] + KeyUtf8(#[source] std::str::Utf8Error), + #[error("a legacy run-event key is not canonical")] + InvalidKey, + #[error("a legacy run-event value is not UTF-8")] + ValueUtf8(#[source] std::str::Utf8Error), + #[error("a legacy run-event value is not valid JSON")] + DecodeEvent(#[source] serde_json::Error), + #[error("a legacy run-event value does not match its key or event contract")] + ValidateEvent(#[source] crate::Error), + #[error("a legacy run history has invalid sequence ordering")] + InvalidSequence, + #[error("a legacy run history does not begin with sequence 1 run.created")] + InvalidFirstEvent, + #[error("replaying a legacy run history")] + Replay(#[source] crate::Error), + #[error("a legacy run-history source counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistorySourceFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistorySourceFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistorySourceFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(strum::IntoStaticStr, thiserror::Error)] +#[strum(serialize_all = "snake_case")] +enum LegacyRunHistoryDiagnosticsFailure { + #[error("reading the legacy run catalog")] + ReadCatalog(#[source] crate::Error), + #[error("opening the legacy run source for diagnostics")] + OpenSource(#[source] crate::Error), + #[error("opening a legacy run-event probe")] + OpenEventProbe(#[source] slatedb::Error), + #[error("reading a legacy run-event probe")] + ReadEventProbe(#[source] slatedb::Error), + #[error("opening the legacy session reverse-row scan")] + OpenSessionScan(#[source] slatedb::Error), + #[error("reading the legacy session reverse-row scan")] + ReadSessionScan(#[source] slatedb::Error), + #[error("a legacy run-history diagnostics counter overflowed")] + CounterOverflow, +} + +impl LegacyRunHistoryDiagnosticsFailure { + fn kind(&self) -> &'static str { + self.into() + } +} + +impl fmt::Debug for LegacyRunHistoryDiagnosticsFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LegacyRunHistoryDiagnosticsFailure") + .field("kind", &self.kind()) + .finish() + } +} + +#[derive(Default)] +struct ImportControls { + #[cfg(test)] + source_after_events: Option, + #[cfg(test)] + after_run_inserted: Option>, +} + +impl ImportControls { + fn source_scan_error(&self, observed_events: u64) -> Option { + #[cfg(test)] + if self.source_after_events == Some(observed_events) { + return Some(slatedb::Error::unavailable( + "injected legacy run-history source failure".to_owned(), + )); + } + let _ = (self, observed_events); + None + } + + #[cfg(test)] + async fn after_run_inserted(&self) { + if let Some(barrier) = &self.after_run_inserted { + barrier.wait().await; + barrier.wait().await; + } + } +} + +struct ValidatedLegacyRunEvent { + run_id: RunId, + payload: EventPayload, + envelope: EventEnvelope, + event_json: String, +} + +struct ValidatedLegacyRunHistory { + run_id: RunId, + events: Vec, + current: CachedRunProjection, +} + +struct LegacyRunHistorySource { + entries: slatedb::DbIterator, + buffered: Option, + observed_events: u64, +} + +impl LegacyRunHistorySource { + async fn open(database: &Database) -> Result { + let source = database + .open_db() + .await + .map_err(LegacyRunHistorySourceFailure::OpenSource)?; + let prefix = SlateKey::new("runs").into_prefix(); + let entries = source + .scan_prefix(prefix) + .await + .map_err(LegacyRunHistorySourceFailure::OpenScan)?; + Ok(Self { + entries, + buffered: None, + observed_events: 0, + }) + } + + async fn next_run( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + let first = if let Some(entry) = self.buffered.take() { + parse_source_event(&entry.key, &entry.value)? + .expect("the buffered source entry belongs to the event namespace") + } else { + let Some(event) = self.next_event(controls).await? else { + return Ok(None); + }; + event + }; + let run_id = first.run_id; + let run_id_text = run_id.to_string(); + let mut events = vec![first]; + loop { + let Some(entry) = self.next_event_entry(controls).await? else { + break; + }; + if event_run_segment(&entry.key) != Some(run_id_text.as_bytes()) { + self.buffered = Some(entry); + break; + } + let event = parse_source_event(&entry.key, &entry.value)? + .expect("an event-namespace entry parses as an event or fails"); + events.push(event); + } + + if events[0].envelope.seq != 1 || events[0].envelope.event.event_name() != "run.created" { + return Err(LegacyRunHistorySourceFailure::InvalidFirstEvent); + } + if events + .windows(2) + .any(|pair| pair[0].envelope.seq >= pair[1].envelope.seq) + { + return Err(LegacyRunHistorySourceFailure::InvalidSequence); + } + let envelopes = events + .iter() + .map(|event| event.envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes) + .map_err(LegacyRunHistorySourceFailure::Replay)?; + let last_seq = events + .last() + .expect("a validated history contains at least one event") + .envelope + .seq; + let current = CachedRunProjection::from_projection(run_id, projection, last_seq); + Ok(Some(ValidatedLegacyRunHistory { + run_id, + events, + current, + })) + } + + async fn next_event( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + let Some(entry) = self.next_event_entry(controls).await? else { + return Ok(None); + }; + parse_source_event(&entry.key, &entry.value) + } + + async fn next_event_entry( + &mut self, + controls: Option<&ImportControls>, + ) -> Result, LegacyRunHistorySourceFailure> { + loop { + if let Some(source) = + controls.and_then(|controls| controls.source_scan_error(self.observed_events)) + { + return Err(LegacyRunHistorySourceFailure::ReadScan(source)); + } + let Some(entry) = self + .entries + .next() + .await + .map_err(LegacyRunHistorySourceFailure::ReadScan)? + else { + return Ok(None); + }; + if event_run_segment(&entry.key).is_none() { + continue; + } + self.observed_events = self + .observed_events + .checked_add(1) + .ok_or(LegacyRunHistorySourceFailure::CounterOverflow)?; + return Ok(Some(entry)); + } + } +} + +impl Database { + /// Strictly imports legacy SlateDB run history into the inactive SQLite + /// run store, committing one complete run at a time. + pub async fn import_legacy_run_history_into( + &self, + pool: &SqlitePool, + ) -> Result { + self.import_legacy_run_history_with_controls(pool, &ImportControls::default()) + .await + } + + async fn import_legacy_run_history_with_controls( + &self, + pool: &SqlitePool, + controls: &ImportControls, + ) -> Result { + let mut report = LegacyRunHistoryImportReport::default(); + let result = self + .run_legacy_run_history_import(pool, controls, &mut report) + .await; + debug_import_outcome(result.as_ref().map_or("failed", |()| "complete"), &report); + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyRunHistoryImportError { report, failure }), + } + } + + async fn run_legacy_run_history_import( + &self, + pool: &SqlitePool, + controls: &ImportControls, + report: &mut LegacyRunHistoryImportReport, + ) -> Result<(), LegacyRunHistoryImportFailure> { + discard_projection_only_rows(pool, report).await?; + let mut source = LegacyRunHistorySource::open(self) + .await + .map_err(LegacyRunHistoryImportFailure::Source)?; + + loop { + let history = match source.next_run(Some(controls)).await { + Ok(history) => history, + Err(error) => { + report.scanned_source_events = source.observed_events; + return Err(LegacyRunHistoryImportFailure::Source(error)); + } + }; + report.scanned_source_events = source.observed_events; + let Some(history) = history else { + break; + }; + import_checked_add(&mut report.scanned_source_runs, 1)?; + import_one_run(pool, controls, history, report).await?; + } + + report.diagnostics = self + .legacy_run_history_diagnostics() + .await + .map_err(LegacyRunHistoryImportFailure::Diagnostics)?; + Ok(()) + } + + /// Verifies every legacy history as an exact SQLite prefix and then + /// independently replays and verifies every SQLite run. + pub async fn verify_legacy_run_history_in( + &self, + pool: &SqlitePool, + ) -> Result { + let mut report = LegacyRunHistoryVerificationReport::default(); + let result = self + .run_legacy_run_history_verification(pool, &mut report) + .await; + debug_verification_outcome(result.as_ref().map_or("failed", |()| "complete"), &report); + match result { + Ok(()) => Ok(report), + Err(failure) => Err(LegacyRunHistoryVerificationError { report, failure }), + } + } + + async fn run_legacy_run_history_verification( + &self, + pool: &SqlitePool, + report: &mut LegacyRunHistoryVerificationReport, + ) -> Result<(), LegacyRunHistoryVerificationFailure> { + let mut source_ids = HashSet::new(); + let mut source = LegacyRunHistorySource::open(self) + .await + .map_err(LegacyRunHistoryVerificationFailure::Source)?; + loop { + let history = match source.next_run(None).await { + Ok(history) => history, + Err(error) => { + report.source_events = source.observed_events; + return Err(LegacyRunHistoryVerificationFailure::Source(error)); + } + }; + report.source_events = source.observed_events; + let Some(history) = history else { + break; + }; + verification_checked_add(&mut report.source_runs, 1)?; + source_ids.insert(history.run_id); + verify_source_prefix(pool, &history).await?; + verification_checked_add(&mut report.matched_prefix_runs, 1)?; + verification_checked_add( + &mut report.matched_prefix_events, + usize_to_verification_count(history.events.len())?, + )?; + } + + let stored_ids: Vec = sqlx::query_scalar("SELECT id FROM runs ORDER BY id ASC") + .fetch_all(pool) + .await + .map_err(LegacyRunHistoryVerificationFailure::ListDestinationRuns)?; + for stored_id in stored_ids { + let run_id = stored_id + .parse::() + .map_err(|_| LegacyRunHistoryVerificationFailure::InvalidDestinationRunId)?; + let mut connection = pool + .acquire() + .await + .map_err(LegacyRunHistoryVerificationFailure::AcquireConnection)?; + let events = + RunSummaryStore::list_events_with_json_on_connection(&mut connection, &run_id) + .await + .map_err(LegacyRunHistoryVerificationFailure::ReadDestination)?; + if events.is_empty() { + return Err(LegacyRunHistoryVerificationFailure::EmptyDestinationHistory); + } + let current = replay_destination(&run_id, &events) + .map_err(LegacyRunHistoryVerificationFailure::ReplayDestination)?; + RunSummaryStore::verify_current_run_on_connection(&mut connection, ¤t) + .await + .map_err(LegacyRunHistoryVerificationFailure::VerifyDestination)?; + + let event_count = usize_to_verification_count(events.len())?; + verification_checked_add(&mut report.target_runs, 1)?; + verification_checked_add(&mut report.target_events, event_count)?; + if !source_ids.contains(&run_id) { + verification_checked_add(&mut report.sql_only_runs, 1)?; + verification_checked_add(&mut report.sql_only_events, event_count)?; + } + } + + report.diagnostics = self + .legacy_run_history_diagnostics() + .await + .map_err(LegacyRunHistoryVerificationFailure::Diagnostics)?; + Ok(()) + } + + async fn legacy_run_history_diagnostics( + &self, + ) -> Result { + let catalog_ids = self + .catalog_index() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)? + .list(&ListRunsQuery::default()) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadCatalog)?; + let mut diagnostics = LegacyRunHistoryDiagnostics { + catalog_markers: u64::try_from(catalog_ids.len()) + .map_err(|_| LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?, + ..LegacyRunHistoryDiagnostics::default() + }; + let source = self + .open_db() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenSource)?; + for run_id in catalog_ids { + let mut events = source + .scan_prefix(keys::run_events_prefix(&run_id)) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenEventProbe)?; + if events + .next() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadEventProbe)? + .is_none() + { + diagnostics.empty_catalog_markers = diagnostics + .empty_catalog_markers + .checked_add(1) + .ok_or(LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?; + } + } + let mut sessions = source + .scan_prefix(keys::sessions_by_id_prefix()) + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::OpenSessionScan)?; + while sessions + .next() + .await + .map_err(LegacyRunHistoryDiagnosticsFailure::ReadSessionScan)? + .is_some() + { + diagnostics.session_reverse_rows = diagnostics + .session_reverse_rows + .checked_add(1) + .ok_or(LegacyRunHistoryDiagnosticsFailure::CounterOverflow)?; + } + Ok(diagnostics) + } +} + +fn parse_source_event( + key: &[u8], + value: &[u8], +) -> Result, LegacyRunHistorySourceFailure> { + let key_text = std::str::from_utf8(key).map_err(LegacyRunHistorySourceFailure::KeyUtf8)?; + let segments = SlateKey::segments(key_text).collect::>(); + if segments.get(2).copied() != Some("events") { + return Ok(None); + } + let ["runs", run_id_text, "events", leaf] = segments.as_slice() else { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + }; + let run_id = run_id_text + .parse::() + .map_err(|_| LegacyRunHistorySourceFailure::InvalidKey)?; + let Some((sequence_text, epoch_ms_text)) = leaf.split_once('-') else { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + }; + if sequence_text.len() != 6 || !sequence_text.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + } + let sequence = sequence_text + .parse::() + .ok() + .filter(|sequence| (1..=keys::MAX_EVENT_SEQ).contains(sequence)) + .ok_or(LegacyRunHistorySourceFailure::InvalidKey)?; + let epoch_ms = epoch_ms_text + .parse::() + .map_err(|_| LegacyRunHistorySourceFailure::InvalidKey)?; + if keys::run_event_key(&run_id, sequence, epoch_ms).as_ref() != key { + return Err(LegacyRunHistorySourceFailure::InvalidKey); + } + + let event_json = std::str::from_utf8(value) + .map_err(LegacyRunHistorySourceFailure::ValueUtf8)? + .to_owned(); + let payload: EventPayload = + serde_json::from_str(&event_json).map_err(LegacyRunHistorySourceFailure::DecodeEvent)?; + payload + .validate(&run_id) + .map_err(LegacyRunHistorySourceFailure::ValidateEvent)?; + let event = + RunEvent::try_from(&payload).map_err(LegacyRunHistorySourceFailure::ValidateEvent)?; + if event.run_id != run_id { + return Err(LegacyRunHistorySourceFailure::ValidateEvent( + crate::Error::RunEventMismatch { + run_id: run_id.to_string(), + seq: sequence, + field: "run_id", + }, + )); + } + Ok(Some(ValidatedLegacyRunEvent { + run_id, + payload, + envelope: EventEnvelope { + seq: sequence, + event, + }, + event_json, + })) +} + +fn event_run_segment(key: &[u8]) -> Option<&[u8]> { + let mut segments = key.split(|byte| *byte == 0); + (segments.next()? == b"runs").then_some(())?; + let run_id = segments.next()?; + (segments.next()? == b"events").then_some(run_id) +} + +async fn discard_projection_only_rows( + pool: &SqlitePool, + report: &mut LegacyRunHistoryImportReport, +) -> Result<(), LegacyRunHistoryImportFailure> { + let mut transaction = pool + .begin() + .await + .map_err(LegacyRunHistoryImportFailure::BeginCleanup)?; + let result = sqlx::query( + r" +DELETE FROM runs +WHERE NOT EXISTS ( + SELECT 1 FROM run_events WHERE run_events.run_id = runs.id +) +", + ) + .execute(&mut *transaction) + .await + .map_err(LegacyRunHistoryImportFailure::DeleteProjectionOnlyRows)?; + let discarded = result.rows_affected(); + transaction + .commit() + .await + .map_err(LegacyRunHistoryImportFailure::CommitCleanup)?; + report.discarded_projection_only_rows = discarded; + Ok(()) +} + +async fn import_one_run( + pool: &SqlitePool, + controls: &ImportControls, + history: ValidatedLegacyRunHistory, + report: &mut LegacyRunHistoryImportReport, +) -> Result<(), LegacyRunHistoryImportFailure> { + #[cfg(not(test))] + let _ = controls; + let mut transaction = pool + .begin() + .await + .map_err(LegacyRunHistoryImportFailure::BeginRunTransaction)?; + let result = async { + let has_destination: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM run_events WHERE run_id = ?)") + .bind(history.run_id.to_string()) + .fetch_one(&mut *transaction) + .await + .map_err(|source| { + LegacyRunHistoryImportFailure::ReadDestination(crate::Error::Sqlite(source)) + })?; + let mut updated = *report; + if has_destination { + let destination = RunSummaryStore::list_events_with_json_on_connection( + &mut transaction, + &history.run_id, + ) + .await + .map_err(LegacyRunHistoryImportFailure::ReadDestination)?; + require_exact_prefix(&history, &destination) + .map_err(|()| LegacyRunHistoryImportFailure::DestinationConflict)?; + let current = replay_destination(&history.run_id, &destination) + .map_err(LegacyRunHistoryImportFailure::ReplayDestination)?; + RunSummaryStore::verify_current_run_on_connection(&mut transaction, ¤t) + .await + .map_err(LegacyRunHistoryImportFailure::VerifyDestination)?; + import_checked_add(&mut updated.verified_existing_runs, 1)?; + import_checked_add( + &mut updated.verified_existing_events, + usize_to_import_count(destination.len())?, + )?; + } else { + RunSummaryStore::insert_imported_run_on_connection(&mut transaction, &history.current) + .await + .map_err(LegacyRunHistoryImportFailure::InsertRun)?; + #[cfg(test)] + controls.after_run_inserted().await; + for event in &history.events { + RunSummaryStore::insert_imported_event_on_connection( + &mut transaction, + &history.run_id, + &event.payload, + &event.envelope, + &event.event_json, + ) + .await + .map_err(LegacyRunHistoryImportFailure::InsertEvent)?; + } + import_checked_add(&mut updated.imported_runs, 1)?; + import_checked_add( + &mut updated.imported_events, + usize_to_import_count(history.events.len())?, + )?; + } + import_checked_add(&mut updated.committed_run_transactions, 1)?; + Ok(updated) + } + .await; + + match result { + Ok(updated) => { + transaction + .commit() + .await + .map_err(LegacyRunHistoryImportFailure::CommitRunTransaction)?; + *report = updated; + Ok(()) + } + Err(prior) => match transaction.rollback().await { + Ok(()) => Err(prior), + Err(source) => Err(LegacyRunHistoryImportFailure::RollbackRunTransaction { + source, + prior: Box::new(prior), + }), + }, + } +} + +async fn verify_source_prefix( + pool: &SqlitePool, + history: &ValidatedLegacyRunHistory, +) -> Result<(), LegacyRunHistoryVerificationFailure> { + let mut connection = pool + .acquire() + .await + .map_err(LegacyRunHistoryVerificationFailure::AcquireConnection)?; + let has_destination: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM run_events WHERE run_id = ?)") + .bind(history.run_id.to_string()) + .fetch_one(&mut *connection) + .await + .map_err(|source| { + LegacyRunHistoryVerificationFailure::ReadDestination(crate::Error::Sqlite(source)) + })?; + if !has_destination { + return Err(LegacyRunHistoryVerificationFailure::MissingDestinationPrefix); + } + let destination = + RunSummaryStore::list_events_with_json_on_connection(&mut connection, &history.run_id) + .await + .map_err(LegacyRunHistoryVerificationFailure::ReadDestination)?; + if destination.len() < history.events.len() { + return Err(LegacyRunHistoryVerificationFailure::MissingDestinationPrefix); + } + require_exact_prefix(history, &destination) + .map_err(|()| LegacyRunHistoryVerificationFailure::DestinationPrefixConflict) +} + +fn require_exact_prefix( + history: &ValidatedLegacyRunHistory, + destination: &[(EventEnvelope, String)], +) -> Result<(), ()> { + if destination.len() < history.events.len() { + return Err(()); + } + if history + .events + .iter() + .zip(destination) + .any(|(source, (target, target_json))| { + source.envelope.seq != target.seq || source.event_json != *target_json + }) + { + return Err(()); + } + Ok(()) +} + +fn replay_destination( + run_id: &RunId, + events: &[(EventEnvelope, String)], +) -> crate::Result { + let envelopes = events + .iter() + .map(|(envelope, _event_json)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let last_seq = envelopes + .last() + .ok_or_else(|| crate::Error::InvalidEvent("run projection requires an event".to_owned()))? + .seq; + Ok(CachedRunProjection::from_projection( + *run_id, projection, last_seq, + )) +} + +fn usize_to_import_count(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyRunHistoryImportFailure::CounterOverflow) +} + +fn import_checked_add(value: &mut u64, amount: u64) -> Result<(), LegacyRunHistoryImportFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyRunHistoryImportFailure::CounterOverflow)?; + Ok(()) +} + +fn usize_to_verification_count(value: usize) -> Result { + u64::try_from(value).map_err(|_| LegacyRunHistoryVerificationFailure::CounterOverflow) +} + +fn verification_checked_add( + value: &mut u64, + amount: u64, +) -> Result<(), LegacyRunHistoryVerificationFailure> { + *value = value + .checked_add(amount) + .ok_or(LegacyRunHistoryVerificationFailure::CounterOverflow)?; + Ok(()) +} + +fn debug_import_outcome(outcome: &'static str, report: &LegacyRunHistoryImportReport) { + debug!( + outcome, + scanned_source_runs = report.scanned_source_runs, + scanned_source_events = report.scanned_source_events, + imported_runs = report.imported_runs, + imported_events = report.imported_events, + verified_existing_runs = report.verified_existing_runs, + verified_existing_events = report.verified_existing_events, + discarded_projection_only_rows = report.discarded_projection_only_rows, + committed_run_transactions = report.committed_run_transactions, + catalog_markers = report.diagnostics.catalog_markers, + empty_catalog_markers = report.diagnostics.empty_catalog_markers, + session_reverse_rows = report.diagnostics.session_reverse_rows, + "Legacy run-history import finished" + ); +} + +fn debug_verification_outcome(outcome: &'static str, report: &LegacyRunHistoryVerificationReport) { + debug!( + outcome, + source_runs = report.source_runs, + source_events = report.source_events, + matched_prefix_runs = report.matched_prefix_runs, + matched_prefix_events = report.matched_prefix_events, + target_runs = report.target_runs, + target_events = report.target_events, + sql_only_runs = report.sql_only_runs, + sql_only_events = report.sql_only_events, + catalog_markers = report.diagnostics.catalog_markers, + empty_catalog_markers = report.diagnostics.empty_catalog_markers, + session_reverse_rows = report.diagnostics.session_reverse_rows, + "Legacy run-history verification finished" + ); +} + +#[cfg(test)] +mod tests { + use std::error::Error as _; + use std::sync::Arc; + use std::time::Duration; + + use chrono::{TimeZone as _, Utc}; + use fabro_types::{ + Graph, RunEvent, RunId, RunProjection, SessionId, WorkflowSettings, test_support, + }; + use fabro_util::error; + use object_store::memory::InMemory; + use tokio::sync::Barrier; + use ulid::Ulid; + + use super::{ + ImportControls, LegacyRunHistoryDiagnostics, LegacyRunHistoryImportReport, + LegacyRunHistoryVerificationReport, parse_source_event, + }; + use crate::keys::SlateKey; + use crate::slate::CachedRunProjection; + use crate::{ + Database, EventEnvelope, EventPayload, RunProjectionReducer, RunSummaryStore, keys, + test_support as store_test_support, + }; + + type TestResult = std::result::Result>; + + struct TestContext { + _directory: tempfile::TempDir, + source: Database, + source_db: slatedb::Db, + sqlite: sqlx::SqlitePool, + } + + impl TestContext { + async fn new() -> TestResult { + let directory = tempfile::tempdir()?; + let sqlite = + fabro_db::Database::connect(directory.path().join("fabro.sqlite3")).await?; + sqlite.migrate().await?; + let sqlite = sqlite.clone_pool(); + let source = Database::new( + Arc::new(InMemory::new()), + "legacy-run-history-import-tests", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + store_test_support::test_run_summary_store(), + ); + let source_db = source.open_db().await?; + Ok(Self { + _directory: directory, + source, + source_db, + sqlite, + }) + } + + async fn put_event( + &self, + run_id: &RunId, + seq: u32, + epoch_ms: i64, + event_json: &str, + ) -> TestResult<()> { + self.source_db + .put( + keys::run_event_key(run_id, seq, epoch_ms), + event_json.as_bytes(), + ) + .await?; + Ok(()) + } + + async fn put_raw(&self, key: impl AsRef<[u8]>, value: &[u8]) -> TestResult<()> { + self.source_db.put(key, value).await?; + Ok(()) + } + + async fn source_entries(&self) -> TestResult, Vec)>> { + let mut entries = self.source_db.scan_prefix(Vec::::new()).await?; + let mut snapshot = Vec::new(); + while let Some(entry) = entries.next().await? { + snapshot.push((entry.key.to_vec(), entry.value.to_vec())); + } + Ok(snapshot) + } + + async fn import(&self) -> TestResult { + Ok(self + .source + .import_legacy_run_history_into(&self.sqlite) + .await?) + } + } + + fn run_id(index: u128) -> RunId { + RunId::from(Ulid::from_parts( + 1_788_000_000_000 + u64::try_from(index).unwrap(), + index, + )) + } + + fn event_value( + run_id: &RunId, + seq: u32, + event: &str, + properties: &serde_json::Value, + ) -> serde_json::Value { + serde_json::json!({ + "id": format!("evt-{seq}-{event}"), + "ts": Utc + .timestamp_millis_opt(1_788_000_000_000 + i64::from(seq)) + .single() + .unwrap() + .to_rfc3339(), + "run_id": run_id.to_string(), + "event": event, + "properties": properties, + }) + } + + fn created_value(run_id: &RunId, title: &str) -> serde_json::Value { + event_value( + run_id, + 1, + "run.created", + &serde_json::json!({ + "title": title, + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "workflow_slug": "test-workflow", + "labels": {}, + "provenance": test_support::test_run_provenance(), + }), + ) + } + + fn submitted_value(run_id: &RunId, seq: u32) -> serde_json::Value { + event_value(run_id, seq, "run.submitted", &serde_json::json!({})) + } + + fn decode_event( + run_id: &RunId, + seq: u32, + event_json: &str, + ) -> TestResult<(EventPayload, EventEnvelope)> { + let payload: EventPayload = serde_json::from_str(event_json)?; + payload.validate(run_id)?; + let event = RunEvent::try_from(&payload)?; + Ok((payload, EventEnvelope { seq, event })) + } + + async fn seed_destination_history( + pool: &sqlx::SqlitePool, + run_id: &RunId, + events: &[(u32, String)], + ) -> TestResult<()> { + let decoded = events + .iter() + .map(|(seq, event_json)| decode_event(run_id, *seq, event_json)) + .collect::>>()?; + let envelopes = decoded + .iter() + .map(|(_payload, envelope)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let current = + CachedRunProjection::from_projection(*run_id, projection, events.last().unwrap().0); + let mut transaction = pool.begin().await?; + RunSummaryStore::insert_imported_run_on_connection(&mut transaction, ¤t).await?; + for ((_, event_json), (payload, envelope)) in events.iter().zip(&decoded) { + RunSummaryStore::insert_imported_event_on_connection( + &mut transaction, + run_id, + payload, + envelope, + event_json, + ) + .await?; + } + transaction.commit().await?; + Ok(()) + } + + async fn append_destination_event( + pool: &sqlx::SqlitePool, + run_id: &RunId, + prior: &[(u32, String)], + next: (u32, String), + ) -> TestResult<()> { + let mut all = prior.to_vec(); + all.push(next.clone()); + let decoded = all + .iter() + .map(|(seq, event_json)| decode_event(run_id, *seq, event_json)) + .collect::>>()?; + let envelopes = decoded + .iter() + .map(|(_payload, envelope)| envelope.clone()) + .collect::>(); + let projection = RunProjection::apply_events(&envelopes)?; + let current = CachedRunProjection::from_projection(*run_id, projection, next.0); + let mut transaction = pool.begin().await?; + RunSummaryStore::append_event_on_connection( + &mut transaction, + prior.last().unwrap().0, + ¤t, + &decoded.last().unwrap().0, + ) + .await?; + transaction.commit().await?; + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_imports_exact_json_gaps_and_count_only_diagnostics() + -> TestResult<()> { + let context = TestContext::new().await?; + let first = run_id(1); + let second = run_id(2); + let empty_marker = run_id(3); + let stale = run_id(4); + let first_created = serde_json::to_string_pretty(&created_value(&first, "first"))?; + let first_submitted = serde_json::to_string(&submitted_value(&first, 4))?; + let second_created = serde_json::to_string(&created_value(&second, "second"))?; + 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.source.catalog_index().await?.add(&first).await?; + context + .source + .catalog_index() + .await? + .add(&empty_marker) + .await?; + context + .source + .put_session_run_index(&SessionId::new(), &first) + .await?; + + let stale_created = serde_json::to_string(&created_value(&stale, "stale"))?; + seed_destination_history(&context.sqlite, &stale, &[(1, stale_created)]).await?; + sqlx::query("DELETE FROM run_events WHERE run_id = ?") + .bind(stale.to_string()) + .execute(&context.sqlite) + .await?; + sqlx::query("UPDATE runs SET summary_json = '{}' WHERE id = ?") + .bind(stale.to_string()) + .execute(&context.sqlite) + .await?; + let source_before = context.source_entries().await?; + + let report = context.import().await?; + + assert_eq!(report, LegacyRunHistoryImportReport { + scanned_source_runs: 2, + scanned_source_events: 3, + imported_runs: 2, + imported_events: 3, + discarded_projection_only_rows: 1, + committed_run_transactions: 2, + diagnostics: LegacyRunHistoryDiagnostics { + catalog_markers: 2, + empty_catalog_markers: 1, + session_reverse_rows: 1, + }, + ..LegacyRunHistoryImportReport::default() + }); + let stored: Vec<(i64, String)> = + sqlx::query_as("SELECT seq, event_json FROM run_events WHERE run_id = ? ORDER BY seq") + .bind(first.to_string()) + .fetch_all(&context.sqlite) + .await?; + assert_eq!(stored, vec![(1, first_created), (4, first_submitted)]); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(first.to_string()) + .fetch_one(&context.sqlite) + .await?, + 4 + ); + assert_eq!(context.source_entries().await?, source_before); + let verification = context + .source + .verify_legacy_run_history_in(&context.sqlite) + .await?; + assert_eq!(verification.target_runs, 2); + assert_eq!(verification.target_events, 3); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_retry_uses_complete_destination_histories_as_progress() + -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(10); + context + .put_event( + &run_id, + 1, + 10, + &serde_json::to_string(&created_value(&run_id, "retry"))?, + ) + .await?; + let first = context.import().await?; + let second = context.import().await?; + + assert_eq!(first.imported_runs, 1); + assert_eq!(second, LegacyRunHistoryImportReport { + scanned_source_runs: 1, + scanned_source_events: 1, + verified_existing_runs: 1, + verified_existing_events: 1, + committed_run_transactions: 1, + ..LegacyRunHistoryImportReport::default() + }); + Ok(()) + } + + #[test] + fn legacy_run_history_rejects_noncanonical_event_key_shapes() -> TestResult<()> { + let run_id = run_id(20); + let event_json = serde_json::to_string(&created_value(&run_id, "key"))?; + let invalid = [ + format!("runs\0{run_id}\0events\0000001-1\0extra").into_bytes(), + b"runs\0not-a-run-id\0events\x00000001-1".to_vec(), + format!("runs\0{run_id}\0events\000001-1").into_bytes(), + format!("runs\0{run_id}\0events\0000000-1").into_bytes(), + format!("runs\0{run_id}\0events\01000000-1").into_bytes(), + format!("runs\0{run_id}\0events\0000001-nope").into_bytes(), + format!("runs\0{run_id}\0events\0000001-01").into_bytes(), + ]; + for key in invalid { + assert!(parse_source_event(&key, event_json.as_bytes()).is_err()); + } + let mut invalid_utf8 = b"runs\0".to_vec(); + invalid_utf8.extend_from_slice(&[0xff, 0xfe]); + invalid_utf8.extend_from_slice(b"\0events\x00000001-1"); + assert!(parse_source_event(&invalid_utf8, event_json.as_bytes()).is_err()); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rejects_invalid_values_sequences_and_replay() -> TestResult<()> { + let invalid_utf8 = TestContext::new().await?; + let first = run_id(30); + invalid_utf8 + .put_raw(keys::run_event_key(&first, 1, 1), &[0xff]) + .await?; + assert!( + invalid_utf8 + .source + .import_legacy_run_history_into(&invalid_utf8.sqlite) + .await + .is_err() + ); + + let invalid_json = TestContext::new().await?; + invalid_json.put_event(&first, 1, 1, "{not-json").await?; + assert!( + invalid_json + .source + .import_legacy_run_history_into(&invalid_json.sqlite) + .await + .is_err() + ); + + let missing_first = TestContext::new().await?; + missing_first + .put_event( + &first, + 2, + 2, + &serde_json::to_string(&submitted_value(&first, 2))?, + ) + .await?; + assert!( + missing_first + .source + .import_legacy_run_history_into(&missing_first.sqlite) + .await + .is_err() + ); + + let wrong_first_event = TestContext::new().await?; + wrong_first_event + .put_event( + &first, + 1, + 1, + &serde_json::to_string(&submitted_value(&first, 1))?, + ) + .await?; + assert!( + wrong_first_event + .source + .import_legacy_run_history_into(&wrong_first_event.sqlite) + .await + .is_err() + ); + + let mismatched_payload = TestContext::new().await?; + let other_run = run_id(31); + mismatched_payload + .put_event( + &first, + 1, + 1, + &serde_json::to_string(&created_value(&other_run, "mismatch"))?, + ) + .await?; + assert!( + mismatched_payload + .source + .import_legacy_run_history_into(&mismatched_payload.sqlite) + .await + .is_err() + ); + + let duplicate = TestContext::new().await?; + let created = serde_json::to_string(&created_value(&first, "duplicate"))?; + duplicate.put_event(&first, 1, 1, &created).await?; + duplicate.put_event(&first, 1, 2, &created).await?; + assert!( + duplicate + .source + .import_legacy_run_history_into(&duplicate.sqlite) + .await + .is_err() + ); + + let unreplayable = TestContext::new().await?; + unreplayable.put_event(&first, 1, 1, &created).await?; + unreplayable.put_event(&first, 2, 2, &created).await?; + assert!( + unreplayable + .source + .import_legacy_run_history_into(&unreplayable.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rolls_back_a_run_when_event_insertion_fails() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(40); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "rollback"))?, + ) + .await?; + sqlx::query( + "CREATE TRIGGER reject_imported_event BEFORE INSERT ON run_events BEGIN SELECT RAISE(FAIL, 'injected'); END", + ) + .execute(&context.sqlite) + .await?; + + let error = context + .source + .import_legacy_run_history_into(&context.sqlite) + .await + .unwrap_err(); + + assert_eq!(error.report().imported_runs, 0); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_interruption_preserves_complete_runs_and_retry_converges() + -> TestResult<()> { + let context = TestContext::new().await?; + let first = run_id(50); + let second = run_id(51); + for run_id in [first, second] { + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "interrupted"))?, + ) + .await?; + } + let controls = ImportControls { + source_after_events: Some(2), + ..ImportControls::default() + }; + + let error = context + .source + .import_legacy_run_history_with_controls(&context.sqlite, &controls) + .await + .unwrap_err(); + assert_eq!(error.report().imported_runs, 1); + assert_eq!(error.report().committed_run_transactions, 1); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 1 + ); + + let retry = context.import().await?; + assert_eq!(retry.imported_runs, 1); + assert_eq!(retry.verified_existing_runs, 1); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 2 + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_cancellation_leaves_no_half_imported_run() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(60); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "cancel"))?, + ) + .await?; + let barrier = Arc::new(Barrier::new(2)); + let source = context.source.clone(); + let sqlite = context.sqlite.clone(); + let task_barrier = Arc::clone(&barrier); + let task = tokio::spawn(async move { + let controls = ImportControls { + after_run_inserted: Some(task_barrier), + ..ImportControls::default() + }; + source + .import_legacy_run_history_with_controls(&sqlite, &controls) + .await + }); + barrier.wait().await; + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM runs") + .fetch_one(&context.sqlite) + .await?, + 0 + ); + assert_eq!(context.import().await?.imported_runs, 1); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_rejects_partial_conflicting_and_corrupt_destinations() + -> TestResult<()> { + let run_id = run_id(70); + let source_created = serde_json::to_string(&created_value(&run_id, "source"))?; + let source_submitted = serde_json::to_string(&submitted_value(&run_id, 2))?; + + let partial = TestContext::new().await?; + partial.put_event(&run_id, 1, 1, &source_created).await?; + partial.put_event(&run_id, 2, 2, &source_submitted).await?; + seed_destination_history(&partial.sqlite, &run_id, &[(1, source_created.clone())]).await?; + assert!( + partial + .source + .import_legacy_run_history_into(&partial.sqlite) + .await + .is_err() + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_events") + .fetch_one(&partial.sqlite) + .await?, + 1 + ); + + let conflicting = TestContext::new().await?; + conflicting + .put_event(&run_id, 1, 1, &source_created) + .await?; + let target_created = serde_json::to_string(&created_value(&run_id, "target"))?; + seed_destination_history(&conflicting.sqlite, &run_id, &[(1, target_created)]).await?; + assert!( + conflicting + .source + .import_legacy_run_history_into(&conflicting.sqlite) + .await + .is_err() + ); + + let corrupt_event = TestContext::new().await?; + corrupt_event + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_event.sqlite, &run_id, &[( + 1, + source_created.clone(), + )]) + .await?; + sqlx::query("UPDATE run_events SET event_name = 'run.failed' WHERE run_id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_event.sqlite) + .await?; + assert!( + corrupt_event + .source + .import_legacy_run_history_into(&corrupt_event.sqlite) + .await + .is_err() + ); + + let corrupt_summary = TestContext::new().await?; + corrupt_summary + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_summary.sqlite, &run_id, &[( + 1, + source_created.clone(), + )]) + .await?; + sqlx::query("UPDATE runs SET title = 'corrupt' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_summary.sqlite) + .await?; + assert!( + corrupt_summary + .source + .import_legacy_run_history_into(&corrupt_summary.sqlite) + .await + .is_err() + ); + + let corrupt_head = TestContext::new().await?; + corrupt_head + .put_event(&run_id, 1, 1, &source_created) + .await?; + seed_destination_history(&corrupt_head.sqlite, &run_id, &[(1, source_created)]).await?; + sqlx::query("UPDATE runs SET source_last_seq = 2 WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_head.sqlite) + .await?; + assert!( + corrupt_head + .source + .import_legacy_run_history_into(&corrupt_head.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_verification_accepts_sql_suffixes_and_sql_only_runs() + -> TestResult<()> { + let context = TestContext::new().await?; + let source_run = run_id(80); + let sql_only_run = run_id(81); + let source_created = serde_json::to_string(&created_value(&source_run, "source"))?; + context + .put_event(&source_run, 1, 1, &source_created) + .await?; + context.import().await?; + let suffix = serde_json::to_string(&submitted_value(&source_run, 2))?; + append_destination_event( + &context.sqlite, + &source_run, + &[(1, source_created)], + (2, suffix), + ) + .await?; + let sql_only_created = serde_json::to_string(&created_value(&sql_only_run, "sql-only"))?; + seed_destination_history(&context.sqlite, &sql_only_run, &[(1, sql_only_created)]).await?; + + let report = context + .source + .verify_legacy_run_history_in(&context.sqlite) + .await?; + + assert_eq!(report, LegacyRunHistoryVerificationReport { + source_runs: 1, + source_events: 1, + matched_prefix_runs: 1, + matched_prefix_events: 1, + target_runs: 2, + target_events: 3, + sql_only_runs: 1, + sql_only_events: 1, + ..LegacyRunHistoryVerificationReport::default() + }); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_verification_fails_on_prefix_and_current_row_corruption() + -> TestResult<()> { + let run_id = run_id(90); + let created = serde_json::to_string(&created_value(&run_id, "verify"))?; + + let changed_prefix = TestContext::new().await?; + changed_prefix.put_event(&run_id, 1, 1, &created).await?; + let changed = serde_json::to_string(&created_value(&run_id, "changed"))?; + seed_destination_history(&changed_prefix.sqlite, &run_id, &[(1, changed)]).await?; + assert!( + changed_prefix + .source + .verify_legacy_run_history_in(&changed_prefix.sqlite) + .await + .is_err() + ); + + let corrupt_row = TestContext::new().await?; + corrupt_row.put_event(&run_id, 1, 1, &created).await?; + seed_destination_history(&corrupt_row.sqlite, &run_id, &[(1, created)]).await?; + sqlx::query("UPDATE runs SET workflow_slug = 'corrupt' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_row.sqlite) + .await?; + assert!( + corrupt_row + .source + .verify_legacy_run_history_in(&corrupt_row.sqlite) + .await + .is_err() + ); + + let corrupt_json = TestContext::new().await?; + let created = serde_json::to_string(&created_value(&run_id, "verify"))?; + corrupt_json.put_event(&run_id, 1, 1, &created).await?; + seed_destination_history(&corrupt_json.sqlite, &run_id, &[(1, created)]).await?; + sqlx::query("UPDATE runs SET summary_json = '{}' WHERE id = ?") + .bind(run_id.to_string()) + .execute(&corrupt_json.sqlite) + .await?; + assert!( + corrupt_json + .source + .verify_legacy_run_history_in(&corrupt_json.sqlite) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_errors_expose_safe_counts_without_event_contents() -> TestResult<()> + { + let context = TestContext::new().await?; + let run_id = run_id(100); + let secret = "DO-NOT-RENDER-THIS-TITLE"; + let mut value = created_value(&run_id, secret); + value + .get_mut("properties") + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("settings"); + context + .put_event(&run_id, 1, 1, &serde_json::to_string(&value)?) + .await?; + + let error = context + .source + .import_legacy_run_history_into(&context.sqlite) + .await + .unwrap_err(); + let rendered = format!("{error:?} {error}"); + let chain = error::collect_chain(&error).join(": "); + assert!(!rendered.contains(secret)); + assert!(!chain.contains(secret)); + assert!(error.source().is_some()); + Ok(()) + } + + #[tokio::test] + async fn legacy_run_history_scan_ignores_non_event_run_namespaces() -> TestResult<()> { + let context = TestContext::new().await?; + let run_id = run_id(110); + context + .put_event( + &run_id, + 1, + 1, + &serde_json::to_string(&created_value(&run_id, "event"))?, + ) + .await?; + context + .put_raw( + SlateKey::new("runs").with(run_id).with("state"), + b"not event JSON", + ) + .await?; + assert_eq!(context.import().await?.imported_events, 1); + Ok(()) + } +} diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 3fd34521d..ad957424d 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -8,6 +8,7 @@ mod error; mod keyed_mutex; mod keys; mod legacy_blob_import; +mod legacy_run_history_import; mod record; mod run_sessions; mod run_state; @@ -37,6 +38,10 @@ pub use legacy_blob_import::{ LegacyBlobImportError, LegacyBlobImportReport, LegacyBlobInventory, LegacyBlobInventoryError, LegacyBlobVerificationError, LegacyBlobVerificationReport, }; +pub use legacy_run_history_import::{ + LegacyRunHistoryDiagnostics, LegacyRunHistoryImportError, LegacyRunHistoryImportReport, + LegacyRunHistoryVerificationError, LegacyRunHistoryVerificationReport, +}; pub use run_sessions::{ ProjectedRunSession, project_run_session, project_run_session_with_context, project_run_sessions, diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 56ea142a4..bef779ac9 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -384,6 +384,19 @@ impl RunSummaryStore { connection: &mut SqliteConnection, run_id: &RunId, ) -> Result> { + Ok( + Self::list_events_with_json_on_connection(connection, run_id) + .await? + .into_iter() + .map(|(envelope, _event_json)| envelope) + .collect(), + ) + } + + pub(crate) async fn list_events_with_json_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + ) -> Result> { let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); query .push(" WHERE run_id = ") @@ -405,7 +418,143 @@ impl RunSummaryStore { actual_last_seq, }); } - decode_event_rows(&rows, run_id) + decode_event_rows_with_json(&rows, run_id) + } + + pub(crate) async fn insert_imported_run_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + ) -> Result<()> { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, entry.last_seq)?; + if !(1..=keys::MAX_EVENT_SEQ).contains(&entry.last_seq) { + return Err(Error::RunHeadMismatch { + run_id: entry.run_id.to_string(), + expected_last_seq: entry.last_seq, + actual_last_seq: None, + }); + } + insert_run_on_connection(connection, &record).await + } + + pub(crate) async fn insert_imported_event_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + payload: &EventPayload, + envelope: &EventEnvelope, + event_json: &str, + ) -> Result<()> { + payload.validate(run_id)?; + let decoded = RunEvent::try_from(payload)?; + if envelope.event != decoded || envelope.event.run_id != *run_id { + return Err(run_event_mismatch(run_id, envelope.seq, "event_json")); + } + if !(1..=keys::MAX_EVENT_SEQ).contains(&envelope.seq) { + return Err(run_event_mismatch(run_id, envelope.seq, "seq")); + } + insert_event_json_on_connection(connection, run_id, envelope, event_json).await + } + + pub(crate) async fn verify_current_run_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + ) -> Result<()> { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, entry.last_seq)?; + let row = sqlx::query( + r" +SELECT id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, + status, archived_at_ms, parent_id, title, workflow_slug, workflow_name, + repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions, + input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, + total_usd_micros, summary_json +FROM runs +WHERE id = ? +", + ) + .bind(entry.run_id.to_string()) + .fetch_optional(connection) + .await? + .ok_or_else(|| Error::RunNotFound(entry.run_id.to_string()))?; + + let run = &record.run; + let diff = run.diff.unwrap_or_default(); + verify_run_field(&row, run, "id", &run.id.to_string())?; + verify_run_field(&row, run, "source_last_seq", &i64::from(record.last_seq))?; + verify_run_field( + &row, + run, + "created_at_ms", + &run.timestamps.created_at.timestamp_millis(), + )?; + verify_run_field( + &row, + run, + "started_at_ms", + &run.timestamps + .started_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "last_event_at_ms", + &run.timestamps + .last_event_at + .unwrap_or(run.timestamps.created_at) + .timestamp_millis(), + )?; + verify_run_field( + &row, + run, + "completed_at_ms", + &run.timestamps + .completed_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "status", + &run.lifecycle.status.kind().to_string(), + )?; + verify_run_field( + &row, + run, + "archived_at_ms", + &run.lifecycle + .archived_at + .map(|value| value.timestamp_millis()), + )?; + verify_run_field( + &row, + run, + "parent_id", + &run.parent_id.map(|value| value.to_string()), + )?; + verify_run_field(&row, run, "title", &run.title)?; + verify_run_field(&row, run, "workflow_slug", &run.workflow.slug)?; + verify_run_field(&row, run, "workflow_name", &record.workflow_name)?; + verify_run_field(&row, run, "repository_name", &record.repository_name)?; + verify_run_field( + &row, + run, + "automation_id", + &run.automation + .as_ref() + .map(|automation| automation.id.clone()), + )?; + verify_run_field(&row, run, "diff_files_changed", &diff.files_changed)?; + verify_run_field(&row, run, "diff_additions", &diff.additions)?; + verify_run_field(&row, run, "diff_deletions", &diff.deletions)?; + verify_run_field(&row, run, "input_tokens", &record.input_tokens)?; + verify_run_field(&row, run, "output_tokens", &record.output_tokens)?; + verify_run_field(&row, run, "reasoning_tokens", &record.reasoning_tokens)?; + verify_run_field(&row, run, "cache_read_tokens", &record.cache_read_tokens)?; + verify_run_field(&row, run, "cache_write_tokens", &record.cache_write_tokens)?; + verify_run_field(&row, run, "total_usd_micros", &record.total_usd_micros)?; + verify_run_field(&row, run, "summary_json", &serde_json::to_string(run)?)?; + Ok(()) } pub(crate) async fn list_events_from_with_limit_on_connection( @@ -625,8 +774,17 @@ async fn insert_event_on_connection( envelope: &EventEnvelope, ) -> Result<()> { let event_json = serde_json::to_string(payload)?; + insert_event_json_on_connection(connection, &record.run.id, envelope, &event_json).await +} + +async fn insert_event_json_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + envelope: &EventEnvelope, + event_json: &str, +) -> Result<()> { sqlx::query(INSERT_EVENT_SQL) - .bind(record.run.id.to_string()) + .bind(run_id.to_string()) .bind(i64::from(envelope.seq)) .bind(envelope.event.event_name()) .bind(envelope.event.node_id.as_deref()) @@ -666,6 +824,34 @@ fn decode_event_rows(rows: &[SqliteRow], run_id: &RunId) -> Result 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)) + }) + .collect() +} + +fn verify_run_field(row: &SqliteRow, run: &Run, field: &'static str, expected: &T) -> Result<()> +where + T: for<'row> sqlx::Decode<'row, Sqlite> + sqlx::Type + PartialEq, +{ + let stored: T = row.try_get(field)?; + if &stored != expected { + return Err(Error::RunSummaryMismatch { + run_id: run.id.to_string(), + field, + }); + } + Ok(()) +} + fn decode_event_row( row: &SqliteRow, expected_run_id: &RunId,