From 6872e2d9b4410085bf2baa3c51c5dde3be2daf11 Mon Sep 17 00:00:00 2001 From: Thomas McRoberts Date: Wed, 8 Jul 2026 07:53:23 +0000 Subject: [PATCH 1/5] nvlink: use switch state machine to rotate certs Instead of rotating the certs via RMS directly in the switch cert monitor, exercise the switch state machine. Signed-off-by: Thomas McRoberts --- crates/api-core/src/setup.rs | 3 - .../src/tests/common/api_fixtures/mod.rs | 3 - crates/api-core/src/tests/nvl_instance.rs | 125 ++----- crates/nvlink-manager/src/lib.rs | 14 - .../nvlink-manager/src/switch_cert_monitor.rs | 319 ++---------------- 5 files changed, 54 insertions(+), 410 deletions(-) diff --git a/crates/api-core/src/setup.rs b/crates/api-core/src/setup.rs index 1716f4626f..b574b77021 100644 --- a/crates/api-core/src/setup.rs +++ b/crates/api-core/src/setup.rs @@ -1493,9 +1493,6 @@ async fn initialize_and_start_controllers<'a>( meter: meter.clone(), config: carbide_config.nvlink_config.clone().unwrap_or_default(), host_health: carbide_config.host_health, - rms_client: api_service.rms_client.clone(), - credential_manager: api_service.credential_manager.clone(), - rack_profiles: carbide_config.rack_profiles.clone(), work_lock_manager_handle: work_lock_manager_handle.clone(), }) .start(join_set, cancel_token.clone())?; diff --git a/crates/api-core/src/tests/common/api_fixtures/mod.rs b/crates/api-core/src/tests/common/api_fixtures/mod.rs index a171ef29bd..3dc0206b5f 100644 --- a/crates/api-core/src/tests/common/api_fixtures/mod.rs +++ b/crates/api-core/src/tests/common/api_fixtures/mod.rs @@ -1455,9 +1455,6 @@ pub async fn create_test_env_with_overrides( db_pool.clone(), test_meter.meter(), config.nvlink_config.clone().unwrap(), - rms_sim.as_rms_client(), - composite_manager.clone(), - config.rack_profiles.clone(), api.work_lock_manager_handle.clone(), ); diff --git a/crates/api-core/src/tests/nvl_instance.rs b/crates/api-core/src/tests/nvl_instance.rs index 7221ffcc86..50f9833a6e 100644 --- a/crates/api-core/src/tests/nvl_instance.rs +++ b/crates/api-core/src/tests/nvl_instance.rs @@ -38,13 +38,12 @@ use common::api_fixtures::{ use db::switch as db_switch; use ipnetwork::IpNetwork; use libnmxc::nmxc_model::{GetGpuInfoListRequest, GetPartitionInfoListRequest, GpuAttr}; -use librms::protos::rack_manager as rms; use model::expected_switch::ExpectedSwitch; use model::instance::config::nvlink::InstanceNvLinkConfig; use model::metadata::Metadata; use model::switch::{ CONTROL_PLANE_STATE_CONFIGURED, FabricManagerState, FabricManagerStatus, NewSwitch, - SwitchConfig, SwitchControllerState, + SwitchConfig, SwitchControllerState, SwitchMaintenanceOperation, }; use model::test_support::{HardwareInfoTemplate, ManagedHostConfig}; use rcgen::{CertifiedKey, generate_simple_self_signed}; @@ -2644,21 +2643,6 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( let switch_id = create_rack_switch_for_nmxc_simulator(&env, &rack_id).await; - if expected_fingerprint_mismatches > 0 { - env.rms_sim - .queue_configure_switch_certificate_response(Ok( - rms::ConfigureSwitchCertificateResponse { - response: Some(rms::NodeBatchResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - ..Default::default() - }), - ..Default::default() - }, - )) - .await; - } - let result = env.run_switch_cert_monitor_iteration().await; assert_eq!(result.observed_endpoints, 1); assert_eq!(result.successful_probes, 1); @@ -2672,92 +2656,39 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( assert_eq!(result.pending_updates, expected_fingerprint_mismatches); assert_eq!(result.apply_errors, 0); - let rms_requests = env - .rms_sim - .submitted_configure_switch_certificate_requests() - .await; - assert_eq!(rms_requests.len(), expected_fingerprint_mismatches); + let mut txn = pool.begin().await.expect("begin txn"); + let switch = db_switch::find_by_id(txn.as_mut(), &switch_id) + .await + .expect("load switch") + .expect("switch"); + txn.commit().await.expect("commit txn"); + if expected_fingerprint_mismatches > 0 { - let request = rms_requests.first().expect("RMS request"); - assert_eq!( - request.services, - vec![rms::SwitchService::ScaleUpFabricManager as i32] - ); - assert!(request.test_hello); - assert_eq!(request.domain.as_deref(), Some(rack_id.as_ref())); - let node = request - .nodes + let request = switch + .switch_maintenance_requested .as_ref() - .expect("RMS request node set") - .nodes - .first() - .expect("RMS request node"); - assert_eq!(node.node_id, switch_id.to_string()); - assert_eq!(node.rack_id, rack_id.to_string()); - - env.rms_sim - .queue_get_configure_switch_certificate_job_status_response(Ok( - rms::GetConfigureSwitchCertificateJobStatusResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - state: "running".to_string(), - ..Default::default() - }, - )) - .await; - let result = env.run_switch_cert_monitor_iteration().await; - assert_eq!(result.observed_endpoints, 1); - assert_eq!(result.successful_probes, 1); + .expect("switch certificate maintenance request"); assert_eq!( - result.fingerprint_mismatches, - expected_fingerprint_mismatches + request.operation, + SwitchMaintenanceOperation::ReconfigureCertificate ); - assert_eq!(result.applied_updates, 0); - assert_eq!(result.pending_updates, expected_fingerprint_mismatches); - assert_eq!(result.apply_errors, 0); - assert_eq!( - env.rms_sim - .submitted_configure_switch_certificate_requests() - .await - .len(), - 1 - ); - - let job_status_requests = env - .rms_sim - .submitted_get_configure_switch_certificate_job_status_requests() - .await; - assert_eq!(job_status_requests.len(), 1); - assert_eq!(job_status_requests[0].job_id, "test-switch-cert-job"); - - env.rms_sim - .queue_get_configure_switch_certificate_job_status_response(Ok( - rms::GetConfigureSwitchCertificateJobStatusResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - state: "completed".to_string(), - ..Default::default() - }, - )) - .await; - let result = env.run_switch_cert_monitor_iteration().await; - assert_eq!(result.observed_endpoints, 1); - assert_eq!(result.successful_probes, 1); - assert_eq!( - result.fingerprint_mismatches, - expected_fingerprint_mismatches - ); - assert_eq!(result.applied_updates, expected_fingerprint_mismatches); - assert_eq!(result.pending_updates, 0); - assert_eq!(result.apply_errors, 0); - assert_eq!( - env.rms_sim - .submitted_configure_switch_certificate_requests() - .await - .len(), - 1 + assert_eq!(request.initiator, "nvlink-switch-cert-monitor"); + } else { + assert!( + switch.switch_maintenance_requested.is_none(), + "matching certificate should not request switch certificate maintenance" ); } + + let rms_requests = env + .rms_sim + .submitted_configure_switch_certificate_requests() + .await; + assert_eq!( + rms_requests.len(), + 0, + "switch cert monitor should queue the switch state machine instead of calling RMS directly" + ); } #[crate::sqlx_test] diff --git a/crates/nvlink-manager/src/lib.rs b/crates/nvlink-manager/src/lib.rs index 5e677c6bcd..ff7d3af343 100644 --- a/crates/nvlink-manager/src/lib.rs +++ b/crates/nvlink-manager/src/lib.rs @@ -27,7 +27,6 @@ use std::io; use std::sync::Arc; use std::time::Duration; -use carbide_secrets::credentials::CredentialManager; use carbide_utils::periodic_timer::PeriodicTimer; use carbide_uuid::machine::MachineId; use carbide_uuid::nvlink::{NvLinkDomainId, NvLinkLogicalPartitionId, NvLinkPartitionId}; @@ -59,7 +58,6 @@ use model::machine::nvlink::{MachineNvLinkGpuStatusObservation, MachineNvLinkSta use model::machine::{HostHealthConfig, LoadSnapshotOptions, ManagedHostStateSnapshot}; use model::nvl_logical_partition::LogicalPartition; use model::nvl_partition::{NvlPartition, NvlPartitionName}; -use model::rack_type::RackProfileConfig; use sqlx::PgPool; #[cfg(feature = "test-support")] pub use switch_cert_monitor::{SwitchCertificateMonitor, SwitchCertificateMonitorIterationResult}; @@ -900,9 +898,6 @@ pub struct NvLinkManager { meter: opentelemetry::metrics::Meter, config: NvLinkConfig, host_health: HostHealthConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, work_lock_manager_handle: WorkLockManagerHandle, } @@ -912,9 +907,6 @@ pub struct NvLinkManagerArgs { pub meter: opentelemetry::metrics::Meter, pub config: NvLinkConfig, pub host_health: HostHealthConfig, - pub rms_client: Option>, - pub credential_manager: Arc, - pub rack_profiles: RackProfileConfig, pub work_lock_manager_handle: WorkLockManagerHandle, } @@ -926,9 +918,6 @@ impl NvLinkManager { meter: args.meter, config: args.config, host_health: args.host_health, - rms_client: args.rms_client, - credential_manager: args.credential_manager, - rack_profiles: args.rack_profiles, work_lock_manager_handle: args.work_lock_manager_handle, } } @@ -953,9 +942,6 @@ impl NvLinkManager { self.db_pool, self.meter, self.config, - self.rms_client, - self.credential_manager, - self.rack_profiles, self.work_lock_manager_handle, ); join_set diff --git a/crates/nvlink-manager/src/switch_cert_monitor.rs b/crates/nvlink-manager/src/switch_cert_monitor.rs index 02f6a4c8b9..f034c490e4 100644 --- a/crates/nvlink-manager/src/switch_cert_monitor.rs +++ b/crates/nvlink-manager/src/switch_cert_monitor.rs @@ -20,18 +20,14 @@ use std::sync::Arc; use std::time::Duration; use std::{fmt, io}; -use carbide_rack::firmware_update::{build_new_node_info, load_switch_firmware_device_info}; -use carbide_rack::rms_node_type::switch_node_type_for_profile; -use carbide_secrets::credentials::CredentialManager; use carbide_utils::metrics::SharedMetricsHolder; use carbide_utils::periodic_timer::PeriodicTimer; -use carbide_uuid::rack::{RackId, RackProfileId}; +use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use chrono::Utc; use db::db_read::PgPoolReader; use db::work_lock_manager::WorkLockManagerHandle; -use librms::protos::rack_manager as rms; -use model::rack_type::RackProfileConfig; +use model::switch::SwitchMaintenanceOperation; use opentelemetry::KeyValue; use opentelemetry::metrics::{Histogram, Meter}; use rustls::{ClientConfig, RootCertStore}; @@ -39,7 +35,6 @@ use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName}; use sha2::{Digest, Sha256}; use sqlx::PgPool; use tokio::net::TcpStream; -use tokio::sync::Mutex; use tokio_rustls::TlsConnector; use tokio_util::sync::CancellationToken; use tracing::Instrument; @@ -59,7 +54,6 @@ struct CertificateInfo { struct SwitchCertificateMonitorTarget { switch_id: SwitchId, rack_id: RackId, - rack_profile_id: Option, endpoint_url: String, } @@ -67,7 +61,6 @@ struct SwitchCertificateMonitorTarget { enum SwitchCertApplyStatus { NotNeeded, Pending, - Applied, Error, Skipped, } @@ -77,7 +70,6 @@ impl SwitchCertApplyStatus { match self { Self::NotNeeded => "not_needed", Self::Pending => "pending", - Self::Applied => "applied", Self::Error => "error", Self::Skipped => "skipped", } @@ -160,11 +152,6 @@ impl fmt::Display for SwitchCertMonitorMetrics { .iter() .filter(|cert| !cert.desired_cert_error.is_empty()) .count(); - let applied_updates = self - .observed_certs - .iter() - .filter(|cert| cert.apply_status == SwitchCertApplyStatus::Applied) - .count(); let pending_updates = self .observed_certs .iter() @@ -172,12 +159,11 @@ impl fmt::Display for SwitchCertMonitorMetrics { .count(); write!( f, - "{{ observed_endpoints: {}, desired_cert_errors: {}, successful_probes: {}, matching_fingerprints: {}, applied_updates: {}, pending_updates: {}, duration: {} }}", + "{{ observed_endpoints: {}, desired_cert_errors: {}, successful_probes: {}, matching_fingerprints: {}, pending_updates: {}, duration: {} }}", self.observed_certs.len(), desired_cert_errors, successful_probes, matching_fingerprints, - applied_updates, pending_updates, self.recording_started_at.elapsed().as_millis(), ) @@ -463,39 +449,8 @@ impl MetricHolder { pub struct SwitchCertificateMonitor { db_pool: PgPool, config: NvLinkConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, metric_holder: Arc, work_lock_manager_handle: WorkLockManagerHandle, - in_flight_certificate_jobs: Mutex>, -} - -#[derive(Clone, Debug)] -struct InFlightSwitchCertJob { - job_id: String, -} - -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] -struct SwitchCertJobKey { - switch_id: String, - rack_id: String, -} - -impl SwitchCertJobKey { - fn from_target(target: &SwitchCertificateMonitorTarget) -> Self { - Self { - switch_id: target.switch_id.to_string(), - rack_id: target.rack_id.to_string(), - } - } -} - -#[derive(Clone, Debug, PartialEq, Eq)] -enum RmsSwitchCertJobState { - Pending(String), - Completed, - Failed(String), } #[derive(Debug, Clone, PartialEq, Eq)] @@ -534,11 +489,7 @@ impl SwitchCertificateMonitorIterationResult { .iter() .filter(|cert| !cert.error.is_empty()) .count(), - applied_updates: metrics - .observed_certs - .iter() - .filter(|cert| cert.apply_status == SwitchCertApplyStatus::Applied) - .count(), + applied_updates: 0, pending_updates: metrics .observed_certs .iter() @@ -562,9 +513,6 @@ impl SwitchCertificateMonitor { db_pool: PgPool, meter: Meter, config: NvLinkConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, work_lock_manager_handle: WorkLockManagerHandle, ) -> Self { let hold_period = config @@ -575,12 +523,8 @@ impl SwitchCertificateMonitor { Self { db_pool, config, - rms_client, - credential_manager, - rack_profiles, metric_holder, work_lock_manager_handle, - in_flight_certificate_jobs: Mutex::new(BTreeMap::new()), } } @@ -769,7 +713,7 @@ impl SwitchCertificateMonitor { rack_id = %rack_id_label, endpoint = %target.endpoint_url, error = %error, - "Failed to request RMS NMX-C switch certificate configuration" + "Failed to request NMX-C switch certificate configuration via switch state machine" ); (SwitchCertApplyStatus::Error, error) } @@ -845,59 +789,9 @@ impl SwitchCertificateMonitor { target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, ) -> Result { - let key = SwitchCertJobKey::from_target(target); - let in_flight_job = self - .in_flight_certificate_jobs - .lock() - .await - .get(&key) - .cloned(); - - if let Some(in_flight_job) = in_flight_job { - let job_state = self - .get_in_flight_certificate_job_state(&in_flight_job.job_id, cancel_token) - .await?; - match job_state { - RmsSwitchCertJobState::Pending(state) => { - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - endpoint = %target.endpoint_url, - job_id = %in_flight_job.job_id, - state = %state, - "RMS NMX-C switch certificate configuration job is still in progress" - ); - Ok(SwitchCertApplyStatus::Pending) - } - RmsSwitchCertJobState::Completed => { - self.in_flight_certificate_jobs.lock().await.remove(&key); - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - endpoint = %target.endpoint_url, - job_id = %in_flight_job.job_id, - "RMS NMX-C switch certificate configuration job completed" - ); - Ok(SwitchCertApplyStatus::Applied) - } - RmsSwitchCertJobState::Failed(error) => { - self.in_flight_certificate_jobs.lock().await.remove(&key); - Err(format!( - "RMS NMX-C switch certificate configuration job {} failed: {}", - in_flight_job.job_id, error - )) - } - } - } else { - let job_id = self - .apply_desired_certificate_with_rms(target, cancel_token) - .await?; - self.in_flight_certificate_jobs - .lock() - .await - .insert(key, InFlightSwitchCertJob { job_id }); - Ok(SwitchCertApplyStatus::Pending) - } + self.request_certificate_reconfiguration_with_switch_state_controller(target, cancel_token) + .await?; + Ok(SwitchCertApplyStatus::Pending) } async fn load_switch_certificate_monitor_targets( @@ -914,7 +808,6 @@ impl SwitchCertificateMonitor { .map(|row| SwitchCertificateMonitorTarget { switch_id: row.switch_id, rack_id: row.rack_id, - rack_profile_id: row.rack_profile_id, endpoint_url: nmx_c_endpoint::nmx_c_endpoint_url_from_nvos_ip( &row.nvos_ip, None, @@ -924,173 +817,42 @@ impl SwitchCertificateMonitor { .collect()) } - async fn apply_desired_certificate_with_rms( + async fn request_certificate_reconfiguration_with_switch_state_controller( &self, target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, - ) -> Result { - let rms_client = self.rms_client.as_ref().ok_or_else(|| { - "RMS client is not configured, so NMX-C switch certificate cannot be applied" - .to_string() - })?; - - let rack_profile_id = target.rack_profile_id.as_ref().ok_or_else(|| { - format!( - "rack {} has no rack_profile_id, so RMS switch node type and topology cannot be resolved", - target.rack_id - ) - })?; - let profile = self - .rack_profiles - .get(rack_profile_id.as_ref()) - .ok_or_else(|| { - format!( - "rack profile {} is not configured, so RMS switch node type and topology cannot be resolved", - rack_profile_id - ) - })?; - let switch_node_type = switch_node_type_for_profile(profile) - .map_err(|error| format!("failed to resolve RMS switch node type: {error}"))?; - - let switch = tokio::select! { + ) -> Result<(), String> { + let mut txn = tokio::select! { _ = cancel_token.cancelled() => { return Err(Self::APPLY_CANCELLED_ERROR.to_string()); } - switch = load_switch_firmware_device_info( - &self.db_pool, - self.credential_manager.as_ref(), - &target.switch_id, - ) => switch + txn = self.db_pool.begin() => txn .map_err(|error| { - format!( - "failed to load switch endpoint info for RMS certificate apply: {error}" - ) + format!("failed to begin transaction for switch certificate maintenance request: {error}") })?, }; - validate_switch_for_rms_certificate_apply(&switch)?; - - let response = tokio::select! { - _ = cancel_token.cancelled() => { - return Err(Self::APPLY_CANCELLED_ERROR.to_string()); - } - response = rms_client.configure_switch_certificate( - rms::ConfigureSwitchCertificateRequest { - nodes: Some(rms::NodeSet { - nodes: vec![build_new_node_info( - &target.rack_id, - &switch, - switch_node_type, - )], - }), - services: vec![rms::SwitchService::ScaleUpFabricManager as i32], - test_hello: true, - domain: Some(target.rack_id.to_string()), - } - ) => response.map_err(|error| { - format!("RMS ConfigureSwitchCertificate failed: {error}") - })?, - }; + db::switch::set_switch_maintenance_requested( + &mut txn, + target.switch_id, + "nvlink-switch-cert-monitor", + SwitchMaintenanceOperation::ReconfigureCertificate, + ) + .await + .map_err(|error| format!("failed to request switch certificate maintenance: {error}"))?; - let batch_response = response.response.ok_or_else(|| { - "RMS ConfigureSwitchCertificate response did not include a batch response".to_string() + txn.commit().await.map_err(|error| { + format!("failed to commit switch certificate maintenance request: {error}") })?; - if batch_response.status != rms::ReturnCode::Success as i32 { - let message = if batch_response.message.trim().is_empty() { - "no error details provided".to_string() - } else { - batch_response.message - }; - return Err(format!( - "RMS ConfigureSwitchCertificate returned status {}: {}", - batch_response.status, message - )); - } - - let switch_id = target.switch_id.to_string(); - let child_job_id = response - .jobs - .iter() - .find(|job| job.node_id == switch_id) - .map(|job| job.job_id.trim()) - .filter(|job_id| !job_id.is_empty()); - let job_id = child_job_id - .or_else(|| { - let parent_job_id = batch_response.job_id.trim(); - if parent_job_id.is_empty() { - None - } else { - Some(parent_job_id) - } - }) - .ok_or_else(|| { - "RMS ConfigureSwitchCertificate response did not include a job id".to_string() - })? - .to_string(); tracing::info!( switch_id = %target.switch_id, rack_id = %target.rack_id, - job_id = %job_id, - "Submitted RMS switch certificate configuration" + endpoint = %target.endpoint_url, + "Requested switch certificate reconfiguration via switch state machine" ); - Ok(job_id) - } - - async fn get_in_flight_certificate_job_state( - &self, - job_id: &str, - cancel_token: &CancellationToken, - ) -> Result { - let rms_client = self.rms_client.as_ref().ok_or_else(|| { - "RMS client is not configured, so NMX-C switch certificate job status cannot be checked" - .to_string() - })?; - - let response = tokio::select! { - _ = cancel_token.cancelled() => { - return Err(Self::APPLY_CANCELLED_ERROR.to_string()); - } - response = rms_client.get_configure_switch_certificate_job_status( - rms::GetConfigureSwitchCertificateJobStatusRequest { - job_id: job_id.to_string(), - }, - ) => response.map_err(|error| { - format!("RMS GetConfigureSwitchCertificateJobStatus failed: {error}") - })?, - }; - - if response.status != rms::ReturnCode::Success as i32 { - return Ok(RmsSwitchCertJobState::Failed(format!( - "RMS GetConfigureSwitchCertificateJobStatus returned status {}: {}", - response.status, - non_empty_or(response.error_message.as_str(), response.message.as_str()) - ))); - } - - let state = response.state.trim().to_ascii_lowercase(); - match state.as_str() { - "completed" | "complete" | "succeeded" | "success" => { - Ok(RmsSwitchCertJobState::Completed) - } - "failed" | "failure" | "error" => Ok(RmsSwitchCertJobState::Failed(non_empty_or( - response.error_message.as_str(), - response.message.as_str(), - ))), - "queued" | "running" | "pending" | "active" | "in_progress" => { - Ok(RmsSwitchCertJobState::Pending(state)) - } - "" => Ok(RmsSwitchCertJobState::Pending("unknown".to_string())), - _ => { - tracing::warn!( - job_id = %job_id, - state = %response.state, - "RMS returned unknown NMX-C switch certificate job state; treating as pending" - ); - Ok(RmsSwitchCertJobState::Pending(response.state)) - } - } + Ok(()) } async fn probe_endpoint_certificate( @@ -1183,27 +945,6 @@ fn desired_server_cert_path(config: &NvLinkConfig, rack_id: &RackId) -> Result Result<(), String> { - if switch.os_ip.as_deref().unwrap_or_default().is_empty() { - return Err(format!( - "switch {} is missing an NVOS IP address for RMS certificate apply", - switch.node_id - )); - } - if switch.os_username.as_deref().unwrap_or_default().is_empty() - || switch.os_password.as_deref().unwrap_or_default().is_empty() - { - return Err(format!( - "switch {} is missing NVOS credentials for RMS certificate apply", - switch.node_id - )); - } - - Ok(()) -} - async fn build_tls_client_config(config: &NvLinkConfig) -> Result { let mut roots = RootCertStore::empty(); let ca_cert_path = config @@ -1295,14 +1036,6 @@ fn cert_expires_within(cert: &CertificateInfo, window: Duration) -> bool { not_after <= warning_threshold } -fn non_empty_or(primary: &str, fallback: &str) -> String { - if primary.trim().is_empty() { - fallback.trim().to_string() - } else { - primary.trim().to_string() - } -} - fn metric_attrs(base_attrs: &[KeyValue], extra_attrs: &[KeyValue]) -> Vec { [base_attrs, extra_attrs].concat() } From ef6ad2e3942b91ee0ab164d57e793daa64ed85ae Mon Sep 17 00:00:00 2001 From: Thomas McRoberts Date: Wed, 8 Jul 2026 15:05:16 +0000 Subject: [PATCH 2/5] nvlink: use component-manager to kick switch FSM ..to configure NVSwitch certs. Signed-off-by: Thomas McRoberts --- Cargo.lock | 1 + .../src/handlers/component_manager.rs | 67 ++++++---------- crates/api-core/src/setup.rs | 1 + .../src/tests/common/api_fixtures/mod.rs | 1 + .../src/component_manager.rs | 76 +++++++++++++++++++ crates/nvlink-manager/Cargo.toml | 1 + crates/nvlink-manager/src/lib.rs | 5 ++ .../nvlink-manager/src/switch_cert_monitor.rs | 50 +++++++----- 8 files changed, 143 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ec987f870f..bcf64b0eef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2504,6 +2504,7 @@ dependencies = [ "carbide-utils", "carbide-uuid", "chrono", + "component-manager", "config-version", "duration-str", "eyre", diff --git a/crates/api-core/src/handlers/component_manager.rs b/crates/api-core/src/handlers/component_manager.rs index 2c0fe34c00..2cbe60a20c 100644 --- a/crates/api-core/src/handlers/component_manager.rs +++ b/crates/api-core/src/handlers/component_manager.rs @@ -28,7 +28,7 @@ use carbide_uuid::machine::MachineId; use carbide_uuid::power_shelf::PowerShelfId; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; -use component_manager::component_manager::ComponentManager; +use component_manager::component_manager::{ComponentManager, SwitchMaintenanceRequestResult}; use component_manager::compute_tray_manager::{ComputeTrayEndpoint, ComputeTrayVendor}; use component_manager::error::ComponentManagerError; use component_manager::nv_switch_manager::SwitchEndpoint; @@ -47,7 +47,6 @@ use model::rack::{FirmwareUpgradeJob, MaintenanceActivity}; use model::switch::SwitchMaintenanceOperation; use tonic::{Code, Request, Response, Status}; -use crate::CarbideError; use crate::api::{Api, log_request_data, log_request_data_redacted}; const MACHINE_POWER_OVERRIDE_SOURCE: &str = "component_power_control"; @@ -303,61 +302,43 @@ fn map_switch_maintenance_operation(action: PowerAction) -> SwitchMaintenanceOpe async fn queue_switch_power_control_via_state_controller( api: &Api, + cm: &ComponentManager, switch_ids: &[SwitchId], action: PowerAction, ) -> Result, Status> { let operation = map_switch_maintenance_operation(action); - queue_switch_maintenance_via_state_controller(api, switch_ids, operation).await + queue_switch_maintenance_via_state_controller(api, cm, switch_ids, operation).await } async fn queue_switch_maintenance_via_state_controller( api: &Api, + cm: &ComponentManager, switch_ids: &[SwitchId], operation: SwitchMaintenanceOperation, ) -> Result, Status> { - let mut txn = api.txn_begin().await?; - let existing = db::switch::find_by( - &mut txn, - db::ObjectColumnFilter::List(db::switch::IdColumn, switch_ids), - ) - .await - .map_err(CarbideError::from)?; - - let by_id: HashMap = - existing.into_iter().map(|sw| (sw.id, sw)).collect(); - let mut results = Vec::with_capacity(switch_ids.len()); - - for switch_id in switch_ids { - let Some(switch) = by_id.get(switch_id) else { - results.push(error_result( - &switch_id.to_string(), - format!("switch {switch_id} not found"), - )); - continue; - }; - - if switch.is_marked_as_deleted() { - results.push(error_result( - &switch_id.to_string(), - format!("switch {switch_id} is marked for deletion"), - )); - continue; - } - - db::switch::set_switch_maintenance_requested( - &mut txn, - *switch_id, - "component-manager", + let results = cm + .request_switch_maintenance_via_state_controller( + &api.database_connection, + switch_ids, operation, + "component-manager", ) .await - .map_err(CarbideError::from)?; + .map_err(component_manager_error_to_status)?; - results.push(success_result(&switch_id.to_string())); - } + Ok(results + .iter() + .map(switch_maintenance_request_result_to_component_result) + .collect()) +} - txn.commit().await?; - Ok(results) +fn switch_maintenance_request_result_to_component_result( + result: &SwitchMaintenanceRequestResult, +) -> rpc::ComponentResult { + match &result.error { + Some(error) => error_result(&result.switch_id.to_string(), error.clone()), + None => success_result(&result.switch_id.to_string()), + } } /// Maps raw proto `ComputeTrayComponent` values to display-name strings. @@ -1241,7 +1222,8 @@ pub(crate) async fn component_power_control( rpc::component_power_control_request::Target::SwitchIds(list) => { if cm.nv_switch_use_state_controller && !bypass_state_controller { let results = - queue_switch_power_control_via_state_controller(api, &list.ids, action).await?; + queue_switch_power_control_via_state_controller(api, cm, &list.ids, action) + .await?; (results, Vec::new()) } else { let endpoints = resolve_switch_endpoints(api, &list.ids).await?; @@ -1458,6 +1440,7 @@ pub(crate) async fn component_configure_switch_certificate( if cm.nv_switch_use_state_controller && !bypass_state_controller { let results = queue_switch_maintenance_via_state_controller( api, + cm, &switch_ids.ids, SwitchMaintenanceOperation::ReconfigureCertificate, ) diff --git a/crates/api-core/src/setup.rs b/crates/api-core/src/setup.rs index b574b77021..d4353c6033 100644 --- a/crates/api-core/src/setup.rs +++ b/crates/api-core/src/setup.rs @@ -1493,6 +1493,7 @@ async fn initialize_and_start_controllers<'a>( meter: meter.clone(), config: carbide_config.nvlink_config.clone().unwrap_or_default(), host_health: carbide_config.host_health, + component_manager: api_service.component_manager.clone().map(Arc::new), work_lock_manager_handle: work_lock_manager_handle.clone(), }) .start(join_set, cancel_token.clone())?; diff --git a/crates/api-core/src/tests/common/api_fixtures/mod.rs b/crates/api-core/src/tests/common/api_fixtures/mod.rs index 3dc0206b5f..075d9c6046 100644 --- a/crates/api-core/src/tests/common/api_fixtures/mod.rs +++ b/crates/api-core/src/tests/common/api_fixtures/mod.rs @@ -1455,6 +1455,7 @@ pub async fn create_test_env_with_overrides( db_pool.clone(), test_meter.meter(), config.nvlink_config.clone().unwrap(), + test_component_manager.clone(), api.work_lock_manager_handle.clone(), ); diff --git a/crates/component-manager/src/component_manager.rs b/crates/component-manager/src/component_manager.rs index e8bb0f7a91..4c50c46879 100644 --- a/crates/component-manager/src/component_manager.rs +++ b/crates/component-manager/src/component_manager.rs @@ -1,11 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::collections::HashMap; use std::sync::Arc; use carbide_redfish::libredfish::RedfishClientPool; +use carbide_uuid::switch::SwitchId; +use db::WithTransaction; use librms::RmsApi; use model::rack_type::RackProfileConfig; +use model::switch::SwitchMaintenanceOperation; use sqlx::PgPool; use crate::compute_tray_manager::{Backend as ComputeBackend, ComputeTrayManager}; @@ -41,7 +45,79 @@ pub struct ComponentManager { pub compute_tray_use_state_controller: bool, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SwitchMaintenanceRequestResult { + pub switch_id: SwitchId, + pub error: Option, +} + impl ComponentManager { + pub async fn request_switch_maintenance_via_state_controller( + &self, + db_pool: &PgPool, + switch_ids: &[SwitchId], + operation: SwitchMaintenanceOperation, + initiator: &str, + ) -> Result, ComponentManagerError> { + if !self.nv_switch_use_state_controller { + return Err(ComponentManagerError::InvalidArgument( + "nv_switch_use_state_controller is disabled; switch maintenance through the state controller is unavailable" + .to_string(), + )); + } + + let switch_ids = switch_ids.to_vec(); + let initiator = initiator.to_string(); + db_pool + .with_txn(|txn| { + Box::pin(async move { + let existing = db::switch::find_by( + txn, + db::ObjectColumnFilter::List(db::switch::IdColumn, &switch_ids), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + let by_id: HashMap = + existing.into_iter().map(|sw| (sw.id, sw)).collect(); + let mut results = Vec::with_capacity(switch_ids.len()); + + for switch_id in &switch_ids { + let Some(switch) = by_id.get(switch_id) else { + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: Some(format!("switch {switch_id} not found")), + }); + continue; + }; + + if switch.is_marked_as_deleted() { + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: Some(format!("switch {switch_id} is marked for deletion")), + }); + continue; + } + + db::switch::set_switch_maintenance_requested( + txn, *switch_id, &initiator, operation, + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: None, + }); + } + + Ok(results) + }) + }) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + } + pub async fn configure_switch_certificate( &self, endpoint: &SwitchEndpoint, diff --git a/crates/nvlink-manager/Cargo.toml b/crates/nvlink-manager/Cargo.toml index 98cb8ed3e1..8ab69ef1df 100644 --- a/crates/nvlink-manager/Cargo.toml +++ b/crates/nvlink-manager/Cargo.toml @@ -21,6 +21,7 @@ carbide-secrets = { path = "../secrets" } carbide-utils = { path = "../utils" } carbide-uuid = { path = "../uuid" } config-version = { path = "../config-version" } +component-manager = { path = "../component-manager" } libnmxc = { path = "../libnmxc" } libnmxm = { path = "../libnmxm" } diff --git a/crates/nvlink-manager/src/lib.rs b/crates/nvlink-manager/src/lib.rs index ff7d3af343..80800d6be1 100644 --- a/crates/nvlink-manager/src/lib.rs +++ b/crates/nvlink-manager/src/lib.rs @@ -32,6 +32,7 @@ use carbide_uuid::machine::MachineId; use carbide_uuid::nvlink::{NvLinkDomainId, NvLinkLogicalPartitionId, NvLinkPartitionId}; use carbide_uuid::rack::RackId; use chrono::Utc; +use component_manager::component_manager::ComponentManager; use config::NvLinkConfig; use config_version::Versioned; use db::machine::find_machine_ids; @@ -898,6 +899,7 @@ pub struct NvLinkManager { meter: opentelemetry::metrics::Meter, config: NvLinkConfig, host_health: HostHealthConfig, + component_manager: Option>, work_lock_manager_handle: WorkLockManagerHandle, } @@ -907,6 +909,7 @@ pub struct NvLinkManagerArgs { pub meter: opentelemetry::metrics::Meter, pub config: NvLinkConfig, pub host_health: HostHealthConfig, + pub component_manager: Option>, pub work_lock_manager_handle: WorkLockManagerHandle, } @@ -918,6 +921,7 @@ impl NvLinkManager { meter: args.meter, config: args.config, host_health: args.host_health, + component_manager: args.component_manager, work_lock_manager_handle: args.work_lock_manager_handle, } } @@ -942,6 +946,7 @@ impl NvLinkManager { self.db_pool, self.meter, self.config, + self.component_manager, self.work_lock_manager_handle, ); join_set diff --git a/crates/nvlink-manager/src/switch_cert_monitor.rs b/crates/nvlink-manager/src/switch_cert_monitor.rs index f034c490e4..4a48fbf058 100644 --- a/crates/nvlink-manager/src/switch_cert_monitor.rs +++ b/crates/nvlink-manager/src/switch_cert_monitor.rs @@ -25,6 +25,7 @@ use carbide_utils::periodic_timer::PeriodicTimer; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use chrono::Utc; +use component_manager::component_manager::ComponentManager; use db::db_read::PgPoolReader; use db::work_lock_manager::WorkLockManagerHandle; use model::switch::SwitchMaintenanceOperation; @@ -449,6 +450,7 @@ impl MetricHolder { pub struct SwitchCertificateMonitor { db_pool: PgPool, config: NvLinkConfig, + component_manager: Option>, metric_holder: Arc, work_lock_manager_handle: WorkLockManagerHandle, } @@ -513,6 +515,7 @@ impl SwitchCertificateMonitor { db_pool: PgPool, meter: Meter, config: NvLinkConfig, + component_manager: Option>, work_lock_manager_handle: WorkLockManagerHandle, ) -> Self { let hold_period = config @@ -523,6 +526,7 @@ impl SwitchCertificateMonitor { Self { db_pool, config, + component_manager, metric_holder, work_lock_manager_handle, } @@ -822,34 +826,46 @@ impl SwitchCertificateMonitor { target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, ) -> Result<(), String> { - let mut txn = tokio::select! { + let Some(component_manager) = self.component_manager.as_ref() else { + return Err( + "component manager is not configured; cannot request switch certificate reconfiguration" + .to_string(), + ); + }; + + let switch_ids = [target.switch_id]; + let results = tokio::select! { _ = cancel_token.cancelled() => { return Err(Self::APPLY_CANCELLED_ERROR.to_string()); } - txn = self.db_pool.begin() => txn - .map_err(|error| { - format!("failed to begin transaction for switch certificate maintenance request: {error}") - })?, + results = component_manager.request_switch_maintenance_via_state_controller( + &self.db_pool, + &switch_ids, + SwitchMaintenanceOperation::ReconfigureCertificate, + "nvlink-switch-cert-monitor", + ) => results.map_err(|error| { + format!( + "component manager failed to request switch certificate reconfiguration: {error}" + ) + })?, }; - db::switch::set_switch_maintenance_requested( - &mut txn, - target.switch_id, - "nvlink-switch-cert-monitor", - SwitchMaintenanceOperation::ReconfigureCertificate, - ) - .await - .map_err(|error| format!("failed to request switch certificate maintenance: {error}"))?; + let Some(result) = results.first() else { + return Err(format!( + "component manager returned no result for switch {} certificate reconfiguration", + target.switch_id + )); + }; - txn.commit().await.map_err(|error| { - format!("failed to commit switch certificate maintenance request: {error}") - })?; + if let Some(error) = &result.error { + return Err(error.clone()); + } tracing::info!( switch_id = %target.switch_id, rack_id = %target.rack_id, endpoint = %target.endpoint_url, - "Requested switch certificate reconfiguration via switch state machine" + "Requested switch certificate reconfiguration via component manager" ); Ok(()) From d92c305887996f30fe4ddf29a79f080e1a5d12f5 Mon Sep 17 00:00:00 2001 From: Thomas McRoberts Date: Tue, 14 Jul 2026 13:32:09 +0000 Subject: [PATCH 3/5] rotate NMX-C certs via rack maintenance ...instead of using the configure_switch-certificate flow. Also introduce a rack maintenance scheduler to ensure atomic and idempotnt rack maintenance. Signed-off-by: Thomas McRoberts --- .../src/handlers/component_manager.rs | 2 +- crates/api-core/src/handlers/rack.rs | 92 ++-- crates/api-core/src/tests/nvl_instance.rs | 54 ++- crates/api-db/src/rack.rs | 19 + crates/api-model/src/rack.rs | 2 +- crates/component-manager/Cargo.toml | 1 + .../src/component_manager.rs | 452 ++++++++++++++++++ .../nvlink-manager/src/switch_cert_monitor.rs | 103 ++-- 8 files changed, 626 insertions(+), 99 deletions(-) diff --git a/crates/api-core/src/handlers/component_manager.rs b/crates/api-core/src/handlers/component_manager.rs index 2cbe60a20c..ee82416289 100644 --- a/crates/api-core/src/handlers/component_manager.rs +++ b/crates/api-core/src/handlers/component_manager.rs @@ -64,7 +64,7 @@ fn unsupported_from_json_firmware_versions(target: &str) -> Status { )) } -fn component_manager_error_to_status(err: ComponentManagerError) -> Status { +pub(crate) fn component_manager_error_to_status(err: ComponentManagerError) -> Status { match err { ComponentManagerError::Unavailable(msg) => Status::unavailable(msg), ComponentManagerError::NotFound(msg) => Status::not_found(msg), diff --git a/crates/api-core/src/handlers/rack.rs b/crates/api-core/src/handlers/rack.rs index da3bacbe1d..89fcdcd680 100644 --- a/crates/api-core/src/handlers/rack.rs +++ b/crates/api-core/src/handlers/rack.rs @@ -22,11 +22,14 @@ use ::rpc::forge::{self as rpc, HealthReportEntry}; use carbide_rack::firmware_object::{ rack_maintenance_access_token_key, rms_access_token_or_noauth, }; -use carbide_secrets::credentials::Credentials; use carbide_uuid::machine::MachineId; use carbide_uuid::power_shelf::PowerShelfId; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; +use component_manager::component_manager::{ + RackMaintenanceAccessToken, RackMaintenanceEligibility, RackMaintenanceRequestOutcome, + request_rack_maintenance_via_state_controller, +}; use db::{ ObjectColumnFilter, WithTransaction, machine as db_machine, power_shelf as db_power_shelf, rack as db_rack, switch as db_switch, @@ -41,6 +44,7 @@ use tonic::{Request, Response, Status}; use crate::CarbideError; use crate::api::{Api, log_request_data, log_request_data_redacted}; use crate::auth::AuthContext; +use crate::handlers::component_manager::component_manager_error_to_status; pub async fn get_rack( api: &Api, @@ -766,58 +770,44 @@ pub(crate) async fn on_demand_rack_maintenance( } } - let access_token_stored = maintenance_access_token.is_some(); - if let Some(token) = maintenance_access_token { - api.credential_manager - .set_credentials( - &rack_maintenance_access_token_key(&rack_id), - &Credentials::UsernamePassword { - username: "access_token".into(), - password: token, - }, - ) - .await - .map_err(|error| CarbideError::Internal { - message: format!("failed to store rack maintenance access token: {error}"), - })?; - } - - let mut updated_config = rack.config.clone(); - updated_config.maintenance_requested = Some(scope); + let maintenance_access_token = + maintenance_access_token.map(|token| RackMaintenanceAccessToken { + credential_manager: api.credential_manager.as_ref(), + token, + }); - let db_result: Result<(), Status> = async { - let mut txn = api.txn_begin().await?; - db_rack::update(&mut txn, &rack_id, &updated_config).await?; - if updated_config - .maintenance_requested - .as_ref() - .is_some_and(|scope| { - scope.should_run(&MaintenanceActivity::FirmwareUpgrade { - firmware_version: None, - components: vec![], - force_update: false, - }) - }) - { - db_rack::update_firmware_upgrade_job(txn.as_mut(), &rack_id, None).await?; - } - txn.commit().await?; - Ok(()) - } + let schedule_result = request_rack_maintenance_via_state_controller( + &api.database_connection, + &rack_id, + scope, + RackMaintenanceEligibility::ReadyOrError, + maintenance_access_token, + ) .await; - if let Err(status) = db_result { - if access_token_stored - && let Err(error) = api - .credential_manager - .delete_credentials(&rack_maintenance_access_token_key(&rack_id)) - .await - { - tracing::warn!( - rack_id = %rack_id, - error = %error, - "failed to delete rack maintenance access token after DB error", - ); - } + + let scheduling_error = match schedule_result { + Ok( + RackMaintenanceRequestOutcome::Scheduled + | RackMaintenanceRequestOutcome::AlreadyPending, + ) => None, + Ok(RackMaintenanceRequestOutcome::Busy) => Some( + CarbideError::InvalidArgument(format!( + "On-demand maintenance for rack {} is already scheduled.", + rack_id, + )) + .into(), + ), + Ok(RackMaintenanceRequestOutcome::Deferred { state }) => Some( + CarbideError::InvalidArgument(format!( + "Rack {} is not in Ready or Error state (current: {:?}). Maintenance can only be requested when the rack is Ready or in Error.", + rack_id, state, + )) + .into(), + ), + Err(error) => Some(component_manager_error_to_status(error)), + }; + + if let Some(status) = scheduling_error { return Err(status); } diff --git a/crates/api-core/src/tests/nvl_instance.rs b/crates/api-core/src/tests/nvl_instance.rs index 50f9833a6e..a0390330c0 100644 --- a/crates/api-core/src/tests/nvl_instance.rs +++ b/crates/api-core/src/tests/nvl_instance.rs @@ -35,15 +35,16 @@ use common::api_fixtures::{ TestEnv, TestManagedHost, create_managed_host_with_hardware_info_template, insert_nvlink_nmxc_endpoint_from_managed_host, }; -use db::switch as db_switch; +use db::{ObjectColumnFilter, rack as db_rack, switch as db_switch}; use ipnetwork::IpNetwork; use libnmxc::nmxc_model::{GetGpuInfoListRequest, GetPartitionInfoListRequest, GpuAttr}; use model::expected_switch::ExpectedSwitch; use model::instance::config::nvlink::InstanceNvLinkConfig; use model::metadata::Metadata; +use model::rack::{MaintenanceActivity, RackState}; use model::switch::{ CONTROL_PLANE_STATE_CONFIGURED, FabricManagerState, FabricManagerStatus, NewSwitch, - SwitchConfig, SwitchControllerState, SwitchMaintenanceOperation, + SwitchConfig, SwitchControllerState, }; use model::test_support::{HardwareInfoTemplate, ManagedHostConfig}; use rcgen::{CertifiedKey, generate_simple_self_signed}; @@ -2639,6 +2640,24 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( .persist(&mut txn) .await .expect("create rack"); + let rack = db_rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db_rack::IdColumn, &rack_id), + ) + .await + .expect("load rack") + .pop() + .expect("rack"); + let updated = db_rack::try_update_controller_state( + txn.as_mut(), + &rack_id, + rack.controller_state.version, + rack.controller_state.version.increment(), + &RackState::Ready, + ) + .await + .expect("set rack ready"); + assert!(updated, "rack should transition to Ready for rotation"); txn.commit().await.expect("commit rack"); let switch_id = create_rack_switch_for_nmxc_simulator(&env, &rack_id).await; @@ -2661,23 +2680,40 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( .await .expect("load switch") .expect("switch"); + let rack = db_rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db_rack::IdColumn, &rack_id), + ) + .await + .expect("load rack") + .pop() + .expect("rack"); txn.commit().await.expect("commit txn"); if expected_fingerprint_mismatches > 0 { - let request = switch - .switch_maintenance_requested + assert!( + switch.switch_maintenance_requested.is_none(), + "certificate rotation should not request per-switch maintenance" + ); + let scope = rack + .config + .maintenance_requested .as_ref() - .expect("switch certificate maintenance request"); + .expect("rack NMX cluster maintenance request"); + assert!(scope.is_full_rack()); assert_eq!( - request.operation, - SwitchMaintenanceOperation::ReconfigureCertificate + scope.activities, + vec![MaintenanceActivity::ConfigureNmxCluster] ); - assert_eq!(request.initiator, "nvlink-switch-cert-monitor"); } else { assert!( switch.switch_maintenance_requested.is_none(), "matching certificate should not request switch certificate maintenance" ); + assert!( + rack.config.maintenance_requested.is_none(), + "matching certificate should not request rack maintenance" + ); } let rms_requests = env @@ -2687,7 +2723,7 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( assert_eq!( rms_requests.len(), 0, - "switch cert monitor should queue the switch state machine instead of calling RMS directly" + "switch cert monitor should queue the rack state machine instead of calling RMS directly" ); } diff --git a/crates/api-db/src/rack.rs b/crates/api-db/src/rack.rs index c2b3cb043f..b7b6be1182 100644 --- a/crates/api-db/src/rack.rs +++ b/crates/api-db/src/rack.rs @@ -58,6 +58,25 @@ where .map_err(|e| DatabaseError::new(query.sql(), e)) } +/// Load one rack while holding a row lock until the caller's transaction +/// completes. +/// +/// Callers that make a read/check/write decision about rack state or config +/// must use this instead of [`find_by`]. In particular, it prevents two +/// concurrent maintenance schedulers from both observing an empty +/// `maintenance_requested` field and then overwriting each other. +pub async fn find_by_id_for_update( + txn: &mut PgConnection, + rack_id: &RackId, +) -> DatabaseResult> { + let query = "SELECT * FROM racks WHERE id = $1 FOR UPDATE"; + sqlx::query_as(query) + .bind(rack_id) + .fetch_optional(txn) + .await + .map_err(|e| DatabaseError::new(query, e)) +} + pub async fn find_ids( txn: impl DbReader<'_>, filter: model::rack::RackSearchFilter, diff --git a/crates/api-model/src/rack.rs b/crates/api-model/src/rack.rs index 55ef4a187b..63825a5880 100644 --- a/crates/api-model/src/rack.rs +++ b/crates/api-model/src/rack.rs @@ -745,7 +745,7 @@ impl std::fmt::Display for MaintenanceActivity { /// Specifies which devices in the rack should be included in an on-demand /// maintenance cycle. When all three device-id lists are empty, the full rack /// is maintained. -#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize, Serialize)] pub struct MaintenanceScope { #[serde(default)] pub machine_ids: Vec, diff --git a/crates/component-manager/Cargo.toml b/crates/component-manager/Cargo.toml index 22bbd6af43..a075dc628f 100644 --- a/crates/component-manager/Cargo.toml +++ b/crates/component-manager/Cargo.toml @@ -47,6 +47,7 @@ tracing = { workspace = true } [dev-dependencies] carbide-api-test-helper = { path = "../api-test-helper" } carbide-macros = { path = "../macros" } +carbide-secrets = { path = "../secrets", features = ["test-support"] } carbide-sqlx-testing = { path = "../sqlx-testing" } carbide-test-support = { path = "../test-support" } toml = { workspace = true } diff --git a/crates/component-manager/src/component_manager.rs b/crates/component-manager/src/component_manager.rs index 4c50c46879..bdb0c249d3 100644 --- a/crates/component-manager/src/component_manager.rs +++ b/crates/component-manager/src/component_manager.rs @@ -3,11 +3,16 @@ use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use carbide_rack::firmware_object::rack_maintenance_access_token_key; use carbide_redfish::libredfish::RedfishClientPool; +use carbide_secrets::credentials::{CredentialManager, Credentials}; +use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use db::WithTransaction; use librms::RmsApi; +use model::rack::{MaintenanceActivity, MaintenanceScope, RackState}; use model::rack_type::RackProfileConfig; use model::switch::SwitchMaintenanceOperation; use sqlx::PgPool; @@ -51,6 +56,184 @@ pub struct SwitchMaintenanceRequestResult { pub error: Option, } +/// Which rack states a maintenance caller is willing to schedule from. +/// +/// Automatic maintenance uses [`ReadyOnly`](Self::ReadyOnly), while the +/// operator-facing API retains its existing ability to recover racks in +/// `Error` via [`ReadyOrError`](Self::ReadyOrError). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RackMaintenanceEligibility { + ReadyOnly, + ReadyOrError, +} + +/// The complete, non-error result of attempting to schedule rack maintenance. +/// +/// Keeping contention and eligibility as outcomes (rather than string errors) +/// lets periodic callers retry without marking certificate rotation failed, +/// while operator-facing callers can map the same result to their public API. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RackMaintenanceRequestOutcome { + /// This call persisted the requested scope. + Scheduled, + /// The exact same scope is already pending. This is an idempotent retry. + AlreadyPending, + /// A different maintenance request is pending and was left untouched. + Busy, + /// No request is pending, but the rack's current state is not eligible. + Deferred { state: RackState }, +} + +/// Credential to persist if this call wins the rack-maintenance scheduling +/// race. The write happens while the rack row is locked so a losing concurrent +/// request cannot replace or delete the winning request's token. +pub struct RackMaintenanceAccessToken<'a> { + pub credential_manager: &'a dyn CredentialManager, + pub token: String, +} + +/// Atomically request rack maintenance through the rack state controller. +/// +/// The rack row is locked before inspecting its state and existing scope, and +/// remains locked through the config write. Exact retries are idempotent; +/// unrelated pending work is never overwritten. Existing work is checked +/// before state eligibility so a retry can still observe `AlreadyPending` +/// after the rack controller has started transitioning the rack. +/// +/// This is a free function because the API supports rack maintenance even when +/// no `ComponentManager` backend is configured. +pub async fn request_rack_maintenance_via_state_controller( + db_pool: &PgPool, + rack_id: &RackId, + scope: MaintenanceScope, + eligibility: RackMaintenanceEligibility, + maintenance_access_token: Option>, +) -> Result { + let rack_id = rack_id.clone(); + let cleanup_rack_id = rack_id.clone(); + let credential_manager = maintenance_access_token + .as_ref() + .map(|credential| credential.credential_manager); + let access_token = maintenance_access_token.map(|credential| credential.token); + let token_stored = Arc::new(AtomicBool::new(false)); + let token_stored_in_txn = token_stored.clone(); + + let result = db_pool + .with_txn(|txn| { + Box::pin(async move { + let rack = db::rack::find_by_id_for_update(txn.as_mut(), &rack_id) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + .ok_or_else(|| { + ComponentManagerError::NotFound(format!("rack {rack_id} not found")) + })?; + + if let Some(existing_scope) = rack.config.maintenance_requested.as_ref() { + return Ok(if existing_scope == &scope { + RackMaintenanceRequestOutcome::AlreadyPending + } else { + RackMaintenanceRequestOutcome::Busy + }); + } + + let state = rack.controller_state.value.clone(); + let eligible = match eligibility { + RackMaintenanceEligibility::ReadyOnly => state == RackState::Ready, + RackMaintenanceEligibility::ReadyOrError => { + matches!(&state, RackState::Ready | RackState::Error { .. }) + } + }; + if !eligible { + return Ok(RackMaintenanceRequestOutcome::Deferred { state }); + } + + if let Some(access_token) = access_token { + let credential_manager = credential_manager + .expect("an access token always carries its credential manager"); + credential_manager + .set_credentials( + &rack_maintenance_access_token_key(&rack_id), + &Credentials::UsernamePassword { + username: "access_token".into(), + password: access_token, + }, + ) + .await + .map_err(|error| { + ComponentManagerError::Internal(format!( + "failed to store rack maintenance access token: {error}" + )) + })?; + token_stored_in_txn.store(true, Ordering::Release); + } + + let reset_firmware_upgrade_job = + scope.should_run(&MaintenanceActivity::FirmwareUpgrade { + firmware_version: None, + components: vec![], + force_update: false, + }); + let mut config = rack.config; + config.maintenance_requested = Some(scope); + db::rack::update(txn.as_mut(), &rack_id, &config) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + // Preserve the operator API's existing behavior, and keep this + // reset atomic with accepting the maintenance request. + if reset_firmware_upgrade_job { + db::rack::update_firmware_upgrade_job(txn.as_mut(), &rack_id, None) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + } + + Ok(RackMaintenanceRequestOutcome::Scheduled) + }) + }) + .await; + + match result { + Ok(Ok(outcome)) => Ok(outcome), + Ok(Err(error)) => { + cleanup_rack_maintenance_access_token( + credential_manager, + token_stored.load(Ordering::Acquire), + &cleanup_rack_id, + ) + .await; + Err(error) + } + Err(error) => { + cleanup_rack_maintenance_access_token( + credential_manager, + token_stored.load(Ordering::Acquire), + &cleanup_rack_id, + ) + .await; + Err(ComponentManagerError::Internal(error.to_string())) + } + } +} + +async fn cleanup_rack_maintenance_access_token( + credential_manager: Option<&dyn CredentialManager>, + token_stored: bool, + rack_id: &RackId, +) { + if token_stored + && let Some(credential_manager) = credential_manager + && let Err(error) = credential_manager + .delete_credentials(&rack_maintenance_access_token_key(rack_id)) + .await + { + tracing::warn!( + rack_id = %rack_id, + error = %error, + "failed to delete rack maintenance access token after scheduling failure", + ); + } +} + impl ComponentManager { pub async fn request_switch_maintenance_via_state_controller( &self, @@ -289,6 +472,13 @@ pub async fn build_component_manager( #[cfg(test)] mod tests { + use carbide_secrets::credentials::{CredentialKey, CredentialReader}; + use carbide_secrets::test_support::credentials::TestCredentialManager; + use carbide_uuid::rack::RackId; + use db::ObjectColumnFilter; + use model::rack::{ + FirmwareUpgradeJob, MaintenanceActivity, MaintenanceScope, RackConfig, RackState, + }; use model::rack_type::{ RackCapabilitiesSet, RackCapabilityCompute, RackCapabilityPowerShelf, RackCapabilitySwitch, RackHardwareTopology, RackProductFamily, RackProfile, @@ -297,6 +487,268 @@ mod tests { use super::*; use crate::config::ComponentManagerConfig; + async fn create_rack_in_state(pool: &PgPool, state: RackState) -> RackId { + let rack_id = RackId::new(uuid::Uuid::new_v4().to_string()); + let mut txn = pool.begin().await.unwrap(); + let rack = db::rack::create(txn.as_mut(), &rack_id, None, &RackConfig::default(), None) + .await + .unwrap(); + if state != RackState::Created { + assert!( + db::rack::try_update_controller_state( + txn.as_mut(), + &rack_id, + rack.controller_state.version, + rack.controller_state.version.increment(), + &state, + ) + .await + .unwrap() + ); + } + txn.commit().await.unwrap(); + rack_id + } + + async fn load_rack(pool: &PgPool, rack_id: &RackId) -> model::rack::Rack { + let mut conn = pool.acquire().await.unwrap(); + db::rack::find_by( + conn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, rack_id), + ) + .await + .unwrap() + .pop() + .unwrap() + } + + fn nmx_scope() -> MaintenanceScope { + MaintenanceScope { + activities: vec![MaintenanceActivity::ConfigureNmxCluster], + ..Default::default() + } + } + + #[carbide_macros::sqlx_test] + async fn rack_maintenance_scheduler_is_atomic_and_idempotent(pool: PgPool) { + let ready_rack = create_rack_in_state(&pool, RackState::Ready).await; + let scope = nmx_scope(); + + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + scope.clone(), + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + assert_eq!( + load_rack(&pool, &ready_rack) + .await + .config + .maintenance_requested, + Some(scope.clone()), + ); + + // Exact retries are idempotent, including after the rack state has + // moved on. A different request is busy and cannot replace it. + let rack = load_rack(&pool, &ready_rack).await; + let mut txn = pool.begin().await.unwrap(); + assert!( + db::rack::try_update_controller_state( + txn.as_mut(), + &ready_rack, + rack.controller_state.version, + rack.controller_state.version.increment(), + &RackState::Discovering, + ) + .await + .unwrap() + ); + txn.commit().await.unwrap(); + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + scope.clone(), + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::AlreadyPending, + ); + + let unrelated_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::PowerSequence], + ..Default::default() + }; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + unrelated_scope, + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Busy, + ); + assert_eq!( + load_rack(&pool, &ready_rack) + .await + .config + .maintenance_requested, + Some(scope), + ); + + // Automatic callers defer non-Ready racks; operator callers may + // explicitly opt into recovering a rack from Error. + let error_state = RackState::Error { + cause: "test".into(), + }; + let error_rack = create_rack_in_state(&pool, error_state.clone()).await; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &error_rack, + nmx_scope(), + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Deferred { state: error_state }, + ); + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &error_rack, + nmx_scope(), + RackMaintenanceEligibility::ReadyOrError, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + + // Two different requests racing for an empty slot serialize on the + // rack row: one wins and the other observes Busy. + let race_rack = create_rack_in_state(&pool, RackState::Ready).await; + let first_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"first"}"#.into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let second_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"second"}"#.into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let credential_manager = TestCredentialManager::default(); + let (first, second) = tokio::join!( + request_rack_maintenance_via_state_controller( + &pool, + &race_rack, + first_scope, + RackMaintenanceEligibility::ReadyOnly, + Some(RackMaintenanceAccessToken { + credential_manager: &credential_manager, + token: "first-token".into(), + }), + ), + request_rack_maintenance_via_state_controller( + &pool, + &race_rack, + second_scope, + RackMaintenanceEligibility::ReadyOnly, + Some(RackMaintenanceAccessToken { + credential_manager: &credential_manager, + token: "second-token".into(), + }), + ), + ); + let first = first.unwrap(); + let second = second.unwrap(); + let winning_token = match (&first, &second) { + (RackMaintenanceRequestOutcome::Scheduled, RackMaintenanceRequestOutcome::Busy) => { + "first-token" + } + (RackMaintenanceRequestOutcome::Busy, RackMaintenanceRequestOutcome::Scheduled) => { + "second-token" + } + unexpected => panic!("expected one Scheduled and one Busy outcome, got {unexpected:?}"), + }; + let stored_credentials = credential_manager + .get_credentials(&CredentialKey::RackMaintenanceAccessToken { + rack_id: race_rack.clone(), + }) + .await + .unwrap() + .expect("the winning request should persist its token"); + assert_eq!( + stored_credentials, + Credentials::UsernamePassword { + username: "access_token".into(), + password: winning_token.into(), + } + ); + + // Firmware bookkeeping is cleared in the same transaction that + // accepts a firmware maintenance request. + let firmware_rack = create_rack_in_state(&pool, RackState::Ready).await; + let mut txn = pool.begin().await.unwrap(); + db::rack::update_firmware_upgrade_job( + txn.as_mut(), + &firmware_rack, + Some(&FirmwareUpgradeJob { + job_id: Some("stale-job".into()), + ..Default::default() + }), + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + let firmware_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some("{}".into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &firmware_rack, + firmware_scope, + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + assert!( + load_rack(&pool, &firmware_rack) + .await + .firmware_upgrade_job + .is_none() + ); + } + fn rms_rack_profiles(profile: RackProfile) -> RackProfileConfig { RackProfileConfig { rack_profiles: [("NVL72".to_string(), profile)].into_iter().collect(), diff --git a/crates/nvlink-manager/src/switch_cert_monitor.rs b/crates/nvlink-manager/src/switch_cert_monitor.rs index 4a48fbf058..0e12f550f9 100644 --- a/crates/nvlink-manager/src/switch_cert_monitor.rs +++ b/crates/nvlink-manager/src/switch_cert_monitor.rs @@ -25,10 +25,13 @@ use carbide_utils::periodic_timer::PeriodicTimer; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use chrono::Utc; -use component_manager::component_manager::ComponentManager; +use component_manager::component_manager::{ + ComponentManager, RackMaintenanceEligibility, RackMaintenanceRequestOutcome, + request_rack_maintenance_via_state_controller, +}; use db::db_read::PgPoolReader; use db::work_lock_manager::WorkLockManagerHandle; -use model::switch::SwitchMaintenanceOperation; +use model::rack::{MaintenanceActivity, MaintenanceScope}; use opentelemetry::KeyValue; use opentelemetry::metrics::{Histogram, Meter}; use rustls::{ClientConfig, RootCertStore}; @@ -717,7 +720,7 @@ impl SwitchCertificateMonitor { rack_id = %rack_id_label, endpoint = %target.endpoint_url, error = %error, - "Failed to request NMX-C switch certificate configuration via switch state machine" + "Failed to request NMX-C cluster configuration via rack state machine" ); (SwitchCertApplyStatus::Error, error) } @@ -793,9 +796,8 @@ impl SwitchCertificateMonitor { target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, ) -> Result { - self.request_certificate_reconfiguration_with_switch_state_controller(target, cancel_token) - .await?; - Ok(SwitchCertApplyStatus::Pending) + self.request_nmx_cluster_configuration_with_rack_state_controller(target, cancel_token) + .await } async fn load_switch_certificate_monitor_targets( @@ -821,54 +823,81 @@ impl SwitchCertificateMonitor { .collect()) } - async fn request_certificate_reconfiguration_with_switch_state_controller( + async fn request_nmx_cluster_configuration_with_rack_state_controller( &self, target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, - ) -> Result<(), String> { - let Some(component_manager) = self.component_manager.as_ref() else { + ) -> Result { + if self.component_manager.is_none() { return Err( - "component manager is not configured; cannot request switch certificate reconfiguration" + "component manager is not configured; cannot request NMX-C cluster configuration" .to_string(), ); - }; + } - let switch_ids = [target.switch_id]; - let results = tokio::select! { + // Empty device lists intentionally select the full rack. ConfigureNmxCluster runs the + // same certificate and fabric-manager workflow as `rack maintenance start + // --activities configure-nmx-cluster`. + let scope = MaintenanceScope { + activities: vec![MaintenanceActivity::ConfigureNmxCluster], + ..Default::default() + }; + let outcome = tokio::select! { _ = cancel_token.cancelled() => { return Err(Self::APPLY_CANCELLED_ERROR.to_string()); } - results = component_manager.request_switch_maintenance_via_state_controller( + outcome = request_rack_maintenance_via_state_controller( &self.db_pool, - &switch_ids, - SwitchMaintenanceOperation::ReconfigureCertificate, - "nvlink-switch-cert-monitor", - ) => results.map_err(|error| { + &target.rack_id, + scope, + RackMaintenanceEligibility::ReadyOnly, + None, + ) => outcome.map_err(|error| { format!( - "component manager failed to request switch certificate reconfiguration: {error}" + "component manager failed to request NMX-C cluster configuration: {error}" ) })?, }; - let Some(result) = results.first() else { - return Err(format!( - "component manager returned no result for switch {} certificate reconfiguration", - target.switch_id - )); - }; - - if let Some(error) = &result.error { - return Err(error.clone()); + match outcome { + RackMaintenanceRequestOutcome::Scheduled => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Requested full-rack NMX-C cluster configuration via component manager" + ); + Ok(SwitchCertApplyStatus::Pending) + } + RackMaintenanceRequestOutcome::AlreadyPending => { + tracing::debug!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Full-rack NMX-C cluster configuration is already pending" + ); + Ok(SwitchCertApplyStatus::Pending) + } + RackMaintenanceRequestOutcome::Busy => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Deferring NMX-C certificate rotation because different rack maintenance is pending" + ); + Ok(SwitchCertApplyStatus::Skipped) + } + RackMaintenanceRequestOutcome::Deferred { state } => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + ?state, + "Deferring NMX-C certificate rotation until the rack is ready" + ); + Ok(SwitchCertApplyStatus::Skipped) + } } - - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - endpoint = %target.endpoint_url, - "Requested switch certificate reconfiguration via component manager" - ); - - Ok(()) } async fn probe_endpoint_certificate( From 1b9076e625fe6676eecb67ef236d02a59e321a01 Mon Sep 17 00:00:00 2001 From: Thomas McRoberts Date: Tue, 14 Jul 2026 16:02:22 +0000 Subject: [PATCH 4/5] rack: attempt to cancel maitenance request ..after vault write failure. If the writing of the RMS token to vault fails, attempt to cancel the just submitted maintenance request to leave the rack in a cleaner state. Signed-off-by: Thomas McRoberts --- crates/api-db/src/rack.rs | 4 +- .../src/component_manager.rs | 262 +++++++++++++----- 2 files changed, 195 insertions(+), 71 deletions(-) diff --git a/crates/api-db/src/rack.rs b/crates/api-db/src/rack.rs index b7b6be1182..7b575848fc 100644 --- a/crates/api-db/src/rack.rs +++ b/crates/api-db/src/rack.rs @@ -58,7 +58,7 @@ where .map_err(|e| DatabaseError::new(query.sql(), e)) } -/// Load one rack while holding a row lock until the caller's transaction +/// Load one active rack while holding a row lock until the caller's transaction /// completes. /// /// Callers that make a read/check/write decision about rack state or config @@ -69,7 +69,7 @@ pub async fn find_by_id_for_update( txn: &mut PgConnection, rack_id: &RackId, ) -> DatabaseResult> { - let query = "SELECT * FROM racks WHERE id = $1 FOR UPDATE"; + let query = "SELECT * FROM racks WHERE id = $1 AND deleted IS NULL FOR UPDATE"; sqlx::query_as(query) .bind(rack_id) .fetch_optional(txn) diff --git a/crates/component-manager/src/component_manager.rs b/crates/component-manager/src/component_manager.rs index bdb0c249d3..51be7015b7 100644 --- a/crates/component-manager/src/component_manager.rs +++ b/crates/component-manager/src/component_manager.rs @@ -3,7 +3,6 @@ use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; use carbide_rack::firmware_object::rack_maintenance_access_token_key; use carbide_redfish::libredfish::RedfishClientPool; @@ -85,8 +84,8 @@ pub enum RackMaintenanceRequestOutcome { } /// Credential to persist if this call wins the rack-maintenance scheduling -/// race. The write happens while the rack row is locked so a losing concurrent -/// request cannot replace or delete the winning request's token. +/// race. It is written only after the scheduling transaction commits, so no +/// database lock is held during the external credential-store operation. pub struct RackMaintenanceAccessToken<'a> { pub credential_manager: &'a dyn CredentialManager, pub token: String, @@ -110,22 +109,19 @@ pub async fn request_rack_maintenance_via_state_controller( maintenance_access_token: Option>, ) -> Result { let rack_id = rack_id.clone(); - let cleanup_rack_id = rack_id.clone(); - let credential_manager = maintenance_access_token - .as_ref() - .map(|credential| credential.credential_manager); - let access_token = maintenance_access_token.map(|credential| credential.token); - let token_stored = Arc::new(AtomicBool::new(false)); - let token_stored_in_txn = token_stored.clone(); + let scheduled_scope = scope.clone(); + let transaction_rack_id = rack_id.clone(); let result = db_pool .with_txn(|txn| { Box::pin(async move { - let rack = db::rack::find_by_id_for_update(txn.as_mut(), &rack_id) + let rack = db::rack::find_by_id_for_update(txn.as_mut(), &transaction_rack_id) .await .map_err(|error| ComponentManagerError::Internal(error.to_string()))? .ok_or_else(|| { - ComponentManagerError::NotFound(format!("rack {rack_id} not found")) + ComponentManagerError::NotFound(format!( + "rack {transaction_rack_id} not found" + )) })?; if let Some(existing_scope) = rack.config.maintenance_requested.as_ref() { @@ -147,26 +143,6 @@ pub async fn request_rack_maintenance_via_state_controller( return Ok(RackMaintenanceRequestOutcome::Deferred { state }); } - if let Some(access_token) = access_token { - let credential_manager = credential_manager - .expect("an access token always carries its credential manager"); - credential_manager - .set_credentials( - &rack_maintenance_access_token_key(&rack_id), - &Credentials::UsernamePassword { - username: "access_token".into(), - password: access_token, - }, - ) - .await - .map_err(|error| { - ComponentManagerError::Internal(format!( - "failed to store rack maintenance access token: {error}" - )) - })?; - token_stored_in_txn.store(true, Ordering::Release); - } - let reset_firmware_upgrade_job = scope.should_run(&MaintenanceActivity::FirmwareUpgrade { firmware_version: None, @@ -175,14 +151,14 @@ pub async fn request_rack_maintenance_via_state_controller( }); let mut config = rack.config; config.maintenance_requested = Some(scope); - db::rack::update(txn.as_mut(), &rack_id, &config) + db::rack::update(txn.as_mut(), &transaction_rack_id, &config) .await .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; // Preserve the operator API's existing behavior, and keep this // reset atomic with accepting the maintenance request. if reset_firmware_upgrade_job { - db::rack::update_firmware_upgrade_job(txn.as_mut(), &rack_id, None) + db::rack::update_firmware_upgrade_job(txn.as_mut(), &transaction_rack_id, None) .await .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; } @@ -192,45 +168,98 @@ pub async fn request_rack_maintenance_via_state_controller( }) .await; - match result { - Ok(Ok(outcome)) => Ok(outcome), - Ok(Err(error)) => { - cleanup_rack_maintenance_access_token( - credential_manager, - token_stored.load(Ordering::Acquire), - &cleanup_rack_id, - ) - .await; - Err(error) - } - Err(error) => { - cleanup_rack_maintenance_access_token( - credential_manager, - token_stored.load(Ordering::Acquire), - &cleanup_rack_id, + let outcome = match result { + Ok(Ok(outcome)) => outcome, + Ok(Err(error)) => return Err(error), + Err(error) => return Err(ComponentManagerError::Internal(error.to_string())), + }; + + if outcome == RackMaintenanceRequestOutcome::Scheduled + && let Some(RackMaintenanceAccessToken { + credential_manager, + token, + }) = maintenance_access_token + && let Err(error) = credential_manager + .set_credentials( + &rack_maintenance_access_token_key(&rack_id), + &Credentials::UsernamePassword { + username: "access_token".into(), + password: token, + }, ) - .await; - Err(ComponentManagerError::Internal(error.to_string())) - } + .await + { + let recovery = + recover_rack_maintenance_after_credential_failure(db_pool, &rack_id, &scheduled_scope) + .await; + return Err(ComponentManagerError::Internal(match recovery { + Ok(recovery) => format!( + "failed to store rack maintenance access token: {error}; recovery: {recovery:?}" + ), + Err(recovery_error) => format!( + "failed to store rack maintenance access token: {error}; failed to recover maintenance request: {recovery_error}" + ), + })); } + + Ok(outcome) } -async fn cleanup_rack_maintenance_access_token( - credential_manager: Option<&dyn CredentialManager>, - token_stored: bool, +#[derive(Debug)] +enum RackMaintenanceCredentialRecovery { + Cleared, + RequestChanged, + ExecutionStarted, + RackNotFound, +} + +async fn recover_rack_maintenance_after_credential_failure( + db_pool: &PgPool, rack_id: &RackId, -) { - if token_stored - && let Some(credential_manager) = credential_manager - && let Err(error) = credential_manager - .delete_credentials(&rack_maintenance_access_token_key(rack_id)) - .await - { - tracing::warn!( - rack_id = %rack_id, - error = %error, - "failed to delete rack maintenance access token after scheduling failure", - ); + scheduled_scope: &MaintenanceScope, +) -> Result { + let rack_id = rack_id.clone(); + let scheduled_scope = scheduled_scope.clone(); + let result = db_pool + .with_txn(|txn| { + Box::pin(async move { + let Some(mut rack) = db::rack::find_by_id_for_update(txn.as_mut(), &rack_id) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + else { + return Ok(RackMaintenanceCredentialRecovery::RackNotFound); + }; + + if rack.config.maintenance_requested.as_ref() != Some(&scheduled_scope) { + return Ok(RackMaintenanceCredentialRecovery::RequestChanged); + } + + let state = rack.controller_state.value.clone(); + if !matches!(&state, RackState::Ready | RackState::Error { .. }) { + // Once execution starts, clearing the scope would make the rack controller + // interpret the default scope as "run all activities". Leave recovery to + // its established missing-credential error path instead. + tracing::warn!( + rack_id = %rack_id, + ?state, + "rack maintenance started before credential storage failed; leaving cleanup to the rack state controller", + ); + return Ok(RackMaintenanceCredentialRecovery::ExecutionStarted); + } + + rack.config.maintenance_requested = None; + db::rack::update(txn.as_mut(), &rack_id, &rack.config) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + Ok(RackMaintenanceCredentialRecovery::Cleared) + }) + }) + .await; + + match result { + Ok(Ok(recovery)) => Ok(recovery), + Ok(Err(error)) => Err(error), + Err(error) => Err(ComponentManagerError::Internal(error.to_string())), } } @@ -472,7 +501,9 @@ pub async fn build_component_manager( #[cfg(test)] mod tests { - use carbide_secrets::credentials::{CredentialKey, CredentialReader}; + use async_trait::async_trait; + use carbide_secrets::SecretsError; + use carbide_secrets::credentials::{CredentialKey, CredentialReader, CredentialWriter}; use carbide_secrets::test_support::credentials::TestCredentialManager; use carbide_uuid::rack::RackId; use db::ObjectColumnFilter; @@ -487,6 +518,45 @@ mod tests { use super::*; use crate::config::ComponentManagerConfig; + struct FailingCredentialManager; + + #[async_trait] + impl CredentialReader for FailingCredentialManager { + async fn get_credentials( + &self, + _key: &CredentialKey, + ) -> Result, SecretsError> { + Ok(None) + } + } + + #[async_trait] + impl CredentialWriter for FailingCredentialManager { + async fn set_credentials( + &self, + _key: &CredentialKey, + _credentials: &Credentials, + ) -> Result<(), SecretsError> { + Err(SecretsError::GenericError( + std::io::Error::other("test credential write failure").into(), + )) + } + + async fn create_credentials( + &self, + _key: &CredentialKey, + _credentials: &Credentials, + ) -> Result<(), SecretsError> { + unreachable!("test only exercises set_credentials") + } + + async fn delete_credentials(&self, _key: &CredentialKey) -> Result<(), SecretsError> { + Ok(()) + } + } + + impl CredentialManager for FailingCredentialManager {} + async fn create_rack_in_state(pool: &PgPool, state: RackState) -> RackId { let rack_id = RackId::new(uuid::Uuid::new_v4().to_string()); let mut txn = pool.begin().await.unwrap(); @@ -747,6 +817,60 @@ mod tests { .firmware_upgrade_job .is_none() ); + + // Soft-deleted racks cannot accept new maintenance work. + let deleted_rack = create_rack_in_state(&pool, RackState::Ready).await; + let mut txn = pool.begin().await.unwrap(); + db::rack::mark_as_deleted(&deleted_rack, txn.as_mut()) + .await + .unwrap(); + txn.commit().await.unwrap(); + assert!(matches!( + request_rack_maintenance_via_state_controller( + &pool, + &deleted_rack, + nmx_scope(), + RackMaintenanceEligibility::ReadyOnly, + None, + ) + .await, + Err(ComponentManagerError::NotFound(_)), + )); + + // Credential storage happens after commit. If it fails before the rack controller + // starts the request, the compensating transaction removes that exact request. + let credential_failure_rack = create_rack_in_state(&pool, RackState::Ready).await; + let credential_failure_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some("{}".into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let error = request_rack_maintenance_via_state_controller( + &pool, + &credential_failure_rack, + credential_failure_scope, + RackMaintenanceEligibility::ReadyOnly, + Some(RackMaintenanceAccessToken { + credential_manager: &FailingCredentialManager, + token: "test-token".into(), + }), + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("recovery: Cleared"), + "unexpected error: {error}" + ); + assert!( + load_rack(&pool, &credential_failure_rack) + .await + .config + .maintenance_requested + .is_none() + ); } fn rms_rack_profiles(profile: RackProfile) -> RackProfileConfig { From b597ba48dc0736817f1ad4c2874e60114ba552af Mon Sep 17 00:00:00 2001 From: Thomas McRoberts Date: Tue, 14 Jul 2026 19:49:36 +0000 Subject: [PATCH 5/5] review feedback Signed-off-by: Thomas McRoberts --- crates/api-core/src/handlers/rack.rs | 2 +- crates/api-db/src/rack.rs | 18 ++--- .../src/component_manager.rs | 70 ++++++++++++------- .../nvlink-manager/src/switch_cert_monitor.rs | 2 +- 4 files changed, 53 insertions(+), 39 deletions(-) diff --git a/crates/api-core/src/handlers/rack.rs b/crates/api-core/src/handlers/rack.rs index 89fcdcd680..e631e827e2 100644 --- a/crates/api-core/src/handlers/rack.rs +++ b/crates/api-core/src/handlers/rack.rs @@ -780,7 +780,7 @@ pub(crate) async fn on_demand_rack_maintenance( &api.database_connection, &rack_id, scope, - RackMaintenanceEligibility::ReadyOrError, + RackMaintenanceEligibility::AllowErrorRecovery, maintenance_access_token, ) .await; diff --git a/crates/api-db/src/rack.rs b/crates/api-db/src/rack.rs index 7b575848fc..8c06a056f1 100644 --- a/crates/api-db/src/rack.rs +++ b/crates/api-db/src/rack.rs @@ -58,23 +58,15 @@ where .map_err(|e| DatabaseError::new(query.sql(), e)) } -/// Load one active rack while holding a row lock until the caller's transaction -/// completes. -/// -/// Callers that make a read/check/write decision about rack state or config -/// must use this instead of [`find_by`]. In particular, it prevents two -/// concurrent maintenance schedulers from both observing an empty -/// `maintenance_requested` field and then overwriting each other. -pub async fn find_by_id_for_update( - txn: &mut PgConnection, - rack_id: &RackId, -) -> DatabaseResult> { - let query = "SELECT * FROM racks WHERE id = $1 AND deleted IS NULL FOR UPDATE"; - sqlx::query_as(query) +/// Lock one active rack until the caller's transaction completes. +pub async fn lock_for_update(txn: &mut PgConnection, rack_id: &RackId) -> DatabaseResult { + let query = "SELECT id FROM racks WHERE id = $1 AND deleted IS NULL FOR UPDATE"; + sqlx::query_as::<_, (RackId,)>(query) .bind(rack_id) .fetch_optional(txn) .await .map_err(|e| DatabaseError::new(query, e)) + .map(|row| row.is_some()) } pub async fn find_ids( diff --git a/crates/component-manager/src/component_manager.rs b/crates/component-manager/src/component_manager.rs index 51be7015b7..2f23280ef1 100644 --- a/crates/component-manager/src/component_manager.rs +++ b/crates/component-manager/src/component_manager.rs @@ -9,7 +9,7 @@ use carbide_redfish::libredfish::RedfishClientPool; use carbide_secrets::credentials::{CredentialManager, Credentials}; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; -use db::WithTransaction; +use db::{ObjectColumnFilter, WithTransaction}; use librms::RmsApi; use model::rack::{MaintenanceActivity, MaintenanceScope, RackState}; use model::rack_type::RackProfileConfig; @@ -57,13 +57,13 @@ pub struct SwitchMaintenanceRequestResult { /// Which rack states a maintenance caller is willing to schedule from. /// -/// Automatic maintenance uses [`ReadyOnly`](Self::ReadyOnly), while the +/// Automatic maintenance uses [`RequireReady`](Self::RequireReady), while the /// operator-facing API retains its existing ability to recover racks in -/// `Error` via [`ReadyOrError`](Self::ReadyOrError). +/// `Error` via [`AllowErrorRecovery`](Self::AllowErrorRecovery). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RackMaintenanceEligibility { - ReadyOnly, - ReadyOrError, + RequireReady, + AllowErrorRecovery, } /// The complete, non-error result of attempting to schedule rack maintenance. @@ -115,14 +115,25 @@ pub async fn request_rack_maintenance_via_state_controller( let result = db_pool .with_txn(|txn| { Box::pin(async move { - let rack = db::rack::find_by_id_for_update(txn.as_mut(), &transaction_rack_id) + if !db::rack::lock_for_update(txn.as_mut(), &transaction_rack_id) .await .map_err(|error| ComponentManagerError::Internal(error.to_string()))? - .ok_or_else(|| { - ComponentManagerError::NotFound(format!( - "rack {transaction_rack_id} not found" - )) - })?; + { + return Err(ComponentManagerError::NotFound(format!( + "rack {transaction_rack_id} not found" + ))); + } + let rack = db::rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, &transaction_rack_id), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + .into_iter() + .next() + .ok_or_else(|| { + ComponentManagerError::NotFound(format!("rack {transaction_rack_id} not found")) + })?; if let Some(existing_scope) = rack.config.maintenance_requested.as_ref() { return Ok(if existing_scope == &scope { @@ -134,8 +145,8 @@ pub async fn request_rack_maintenance_via_state_controller( let state = rack.controller_state.value.clone(); let eligible = match eligibility { - RackMaintenanceEligibility::ReadyOnly => state == RackState::Ready, - RackMaintenanceEligibility::ReadyOrError => { + RackMaintenanceEligibility::RequireReady => state == RackState::Ready, + RackMaintenanceEligibility::AllowErrorRecovery => { matches!(&state, RackState::Ready | RackState::Error { .. }) } }; @@ -223,9 +234,20 @@ async fn recover_rack_maintenance_after_credential_failure( let result = db_pool .with_txn(|txn| { Box::pin(async move { - let Some(mut rack) = db::rack::find_by_id_for_update(txn.as_mut(), &rack_id) + if !db::rack::lock_for_update(txn.as_mut(), &rack_id) .await .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + { + return Ok(RackMaintenanceCredentialRecovery::RackNotFound); + } + let Some(mut rack) = db::rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, &rack_id), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + .into_iter() + .next() else { return Ok(RackMaintenanceCredentialRecovery::RackNotFound); }; @@ -609,7 +631,7 @@ mod tests { &pool, &ready_rack, scope.clone(), - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await @@ -645,7 +667,7 @@ mod tests { &pool, &ready_rack, scope.clone(), - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await @@ -662,7 +684,7 @@ mod tests { &pool, &ready_rack, unrelated_scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await @@ -688,7 +710,7 @@ mod tests { &pool, &error_rack, nmx_scope(), - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await @@ -700,7 +722,7 @@ mod tests { &pool, &error_rack, nmx_scope(), - RackMaintenanceEligibility::ReadyOrError, + RackMaintenanceEligibility::AllowErrorRecovery, None, ) .await @@ -733,7 +755,7 @@ mod tests { &pool, &race_rack, first_scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, Some(RackMaintenanceAccessToken { credential_manager: &credential_manager, token: "first-token".into(), @@ -743,7 +765,7 @@ mod tests { &pool, &race_rack, second_scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, Some(RackMaintenanceAccessToken { credential_manager: &credential_manager, token: "second-token".into(), @@ -804,7 +826,7 @@ mod tests { &pool, &firmware_rack, firmware_scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await @@ -830,7 +852,7 @@ mod tests { &pool, &deleted_rack, nmx_scope(), - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) .await, @@ -852,7 +874,7 @@ mod tests { &pool, &credential_failure_rack, credential_failure_scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, Some(RackMaintenanceAccessToken { credential_manager: &FailingCredentialManager, token: "test-token".into(), diff --git a/crates/nvlink-manager/src/switch_cert_monitor.rs b/crates/nvlink-manager/src/switch_cert_monitor.rs index 0e12f550f9..b14ec5ff47 100644 --- a/crates/nvlink-manager/src/switch_cert_monitor.rs +++ b/crates/nvlink-manager/src/switch_cert_monitor.rs @@ -850,7 +850,7 @@ impl SwitchCertificateMonitor { &self.db_pool, &target.rack_id, scope, - RackMaintenanceEligibility::ReadyOnly, + RackMaintenanceEligibility::RequireReady, None, ) => outcome.map_err(|error| { format!(