diff --git a/dev-tools/omdb/src/bin/omdb/db/saga.rs b/dev-tools/omdb/src/bin/omdb/db/saga.rs index 3dd8e731a3c..88c3df1cf86 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)), } } } @@ -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,23 +376,20 @@ 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 => { + 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#" @@ -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 Saga::try_from(row) { + 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..5a969a8773a 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,39 +253,34 @@ 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() - && 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()?, - }) + 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, + ))), + } } } @@ -304,13 +297,54 @@ 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. 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, @@ -318,12 +352,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 { @@ -344,12 +376,12 @@ impl Saga { time_created: now, name: name.to_string(), saga_dag: dag, + // 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(), adopt_time: now, - // A newly-created saga is never abandoned. - abandon_metadata: None, } } @@ -372,12 +404,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), } } } @@ -386,10 +416,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, @@ -401,46 +428,34 @@ 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:?}" - ))); - } - // The abandon metadata must be present exactly when the saga is // abandoned. - let abandon_metadata = 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" - )) - })?; - - Some(metadata) + let saga_state = match (saga_state, abandon_metadata) { + (SagaState::Abandoned, Some(metadata)) => { + SagaExecState::Abandoned(metadata) + } + (SagaState::Abandoned, None) => { + return Err(Error::internal_error(&format!( + "saga {id}: abandoned but has no abandonment metadata" + ))); } - 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::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:?}" + ))); } }; @@ -454,22 +469,25 @@ impl TryFrom for Saga { current_sec, adopt_generation, adopt_time, - abandon_metadata, }) } } /// 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 { @@ -478,7 +496,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, @@ -489,6 +507,243 @@ 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. +impl diesel::Insertable for &Saga { + type Values = >::Values; + + fn values(self) -> Self::Values { + >::values( + self.insert_values(), + ) + } +} + +// 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 +/// a [`Saga`]. +/// +/// 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; + + fn try_from(loaded: LoadedSaga) -> Result { + Saga::try_from(loaded.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)] @@ -608,10 +863,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 @@ -619,8 +877,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 a5f218cd0c4..b54e72a0a14 100644 --- a/nexus/db-queries/src/db/datastore/saga.rs +++ b/nexus/db-queries/src/db/datastore/saga.rs @@ -110,10 +110,8 @@ 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); 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 +178,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 +202,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 +231,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) @@ -271,14 +269,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))?; @@ -406,9 +404,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::SagaRow; - use nexus_db_model::SagaState; use nexus_db_model::{SagaNodeEvent, SecId}; use omicron_common::api::external::Generation; use omicron_test_utils::dev; @@ -446,12 +443,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"); @@ -466,12 +458,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. @@ -727,19 +720,17 @@ 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); - 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()); @@ -875,8 +866,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(); @@ -885,7 +876,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(); @@ -902,12 +893,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 +917,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 +926,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 == SagaExecState::Running + || saga.saga_state == SagaExecState::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 +982,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");