Accept live repair status reports from Crucible

Allow any Upstairs to notify Nexus about the start or completion (plus
status) of live repairs. The motivation for this was to be used in the
final stage of region replacement to notify Nexus that the replacement
has finished, but more generally this can be used to keep track of how
many times repair occurs for each region.

Fixes #5120

common/src/api/internal/nexus.rs

@@ -9,6 +9,10 @@ use crate::api::external::{
     InstanceState, IpNet, SemverVersion, Vni,
 };
 use chrono::{DateTime, Utc};
+use omicron_uuid_kinds::DownstairsRegionKind;
+use omicron_uuid_kinds::LiveRepairKind;
+use omicron_uuid_kinds::TypedUuid;
+use omicron_uuid_kinds::UpstairsSessionKind;
 use parse_display::{Display, FromStr};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -251,3 +255,24 @@ pub enum HostIdentifier {
     Ip(IpNet),
     Vpc(Vni),
 }
+
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
+pub struct DownstairsUnderRepair {
+    pub region_uuid: TypedUuid<DownstairsRegionKind>,
+    pub target_addr: std::net::SocketAddrV6,
+}
+
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
+pub struct RepairStartInfo {
+    pub session_id: TypedUuid<UpstairsSessionKind>,
+    pub repair_id: TypedUuid<LiveRepairKind>,
+    pub repairs: Vec<DownstairsUnderRepair>,
+}
+
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
+pub struct RepairFinishInfo {
+    pub session_id: TypedUuid<UpstairsSessionKind>,
+    pub repair_id: TypedUuid<LiveRepairKind>,
+    pub repairs: Vec<DownstairsUnderRepair>,
+    pub aborted: bool,
+}

nexus/Cargo.toml

@@ -76,7 +76,6 @@ tokio-postgres = { workspace = true, features = ["with-serde_json-1"] }
 tough.workspace = true
 trust-dns-resolver.workspace = true
 uuid.workspace = true
-
 nexus-blueprint-execution.workspace = true
 nexus-defaults.workspace = true
 nexus-db-model.workspace = true
@@ -93,6 +92,7 @@ rustls = { workspace = true }
 rustls-pemfile = { workspace = true }
 update-common.workspace = true
 omicron-workspace-hack.workspace = true
+omicron-uuid-kinds.workspace = true
 
 [dev-dependencies]
 async-bb8-diesel.workspace = true

nexus/db-model/src/lib.rs

@@ -39,6 +39,7 @@ mod ipv4net;
 pub mod ipv6;
 mod ipv6net;
 mod l4_port_range;
+mod live_repair;
 mod macaddr;
 mod name;
 mod network_interface;
@@ -139,6 +140,7 @@ pub use ipv4net::*;
 pub use ipv6::*;
 pub use ipv6net::*;
 pub use l4_port_range::*;
+pub use live_repair::*;
 pub use name::*;
 pub use network_interface::*;
 pub use oximeter_info::*;

nexus/db-model/src/live_repair.rs

