Implement record based Crucible reference counting

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

@@ -2564,26 +2564,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/region.rs

@@ -46,6 +46,10 @@ pub struct Region {
 
     // A region may be read-only
     read_only: bool,
+
+    // Shared read-only regions require a "deleting" flag to avoid a
+    // use-after-free scenario
+    deleting: bool,
 }
 
 impl Region {
@@ -67,6 +71,7 @@ impl Region {
             extent_count: extent_count as i64,
             port: Some(port.into()),
             read_only,
+            deleting: false,
         }
     }
 
@@ -99,4 +104,7 @@ impl Region {
     pub fn read_only(&self) -> bool {
         self.read_only
     }
+    pub fn deleting(&self) -> bool {
+        self.deleting
+    }
 }

nexus/db-model/src/schema.rs

@@ -1063,6 +1063,8 @@ table! {
         port -> Nullable<Int4>,
 
         read_only -> Bool,
+
+        deleting -> Bool,
     }
 }
 

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(109, 0, 0);
+pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(113, 0, 0);
 
 /// List of all past database schema versions, in *reverse* order
 ///
@@ -29,7 +29,11 @@ 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(109, "crucible-ref-count-records"),
+        KnownVersion::new(113, "crucible-ref-count-records"),
+        KnownVersion::new(112, "blueprint-dataset"),
+        KnownVersion::new(111, "drop-omicron-zone-underlay-address"),
+        KnownVersion::new(110, "clickhouse-policy"),
+        KnownVersion::new(109, "inv-clickhouse-keeper-membership"),
         KnownVersion::new(108, "internet-gateway"),
         KnownVersion::new(107, "add-instance-boot-disk"),
         KnownVersion::new(106, "dataset-kinds-update"),

nexus/db-model/src/volume_resource_usage.rs

@@ -29,13 +29,14 @@ 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
 /// (see omdb's validate volume references command) but it's information that is
-/// tossed out, as Nexus knows what volumes use what resources! Instead, record
-/// what read-only resources a volume uses here.
+/// tossed out, as Nexus knows what volumes use what resources! Instead of
+/// throwing away that knowledge and only incrementing and decrementing an
+/// integer, record what read-only resources a volume uses in this table.
 ///
 /// Note: users should not use this object directly, and instead use the
 /// [`VolumeResourceUsage`] enum, which is type-safe and will convert to and
@@ -101,31 +102,40 @@ impl VolumeResourceUsageRecord {
     }
 }
 
-impl From<VolumeResourceUsageRecord> for VolumeResourceUsage {
-    fn from(record: VolumeResourceUsageRecord) -> VolumeResourceUsage {
+impl TryFrom<VolumeResourceUsageRecord> for VolumeResourceUsage {
+    type Error = String;
+
+    fn try_from(
+        record: VolumeResourceUsageRecord,
+    ) -> Result<VolumeResourceUsage, String> {
         match record.usage_type {
             VolumeResourceUsageType::ReadOnlyRegion => {
-                VolumeResourceUsage::ReadOnlyRegion {
-                    region_id: record
-                        .region_id
-                        .expect("valid read-only region usage record"),
-                }
+                let Some(region_id) = record.region_id else {
+                    return Err("valid read-only region usage record".into());
+                };
+
+                Ok(VolumeResourceUsage::ReadOnlyRegion { region_id })
             }
 
             VolumeResourceUsageType::RegionSnapshot => {
-                VolumeResourceUsage::RegionSnapshot {
-                    dataset_id: record
-                        .region_snapshot_dataset_id
-                        .expect("valid region snapshot usage record"),
-
-                    region_id: record
-                        .region_snapshot_region_id
-                        .expect("valid region snapshot usage record"),
-
-                    snapshot_id: record
-                        .region_snapshot_snapshot_id
-                        .expect("valid region snapshot usage record"),
-                }
+                let Some(dataset_id) = record.region_snapshot_dataset_id else {
+                    return Err("valid region snapshot usage record".into());
+                };
+
+                let Some(region_id) = record.region_snapshot_region_id else {
+                    return Err("valid region snapshot usage record".into());
+                };
+
+                let Some(snapshot_id) = record.region_snapshot_snapshot_id
+                else {
+                    return Err("valid region snapshot usage record".into());
+                };
+
+                Ok(VolumeResourceUsage::RegionSnapshot {
+                    dataset_id,
+                    region_id,
+                    snapshot_id,
+                })
             }
         }
     }

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))
+    }
 }