[nexus] Make project creation unwind safe, add tests (#2087)

- Defines undo actions for project creation saga
- Adds test for the project creation saga

Part of #2052

Co-authored-by: Alex Plotnick <[email protected]>

nexus/src/app/project.rs

@@ -85,8 +85,10 @@ impl super::Nexus {
                 saga_params,
             )
             .await?;
-        let db_project = saga_outputs
-            .lookup_node_output::<db::model::Project>("project")
+        let (_authz_project, db_project) = saga_outputs
+            .lookup_node_output::<(authz::Project, db::model::Project)>(
+                "project",
+            )
             .map_err(|e| Error::internal_error(&format!("{:#}", &e)))
             .internal_context("looking up output from project create saga")?;
         Ok(db_project)

nexus/src/app/sagas/project_create.rs

@@ -8,7 +8,6 @@ use super::NexusSaga;
 use crate::app::sagas;
 use crate::app::sagas::declare_saga_actions;
 use crate::context::OpContext;
-use crate::db::lookup::LookupPath;
 use crate::external_api::params;
 use crate::{authn, authz, db};
 use nexus_defaults as defaults;
@@ -33,6 +32,7 @@ declare_saga_actions! {
     project_create;
     PROJECT_CREATE_RECORD -> "project" {
         + spc_create_record
+        - spc_create_record_undo
     }
     PROJECT_CREATE_VPC_PARAMS -> "vpc_create_params" {
         + spc_create_vpc_params
@@ -74,7 +74,7 @@ impl NexusSaga for SagaProjectCreate {
 
 async fn spc_create_record(
     sagactx: NexusActionContext,
-) -> Result<db::model::Project, ActionError> {
+) -> Result<(authz::Project, db::model::Project), ActionError> {
     let osagactx = sagactx.user_data();
     let params = sagactx.saga_params::<Params>()?;
     let opctx = OpContext::for_saga_action(&sagactx, &params.serialized_authn);
@@ -88,25 +88,42 @@ async fn spc_create_record(
         .map_err(ActionError::action_failed)
 }
 
+async fn spc_create_record_undo(
+    sagactx: NexusActionContext,
+) -> Result<(), anyhow::Error> {
+    let osagactx = sagactx.user_data();
+    let params = sagactx.saga_params::<Params>()?;
+    let opctx = OpContext::for_saga_action(&sagactx, &params.serialized_authn);
+
+    let (_authz_project, project) =
+        sagactx.lookup::<(authz::Project, db::model::Project)>("project")?;
+
+    let (.., authz_project, project) =
+        db::lookup::LookupPath::new(&opctx, osagactx.datastore())
+            .project_id(project.id())
+            .fetch_for(authz::Action::Delete)
+            .await?;
+
+    osagactx
+        .datastore()
+        .project_delete(&opctx, &authz_project, &project)
+        .await?;
+    Ok(())
+}
+
 async fn spc_create_vpc_params(
     sagactx: NexusActionContext,
 ) -> Result<sagas::vpc_create::Params, ActionError> {
-    let osagactx = sagactx.user_data();
     let params = sagactx.saga_params::<Params>()?;
     let opctx = OpContext::for_saga_action(&sagactx, &params.serialized_authn);
 
-    let project_id = sagactx.lookup::<db::model::Project>("project")?.id();
+    let (authz_project, _project) =
+        sagactx.lookup::<(authz::Project, db::model::Project)>("project")?;
     let ipv6_prefix = Some(
         defaults::random_vpc_ipv6_prefix()
             .map_err(ActionError::action_failed)?,
     );
 
-    let (.., authz_project) = LookupPath::new(&opctx, osagactx.datastore())
-        .project_id(project_id)
-        .lookup_for(authz::Action::CreateChild)
-        .await
-        .map_err(ActionError::action_failed)?;
-
     let vpc_create = params::VpcCreate {
         identity: IdentityMetadataCreateParams {
             name: "default".parse().unwrap(),
@@ -125,3 +142,167 @@ async fn spc_create_vpc_params(
     };
     Ok(saga_params)
 }
+
+#[cfg(test)]
+mod test {
+    use crate::{
+        app::saga::create_saga_dag, app::sagas::project_create::Params,
+        app::sagas::project_create::SagaProjectCreate, authn::saga::Serialized,
+        authz, context::OpContext, db::datastore::DataStore,
+        external_api::params,
+    };
+    use async_bb8_diesel::{AsyncRunQueryDsl, OptionalExtension};
+    use diesel::{ExpressionMethods, QueryDsl, SelectableHelper};
+    use dropshot::test_util::ClientTestContext;
+    use nexus_test_utils::resource_helpers::create_organization;
+    use nexus_test_utils::resource_helpers::populate_ip_pool;
+    use nexus_test_utils_macros::nexus_test;
+    use omicron_common::api::external::IdentityMetadataCreateParams;
+    use omicron_common::api::external::NameOrId;
+    use uuid::Uuid;
+
+    type ControlPlaneTestContext =
+        nexus_test_utils::ControlPlaneTestContext<crate::Server>;
+
+    const ORG_NAME: &str = "test-org";
+
+    async fn create_org(client: &ClientTestContext) -> Uuid {
+        populate_ip_pool(&client, "default", None).await;
+        let org = create_organization(&client, ORG_NAME).await;
+        org.identity.id
+    }
+
+    // Helper for creating project create parameters
+    fn new_test_params(
+        opctx: &OpContext,
+        authz_org: authz::Organization,
+    ) -> Params {
+        Params {
+            serialized_authn: Serialized::for_opctx(opctx),
+            project_create: params::ProjectCreate {
+                identity: IdentityMetadataCreateParams {
+                    name: "my-project".parse().unwrap(),
+                    description: "My Project".to_string(),
+                },
+            },
+            authz_org,
+        }
+    }
+
+    fn test_opctx(cptestctx: &ControlPlaneTestContext) -> OpContext {
+        OpContext::for_tests(
+            cptestctx.logctx.log.new(o!()),
+            cptestctx.server.apictx.nexus.datastore().clone(),
+        )
+    }
+
+    async fn get_authz_org(
+        cptestctx: &ControlPlaneTestContext,
+        org_id: Uuid,
+        action: authz::Action,
+    ) -> authz::Organization {
+        let nexus = &cptestctx.server.apictx.nexus;
+        let org_selector =
+            params::OrganizationSelector { organization: NameOrId::Id(org_id) };
+        let opctx = test_opctx(&cptestctx);
+        let (.., authz_org) = nexus
+            .organization_lookup(&opctx, &org_selector)
+            .expect("Invalid parameters constructing organization lookup")
+            .lookup_for(action)
+            .await
+            .expect("Organization does not exist");
+        authz_org
+    }
+
+    async fn verify_clean_slate(datastore: &DataStore) {
+        assert!(no_projects_exist(datastore).await);
+        crate::app::sagas::vpc_create::test::verify_clean_slate(datastore)
+            .await;
+    }
+
+    async fn no_projects_exist(datastore: &DataStore) -> bool {
+        use crate::db::model::Project;
+        use crate::db::schema::project::dsl;
+
+        dsl::project
+            .filter(dsl::time_deleted.is_null())
+            .select(Project::as_select())
+            .first_async::<Project>(datastore.pool_for_tests().await.unwrap())
+            .await
+            .optional()
+            .unwrap()
+            .map(|project| {
+                eprintln!("Project exists: {project:?}");
+            })
+            .is_none()
+    }
+
+    #[nexus_test(server = crate::Server)]
+    async fn test_saga_basic_usage_succeeds(
+        cptestctx: &ControlPlaneTestContext,
+    ) {
+        let client = &cptestctx.external_client;
+        let nexus = &cptestctx.server.apictx.nexus;
+        let org_id = create_org(&client).await;
+
+        // Before running the test, confirm we have no records of any projects.
+        verify_clean_slate(nexus.datastore()).await;
+
+        // Build the saga DAG with the provided test parameters
+        let opctx = test_opctx(&cptestctx);
+        let authz_org =
+            get_authz_org(&cptestctx, org_id, authz::Action::CreateChild).await;
+        let params = new_test_params(&opctx, authz_org);
+        let dag = create_saga_dag::<SagaProjectCreate>(params).unwrap();
+        let runnable_saga = nexus.create_runnable_saga(dag).await.unwrap();
+
+        // Actually run the saga
+        nexus.run_saga(runnable_saga).await.unwrap();
+    }
+
+    #[nexus_test(server = crate::Server)]
+    async fn test_action_failure_can_unwind(
+        cptestctx: &ControlPlaneTestContext,
+    ) {
+        let log = &cptestctx.logctx.log;
+
+        let client = &cptestctx.external_client;
+        let nexus = &cptestctx.server.apictx.nexus;
+        let org_id = create_org(&client).await;
+
+        // Build the saga DAG with the provided test parameters
+        let opctx = test_opctx(&cptestctx);
+        let authz_org =
+            get_authz_org(&cptestctx, org_id, authz::Action::CreateChild).await;
+        let params = new_test_params(&opctx, authz_org);
+        let dag = create_saga_dag::<SagaProjectCreate>(params).unwrap();
+
+        for node in dag.get_nodes() {
+            // Create a new saga for this node.
+            info!(
+                log,
+                "Creating new saga which will fail at index {:?}", node.index();
+                "node_name" => node.name().as_ref(),
+                "label" => node.label(),
+            );
+
+            let runnable_saga =
+                nexus.create_runnable_saga(dag.clone()).await.unwrap();
+
+            // Inject an error instead of running the node.
+            //
+            // This should cause the saga to unwind.
+            nexus
+                .sec()
+                .saga_inject_error(runnable_saga.id(), node.index())
+                .await
+                .unwrap();
+            nexus
+                .run_saga(runnable_saga)
+                .await
+                .expect_err("Saga should have failed");
+
+            verify_clean_slate(nexus.datastore()).await;
+        }
+    }
+}

nexus/src/app/sagas/vpc_create.rs

@@ -409,7 +409,7 @@ async fn svc_notify_sleds(
 }
 
 #[cfg(test)]
-mod test {
+pub(crate) mod test {
     use crate::{
         app::saga::create_saga_dag, app::sagas::vpc_create::Params,
         app::sagas::vpc_create::SagaVpcCreate, authn::saga::Serialized, authz,
@@ -551,7 +551,7 @@ mod test {
             .expect("Failed to delete VPC");
     }
 
-    async fn verify_clean_slate(datastore: &DataStore) {
+    pub(crate) async fn verify_clean_slate(datastore: &DataStore) {
         assert!(no_vpcs_exist(datastore).await);
         assert!(no_routers_exist(datastore).await);
         assert!(no_routes_exist(datastore).await);

nexus/src/db/datastore/project.rs

@@ -94,14 +94,14 @@ impl DataStore {
         opctx: &OpContext,
         org: &authz::Organization,
         project: Project,
-    ) -> CreateResult<Project> {
+    ) -> CreateResult<(authz::Project, Project)> {
         use db::schema::project::dsl;
 
         opctx.authorize(authz::Action::CreateChild, org).await?;
 
         let name = project.name().as_str().to_string();
         let organization_id = project.organization_id;
-        Organization::insert_resource(
+        let db_project = Organization::insert_resource(
             organization_id,
             diesel::insert_into(dsl::project).values(project),
         )
@@ -118,7 +118,16 @@ impl DataStore {
                     ErrorHandler::Conflict(ResourceType::Project, &name),
                 )
             }
-        })
+        })?;
+
+        Ok((
+            authz::Project::new(
+                org.clone(),
+                db_project.id(),
+                LookupType::ByName(db_project.name().to_string()),
+            ),
+            db_project,
+        ))
     }
 
     generate_fn_to_ensure_none_in_project!(instance, name, String);

nexus/src/db/queries/network_interface.rs

@@ -1728,7 +1728,7 @@ mod tests {
                 .lookup_for(authz::Action::CreateChild)
                 .await
                 .unwrap();
-            let project = db_datastore
+            let (.., project) = db_datastore
                 .project_create(&opctx, &authz_org, project)
                 .await
                 .unwrap();