Only perform region replacement for r/w regions (#7243)

Region replacement does not work for read-only regions (there is a check
in the region_replacement_start saga that bails out if the supplied
region is read-only) - this is tracked by #6172. The current plan to
make read-only region replacement work is to use the same machinery as
region snapshot replacements (as they're both read-only), so this commit
changes the current query that finds expunged regions to replace to only
return read/write regions. This can be changed back in the future if the
plan for #6172 changes.

Author: jmpesp

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

@@ -422,8 +422,8 @@ impl DataStore {
         }
     }
 
-    /// Find regions on expunged disks
-    pub async fn find_regions_on_expunged_physical_disks(
+    /// Find read/write regions on expunged disks
+    pub async fn find_read_write_regions_on_expunged_physical_disks(
         &self,
         opctx: &OpContext,
     ) -> LookupResult<Vec<Region>> {
@@ -450,6 +450,8 @@ impl DataStore {
                     ))
                     .select(dataset_dsl::id)
             ))
+            // only return read-write regions here
+            .filter(region_dsl::read_only.eq(false))
             .select(Region::as_select())
             .load_async(&*conn)
             .await

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

@@ -37,6 +37,13 @@ impl DataStore {
         opctx: &OpContext,
         region: &Region,
     ) -> Result<Uuid, Error> {
+        if region.read_only() {
+            return Err(Error::invalid_request(format!(
+                "region {} is read-only",
+                region.id(),
+            )));
+        }
+
         let request = RegionReplacement::for_region(region);
         let request_id = request.id;
 

nexus/src/app/background/tasks/region_replacement.rs

@@ -68,17 +68,18 @@ impl BackgroundTask for RegionReplacementDetector {
 
             let mut status = RegionReplacementStatus::default();
 
-            // Find regions on expunged physical disks
+            // Find read/write regions on expunged physical disks
             let regions_to_be_replaced = match self
                 .datastore
-                .find_regions_on_expunged_physical_disks(opctx)
+                .find_read_write_regions_on_expunged_physical_disks(opctx)
                 .await
             {
                 Ok(regions) => regions,
 
                 Err(e) => {
                     let s = format!(
-                        "find_regions_on_expunged_physical_disks failed: {e}"
+                        "find_read_write_regions_on_expunged_physical_disks \
+                        failed: {e}"
                     );
                     error!(&log, "{s}");
                     status.errors.push(s);

nexus/tests/integration_tests/disks.rs

@@ -2577,7 +2577,7 @@ async fn test_disk_expunge(cptestctx: &ControlPlaneTestContext) {
 
     // All three regions should be returned
     let expunged_regions = datastore
-        .find_regions_on_expunged_physical_disks(&opctx)
+        .find_read_write_regions_on_expunged_physical_disks(&opctx)
         .await
         .unwrap();