Accept notifications from Crucible

clients/nexus-client/src/lib.rs

@@ -35,8 +35,8 @@ progenitor::generate_api!(
         NewPasswordHash = omicron_passwords::NewPasswordHash,
 
         TypedUuidForUpstairsKind = omicron_uuid_kinds::TypedUuid<omicron_uuid_kinds::UpstairsKind>,
+        TypedUuidForUpstairsRepairKind = omicron_uuid_kinds::TypedUuid<omicron_uuid_kinds::UpstairsRepairKind>,
         TypedUuidForUpstairsSessionKind = omicron_uuid_kinds::TypedUuid<omicron_uuid_kinds::UpstairsSessionKind>,
-        TypedUuidForLiveRepairKind = omicron_uuid_kinds::TypedUuid<omicron_uuid_kinds::LiveRepairKind>,
     }
 );
 

common/Cargo.toml

@@ -37,6 +37,7 @@ toml.workspace = true
 uuid.workspace = true
 parse-display.workspace = true
 progenitor.workspace = true
+progenitor-client.workspace = true
 omicron-workspace-hack.workspace = true
 once_cell.workspace = true
 regress.workspace = true

common/src/api/internal/nexus.rs

@@ -10,8 +10,8 @@ use crate::api::external::{
 };
 use chrono::{DateTime, Utc};
 use omicron_uuid_kinds::DownstairsRegionKind;
-use omicron_uuid_kinds::LiveRepairKind;
 use omicron_uuid_kinds::TypedUuid;
+use omicron_uuid_kinds::UpstairsRepairKind;
 use omicron_uuid_kinds::UpstairsSessionKind;
 use parse_display::{Display, FromStr};
 use schemars::JsonSchema;
