Explicitly separate region deletion code

The garbage collection of read/write regions must be separate from
read-only regions:

- read/write regions are garbage collected by either being deleted
  during a volume soft delete, or by appearing later during the "find
  deleted volume regions" section of the volume delete saga

- read-only regions are garbage collected only in the volume soft delete
  code, when there are no more references to them

`find_deleted_volume_regions` was changed to only operate on read/write
regions, and no longer returns the optional RegionSnapshot object: that
check was moved from the volume delete saga into the function, as it
didn't make sense that it was separated.

This commit also adds checks to validate that invariants related to
volumes are not violated during tests. One invalid test was deleted
(regions will never be deleted when they're in use!)

In order to properly test the separate region deletion routines, the
first part of the fixes for dealing with deleted volumes during region
snapshot replacement were brought in from that branch: these are the
changes to region_snapshot_replacement_step.rs and
region_snapshot_replacement_start.rs.

dev-tools/omdb/src/bin/omdb/db.rs

@@ -2451,26 +2451,22 @@ async fn cmd_db_region_find_deleted(
     struct Row {
         dataset_id: Uuid,
         region_id: Uuid,
-        region_snapshot_id: String,
-        volume_id: Uuid,
+        volume_id: String,
     }
 
     let rows: Vec<Row> = datasets_regions_volumes
         .into_iter()
         .map(|row| {
-            let (dataset, region, region_snapshot, volume) = row;
+            let (dataset, region, volume) = row;
 
             Row {
                 dataset_id: dataset.id(),
                 region_id: region.id(),
-                region_snapshot_id: if let Some(region_snapshot) =
-                    region_snapshot
-                {
-                    region_snapshot.snapshot_id.to_string()
+                volume_id: if let Some(volume) = volume {
+                    volume.id().to_string()
                 } else {
                     String::from("")
                 },
-                volume_id: volume.id(),
             }
         })
         .collect();

nexus/db-model/src/volume_resource_usage.rs

@@ -29,7 +29,7 @@ impl_enum_type!(
 ///
 /// Read-only resources can be used by many volumes, and because of this they
 /// need to have a reference count so they can be deleted when they're not
-/// referenced anymore. The region_snapshot table uses a `volume_references`
+/// referenced anymore. The region_snapshot table used a `volume_references`
 /// column, which counts how many uses there are. The region table does not have
 /// this column, and more over a simple integer works for reference counting but
 /// does not tell you _what_ volume that use is from. This can be determined

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

@@ -362,6 +362,12 @@ impl DataStore {
                             )
                             .execute_async(&conn).await?;
                     }
+
+                    // Whenever a region is hard-deleted, validate invariants
+                    // for all volumes
+                    #[cfg(any(test, feature = "testing"))]
+                    Self::validate_volume_invariants(&conn).await?;
+
                     Ok(())
                 }
             })

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

@@ -64,14 +64,25 @@ impl DataStore {
     ) -> DeleteResult {
         use db::schema::region_snapshot::dsl;
 
-        diesel::delete(dsl::region_snapshot)
+        let conn = self.pool_connection_unauthorized().await?;
+
+        let result = diesel::delete(dsl::region_snapshot)
             .filter(dsl::dataset_id.eq(dataset_id))
             .filter(dsl::region_id.eq(region_id))
             .filter(dsl::snapshot_id.eq(snapshot_id))
-            .execute_async(&*self.pool_connection_unauthorized().await?)
+            .execute_async(&*conn)
             .await
             .map(|_rows_deleted| ())
-            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server));
+
+        // Whenever a region snapshot is hard-deleted, validate invariants for
+        // all volumes
+        #[cfg(any(test, feature = "testing"))]
+        Self::validate_volume_invariants(&conn)
+            .await
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?;
+
+        result
     }
 
     /// Find region snapshots on expunged disks

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

@@ -322,4 +322,23 @@ impl DataStore {
             .optional()
             .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))
     }
+
+    /// Get a snapshot, returning None if it does not exist (instead of a
+    /// NotFound error).
+    pub async fn snapshot_get(
+        &self,
+        opctx: &OpContext,
+        snapshot_id: Uuid,
+    ) -> LookupResult<Option<Snapshot>> {
+        let conn = self.pool_connection_authorized(opctx).await?;
+
+        use db::schema::snapshot::dsl;
+        dsl::snapshot
+            .filter(dsl::id.eq(snapshot_id))
+            .select(Snapshot::as_select())
+            .first_async(&*conn)
+            .await
+            .optional()
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))
+    }
 }