@@ -0,0 +1,95 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+use super::impl_enum_type;
+use crate::ipv6;
+use crate::schema::live_repair_notification;
+use crate::typed_uuid::DbTypedUuid;
+use crate::SqlU16;
+use chrono::{DateTime, Utc};
+use omicron_uuid_kinds::DownstairsRegionKind;
+use omicron_uuid_kinds::LiveRepairKind;
+use omicron_uuid_kinds::TypedUuid;
+use omicron_uuid_kinds::UpstairsKind;
+use omicron_uuid_kinds::UpstairsSessionKind;
+use serde::{Deserialize, Serialize};
+use std::net::SocketAddrV6;
+
+impl_enum_type!(
+    #[derive(SqlType, Debug, QueryId)]
+    #[diesel(postgres_type(name = "live_repair_notification_type", schema = "public"))]
+    pub struct LiveRepairNotificationTypeEnum;
+
+    #[derive(Copy, Clone, Debug, AsExpression, FromSqlRow, Serialize, Deserialize, PartialEq)]
+    #[diesel(sql_type = LiveRepairNotificationTypeEnum)]
+    pub enum LiveRepairNotificationType;
+
+    // Notification types
+    Started => b"started"
+    Succeeded => b"succeeded"
+    Failed => b"failed"
+);
+
+/// A record of Crucible live repair notifications: when a live repair started,
+/// succeeded, failed, etc.
+///
+/// Each live repair attempt is uniquely identified by the repair ID, upstairs
+/// ID, session ID, and region ID. How those change tells Nexus about what is
+/// going on:
+///
+/// - if all IDs are the same for different requests, Nexus knows that the
+///   client is retrying the notification.
+///
+/// - if the upstairs ID, session ID, and region ID are all the same, but the
+///   repair ID is different, then the same Upstairs is trying to repair that
+///   region again. This could be due to a failed first attempt, or that
+///   downstairs may have been kicked out again.
+///
+/// - if the upstairs ID and region ID are the same, but the session ID and
+///   repair ID are different, then a different session of the same Upstairs is
+///   trying to repair that Downstairs. Session IDs change each time the
+///   Upstairs is created, so it could have crashed, or it could have been
+///   migrated and the destination Propolis' Upstairs is attempting to repair
+///   the same region.
+#[derive(Queryable, Insertable, Debug, Clone, Selectable)]
+#[diesel(table_name = live_repair_notification)]
+pub struct LiveRepairNotification {
+    pub time: DateTime<Utc>,
+
+    pub repair_id: DbTypedUuid<LiveRepairKind>,
+    pub upstairs_id: DbTypedUuid<UpstairsKind>,
+    pub session_id: DbTypedUuid<UpstairsSessionKind>,
+
+    pub region_id: DbTypedUuid<DownstairsRegionKind>,
+    pub target_ip: ipv6::Ipv6Addr,
+    pub target_port: SqlU16,
+
+    pub notification_type: LiveRepairNotificationType,
+}
+
+impl LiveRepairNotification {
+    pub fn new(
+        repair_id: TypedUuid<LiveRepairKind>,
+        upstairs_id: TypedUuid<UpstairsKind>,
+        session_id: TypedUuid<UpstairsSessionKind>,
+        region_id: TypedUuid<DownstairsRegionKind>,
+        target_addr: SocketAddrV6,
+        notification_type: LiveRepairNotificationType,
+    ) -> Self {
+        Self {
+            time: Utc::now(),
+            repair_id: repair_id.into(),
+            upstairs_id: upstairs_id.into(),
+            session_id: session_id.into(),
+            region_id: region_id.into(),
+            target_ip: target_addr.ip().into(),
+            target_port: target_addr.port().into(),
+            notification_type,
+        }
+    }
+
+    pub fn address(&self) -> SocketAddrV6 {
+        SocketAddrV6::new(*self.target_ip, *self.target_port, 0, 0)
+    }
+}

nexus/db-model/src/schema.rs

@@ -13,7 +13,7 @@ use omicron_common::api::external::SemverVersion;
 ///
 /// This should be updated whenever the schema is changed. For more details,
 /// refer to: schema/crdb/README.adoc
-pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(36, 0, 0);
+pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(37, 0, 0);
 
 table! {
     disk (id) {
@@ -1517,6 +1517,22 @@ table! {
     }
 }
 
+table! {
+    live_repair_notification (repair_id, upstairs_id, session_id, region_id, notification_type) {
+        time -> Timestamptz,
+
+        repair_id -> Uuid,
+        upstairs_id -> Uuid,
+        session_id -> Uuid,
+
+        region_id -> Uuid,
+        target_ip -> Inet,
+        target_port -> Int4,
+
+        notification_type -> crate::LiveRepairNotificationTypeEnum,
+    }
+}
+
 table! {
     db_metadata (singleton) {
         singleton -> Bool,

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

@@ -6,10 +6,13 @@
 
 use super::DataStore;
 use crate::db;
+use crate::db::datastore::OpContext;
 use crate::db::error::public_error_from_diesel;
 use crate::db::error::ErrorHandler;
 use crate::db::identity::Asset;
 use crate::db::model::Dataset;
+use crate::db::model::LiveRepairNotification;
+use crate::db::model::LiveRepairNotificationType;
 use crate::db::model::Region;
 use crate::db::model::RegionSnapshot;
 use crate::db::model::Volume;
@@ -809,6 +812,101 @@ impl DataStore {
                 public_error_from_diesel(e, ErrorHandler::Server)
             })
     }
+
+    // An Upstairs is created as part of a Volume hierarchy if the Volume
+    // Construction Request includes a "Region" variant. This may be at any
+    // layer of the Volume, and some notifications will come from an Upstairs
+    // instead of the top level of the Volume. The following functions have an
+    // Upstairs ID instead of a Volume ID for this reason.
+
+    /// Record when an Upstairs notifies us about a live repair. If that record
+    /// (uniquely identified by the four IDs passed in plus the notification
+    /// type) exists already, do nothing.
+    pub async fn live_repair_notification(
+        &self,
+        opctx: &OpContext,
+        record: LiveRepairNotification,
+    ) -> Result<(), Error> {
+        use db::schema::live_repair_notification::dsl;
+
+        let conn = self.pool_connection_authorized(opctx).await?;
+        let err = OptionalError::new();
+
+        self.transaction_retry_wrapper("live_repair_notification")
+            .transaction(&conn, |conn| {
+                let record = record.clone();
+                let err = err.clone();
+
+                async move {
+                    match &record.notification_type {
+                        LiveRepairNotificationType::Started => {
+                            // Proceed - the insertion can succeed or fail below
+                            // based on the table's primary key
+                        }
+
+                        LiveRepairNotificationType::Succeeded
+                        | LiveRepairNotificationType::Failed => {
+                            // However, Nexus must accept only one "finished"
+                            // status - an Upstairs cannot change this and must
+                            // instead perform another repair with a new repair
+                            // ID.
+                            let maybe_existing_finish_record: Option<
+                                LiveRepairNotification,
+                            > = dsl::live_repair_notification
+                                .filter(dsl::repair_id.eq(record.repair_id))
+                                .filter(dsl::upstairs_id.eq(record.upstairs_id))
+                                .filter(dsl::session_id.eq(record.session_id))
+                                .filter(dsl::region_id.eq(record.region_id))
+                                .filter(dsl::notification_type.eq_any(vec![
+                                    LiveRepairNotificationType::Succeeded,
+                                    LiveRepairNotificationType::Failed,
+                                ]))
+                                .get_result_async(&conn)
+                                .await
+                                .optional()?;
+
+                            if let Some(existing_finish_record) =
+                                maybe_existing_finish_record
+                            {
+                                if existing_finish_record.notification_type
+                                    != record.notification_type
+                                {
+                                    return Err(err.bail(Error::conflict(
+                                        "existing finish record does not match",
+                                    )));
+                                } else {
+                                    // inserting the same record, bypass
+                                    return Ok(());
+                                }
+                            }
+                        }
+                    }
+
+                    diesel::insert_into(dsl::live_repair_notification)
+                        .values(record)
+                        .on_conflict((
+                            dsl::repair_id,
+                            dsl::upstairs_id,
+                            dsl::session_id,
+                            dsl::region_id,
+                            dsl::notification_type,
+                        ))
+                        .do_nothing()
+                        .execute_async(&conn)
+                        .await?;
+
+                    Ok(())
+                }
+            })
+            .await
+            .map_err(|e| {
+                if let Some(err) = err.take() {
+                    err
+                } else {
+                    public_error_from_diesel(e, ErrorHandler::Server)
+                }
+            })
+    }
 }
 
 #[derive(Default, Clone, Debug, Serialize, Deserialize)]

nexus/db-queries/src/db/pool_connection.rs

@@ -51,6 +51,7 @@ static CUSTOM_TYPE_KEYS: &'static [&'static str] = &[
     "ip_attach_state",
     "ip_kind",
     "ip_pool_resource_type",
+    "live_repair_notification_type",
     "network_interface_kind",
     "physical_disk_kind",
     "producer_kind",