@@ -256,6 +256,13 @@ pub enum HostIdentifier {
     Vpc(Vni),
 }
 
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum UpstairsRepairType {
+    Live,
+    Reconciliation,
+}
+
 #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
 pub struct DownstairsUnderRepair {
     pub region_uuid: TypedUuid<DownstairsRegionKind>,
@@ -264,15 +271,26 @@ pub struct DownstairsUnderRepair {
 
 #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
 pub struct RepairStartInfo {
+    pub time: DateTime<Utc>,
     pub session_id: TypedUuid<UpstairsSessionKind>,
-    pub repair_id: TypedUuid<LiveRepairKind>,
+    pub repair_id: TypedUuid<UpstairsRepairKind>,
+    pub repair_type: UpstairsRepairType,
     pub repairs: Vec<DownstairsUnderRepair>,
 }
 
 #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
 pub struct RepairFinishInfo {
+    pub time: DateTime<Utc>,
     pub session_id: TypedUuid<UpstairsSessionKind>,
-    pub repair_id: TypedUuid<LiveRepairKind>,
+    pub repair_id: TypedUuid<UpstairsRepairKind>,
+    pub repair_type: UpstairsRepairType,
     pub repairs: Vec<DownstairsUnderRepair>,
     pub aborted: bool,
 }
+
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
+pub struct RepairProgress {
+    pub time: DateTime<Utc>,
+    pub current_item: i64,
+    pub total_items: i64,
+}

common/src/lib.rs

@@ -79,3 +79,84 @@ impl slog::KV for FileKv {
 }
 
 pub const OMICRON_DPD_TAG: &str = "omicron";
+
+use futures::Future;
+use slog::warn;
+
+/// Retry a progenitor client operation until a known result is returned.
+///
+/// Saga execution relies on the outcome of an external call being known: since
+/// they are idempotent, reissue the external call until a known result comes
+/// back. Retry if a communication error is seen, or if another retryable error
+/// is seen.
+///
+/// Note that retrying is only valid if the call itself is idempotent.
+pub async fn retry_until_known_result<F, T, E, Fut>(
+    log: &slog::Logger,
+    mut f: F,
+) -> Result<T, progenitor_client::Error<E>>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, progenitor_client::Error<E>>>,
+    E: std::fmt::Debug,
+{
+    backoff::retry_notify(
+        backoff::retry_policy_internal_service(),
+        move || {
+            let fut = f();
+            async move {
+                match fut.await {
+                    Err(progenitor_client::Error::CommunicationError(e)) => {
+                        warn!(
+                            log,
+                            "saw transient communication error {}, retrying...",
+                            e,
+                        );
+
+                        Err(backoff::BackoffError::transient(
+                            progenitor_client::Error::CommunicationError(e),
+                        ))
+                    }
+
+                    Err(progenitor_client::Error::ErrorResponse(
+                        response_value,
+                    )) => {
+                        match response_value.status() {
+                            // Retry on 503 or 429
+                            http::StatusCode::SERVICE_UNAVAILABLE
+                            | http::StatusCode::TOO_MANY_REQUESTS => {
+                                Err(backoff::BackoffError::transient(
+                                    progenitor_client::Error::ErrorResponse(
+                                        response_value,
+                                    ),
+                                ))
+                            }
+
+                            // Anything else is a permanent error
+                            _ => Err(backoff::BackoffError::Permanent(
+                                progenitor_client::Error::ErrorResponse(
+                                    response_value,
+                                ),
+                            )),
+                        }
+                    }
+
+                    Err(e) => {
+                        warn!(log, "saw permanent error {}, aborting", e,);
+
+                        Err(backoff::BackoffError::Permanent(e))
+                    }
+
+                    Ok(v) => Ok(v),
+                }
+            }
+        },
+        |error: progenitor_client::Error<_>, delay| {
+            warn!(
+                log,
+                "failed external call ({:?}), will retry in {:?}", error, delay,
+            );
+        },
+    )
+    .await
+}

nexus/Cargo.toml

@@ -76,7 +76,7 @@ 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
 nexus-db-queries.workspace = true

nexus/db-model/src/lib.rs

@@ -39,7 +39,6 @@ mod ipv4net;
 pub mod ipv6;
 mod ipv6net;
 mod l4_port_range;
-mod live_repair;
 mod macaddr;
 mod name;
 mod network_interface;
@@ -85,6 +84,7 @@ mod switch;
 mod tuf_repo;
 mod typed_uuid;
 mod unsigned;
+mod upstairs_repair;
 mod user_builtin;
 mod utilization;
 mod virtual_provisioning_collection;
@@ -141,7 +141,6 @@ 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::*;
@@ -176,6 +175,7 @@ pub use switch_interface::*;
 pub use switch_port::*;
 pub use tuf_repo::*;
 pub use typed_uuid::to_db_typed_uuid;
+pub use upstairs_repair::*;
 pub use user_builtin::*;
 pub use utilization::*;
 pub use virtual_provisioning_collection::*;

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(37, 0, 0);
+pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(38, 0, 0);
 
 table! {
     disk (id) {
@@ -1520,18 +1520,28 @@ table! {
 }
 
 table! {
-    live_repair_notification (repair_id, upstairs_id, session_id, region_id, notification_type) {
+    upstairs_repair_notification (repair_id, upstairs_id, session_id, region_id, notification_type) {
         time -> Timestamptz,
 
         repair_id -> Uuid,
+        repair_type -> crate::UpstairsRepairTypeEnum,
         upstairs_id -> Uuid,
         session_id -> Uuid,
 
         region_id -> Uuid,
         target_ip -> Inet,
         target_port -> Int4,
 
-        notification_type -> crate::LiveRepairNotificationTypeEnum,
+        notification_type -> crate::UpstairsRepairNotificationTypeEnum,
+    }
+}
+
+table! {
+    upstairs_repair_progress (repair_id, time, current_item, total_items) {
+        repair_id -> Uuid,
+        time -> Timestamptz,
+        current_item -> Int8,
+        total_items -> Int8,
     }
 }