[nexus] Add tests for instance_create unwind safety

common/Cargo.toml

@@ -25,7 +25,7 @@ serde_json = "1.0"
 serde_with = "2.0.1"
 slog = { version = "2.5", features = [ "max_level_trace", "release_max_level_debug" ] }
 smf = "0.2"
-steno = "0.2"
+steno = { git = "https://github.com/oxidecomputer/steno", branch = "node-access" }
 thiserror = "1.0"
 tokio = { version = "1.21", features = [ "full" ] }
 tokio-postgres = { version = "0.7", features = [ "with-chrono-0_4", "with-uuid-1" ] }

nexus/Cargo.toml

@@ -52,7 +52,7 @@ serde_urlencoded = "0.7.1"
 serde_with = "2.0.1"
 sled-agent-client = { path = "../sled-agent-client" }
 slog-dtrace = "0.2"
-steno = "0.2"
+steno = { git = "https://github.com/oxidecomputer/steno", branch = "node-access" }
 tempfile = "3.3"
 thiserror = "1.0"
 toml = "0.5.9"
@@ -121,7 +121,9 @@ itertools = "0.10.5"
 nexus-test-utils-macros = { path = "test-utils-macros" }
 nexus-test-utils = { path = "test-utils" }
 omicron-test-utils = { path = "../test-utils" }
+omicron-sled-agent = { path = "../sled-agent" }
 openapiv3 = "1.0"
+petgraph = "0.6.2"
 regex = "1.6.0"
 subprocess = "0.2.9"
 term = "0.7"

nexus/db-model/Cargo.toml

@@ -25,7 +25,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 uuid = { version = "1.2.1", features = ["serde", "v4"] }
 
-steno = "0.2"
+steno = { git = "https://github.com/oxidecomputer/steno", branch = "node-access" }
 
 db-macros = { path = "../db-macros" }
 omicron-common = { path = "../../common" }

nexus/src/app/mod.rs

@@ -33,7 +33,7 @@ mod organization;
 mod oximeter;
 mod project;
 mod rack;
-mod saga;
+pub mod saga;
 mod session;
 mod silo;
 mod sled;
@@ -46,7 +46,7 @@ mod vpc_subnet;
 
 // Sagas are not part of the "Nexus" implementation, but they are
 // application logic.
-mod sagas;
+pub mod sagas;
 
 // TODO: When referring to API types, we should try to include
 // the prefix unless it is unambiguous.

nexus/src/app/saga.rs

@@ -11,6 +11,7 @@ use crate::authz;
 use crate::context::OpContext;
 use crate::saga_interface::SagaContext;
 use anyhow::Context;
+use futures::future::BoxFuture;
 use futures::StreamExt;
 use omicron_common::api::external;
 use omicron_common::api::external::DataPageParams;
@@ -24,9 +25,32 @@ use steno::DagBuilder;
 use steno::SagaDag;
 use steno::SagaId;
 use steno::SagaName;
+use steno::SagaResult;
 use steno::SagaResultOk;
 use uuid::Uuid;
 
+pub struct RunnableSaga {
+    id: SagaId,
+    fut: BoxFuture<'static, SagaResult>,
+}
+
+impl RunnableSaga {
+    pub fn id(&self) -> SagaId {
+        self.id
+    }
+}
+
+pub fn create_saga_dag<N: NexusSaga>(
+    params: N::Params,
+) -> Result<SagaDag, Error> {
+    let builder = DagBuilder::new(SagaName::new(N::NAME));
+    let dag = N::make_saga_dag(&params, builder)?;
+    let params = serde_json::to_value(&params).map_err(|e| {
+        SagaInitError::SerializeError(String::from("saga params"), e)
+    })?;
+    Ok(SagaDag::new(dag, params))
+}
+
 impl super::Nexus {
     pub async fn sagas_list(
         &self,
@@ -66,27 +90,18 @@ impl super::Nexus {
             })?
     }
 
-    /// Given a saga type and parameters, create a new saga and execute it.
-    pub(crate) async fn execute_saga<N: NexusSaga>(
+    pub async fn create_runnable_saga(
         self: &Arc<Self>,
-        params: N::Params,
-    ) -> Result<SagaResultOk, Error> {
-        let saga = {
-            let builder = DagBuilder::new(SagaName::new(N::NAME));
-            let dag = N::make_saga_dag(&params, builder)?;
-            let params = serde_json::to_value(&params).map_err(|e| {
-                SagaInitError::SerializeError(String::from("saga params"), e)
-            })?;
-            SagaDag::new(dag, params)
-        };
-
+        dag: SagaDag,
+    ) -> Result<RunnableSaga, Error> {
+        // Construct the context necessary to execute this saga.
         let saga_id = SagaId(Uuid::new_v4());
         let saga_logger = self.log.new(o!(
-            "saga_name" => saga.saga_name().to_string(),
+            "saga_name" => dag.saga_name().to_string(),
             "saga_id" => saga_id.to_string()
         ));
         let saga_context = Arc::new(Arc::new(SagaContext::new(
-            Arc::clone(self),
+            self.clone(),
             saga_logger,
             Arc::clone(&self.authz),
         )));
@@ -95,7 +110,7 @@ impl super::Nexus {
             .saga_create(
                 saga_id,
                 saga_context,
-                Arc::new(saga),
+                Arc::new(dag),
                 ACTION_REGISTRY.clone(),
             )
             .await
@@ -106,14 +121,20 @@ impl super::Nexus {
                 // Steno.
                 Error::internal_error(&format!("{:#}", error))
             })?;
+        Ok(RunnableSaga { id: saga_id, fut: future })
+    }
 
+    pub async fn run_saga(
+        &self,
+        runnable_saga: RunnableSaga,
+    ) -> Result<SagaResultOk, Error> {
         self.sec_client
-            .saga_start(saga_id)
+            .saga_start(runnable_saga.id)
             .await
             .context("starting saga")
             .map_err(|error| Error::internal_error(&format!("{:#}", error)))?;
 
-        let result = future.await;
+        let result = runnable_saga.fut.await;
         result.kind.map_err(|saga_error| {
             saga_error
                 .error_source
@@ -125,4 +146,23 @@ impl super::Nexus {
                 ))
         })
     }
+
+    pub fn sec(&self) -> &steno::SecClient {
+        &self.sec_client
+    }
+
+    /// Given a saga type and parameters, create a new saga and execute it.
+    pub(crate) async fn execute_saga<N: NexusSaga>(
+        self: &Arc<Self>,
+        params: N::Params,
+    ) -> Result<SagaResultOk, Error> {
+        // Construct the DAG specific to this saga.
+        let dag = create_saga_dag::<N>(params)?;
+
+        // Register the saga with the saga executor.
+        let runnable_saga = self.create_runnable_saga(dag).await?;
+
+        // Actually run the saga to completion.
+        self.run_saga(runnable_saga).await
+    }
 }