Handle region snapshot replacement volume deletes

nexus/db-model/src/region_snapshot_replacement.rs

@@ -81,12 +81,12 @@ impl std::str::FromStr for RegionSnapshotReplacementState {
 ///          |                        |
 ///          v                        ---
 ///                                   ---
-///       Running                     |
-///                                   |
-///          |                        |
-///          v                        |
-///                                   | responsibility of region snapshot
-///     Completing                    | replacement finish saga
+///       Running  <--                |
+///                  |                |
+///          |       |                |
+///          v       |                |
+///                  |                | responsibility of region snapshot
+///     Completing --                 | replacement finish saga
 ///                                   |
 ///          |                        |
 ///          v                        |

nexus/db-model/src/schema_versions.rs

@@ -17,7 +17,7 @@ use std::collections::BTreeMap;
 ///
 /// This must be updated when you change the database schema.  Refer to
 /// schema/crdb/README.adoc in the root of this repository for details.
-pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(115, 0, 0);
+pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(117, 0, 0);
 
 /// List of all past database schema versions, in *reverse* order
 ///
@@ -29,7 +29,9 @@ static KNOWN_VERSIONS: Lazy<Vec<KnownVersion>> = Lazy::new(|| {
         // |  leaving the first copy as an example for the next person.
         // v
         // KnownVersion::new(next_int, "unique-dirname-with-the-sql-files"),
-        KnownVersion::new(115, "add-completing-and-new-region-volume"),
+        KnownVersion::new(117, "add-completing-and-new-region-volume"),
+        KnownVersion::new(116, "bp-physical-disk-disposition"),
+        KnownVersion::new(115, "inv-omicron-physical-disks-generation"),
         KnownVersion::new(114, "crucible-ref-count-records"),
         KnownVersion::new(113, "add-tx-eq"),
         KnownVersion::new(112, "blueprint-dataset"),

nexus/db-queries/src/db/datastore/region_replacement.rs

@@ -22,7 +22,6 @@ use crate::db::update_and_check::UpdateAndCheck;
 use crate::db::update_and_check::UpdateStatus;
 use crate::db::TransactionError;
 use crate::transaction_retry::OptionalError;
-use async_bb8_diesel::AsyncConnection;
 use async_bb8_diesel::AsyncRunQueryDsl;
 use diesel::prelude::*;
 use omicron_common::api::external::Error;
@@ -54,9 +53,11 @@ impl DataStore {
         request: RegionReplacement,
     ) -> Result<(), Error> {
         let err = OptionalError::new();
-        self.pool_connection_authorized(opctx)
-            .await?
-            .transaction_async(|conn| {
+        let conn = self.pool_connection_authorized(opctx).await?;
+
+        self.transaction_retry_wrapper("insert_region_replacement_request")
+            .transaction(&conn, |conn| {
+                let request = request.clone();
                 let err = err.clone();
                 async move {
                     use db::schema::region_replacement::dsl;

nexus/db-queries/src/db/datastore/region_snapshot_replacement.rs

@@ -22,7 +22,6 @@ use crate::db::update_and_check::UpdateAndCheck;
 use crate::db::update_and_check::UpdateStatus;
 use crate::db::TransactionError;
 use crate::transaction_retry::OptionalError;
-use async_bb8_diesel::AsyncConnection;
 use async_bb8_diesel::AsyncRunQueryDsl;
 use diesel::prelude::*;
 use omicron_common::api::external::Error;
@@ -94,44 +93,48 @@ impl DataStore {
         volume_id: Uuid,
     ) -> Result<(), Error> {
         let err = OptionalError::new();
-        self.pool_connection_authorized(opctx)
-            .await?
-            .transaction_async(|conn| {
-                let err = err.clone();
-                async move {
-                    use db::schema::region_snapshot_replacement::dsl;
+        let conn = self.pool_connection_authorized(opctx).await?;
 
-                    // An associated volume repair record isn't _strictly_
-                    // needed: snapshot volumes should never be directly
-                    // constructed, and therefore won't ever have an associated
-                    // Upstairs that receives a volume replacement request.
-                    // However it's being done in an attempt to be overly
-                    // cautious, and it validates that the volume exist:
-                    // otherwise it would be possible to create a region
-                    // snapshot replacement request for a volume that didn't
-                    // exist!
-
-                    Self::volume_repair_insert_in_txn(
-                        &conn, err, volume_id, request.id,
-                    )
-                    .await?;
+        self.transaction_retry_wrapper(
+            "insert_region_snapshot_replacement_request_with_volume_id",
+        )
+        .transaction(&conn, |conn| {
+            let request = request.clone();
+            let err = err.clone();
+            async move {
+                use db::schema::region_snapshot_replacement::dsl;
 
-                    diesel::insert_into(dsl::region_snapshot_replacement)
-                        .values(request)
-                        .execute_async(&conn)
-                        .await?;
+                // An associated volume repair record isn't _strictly_
+                // needed: snapshot volumes should never be directly
+                // constructed, and therefore won't ever have an associated
+                // Upstairs that receives a volume replacement request.
+                // However it's being done in an attempt to be overly
+                // cautious, and it validates that the volume exist:
+                // otherwise it would be possible to create a region
+                // snapshot replacement request for a volume that didn't
+                // exist!
 
-                    Ok(())
-                }
-            })
-            .await
-            .map_err(|e| {
-                if let Some(err) = err.take() {
-                    err
-                } else {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                }
-            })
+                Self::volume_repair_insert_in_txn(
+                    &conn, err, volume_id, request.id,
+                )
+                .await?;
+
+                diesel::insert_into(dsl::region_snapshot_replacement)
+                    .values(request)
+                    .execute_async(&conn)
+                    .await?;
+
+                Ok(())
+            }
+        })
+        .await
+        .map_err(|e| {
+            if let Some(err) = err.take() {
+                err
+            } else {
+                public_error_from_diesel(e, ErrorHandler::Server)
+            }
+        })
     }
 
     pub async fn get_region_snapshot_replacement_request_by_id(
@@ -685,16 +688,25 @@ impl DataStore {
         region_snapshot_replacement_id: Uuid,
         operating_saga_id: Uuid,
     ) -> Result<(), Error> {
-        type TxnError = TransactionError<Error>;
-
         let err = OptionalError::new();
         let conn = self.pool_connection_authorized(opctx).await?;
 
-        self.transaction_retry_wrapper("set_region_snapshot_replacement_complete")
-            .transaction(&conn, |conn| {
-                let err = err.clone();
-                async move {
-                    use db::schema::volume_repair::dsl as volume_repair_dsl;
+        self.transaction_retry_wrapper(
+            "set_region_snapshot_replacement_complete",
+        )
+        .transaction(&conn, |conn| {
+            let err = err.clone();
+            async move {
+                use db::schema::volume_repair::dsl as volume_repair_dsl;
+
+                diesel::delete(
+                    volume_repair_dsl::volume_repair.filter(
+                        volume_repair_dsl::repair_id
+                            .eq(region_snapshot_replacement_id),
+                    ),
+                )
+                .execute_async(&conn)
+                .await?;
 
                 use db::schema::region_snapshot_replacement::dsl;
 
@@ -716,22 +728,81 @@ impl DataStore {
                     .execute_and_check(&conn)
                     .await?;
 
+                match result.status {
+                    UpdateStatus::Updated => Ok(()),
+                    UpdateStatus::NotUpdatedButExists => {
+                        let record = result.found;
+
+                        if record.replacement_state
+                            == RegionSnapshotReplacementState::Complete
+                        {
+                            Ok(())
+                        } else {
+                            Err(err.bail(Error::conflict(format!(
+                                "region snapshot replacement {} set to {:?} \
+                                    (operating saga id {:?})",
+                                region_snapshot_replacement_id,
+                                record.replacement_state,
+                                record.operating_saga_id,
+                            ))))
+                        }
+                    }
+                }
+            }
+        })
+        .await
+        .map_err(|e| match err.take() {
+            Some(error) => error,
+            None => public_error_from_diesel(e, ErrorHandler::Server),
+        })
+    }
+
+    /// Transition a RegionSnapshotReplacement record from Requested to Complete
+    /// - this is required when the region snapshot is hard-deleted, which means
+    /// that all volume references are gone and no replacement is required. Also
+    /// removes the `volume_repair` record that is taking a "lock" on the
+    /// Volume.
+    pub async fn set_region_snapshot_replacement_complete_from_requested(
+        &self,
+        opctx: &OpContext,
+        region_snapshot_replacement_id: Uuid,
+    ) -> Result<(), Error> {
+        type TxnError = TransactionError<Error>;
+
+        let err = OptionalError::new();
+        let conn = self.pool_connection_authorized(opctx).await?;
+
+        self.transaction_retry_wrapper("set_region_snapshot_replacement_complete")
+            .transaction(&conn, |conn| {
+                let err = err.clone();
+                async move {
+                    use db::schema::volume_repair::dsl as volume_repair_dsl;
                     use db::schema::region_snapshot_replacement::dsl;
 
+                    diesel::delete(
+                        volume_repair_dsl::volume_repair.filter(
+                            volume_repair_dsl::repair_id
+                                .eq(region_snapshot_replacement_id),
+                        ),
+                    )
+                    .execute_async(&conn)
+                    .await?;
+
                     let result = diesel::update(dsl::region_snapshot_replacement)
                         .filter(dsl::id.eq(region_snapshot_replacement_id))
                         .filter(
                             dsl::replacement_state
-                                .eq(RegionSnapshotReplacementState::Running),
+                                .eq(RegionSnapshotReplacementState::Requested),
                         )
                         .filter(dsl::operating_saga_id.is_null())
-                        .set((dsl::replacement_state
-                            .eq(RegionSnapshotReplacementState::Complete),))
+                        .filter(dsl::new_region_volume_id.is_null())
+                        .set(dsl::replacement_state
+                            .eq(RegionSnapshotReplacementState::Complete))
                         .check_if_exists::<RegionSnapshotReplacement>(
                             region_snapshot_replacement_id,
                         )
                         .execute_and_check(&conn)
-                        .await?;
+                            .await?;
 
                     match result.status {
                         UpdateStatus::Updated => Ok(()),
@@ -750,7 +821,7 @@ impl DataStore {
                                     region_snapshot_replacement_id,
                                     record.replacement_state,
                                     record.operating_saga_id,
-                                ),
+                                )
                                 ))))
                             }
                         }
@@ -1220,11 +1291,16 @@ impl DataStore {
         opctx: &OpContext,
         region_snapshot_replacement_step: RegionSnapshotReplacementStep,
     ) -> Result<(), Error> {
-        type TxnError = TransactionError<Error>;
+        let conn = self.pool_connection_authorized(opctx).await?;
+        let err = OptionalError::new();
 
-        self.pool_connection_authorized(opctx)
-            .await?
-            .transaction_async(|conn| async move {
+        self.transaction_retry_wrapper(
+            "set_region_snapshot_replacement_complete",
+        )
+        .transaction(&conn, |conn| {
+            let err = err.clone();
+
+            async move {
                 use db::schema::volume_repair::dsl as volume_repair_dsl;
 
                 diesel::delete(
@@ -1267,27 +1343,23 @@ impl DataStore {
                         {
                             Ok(())
                         } else {
-                            Err(TxnError::CustomError(Error::conflict(
-                                format!(
-                                    "region snapshot replacement step {} set \
+                            Err(err.bail(Error::conflict(format!(
+                                "region snapshot replacement step {} set \
                                     to {:?} (operating saga id {:?})",
-                                    region_snapshot_replacement_step.id,
-                                    record.replacement_state,
-                                    record.operating_saga_id,
-                                ),
-                            )))
+                                region_snapshot_replacement_step.id,
+                                record.replacement_state,
+                                record.operating_saga_id,
+                            ))))
                         }
                     }
                 }
-            })
-            .await
-            .map_err(|e| match e {
-                TxnError::CustomError(error) => error,
-
-                TxnError::Database(error) => {
-                    public_error_from_diesel(error, ErrorHandler::Server)
-                }
-            })
+            }
+        })
+        .await
+        .map_err(|e| match err.take() {
+            Some(error) => error,
+            None => public_error_from_diesel(e, ErrorHandler::Server),
+        })
     }
 }