From 62c30f084171c033b7762d9f5ef0c358034bf217 Mon Sep 17 00:00:00 2001 From: karencfv Date: Mon, 20 Jul 2026 20:51:21 +1200 Subject: [PATCH 1/4] [datastore] impl Selectable and Queryable for Saga --- dev-tools/omdb/src/bin/omdb/db/saga.rs | 62 +++-- nexus/db-model/src/saga_types.rs | 269 +++++++++++++++++++--- nexus/db-queries/src/db/datastore/saga.rs | 79 +++---- 3 files changed, 300 insertions(+), 110 deletions(-) diff --git a/dev-tools/omdb/src/bin/omdb/db/saga.rs b/dev-tools/omdb/src/bin/omdb/db/saga.rs index 3dd8e731a3c..dd4ee2e18b9 100644 --- a/dev-tools/omdb/src/bin/omdb/db/saga.rs +++ b/dev-tools/omdb/src/bin/omdb/db/saga.rs @@ -240,12 +240,11 @@ You should only do this if: if !args.bypass_sec_check { let saga: Saga = { use nexus_db_schema::schema::saga::dsl; - let row = dsl::saga + dsl::saga .filter(dsl::id.eq(args.saga_id)) - .select(nexus_db_model::SagaRow::as_select()) - .first_async(&*conn) - .await?; - Saga::try_from(row)? + .select(Saga::as_select()) + .first_async::(&*conn) + .await? }; let status = get_saga_sec_status(omdb, opctx, &saga).await; @@ -377,14 +376,11 @@ async fn cmd_sagas_abandon( let should_print_color = should_colorize(omdb.output.color, supports_color::Stream::Stdout); let conn = datastore.pool_connection_for_tests().await?; - let saga: Saga = { - let row = dsl::saga - .filter(dsl::id.eq(args.saga_id)) - .select(nexus_db_model::SagaRow::as_select()) - .first_async(&*conn) - .await?; - Saga::try_from(row)? - }; + let saga: Saga = dsl::saga + .filter(dsl::id.eq(args.saga_id)) + .select(Saga::as_select()) + .first_async::(&*conn) + .await?; match saga.saga_state { SagaState::Done => { @@ -424,14 +420,11 @@ execute even if it is abandoned. You should only proceed if: // Before doing anything: find the current SEC for the saga, and ping it to // ensure that the Nexus is down. if !args.bypass_sec_check { - let saga = { - let row = dsl::saga - .filter(dsl::id.eq(args.saga_id)) - .select(nexus_db_model::SagaRow::as_select()) - .first_async(&*conn) - .await?; - Saga::try_from(row)? - }; + let saga = dsl::saga + .filter(dsl::id.eq(args.saga_id)) + .select(Saga::as_select()) + .first_async::(&*conn) + .await?; let status = get_saga_sec_status(omdb, opctx, &saga).await; status.display_message(should_print_color); @@ -481,22 +474,22 @@ async fn get_all_sagas_in_state( let records_batch = paginated(dsl::saga, dsl::id, &p.current_pagparams()) .filter(dsl::saga_state.eq(state)) - .select(nexus_db_model::SagaRow::as_select()) - .load_async::(&**conn) + .select(nexus_db_model::LoadedSaga::as_select()) + .load_async::(&**conn) .await .context("fetching sagas")?; paginator = p - .found_batch(&records_batch, &|s: &nexus_db_model::SagaRow| s.id()); + .found_batch(&records_batch, &|s: &nexus_db_model::LoadedSaga| { + s.id() + }); for row in records_batch { - match Saga::try_from(row.clone()) { - Ok(saga_row) => sagas.push(saga_row), + let saga_id = row.id(); + match row.validate() { + Ok(saga) => sagas.push(saga), Err(e) => { - eprintln!( - "WARNING: Skipping saga with id {}: {e}", - row.id() - ) + eprintln!("WARNING: Skipping saga with id {saga_id}: {e}") } }; } @@ -785,13 +778,12 @@ async fn cmd_sagas_show( let saga = { use nexus_db_schema::schema::saga::dsl; - let row = dsl::saga + dsl::saga .filter(dsl::id.eq(saga_id)) - .select(nexus_db_model::SagaRow::as_select()) - .first_async(&*conn) + .select(Saga::as_select()) + .first_async::(&*conn) .await - .with_context(|| format!("error fetching saga {saga_id}"))?; - Saga::try_from(row)? + .with_context(|| format!("error fetching saga {saga_id}"))? }; print_saga_nodes(Some(saga), nodes); diff --git a/nexus/db-model/src/saga_types.rs b/nexus/db-model/src/saga_types.rs index 080de268d78..839b95bf4b4 100644 --- a/nexus/db-model/src/saga_types.rs +++ b/nexus/db-model/src/saga_types.rs @@ -224,19 +224,17 @@ impl From for SagaState { /// Represents a raw row in the "saga" table. /// /// This is the type Diesel reads and writes directly. It can represent states -/// that should never produced (for example, abandonment metadata columns set -/// without the saga being `Abandoned`), so it isn't constructed directly. -/// Reads produce it via `Queryable` and immediately validate it into [`Saga`] -/// (`Saga::try_from`). Writes lower a validated [`Saga`] into it via -/// `SagaRow::from(&saga)`. +/// that should never be produced (for example, abandonment metadata columns +/// set without the saga being `Abandoned`), so it shouldn't be constructed +/// directly. /// -/// This is `pub` only because Diesel needs to name the row type at query sites -/// in other crates (like omdb). It's `#[doc(hidden)]` to signal that it's an -/// internal type and not part of the supported API. -#[doc(hidden)] +/// Validated reads go through [`Saga`], and raw reads through [`LoadedSaga`] +/// (both `Selectable`). Writes go through the `Insertable` impl on `&Saga`. +/// All three route through `SagaRow` internally, validating via +/// `Saga::try_from` and lowering via `SagaRow::from(&saga)`. #[derive(Queryable, Insertable, Clone, Debug, Selectable, PartialEq)] #[diesel(table_name = saga)] -pub struct SagaRow { +pub(crate) struct SagaRow { id: SagaId, creator: SecId, time_created: chrono::DateTime, @@ -255,26 +253,10 @@ pub struct SagaRow { } impl SagaRow { - pub fn id(&self) -> SagaId { + fn id(&self) -> SagaId { self.id } - pub fn current_sec(&self) -> Option { - self.current_sec - } - - pub fn saga_state(&self) -> SagaState { - self.saga_state - } - - pub fn creator(&self) -> SecId { - self.creator - } - - pub fn adopt_generation(&self) -> super::Generation { - self.adopt_generation - } - fn is_abandon_metadata_empty(&self) -> bool { self.abandon_comment.is_none() && self.abandon_reason.is_none() @@ -489,6 +471,239 @@ impl From<&Saga> for SagaRow { } } +// The `saga` table's columns, in the order `SagaRow` declares them. Named via +// the public schema so `Saga`/`LoadedSaga` can use them as their +// `Selectable::SelectExpression`. We do this to avoid making `SagaRow` public. +type SagaColumns = ( + saga::id, + saga::creator, + saga::time_created, + saga::name, + saga::saga_dag, + saga::saga_state, + saga::current_sec, + saga::adopt_generation, + saga::adopt_time, + saga::abandon_time, + saga::abandon_reason, + saga::abandon_comment, +); + +fn saga_columns() -> SagaColumns { + ( + saga::id, + saga::creator, + saga::time_created, + saga::name, + saga::saga_dag, + saga::saga_state, + saga::current_sec, + saga::adopt_generation, + saga::adopt_time, + saga::abandon_time, + saga::abandon_reason, + saga::abandon_comment, + ) +} + +// The Rust-side tuple that [`SagaColumns`] deserializes into, mirroring +// `SagaRow`'s fields. Used as `Queryable::Row` for `Saga` and `LoadedSaga`. +// Spelled out in public types so it doesn't leak the crate-private `SagaRow`. +type SagaRowColumns = ( + SagaId, + SecId, + DateTime, + String, + serde_json::Value, + SagaState, + Option, + super::Generation, + DateTime, + Option>, + Option, + Option, +); + +impl SagaRow { + // Reassemble a raw row from the loaded column tuple, so the (private) + // validation logic in `TryFrom` can be reused by the `Queryable` + // impls without exposing `SagaRow` in any signature. + fn from_columns(columns: SagaRowColumns) -> Self { + let ( + id, + creator, + time_created, + name, + saga_dag, + saga_state, + current_sec, + adopt_generation, + adopt_time, + abandon_time, + abandon_reason, + abandon_comment, + ) = columns; + SagaRow { + id, + creator, + time_created, + name, + saga_dag, + saga_state, + current_sec, + adopt_generation, + adopt_time, + abandon_time, + abandon_reason, + abandon_comment, + } + } +} + +// Allow `Saga` to be selected and loaded directly, so query sites can use +// `.select(Saga::as_select())` and `.load::()` without naming the private +// `SagaRow`. `Queryable::build` is fallible, so validation happens as part of +// deserialization via `Saga::try_from`. +// +// Because validation happens in `build`, loading a `Saga` is all-or-nothing: a +// row that fails validation fails the entire query. Callers that need to skip +// individual invalid rows should load [`LoadedSaga`] instead. +impl diesel::Selectable for Saga { + type SelectExpression = SagaColumns; + + fn construct_selection() -> Self::SelectExpression { + saga_columns() + } +} + +impl diesel::deserialize::Queryable for Saga +where + SagaRowColumns: diesel::deserialize::FromStaticSqlRow, +{ + type Row = SagaRowColumns; + + fn build(row: Self::Row) -> deserialize::Result { + Ok(Saga::try_from(SagaRow::from_columns(row))?) + } +} + +/// The `saga` column assignments produced when inserting a `Saga`. +/// +/// This is the type of the `(col.eq(value), ...)` tuple built by +/// `Saga::insert_values`. It's named via the public `diesel::dsl::Eq` so that +/// the `Insertable` impl on `&Saga` can delegate its `Values` to it without +/// naming the crate-private `SagaRow`. +pub type SagaInsertValues = ( + diesel::dsl::Eq, + diesel::dsl::Eq, + diesel::dsl::Eq>, + diesel::dsl::Eq, + diesel::dsl::Eq, + diesel::dsl::Eq, + diesel::dsl::Eq>, + diesel::dsl::Eq, + diesel::dsl::Eq>, + diesel::dsl::Eq>>, + diesel::dsl::Eq>, + diesel::dsl::Eq>, +); + +impl Saga { + // The column assignments for inserting this saga. `SagaRow::from(self)` + // expands the all-or-none `abandon_metadata` back into its three nullable + // columns. This backs the `Insertable` impl below; callers just use + // `.values(saga)`. + fn insert_values(&self) -> SagaInsertValues { + use diesel::ExpressionMethods; + + let SagaRow { + id, + creator, + time_created, + name, + saga_dag, + saga_state, + current_sec, + adopt_generation, + adopt_time, + abandon_time, + abandon_reason, + abandon_comment, + } = SagaRow::from(self); + ( + saga::id.eq(id), + saga::creator.eq(creator), + saga::time_created.eq(time_created), + saga::name.eq(name), + saga::saga_dag.eq(saga_dag), + saga::saga_state.eq(saga_state), + saga::current_sec.eq(current_sec), + saga::adopt_generation.eq(adopt_generation), + saga::adopt_time.eq(adopt_time), + saga::abandon_time.eq(abandon_time), + saga::abandon_reason.eq(abandon_reason), + saga::abandon_comment.eq(abandon_comment), + ) + } +} + +// Allow a validated `Saga` to be inserted directly. `Values` delegates to the +// public `SagaInsertValues` tuple, which is what keeps `SagaRow` crate-private. +// `UndecoratedInsertRecord` enables the batch form. +impl diesel::Insertable for &Saga { + type Values = >::Values; + + fn values(self) -> Self::Values { + >::values( + self.insert_values(), + ) + } +} + +// TODO-K: DO I really need this? +impl diesel::query_builder::UndecoratedInsertRecord for &Saga {} + +/// A raw saga row loaded from the database that has not yet been validated into +/// a [`Saga`]. +/// +/// Structural deserialization of the raw columns never fails here. Call +/// [`LoadedSaga::validate`] to check that the abandon metadata is consistent +/// with the saga state and obtain a [`Saga`]. This lets batch readers skip +/// individual invalid rows instead of failing the whole load, which loading +/// [`Saga`] directly would do. +pub struct LoadedSaga(SagaRow); + +impl LoadedSaga { + pub fn id(&self) -> SagaId { + self.0.id() + } + + /// Validate the loaded row into a [`Saga`], checking that the abandon + /// metadata is consistent with the saga state. + pub fn validate(self) -> Result { + Saga::try_from(self.0) + } +} + +impl diesel::Selectable for LoadedSaga { + type SelectExpression = SagaColumns; + + fn construct_selection() -> Self::SelectExpression { + saga_columns() + } +} + +impl diesel::deserialize::Queryable for LoadedSaga +where + SagaRowColumns: diesel::deserialize::FromStaticSqlRow, +{ + type Row = SagaRowColumns; + + fn build(row: Self::Row) -> deserialize::Result { + Ok(LoadedSaga(SagaRow::from_columns(row))) + } +} + /// Represents a row in the "SagaNodeEvent" table #[derive(Queryable, Insertable, Clone, Debug, Selectable, PartialEq)] #[diesel(table_name = saga_node_event)] diff --git a/nexus/db-queries/src/db/datastore/saga.rs b/nexus/db-queries/src/db/datastore/saga.rs index a5f218cd0c4..2402f65d360 100644 --- a/nexus/db-queries/src/db/datastore/saga.rs +++ b/nexus/db-queries/src/db/datastore/saga.rs @@ -110,10 +110,11 @@ impl DataStore { ) -> Result<(), Error> { use nexus_db_schema::schema::saga::dsl; - // Lower the validated saga into its raw row for insertion. - let row = db::saga_types::SagaRow::from(saga); + // The `Insertable` impl on `&Saga` expands the validated saga into its + // column assignments (spreading `abandon_metadata` across the three + // columns). diesel::insert_into(dsl::saga) - .values(row) + .values(saga) .execute_async(&*self.pool_connection_unauthorized().await?) .await .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?; @@ -180,7 +181,7 @@ impl DataStore { .filter(dsl::id.eq(saga_id)) .filter(dsl::current_sec.eq(current_sec)) .set::(new_state.clone().into()) - .check_if_exists::(saga_id) + .check_if_exists::(saga_id) .execute_and_check(&*self.pool_connection_unauthorized().await?) .await .map_err(|e| { @@ -204,8 +205,8 @@ impl DataStore { saga_id, new_state, current_sec, - result.found.current_sec(), - result.found.saga_state(), + result.found.current_sec, + result.found.saga_state, ))) } } @@ -233,8 +234,8 @@ impl DataStore { .eq_any(SagaState::RECOVERY_CANDIDATE_STATES), ) .filter(dsl::current_sec.eq(sec_id)) - .select(db::saga_types::SagaRow::as_select()) - .load_async::(&*conn) + .select(db::saga_types::LoadedSaga::as_select()) + .load_async::(&*conn) .await .map_err(|e| { public_error_from_diesel(e, ErrorHandler::Server) @@ -245,7 +246,7 @@ impl DataStore { // Validate each row into a `Saga` as we collect it. for row in batch { let saga_id = row.id(); - match db::saga_types::Saga::try_from(row) { + match row.validate() { Ok(saga) => sagas.push(saga), Err(e) => { warn!( @@ -271,14 +272,14 @@ impl DataStore { use nexus_db_schema::schema::saga::dsl; let conn = self.pool_connection_authorized(opctx).await?; - let rows: Vec = dsl::saga + let rows: Vec = dsl::saga .filter( dsl::saga_state .eq_any(vec![SagaState::Running, SagaState::Unwinding]), ) .filter(dsl::time_created.lt(time_limit)) .limit(500) - .select(db::saga_types::SagaRow::as_select()) + .select(db::saga_types::LoadedSaga::as_select()) .load_async(&*conn) .await .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?; @@ -287,7 +288,7 @@ impl DataStore { .into_iter() .filter_map(|row| { let saga_id = row.id(); - db::saga_types::Saga::try_from(row) + row.validate() .inspect_err(|e| { warn!( &opctx.log, @@ -407,7 +408,6 @@ mod test { use db::queries::ALLOW_FULL_TABLE_SCAN_SQL; use nexus_db_model::Saga; use nexus_db_model::SagaReasonAbandoned; - use nexus_db_model::SagaRow; use nexus_db_model::SagaState; use nexus_db_model::{SagaNodeEvent, SecId}; use omicron_common::api::external::Generation; @@ -446,12 +446,7 @@ mod test { .await .expect("Failed to access db connection"); diesel::insert_into(nexus_db_schema::schema::saga::dsl::saga) - .values( - inserted_sagas - .iter() - .map(db::saga_types::SagaRow::from) - .collect::>(), - ) + .values(inserted_sagas.iter().collect::>()) .execute_async(&*conn) .await .expect("Failed to insert test setup data"); @@ -727,13 +722,12 @@ mod test { let saga_id: db::saga_types::SagaId = node_cx.saga_id.into(); let found_saga = { use nexus_db_schema::schema::saga::dsl; - let row = dsl::saga + dsl::saga .filter(dsl::id.eq(saga_id)) - .select(SagaRow::as_select()) - .first_async(&*conn) + .select(Saga::as_select()) + .first_async::(&*conn) .await - .unwrap(); - Saga::try_from(row).expect("row is a valid saga") + .expect("row is a valid saga") }; assert_eq!(found_saga.saga_state, SagaState::Abandoned); @@ -902,12 +896,7 @@ mod test { use nexus_db_schema::schema::saga::dsl; let conn = datastore.pool_connection_for_tests().await.unwrap(); diesel::insert_into(dsl::saga) - .values( - sagas_to_insert - .iter() - .map(db::saga_types::SagaRow::from) - .collect::>(), - ) + .values(sagas_to_insert.iter().collect::>()) .execute_async(&*conn) .await .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server)) @@ -931,8 +920,8 @@ mod test { use nexus_db_schema::schema::saga::dsl; conn.batch_execute_async(ALLOW_FULL_TABLE_SCAN_SQL).await?; dsl::saga - .select(nexus_db_model::SagaRow::as_select()) - .load_async(&conn) + .select(Saga::as_select()) + .load_async::(&conn) .await }) .await @@ -940,18 +929,18 @@ mod test { for saga in all_sagas { println!("checking saga: {:?}", saga); - let current_sec = saga.current_sec().unwrap(); - if sagas_affected.contains(&saga.id()) { - assert!(saga.creator() == sec_b || saga.creator() == sec_c); + let current_sec = saga.current_sec.unwrap(); + if sagas_affected.contains(&saga.id) { + assert!(saga.creator == sec_b || saga.creator == sec_c); assert_eq!(current_sec, sec_a); - assert_eq!(*saga.adopt_generation(), Generation::from(2)); + assert_eq!(*saga.adopt_generation, Generation::from(2)); assert!( - saga.saga_state() == SagaState::Running - || saga.saga_state() == SagaState::Unwinding + saga.saga_state == SagaState::Running + || saga.saga_state == SagaState::Unwinding ); - } else if sagas_unaffected.contains(&saga.id()) { - assert_eq!(current_sec, saga.creator()); - assert_eq!(*saga.adopt_generation(), Generation::from(1)); + } else if sagas_unaffected.contains(&saga.id) { + assert_eq!(current_sec, saga.creator); + assert_eq!(*saga.adopt_generation, Generation::from(1)); // Its SEC and state could be anything since we've deliberately // included sagas with various states and SECs that should not // be affected by the reassignment. @@ -996,13 +985,7 @@ mod test { let conn = datastore.pool_connection_for_tests().await.unwrap(); diesel::insert_into(nexus_db_schema::schema::saga::dsl::saga) - .values(vec![ - db::saga_types::SagaRow::from(&running), - db::saga_types::SagaRow::from(&running2), - db::saga_types::SagaRow::from(&unwinding), - db::saga_types::SagaRow::from(&done), - db::saga_types::SagaRow::from(&abandoned), - ]) + .values(vec![&running, &running2, &unwinding, &done, &abandoned]) .execute_async(&*conn) .await .expect("Failed to insert test setup data"); From 97f4800d539242955cfa75474d3575a6ca22bedb Mon Sep 17 00:00:00 2001 From: karencfv Date: Tue, 21 Jul 2026 13:26:25 +1200 Subject: [PATCH 2/4] refactor Saga so it is impossible to represent invalid states --- dev-tools/omdb/src/bin/omdb/db/saga.rs | 10 +- nexus/db-model/src/saga_types.rs | 143 +++++++++++++++------- nexus/db-queries/src/db/datastore/saga.rs | 28 ++--- 3 files changed, 120 insertions(+), 61 deletions(-) diff --git a/dev-tools/omdb/src/bin/omdb/db/saga.rs b/dev-tools/omdb/src/bin/omdb/db/saga.rs index dd4ee2e18b9..36af8052296 100644 --- a/dev-tools/omdb/src/bin/omdb/db/saga.rs +++ b/dev-tools/omdb/src/bin/omdb/db/saga.rs @@ -22,6 +22,7 @@ use internal_dns_resolver::ResolveError; use internal_dns_types::names::ServiceName; use nexus_db_lookup::DataStoreConnection; use nexus_db_model::Saga; +use nexus_db_model::SagaExecState; use nexus_db_model::SagaNodeEvent; use nexus_db_model::SagaState; use nexus_db_model::SecId; @@ -185,7 +186,6 @@ impl From for SagaRow { current_sec, adopt_generation: _, adopt_time: _, - abandon_metadata: _, } = saga; Self { id: id.0.into(), @@ -196,7 +196,7 @@ impl From for SagaRow { }, time_created, name, - state: format!("{saga_state:?}"), + state: format!("{:?}", SagaState::from(saga_state)), } } } @@ -383,13 +383,13 @@ async fn cmd_sagas_abandon( .await?; match saga.saga_state { - SagaState::Done => { + SagaExecState::Done => { bail!("saga {} is already done executing", args.saga_id); } - SagaState::Abandoned => { + SagaExecState::Abandoned(_) => { bail!("saga {} is already abandoned", args.saga_id); } - SagaState::Running | SagaState::Unwinding => {} + SagaExecState::Running | SagaExecState::Unwinding => {} } let text = r#" diff --git a/nexus/db-model/src/saga_types.rs b/nexus/db-model/src/saga_types.rs index 839b95bf4b4..5394ea3b2a2 100644 --- a/nexus/db-model/src/saga_types.rs +++ b/nexus/db-model/src/saga_types.rs @@ -286,13 +286,55 @@ pub struct AbandonMetadata { pub comment: String, } +/// The execution state of a [`Saga`], in memory. +/// +/// This is the validated, domain-side counterpart to the flat `saga_state` +/// column ([`SagaState`]) plus the three nullable abandon columns. Bundling the +/// abandonment metadata into the `Abandoned` variant makes the invalid +/// combinations unrepresentable. Only [`Abandoned`](Self::Abandoned) can carry +/// metadata, and it always must. Unlike [`SagaState`], this type is never +/// stored directly; it is split back into the columns by +/// `From<&Saga> for SagaRow`. +#[derive(Clone, Debug, PartialEq)] +pub enum SagaExecState { + Running, + Unwinding, + Done, + Abandoned(AbandonMetadata), +} + +impl From for SagaState { + fn from(value: SagaExecState) -> Self { + match value { + SagaExecState::Running => SagaState::Running, + SagaExecState::Unwinding => SagaState::Unwinding, + SagaExecState::Done => SagaState::Done, + SagaExecState::Abandoned(_) => SagaState::Abandoned, + } + } +} + +impl From for SagaExecState { + fn from(value: steno::SagaCachedState) -> Self { + // Steno has no concept of an abandoned saga, so this only ever produces + // the non-abandoned variants. + match value { + steno::SagaCachedState::Running => SagaExecState::Running, + steno::SagaCachedState::Unwinding => SagaExecState::Unwinding, + steno::SagaCachedState::Done => SagaExecState::Done, + } + } +} + /// A saga loaded and validated from the database or built in memory for /// insertion. /// /// Compared to [`SagaRow`], the three abandon metadata columns are bundled -/// into a single `abandon` field, kept all or none. Reads go through -/// `TryFrom`, which rejects rows whose metadata is inconsistent with -/// `saga_state` with an [`Error::internal_error`]. +/// into a the saga_state field as part of the `SagaExecState::Abandoned()` +/// variant so invalid state/metadata combinations can't be represented. +/// Reads from the database go through `TryFrom`, which rejects rows +/// whose metadata is inconsistent with `saga_state` with an +/// [`Error::internal_error`]. #[derive(Clone, Debug, PartialEq)] pub struct Saga { pub id: SagaId, @@ -300,12 +342,10 @@ pub struct Saga { pub time_created: DateTime, pub name: String, pub saga_dag: serde_json::Value, - pub saga_state: SagaState, + pub saga_state: SagaExecState, pub current_sec: Option, pub adopt_generation: super::Generation, pub adopt_time: DateTime, - // Abandon metadata, present only when `saga_state` is `Abandoned`. - pub abandon_metadata: Option, } impl Saga { @@ -326,12 +366,12 @@ impl Saga { time_created: now, name: name.to_string(), saga_dag: dag, + // A newly-created saga is never abandoned; `SagaExecState::from` + // only ever yields the non-abandoned variants. saga_state: state.into(), current_sec: Some(creator), adopt_generation: Generation::new().into(), adopt_time: now, - // A newly-created saga is never abandoned. - abandon_metadata: None, } } @@ -354,12 +394,10 @@ impl Saga { time_created: now, name, saga_dag: dag, - saga_state: SagaState::Abandoned, + saga_state: SagaExecState::Abandoned(abandon_metadata), current_sec: Some(creator), adopt_generation: Generation::new().into(), adopt_time: now, - // An abandoned saga must always contain abandonment metadata. - abandon_metadata: Some(abandon_metadata), } } } @@ -400,9 +438,12 @@ impl TryFrom for Saga { ))); } + // TODO-K: Do a try_from for SagaState? + // The abandon metadata must be present exactly when the saga is - // abandoned. - let abandon_metadata = match saga_state { + // abandoned. Fold the flat column plus metadata into the validated + // `SagaExecState`. + let saga_state = match saga_state { SagaState::Abandoned => { let metadata = valid_abandon_metadata.ok_or_else(|| { Error::internal_error(&format!( @@ -410,19 +451,19 @@ impl TryFrom for Saga { )) })?; - Some(metadata) + SagaExecState::Abandoned(metadata) + } + SagaState::Running => { + reject_unexpected_metadata(id, valid_abandon_metadata)?; + SagaExecState::Running } - SagaState::Done | SagaState::Running | SagaState::Unwinding => { - if let Some(metadata) = valid_abandon_metadata { - return Err(Error::internal_error(&format!( - "saga {id}: has abandonment metadata but is not \ - abandoned. abandon_time: {:?} abandon_reason {:?} \ - abandon_comment: {:?}", - metadata.time, metadata.reason, metadata.comment - ))); - } - - None + SagaState::Unwinding => { + reject_unexpected_metadata(id, valid_abandon_metadata)?; + SagaExecState::Unwinding + } + SagaState::Done => { + reject_unexpected_metadata(id, valid_abandon_metadata)?; + SagaExecState::Done } }; @@ -436,22 +477,41 @@ impl TryFrom for Saga { current_sec, adopt_generation, adopt_time, - abandon_metadata, }) } } +// TODO-K: This is a bit shit, fix +// A non-abandoned saga must not carry abandonment metadata. +fn reject_unexpected_metadata( + id: SagaId, + metadata: Option, +) -> Result<(), Error> { + if let Some(AbandonMetadata { time, reason, comment }) = metadata { + return Err(Error::internal_error(&format!( + "saga {id}: has abandonment metadata but is not abandoned. \ + abandon_time: {time:?} abandon_reason {reason:?} \ + abandon_comment: {comment:?}" + ))); + } + Ok(()) +} + /// Lowers a validated [`Saga`] into a raw [`SagaRow`] for insertion. This -/// expands the all or none `abandon_metadata` back into the three nullable -/// columns. +/// splits the `SagaExecState` back into the flat `saga_state` column plus the +/// three nullable abandon columns. impl From<&Saga> for SagaRow { fn from(saga: &Saga) -> Self { let (abandon_time, abandon_reason, abandon_comment) = - match &saga.abandon_metadata { - Some(AbandonMetadata { time, reason, comment }) => { - (Some(*time), Some(*reason), Some(comment.clone())) - } - None => (None, None, None), + match &saga.saga_state { + SagaExecState::Abandoned(AbandonMetadata { + time, + reason, + comment, + }) => (Some(*time), Some(*reason), Some(comment.clone())), + SagaExecState::Running + | SagaExecState::Unwinding + | SagaExecState::Done => (None, None, None), }; SagaRow { @@ -460,7 +520,7 @@ impl From<&Saga> for SagaRow { time_created: saga.time_created, name: saga.name.clone(), saga_dag: saga.saga_dag.clone(), - saga_state: saga.saga_state, + saga_state: saga.saga_state.clone().into(), current_sec: saga.current_sec, adopt_generation: saga.adopt_generation, adopt_time: saga.adopt_time, @@ -649,7 +709,6 @@ impl Saga { // Allow a validated `Saga` to be inserted directly. `Values` delegates to the // public `SagaInsertValues` tuple, which is what keeps `SagaRow` crate-private. -// `UndecoratedInsertRecord` enables the batch form. impl diesel::Insertable for &Saga { type Values = >::Values; @@ -660,7 +719,7 @@ impl diesel::Insertable for &Saga { } } -// TODO-K: DO I really need this? +// Enables the batch form of `Insertable`. impl diesel::query_builder::UndecoratedInsertRecord for &Saga {} /// A raw saga row loaded from the database that has not yet been validated into @@ -823,10 +882,13 @@ mod test { ); let saga = Saga::try_from(row) .expect("abandoned saga with full metadata should be valid"); - assert_eq!(saga.saga_state, SagaState::Abandoned); assert_eq!( - saga.abandon_metadata, - Some(AbandonMetadata { time, reason, comment: comment.clone() }) + saga.saga_state, + SagaExecState::Abandoned(AbandonMetadata { + time, + reason, + comment: comment.clone() + }) ); // Not abandoned and empty metadata is valid. No metadata is propagated @@ -834,8 +896,7 @@ mod test { let row = fake_saga_row(SagaState::Running, None, None, None); let saga = Saga::try_from(row) .expect("non-abandoned saga without metadata should be valid"); - assert_eq!(saga.saga_state, SagaState::Running); - assert_eq!(saga.abandon_metadata, None); + assert_eq!(saga.saga_state, SagaExecState::Running); // Abandoned and empty metadata is invalid. let row = fake_saga_row(SagaState::Abandoned, None, None, None); diff --git a/nexus/db-queries/src/db/datastore/saga.rs b/nexus/db-queries/src/db/datastore/saga.rs index 2402f65d360..a1a75cb7b46 100644 --- a/nexus/db-queries/src/db/datastore/saga.rs +++ b/nexus/db-queries/src/db/datastore/saga.rs @@ -110,9 +110,6 @@ impl DataStore { ) -> Result<(), Error> { use nexus_db_schema::schema::saga::dsl; - // The `Insertable` impl on `&Saga` expands the validated saga into its - // column assignments (spreading `abandon_metadata` across the three - // columns). diesel::insert_into(dsl::saga) .values(saga) .execute_async(&*self.pool_connection_unauthorized().await?) @@ -246,6 +243,7 @@ impl DataStore { // Validate each row into a `Saga` as we collect it. for row in batch { let saga_id = row.id(); + // TODO-K: Do I still need to do this? match row.validate() { Ok(saga) => sagas.push(saga), Err(e) => { @@ -407,8 +405,8 @@ mod test { use chrono::TimeDelta; use db::queries::ALLOW_FULL_TABLE_SCAN_SQL; use nexus_db_model::Saga; + use nexus_db_model::SagaExecState; use nexus_db_model::SagaReasonAbandoned; - use nexus_db_model::SagaState; use nexus_db_model::{SagaNodeEvent, SecId}; use omicron_common::api::external::Generation; use omicron_test_utils::dev; @@ -461,12 +459,13 @@ mod test { assert!( !observed_sagas .iter() - .any(|s| s.saga_state == SagaState::Abandoned) + .any(|s| matches!(s.saga_state, SagaExecState::Abandoned(_))) ); // Remove the abandoned saga from the inserted set so that it can be // compared to the observed set. - inserted_sagas.retain(|s| s.saga_state != SagaState::Abandoned); + inserted_sagas + .retain(|s| !matches!(s.saga_state, SagaExecState::Abandoned(_))); // The observed list is sorted by ID, so sort the inserted list that way // too so that the lists can be tested for equality. @@ -730,10 +729,9 @@ mod test { .expect("row is a valid saga") }; - assert_eq!(found_saga.saga_state, SagaState::Abandoned); - let abandon = found_saga - .abandon_metadata - .expect("an abandoned saga should have abandonment metadata"); + let SagaExecState::Abandoned(abandon) = found_saga.saga_state else { + panic!("an abandoned saga should be in the Abandoned state"); + }; assert_eq!(abandon.reason, SagaReasonAbandoned::Unrecoverable); assert_eq!(abandon.comment, "test".to_string()); @@ -869,8 +867,8 @@ mod test { .iter() .filter_map(|saga| { ((saga.creator == sec_b || saga.creator == sec_c) - && (saga.saga_state == SagaState::Running - || saga.saga_state == SagaState::Unwinding)) + && (saga.saga_state == SagaExecState::Running + || saga.saga_state == SagaExecState::Unwinding)) .then(|| saga.id) }) .collect(); @@ -879,7 +877,7 @@ mod test { .filter_map(|saga| { (saga.creator == sec_a || saga.creator == sec_d - || saga.saga_state == SagaState::Done) + || saga.saga_state == SagaExecState::Done) .then(|| saga.id) }) .collect(); @@ -935,8 +933,8 @@ mod test { assert_eq!(current_sec, sec_a); assert_eq!(*saga.adopt_generation, Generation::from(2)); assert!( - saga.saga_state == SagaState::Running - || saga.saga_state == SagaState::Unwinding + saga.saga_state == SagaExecState::Running + || saga.saga_state == SagaExecState::Unwinding ); } else if sagas_unaffected.contains(&saga.id) { assert_eq!(current_sec, saga.creator); From f1ecedc0f310c2cd097a43869905f0c0b4c2010f Mon Sep 17 00:00:00 2001 From: karencfv Date: Tue, 21 Jul 2026 18:11:21 +1200 Subject: [PATCH 3/4] clean up --- dev-tools/omdb/src/bin/omdb/db/saga.rs | 2 +- nexus/db-model/src/saga_types.rs | 138 ++++++++++------------ nexus/db-queries/src/db/datastore/saga.rs | 5 +- 3 files changed, 64 insertions(+), 81 deletions(-) diff --git a/dev-tools/omdb/src/bin/omdb/db/saga.rs b/dev-tools/omdb/src/bin/omdb/db/saga.rs index 36af8052296..88c3df1cf86 100644 --- a/dev-tools/omdb/src/bin/omdb/db/saga.rs +++ b/dev-tools/omdb/src/bin/omdb/db/saga.rs @@ -486,7 +486,7 @@ async fn get_all_sagas_in_state( for row in records_batch { let saga_id = row.id(); - match row.validate() { + match Saga::try_from(row) { Ok(saga) => sagas.push(saga), Err(e) => { eprintln!("WARNING: Skipping saga with id {saga_id}: {e}") diff --git a/nexus/db-model/src/saga_types.rs b/nexus/db-model/src/saga_types.rs index 5394ea3b2a2..28020092d95 100644 --- a/nexus/db-model/src/saga_types.rs +++ b/nexus/db-model/src/saga_types.rs @@ -257,19 +257,31 @@ impl SagaRow { self.id } - fn is_abandon_metadata_empty(&self) -> bool { - self.abandon_comment.is_none() - && self.abandon_reason.is_none() - && self.abandon_time.is_none() - } - - // Returns the abandonment metadata iff all three columns are set. - fn valid_abandon_metadata(&self) -> Option { - Some(AbandonMetadata { - time: self.abandon_time?, - reason: self.abandon_reason?, - comment: self.abandon_comment.clone()?, - }) + /// Collapses the three nullable abandon columns into all-or-nothing + /// metadata. + fn abandon_metadata(&self) -> Result, Error> { + match ( + self.abandon_time, + self.abandon_reason, + self.abandon_comment.clone(), + ) { + (None, None, None) => Ok(None), + (Some(time), Some(reason), Some(comment)) => { + Ok(Some(AbandonMetadata { time, reason, comment })) + } + // A partially-populated set is impossible per the + // `abandoned_requires_metadata` and + // `not_abandoned_requires_no_metadata` CHECK constraints, so treat + // it as corruption. + _ => Err(Error::internal_error(&format!( + "saga {}: abandonment metadata is partially populated. \ + abandon_time: {:?} abandon_reason: {:?} abandon_comment: {:?}", + self.id, + self.abandon_time, + self.abandon_reason, + self.abandon_comment, + ))), + } } } @@ -406,10 +418,7 @@ impl TryFrom for Saga { type Error = Error; fn try_from(row: SagaRow) -> Result { - let is_abandon_metadata_empty = row.is_abandon_metadata_empty(); - // If present and valid, convert the three nullable abandonment columns - // into `AbandonMetadata`. - let valid_abandon_metadata = row.valid_abandon_metadata(); + let abandon_metadata = row.abandon_metadata()?; let SagaRow { id, @@ -421,49 +430,35 @@ impl TryFrom for Saga { current_sec, adopt_generation, adopt_time, - abandon_time, - abandon_reason, - abandon_comment, + abandon_time: _, + abandon_reason: _, + abandon_comment: _, } = row; - // A partially-populated set is impossible per the - // `abandoned_requires_metadata` and - // `not_abandoned_requires_no_metadata` CHECK constraints, so treat - // it as corruption. - if !is_abandon_metadata_empty && valid_abandon_metadata.is_none() { - return Err(Error::internal_error(&format!( - "saga {id}: abandonment metadata is partially populated. \ - abandon_time: {abandon_time:?} abandon_reason {abandon_reason:?} \ - abandon_comment: {abandon_comment:?}" - ))); - } - - // TODO-K: Do a try_from for SagaState? - // The abandon metadata must be present exactly when the saga is // abandoned. Fold the flat column plus metadata into the validated // `SagaExecState`. - let saga_state = match saga_state { - SagaState::Abandoned => { - let metadata = valid_abandon_metadata.ok_or_else(|| { - Error::internal_error(&format!( - "saga {id}: abandoned but has no abandonment metadata" - )) - })?; - + let saga_state = match (saga_state, abandon_metadata) { + (SagaState::Abandoned, Some(metadata)) => { SagaExecState::Abandoned(metadata) } - SagaState::Running => { - reject_unexpected_metadata(id, valid_abandon_metadata)?; - SagaExecState::Running - } - SagaState::Unwinding => { - reject_unexpected_metadata(id, valid_abandon_metadata)?; - SagaExecState::Unwinding + (SagaState::Abandoned, None) => { + return Err(Error::internal_error(&format!( + "saga {id}: abandoned but has no abandonment metadata" + ))); } - SagaState::Done => { - reject_unexpected_metadata(id, valid_abandon_metadata)?; - SagaExecState::Done + (SagaState::Running, None) => SagaExecState::Running, + (SagaState::Unwinding, None) => SagaExecState::Unwinding, + (SagaState::Done, None) => SagaExecState::Done, + ( + SagaState::Running | SagaState::Unwinding | SagaState::Done, + Some(AbandonMetadata { time, reason, comment }), + ) => { + return Err(Error::internal_error(&format!( + "saga {id}: has abandonment metadata but is not abandoned. \ + abandon_time: {time:?} abandon_reason: {reason:?} \ + abandon_comment: {comment:?}" + ))); } }; @@ -481,22 +476,6 @@ impl TryFrom for Saga { } } -// TODO-K: This is a bit shit, fix -// A non-abandoned saga must not carry abandonment metadata. -fn reject_unexpected_metadata( - id: SagaId, - metadata: Option, -) -> Result<(), Error> { - if let Some(AbandonMetadata { time, reason, comment }) = metadata { - return Err(Error::internal_error(&format!( - "saga {id}: has abandonment metadata but is not abandoned. \ - abandon_time: {time:?} abandon_reason {reason:?} \ - abandon_comment: {comment:?}" - ))); - } - Ok(()) -} - /// Lowers a validated [`Saga`] into a raw [`SagaRow`] for insertion. This /// splits the `SagaExecState` back into the flat `saga_state` column plus the /// three nullable abandon columns. @@ -725,22 +704,27 @@ impl diesel::query_builder::UndecoratedInsertRecord for &Saga {} /// A raw saga row loaded from the database that has not yet been validated into /// a [`Saga`]. /// -/// Structural deserialization of the raw columns never fails here. Call -/// [`LoadedSaga::validate`] to check that the abandon metadata is consistent -/// with the saga state and obtain a [`Saga`]. This lets batch readers skip -/// individual invalid rows instead of failing the whole load, which loading -/// [`Saga`] directly would do. +/// Its `Queryable::build` is infallible, so it defers the abandon-metadata +/// check to `Saga::try_from(loaded_saga)`, run per row after loading. Because +/// Diesel's `load` is all-or-nothing, validating there (as `Saga` does in its +/// own `build`) fails the whole batch on one bad row. Deferring it lets batch +/// readers skip individual invalid rows instead. pub struct LoadedSaga(SagaRow); impl LoadedSaga { pub fn id(&self) -> SagaId { self.0.id() } +} + +/// Validate a loaded row into a [`Saga`], checking that the abandon metadata is +/// consistent with the saga state. Mirrors [`TryFrom`], which does the +/// same check on the raw row. +impl TryFrom for Saga { + type Error = Error; - /// Validate the loaded row into a [`Saga`], checking that the abandon - /// metadata is consistent with the saga state. - pub fn validate(self) -> Result { - Saga::try_from(self.0) + fn try_from(loaded: LoadedSaga) -> Result { + Saga::try_from(loaded.0) } } diff --git a/nexus/db-queries/src/db/datastore/saga.rs b/nexus/db-queries/src/db/datastore/saga.rs index a1a75cb7b46..b54e72a0a14 100644 --- a/nexus/db-queries/src/db/datastore/saga.rs +++ b/nexus/db-queries/src/db/datastore/saga.rs @@ -243,8 +243,7 @@ impl DataStore { // Validate each row into a `Saga` as we collect it. for row in batch { let saga_id = row.id(); - // TODO-K: Do I still need to do this? - match row.validate() { + match db::saga_types::Saga::try_from(row) { Ok(saga) => sagas.push(saga), Err(e) => { warn!( @@ -286,7 +285,7 @@ impl DataStore { .into_iter() .filter_map(|row| { let saga_id = row.id(); - row.validate() + db::saga_types::Saga::try_from(row) .inspect_err(|e| { warn!( &opctx.log, From aaf34cd1069d550fbf1320474afef4341ed6ac0d Mon Sep 17 00:00:00 2001 From: karencfv Date: Tue, 21 Jul 2026 18:30:52 +1200 Subject: [PATCH 4/4] fix some comments --- nexus/db-model/src/saga_types.rs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/nexus/db-model/src/saga_types.rs b/nexus/db-model/src/saga_types.rs index 28020092d95..5a969a8773a 100644 --- a/nexus/db-model/src/saga_types.rs +++ b/nexus/db-model/src/saga_types.rs @@ -257,8 +257,7 @@ impl SagaRow { self.id } - /// Collapses the three nullable abandon columns into all-or-nothing - /// metadata. + // Returns the abandonment metadata iff all three columns are set. fn abandon_metadata(&self) -> Result, Error> { match ( self.abandon_time, @@ -303,9 +302,8 @@ pub struct AbandonMetadata { /// This is the validated, domain-side counterpart to the flat `saga_state` /// column ([`SagaState`]) plus the three nullable abandon columns. Bundling the /// abandonment metadata into the `Abandoned` variant makes the invalid -/// combinations unrepresentable. Only [`Abandoned`](Self::Abandoned) can carry -/// metadata, and it always must. Unlike [`SagaState`], this type is never -/// stored directly; it is split back into the columns by +/// combinations unrepresentable. Unlike [`SagaState`], this type is never +/// stored directly. It is split back into the columns by /// `From<&Saga> for SagaRow`. #[derive(Clone, Debug, PartialEq)] pub enum SagaExecState { @@ -378,8 +376,8 @@ impl Saga { time_created: now, name: name.to_string(), saga_dag: dag, - // A newly-created saga is never abandoned; `SagaExecState::from` - // only ever yields the non-abandoned variants. + // A newly-created saga is never abandoned. Steno's + // `SagaCachedState` only contains non-abandoned variants. saga_state: state.into(), current_sec: Some(creator), adopt_generation: Generation::new().into(), @@ -436,8 +434,7 @@ impl TryFrom for Saga { } = row; // The abandon metadata must be present exactly when the saga is - // abandoned. Fold the flat column plus metadata into the validated - // `SagaExecState`. + // abandoned. let saga_state = match (saga_state, abandon_metadata) { (SagaState::Abandoned, Some(metadata)) => { SagaExecState::Abandoned(metadata)