Clean up volume usage records created by sagas

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

@@ -467,9 +467,18 @@ impl DataStore {
     ) -> DeleteResult {
         use db::schema::volume::dsl;
 
+        let conn = self.pool_connection_unauthorized().await?;
+
+        // If running integration tests, assert that the volume we're about to
+        // hard delete has no volume usage records.
+        #[cfg(any(test, feature = "testing"))]
+        Self::validate_volume_has_no_usage_records(&conn, volume_id)
+            .await
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?;
+
         diesel::delete(dsl::volume)
             .filter(dsl::id.eq(to_db_typed_uuid(volume_id)))
-            .execute_async(&*self.pool_connection_unauthorized().await?)
+            .execute_async(&*conn)
             .await
             .map(|_| ())
             .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))
@@ -4061,6 +4070,32 @@ impl DataStore {
 
         Ok(())
     }
+
+    async fn validate_volume_has_no_usage_records(
+        conn: &async_bb8_diesel::Connection<DbConnection>,
+        volume_id: VolumeUuid,
+    ) -> Result<(), diesel::result::Error> {
+        use db::schema::volume_resource_usage::dsl;
+
+        let matching_usage_records: Vec<VolumeResourceUsage> =
+            dsl::volume_resource_usage
+                .filter(dsl::volume_id.eq(to_db_typed_uuid(volume_id)))
+                .select(VolumeResourceUsageRecord::as_select())
+                .get_results_async(conn)
+                .await?
+                .into_iter()
+                .map(|r| r.try_into().unwrap())
+                .collect();
+
+        if !matching_usage_records.is_empty() {
+            return Err(Self::volume_invariant_violated(format!(
+                "volume {volume_id} has matching usage records: {:?}",
+                matching_usage_records,
+            )));
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

nexus/src/app/sagas/region_snapshot_replacement_start.rs

@@ -911,6 +911,15 @@ async fn rsrss_new_region_volume_create_undo(
 
     let new_region_volume_id =
         sagactx.lookup::<VolumeUuid>("new_region_volume_id")?;
+
+    if osagactx.datastore().volume_get(new_region_volume_id).await?.is_some() {
+        // All the knowledge to unwind the resources created by this saga is in
+        // this saga, but use soft delete in order to keep volume resource usage
+        // records consistent (they would have been added in the volume create).
+        // Make sure to only call this if the volume still exists.
+        osagactx.datastore().soft_delete_volume(new_region_volume_id).await?;
+    }
+
     osagactx.datastore().volume_hard_delete(new_region_volume_id).await?;
 
     Ok(())
@@ -1008,6 +1017,15 @@ async fn rsrss_create_fake_volume_undo(
     // Delete the fake volume.
 
     let new_volume_id = sagactx.lookup::<VolumeUuid>("new_volume_id")?;
+
+    if osagactx.datastore().volume_get(new_volume_id).await?.is_some() {
+        // All the knowledge to unwind the resources created by this saga is in
+        // this saga, but use soft delete in order to keep volume resource usage
+        // records consistent (they would have been added in the volume create).
+        // Make sure to only call this if the volume still exists.
+        osagactx.datastore().soft_delete_volume(new_volume_id).await?;
+    }
+
     osagactx.datastore().volume_hard_delete(new_volume_id).await?;
 
     Ok(())