Fix ensuring snapshot creation and deletion (#3947)

jmpesp · web-flow · commit 58e8c67e3449 · 2023-08-30T12:28:57.000-04:00
If `project_ensure_snapshot` is passed a snapshot where the project and
name matches a record that exists already, but the ID is different, then

    .on_conflict((dsl::project_id, dsl::name))
    .filter_target(dsl::time_deleted.is_null())
    .do_update()
    .set(dsl::time_modified.eq(dsl::time_modified)),

will incorrectly modify the existing record, leading callers of that
function to erroneously believe they were the one to create that
snapshot record. Change `project_ensure_snapshot` to check if an
existing snapshot record has a matching project and name, and return
ObjectAlreadyExists if so.

It's not appropriate to delete a snapshot in any state: if it's in state
Creating, then a `snapshot_create` saga is working on it and
concurrently running `snapshot_delete` will cause a conflict. Add an
argument to `project_delete_snapshot` that describes the states that a
snapshot can be deleted in, depending on on the call site.
diff --git a/nexus/db-queries/src/db/datastore/snapshot.rs b/nexus/db-queries/src/db/datastore/snapshot.rs
@@ -19,9 +19,14 @@ use crate::db::model::Snapshot;
 use crate::db::model::SnapshotState;
 use crate::db::pagination::paginated;
 use crate::db::update_and_check::UpdateAndCheck;
+use crate::db::TransactionError;
+use async_bb8_diesel::AsyncConnection;
 use async_bb8_diesel::AsyncRunQueryDsl;
+use async_bb8_diesel::ConnectionError;
 use chrono::Utc;
 use diesel::prelude::*;
+use diesel::result::Error as DieselError;
+use nexus_types::identity::Resource;
 use omicron_common::api::external::http_pagination::PaginatedBy;
 use omicron_common::api::external::CreateResult;
 use omicron_common::api::external::Error;
@@ -43,28 +48,127 @@ impl DataStore {
         let gen = snapshot.gen;
         opctx.authorize(authz::Action::CreateChild, authz_project).await?;
 
-        use db::schema::snapshot::dsl;
+        #[derive(Debug, thiserror::Error)]
+        pub enum CustomError {
+            #[error("Resource already exists")]
+            ResourceAlreadyExists,
+
+            #[error("saw AsyncInsertError")]
+            InsertError(AsyncInsertError),
+        }
+
+        type TxnError = TransactionError<CustomError>;
+
+        let snapshot_name = snapshot.name().to_string();
         let project_id = snapshot.project_id;
-        let snapshot: Snapshot = Project::insert_resource(
-            project_id,
-            diesel::insert_into(dsl::snapshot)
-                .values(snapshot)
-                .on_conflict((dsl::project_id, dsl::name))
-                .filter_target(dsl::time_deleted.is_null())
-                .do_update()
-                .set(dsl::time_modified.eq(dsl::time_modified)),
-        )
-        .insert_and_get_result_async(self.pool_authorized(opctx).await?)
-        .await
-        .map_err(|e| match e {
-            AsyncInsertError::CollectionNotFound => Error::ObjectNotFound {
-                type_name: ResourceType::Project,
-                lookup_type: LookupType::ById(project_id),
-            },
-            AsyncInsertError::DatabaseError(e) => {
-                public_error_from_diesel_pool(e, ErrorHandler::Server)
-            }
-        })?;
+
+        let snapshot: Snapshot = self
+            .pool_authorized(opctx)
+            .await?
+            .transaction_async(|conn| async move {
+                use db::schema::snapshot::dsl;
+
+                // If an undeleted snapshot exists in the database with the
+                // same name and project but a different id to the snapshot
+                // this function was passed as an argument, then return an
+                // error here.
+                //
+                // As written below,
+                //
+                //    .on_conflict((dsl::project_id, dsl::name))
+                //    .filter_target(dsl::time_deleted.is_null())
+                //    .do_update()
+                //    .set(dsl::time_modified.eq(dsl::time_modified))
+                //
+                // will set any existing record's `time_modified` if the
+                // project id and name match, even if the snapshot ID does
+                // not match. diesel supports adding a filter below like so
+                // (marked with >>):
+                //
+                //    .on_conflict((dsl::project_id, dsl::name))
+                //    .filter_target(dsl::time_deleted.is_null())
+                //    .do_update()
+                //    .set(dsl::time_modified.eq(dsl::time_modified))
+                // >> .filter(dsl::id.eq(snapshot.id()))
+                //
+                // which will restrict the `insert_into`'s set so that it
+                // only applies if the snapshot ID matches. But,
+                // AsyncInsertError does not have a ObjectAlreadyExists
+                // variant, so this will be returned as CollectionNotFound
+                // due to the `insert_into` failing.
+                //
+                // If this function is passed a snapshot with an ID that
+                // does not match, but a project and name that does, return
+                // ObjectAlreadyExists here.
+
+                let existing_snapshot_id: Option<Uuid> = match dsl::snapshot
+                    .filter(dsl::time_deleted.is_null())
+                    .filter(dsl::name.eq(snapshot.name().to_string()))
+                    .filter(dsl::project_id.eq(snapshot.project_id))
+                    .select(dsl::id)
+                    .limit(1)
+                    .first_async(&conn)
+                    .await
+                {
+                    Ok(v) => Ok(Some(v)),
+                    Err(ConnectionError::Query(DieselError::NotFound)) => {
+                        Ok(None)
+                    }
+                    Err(e) => Err(e),
+                }?;
+
+                if let Some(existing_snapshot_id) = existing_snapshot_id {
+                    if existing_snapshot_id != snapshot.id() {
+                        return Err(TransactionError::CustomError(
+                            CustomError::ResourceAlreadyExists,
+                        ));
+                    }
+                }
+
+                Project::insert_resource(
+                    project_id,
+                    diesel::insert_into(dsl::snapshot)
+                        .values(snapshot)
+                        .on_conflict((dsl::project_id, dsl::name))
+                        .filter_target(dsl::time_deleted.is_null())
+                        .do_update()
+                        .set(dsl::time_modified.eq(dsl::time_modified)),
+                )
+                .insert_and_get_result_async(&conn)
+                .await
+                .map_err(|e| {
+                    TransactionError::CustomError(CustomError::InsertError(e))
+                })
+            })
+            .await
+            .map_err(|e: TxnError| match e {
+                TxnError::CustomError(e) => match e {
+                    CustomError::ResourceAlreadyExists => {
+                        Error::ObjectAlreadyExists {
+                            type_name: ResourceType::Snapshot,
+                            object_name: snapshot_name,
+                        }
+                    }
+                    CustomError::InsertError(e) => match e {
+                        AsyncInsertError::CollectionNotFound => {
+                            Error::ObjectNotFound {
+                                type_name: ResourceType::Project,
+                                lookup_type: LookupType::ById(project_id),
+                            }
+                        }
+                        AsyncInsertError::DatabaseError(e) => {
+                            public_error_from_diesel_pool(
+                                e,
+                                ErrorHandler::Server,
+                            )
+                        }
+                    },
+                },
+
+                TxnError::Pool(e) => {
+                    public_error_from_diesel_pool(e, ErrorHandler::Server)
+                }
+            })?;
 
         bail_unless!(
             snapshot.state == SnapshotState::Creating,
@@ -141,32 +245,50 @@ impl DataStore {
         opctx: &OpContext,
         authz_snapshot: &authz::Snapshot,
         db_snapshot: &Snapshot,
+        ok_to_delete_states: Vec<SnapshotState>,
     ) -> Result<Uuid, Error> {
         opctx.authorize(authz::Action::Delete, authz_snapshot).await?;
 
         let now = Utc::now();
 
-        // A snapshot can be deleted in any state. It's never attached to an
-        // instance, and any disk launched from it will copy and modify the volume
-        // construction request it's based on.
+        // A snapshot can be deleted in states Ready and Faulted. It's never
+        // attached to an instance, and any disk launched from it will copy and
+        // modify the volume construction request it's based on. However, if its
+        // state is Creating then the snapshot_create saga is currently running
+        // and this delete action would disrupt that. If its in state Destroyed,
+        // then it was already deleted.
 
         let snapshot_id = authz_snapshot.id();
         let gen = db_snapshot.gen;
 
         use db::schema::snapshot::dsl;
 
-        diesel::update(dsl::snapshot)
+        let updated_rows = diesel::update(dsl::snapshot)
             .filter(dsl::time_deleted.is_null())
             .filter(dsl::gen.eq(gen))
             .filter(dsl::id.eq(snapshot_id))
-            .set(dsl::time_deleted.eq(now))
+            .filter(dsl::state.eq_any(ok_to_delete_states))
+            .set((
+                dsl::time_deleted.eq(now),
+                dsl::state.eq(SnapshotState::Destroyed),
+            ))
             .check_if_exists::<Snapshot>(snapshot_id)
             .execute_async(self.pool_authorized(&opctx).await?)
             .await
             .map_err(|e| {
                 public_error_from_diesel_pool(e, ErrorHandler::Server)
             })?;
 
+        if updated_rows == 0 {
+            // Either:
+            //
+            // - the snapshot was already deleted
+            // - the generation number changed
+            // - the state of the snapshot isn't one of `ok_to_delete_states`
+
+            return Err(Error::invalid_request("snapshot cannot be deleted"));
+        }
+
         Ok(snapshot_id)
     }
 }
diff --git a/nexus/src/app/sagas/snapshot_create.rs b/nexus/src/app/sagas/snapshot_create.rs
@@ -594,7 +594,16 @@ async fn ssc_create_snapshot_record_undo(
 
     osagactx
         .datastore()
-        .project_delete_snapshot(&opctx, &authz_snapshot, &db_snapshot)
+        .project_delete_snapshot(
+            &opctx,
+            &authz_snapshot,
+            &db_snapshot,
+            vec![
+                db::model::SnapshotState::Creating,
+                db::model::SnapshotState::Ready,
+                db::model::SnapshotState::Faulted,
+            ],
+        )
         .await?;
 
     Ok(())
diff --git a/nexus/src/app/sagas/snapshot_delete.rs b/nexus/src/app/sagas/snapshot_delete.rs
@@ -116,6 +116,11 @@ async fn ssd_delete_snapshot_record(
             &opctx,
             &params.authz_snapshot,
             &params.snapshot,
+            vec![
+                db::model::SnapshotState::Ready,
+                db::model::SnapshotState::Faulted,
+                db::model::SnapshotState::Destroyed,
+            ],
         )
         .await
         .map_err(ActionError::action_failed)?;
diff --git a/nexus/tests/integration_tests/snapshots.rs b/nexus/tests/integration_tests/snapshots.rs
@@ -953,6 +953,7 @@ async fn test_create_snapshot_record_idempotent(
     let datastore = nexus.datastore();
 
     let project_id = create_org_and_project(&client).await;
+    let disk_id = Uuid::new_v4();
 
     let snapshot = db::model::Snapshot {
         identity: db::model::SnapshotIdentity {
@@ -968,7 +969,7 @@ async fn test_create_snapshot_record_idempotent(
         },
 
         project_id,
-        disk_id: Uuid::new_v4(),
+        disk_id,
         volume_id: Uuid::new_v4(),
         destination_volume_id: Uuid::new_v4(),
 
@@ -995,12 +996,51 @@ async fn test_create_snapshot_record_idempotent(
         .unwrap();
 
     let snapshot_created_2 = datastore
-        .project_ensure_snapshot(&opctx, &authz_project, snapshot)
+        .project_ensure_snapshot(&opctx, &authz_project, snapshot.clone())
         .await
         .unwrap();
 
     assert_eq!(snapshot_created_1.id(), snapshot_created_2.id());
 
+    // Test that attempting to ensure a snapshot with the same project + name
+    // but a different ID will not work: if a user tries to fire off multiple
+    // snapshot creations for the same project + name, they will get different
+    // IDs, so this should be rejected, ensuring only one snapshot create saga
+    // would apply for the user's multiple requests.
+
+    let dupe_snapshot = db::model::Snapshot {
+        identity: db::model::SnapshotIdentity {
+            id: Uuid::new_v4(),
+            name: external::Name::try_from("snapshot".to_string())
+                .unwrap()
+                .into(),
+            description: "snapshot".into(),
+
+            time_created: Utc::now(),
+            time_modified: Utc::now(),
+            time_deleted: None,
+        },
+
+        project_id,
+        disk_id,
+        volume_id: Uuid::new_v4(),
+        destination_volume_id: Uuid::new_v4(),
+
+        gen: db::model::Generation::new(),
+        state: db::model::SnapshotState::Creating,
+        block_size: db::model::BlockSize::Traditional,
+        size: external::ByteCount::try_from(1024u32).unwrap().into(),
+    };
+
+    let dupe_snapshot_created_err = datastore
+        .project_ensure_snapshot(&opctx, &authz_project, dupe_snapshot)
+        .await;
+
+    assert!(matches!(
+        dupe_snapshot_created_err.unwrap_err(),
+        external::Error::ObjectAlreadyExists { .. },
+    ));
+
     // Test project_delete_snapshot is idempotent
 
     let (.., authz_snapshot, db_snapshot) = LookupPath::new(&opctx, &datastore)
@@ -1010,12 +1050,31 @@ async fn test_create_snapshot_record_idempotent(
         .unwrap();
 
     datastore
-        .project_delete_snapshot(&opctx, &authz_snapshot, &db_snapshot)
+        .project_delete_snapshot(
+            &opctx,
+            &authz_snapshot,
+            &db_snapshot,
+            vec![
+                // This must match what is in the snapshot_delete saga!
+                db::model::SnapshotState::Ready,
+                db::model::SnapshotState::Faulted,
+                db::model::SnapshotState::Destroyed,
+            ],
+        )
         .await
         .unwrap();
 
     datastore
-        .project_delete_snapshot(&opctx, &authz_snapshot, &db_snapshot)
+        .project_delete_snapshot(
+            &opctx,
+            &authz_snapshot,
+            &db_snapshot,
+            vec![
+                db::model::SnapshotState::Ready,
+                db::model::SnapshotState::Faulted,
+                db::model::SnapshotState::Destroyed,
+            ],
+        )
         .await
         .unwrap();
 }
