diff --git a/crates/api-core/src/tests/rack_state_controller/handler.rs b/crates/api-core/src/tests/rack_state_controller/handler.rs index 6209c6f57d..14b6a0c565 100644 --- a/crates/api-core/src/tests/rack_state_controller/handler.rs +++ b/crates/api-core/src/tests/rack_state_controller/handler.rs @@ -271,6 +271,30 @@ async fn set_machine_host_reprovision_state( Ok(()) } +async fn set_machine_power_states( + pool: &sqlx::PgPool, + machine_id: &MachineId, + desired_power_state: model::power_manager::PowerState, + actual_power_state: model::power_manager::PowerState, +) -> Result<(), Box> { + let mut txn = pool.begin().await?; + let power_options = db::power_options::get_by_ids(&[*machine_id], txn.as_mut()) + .await? + .pop() + .expect("machine should have power options"); + let mut power_options = db::power_options::update_desired_state( + machine_id, + desired_power_state, + &power_options.desired_power_state_version, + txn.as_mut(), + ) + .await?; + power_options.last_fetched_power_state = actual_power_state; + db::power_options::persist(&power_options, txn.as_mut()).await?; + txn.commit().await?; + Ok(()) +} + fn waiting_for_rack_firmware_upgrade_state() -> model::machine::HostReprovisionState { model::machine::HostReprovisionState::WaitingForRackFirmwareUpgrade } @@ -1644,6 +1668,117 @@ async fn test_ingestion_transitions_to_firmware_upgrade_and_submits_rack_profile Ok(()) } +#[crate::sqlx_test] +async fn test_firmware_upgrade_start_rejects_desired_off_machine_before_rms_submission( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let env = create_test_env_with_overrides( + pool.clone(), + TestEnvOverrides { + config: Some(config_with_rack_profiles()), + ..Default::default() + }, + ) + .await; + let (rack_id, host) = create_single_compute_rack(&env, &pool).await?; + set_machine_power_states( + &pool, + &host.host_snapshot.id, + model::power_manager::PowerState::Off, + model::power_manager::PowerState::On, + ) + .await?; + + let config = RackConfig { + maintenance_requested: Some(MaintenanceScope { + machine_ids: vec![host.host_snapshot.id], + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"fw-json"}"#.to_string()), + components: vec!["BMC".to_string()], + force_update: false, + }], + ..Default::default() + }), + ..Default::default() + }; + let mut txn = pool.acquire().await?; + db_rack::update(txn.as_mut(), &rack_id, &config).await?; + drop(txn); + env.api + .credential_manager + .set_credentials( + &CredentialKey::RackMaintenanceAccessToken { + rack_id: rack_id.clone(), + }, + &Credentials::UsernamePassword { + username: "access_token".to_string(), + password: "token".to_string(), + }, + ) + .await + .map_err(|error| eyre::eyre!("failed to set maintenance access token: {}", error))?; + + let mut rack = get_db_rack(env.db_reader().as_mut(), &rack_id).await; + let handler = RackStateHandler::default(); + let mut services = env.rack_state_handler_services(); + let mut metrics = RackMetrics::default(); + let mut db_writes = DbWriteBatch::default(); + let mut ctx = StateHandlerContext:: { + services: &mut services, + metrics: &mut metrics, + pending_db_writes: &mut db_writes, + }; + let fw_state = RackState::Maintenance { + maintenance_state: RackMaintenanceState::FirmwareUpgrade { + rack_firmware_upgrade: FirmwareUpgradeState::Start, + }, + }; + + let mut outcome = handler + .handle_object_state(&rack_id, &mut rack, &fw_state, &mut ctx) + .await?; + if let Some(txn) = outcome.take_transaction() { + txn.commit().await?; + } + + let StateHandlerOutcome::Transition { + next_state: RackState::Error { cause }, + .. + } = outcome + else { + panic!("desired-Off target should fail rack firmware start"); + }; + assert!(cause.contains(&host.host_snapshot.id.to_string())); + assert!(cause.contains("desired power state is Off")); + assert!( + env.rms_sim + .submitted_apply_firmware_object_requests() + .await + .is_empty() + ); + + let rack = get_db_rack(env.db_reader().as_mut(), &rack_id).await; + assert!(rack.config.maintenance_requested.is_none()); + let machine = db::machine::find_one( + &pool, + &host.host_snapshot.id, + model::machine::machine_search_config::MachineSearchConfig::default(), + ) + .await? + .expect("machine should exist"); + assert!(machine.host_reprovision_requested.is_none()); + let token = env + .test_credential_manager + .get_credentials(&CredentialKey::RackMaintenanceAccessToken { + rack_id: rack_id.clone(), + }) + .await + .map_err(|error| eyre::eyre!("failed to get maintenance access token: {}", error))?; + assert!(token.is_none()); + + Ok(()) +} + #[crate::sqlx_test] async fn test_firmware_upgrade_start_submits_json_and_deletes_access_token( pool: sqlx::PgPool, @@ -1883,6 +2018,152 @@ async fn test_firmware_upgrade_start_missing_profile_deletes_access_token( Ok(()) } +/// A machine that cannot consume its rack reprovision request because desired +/// power is Off fails the rack job without clearing requests already owned by +/// active device reprovisioning state machines. +#[crate::sqlx_test] +async fn test_firmware_upgrade_wait_for_complete_recovers_power_blocked_machine( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let env = create_test_env_with_overrides( + pool.clone(), + TestEnvOverrides { + config: Some(config_with_rack_profiles()), + ..Default::default() + }, + ) + .await; + let (rack_id, blocked_host, active_host) = create_two_compute_rack(&env, &pool).await?; + set_machine_power_states( + &pool, + &blocked_host.host_snapshot.id, + model::power_manager::PowerState::Off, + model::power_manager::PowerState::Off, + ) + .await?; + + let scope = MaintenanceScope { + machine_ids: vec![blocked_host.host_snapshot.id, active_host.host_snapshot.id], + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"fw-json"}"#.to_string()), + components: vec!["BMC".to_string()], + force_update: false, + }], + ..Default::default() + }; + let job = FirmwareUpgradeJob { + job_id: Some("parent-job".to_string()), + status: Some("in_progress".to_string()), + started_at: Some(chrono::Utc::now()), + ..Default::default() + }; + let initiator = format!("rack-{rack_id}"); + let mut txn = pool.begin().await?; + let config = RackConfig { + maintenance_requested: Some(scope), + ..Default::default() + }; + db_rack::update(txn.as_mut(), &rack_id, &config).await?; + db_rack::update_firmware_upgrade_job(txn.as_mut(), &rack_id, Some(&job)).await?; + db::host_machine_update::trigger_host_reprovisioning_request( + txn.as_mut(), + &initiator, + &blocked_host.host_snapshot.id, + ) + .await?; + db::host_machine_update::trigger_host_reprovisioning_request( + txn.as_mut(), + &initiator, + &active_host.host_snapshot.id, + ) + .await?; + txn.commit().await?; + env.api + .credential_manager + .set_credentials( + &CredentialKey::RackMaintenanceAccessToken { + rack_id: rack_id.clone(), + }, + &Credentials::UsernamePassword { + username: "access_token".to_string(), + password: "token".to_string(), + }, + ) + .await + .map_err(|error| eyre::eyre!("failed to set maintenance access token: {}", error))?; + set_machine_host_reprovision_state( + &pool, + &active_host.host_snapshot.id, + waiting_for_rack_firmware_upgrade_state(), + ) + .await?; + + let mut rack = get_db_rack(env.db_reader().as_mut(), &rack_id).await; + let handler = RackStateHandler::default(); + let mut services = env.rack_state_handler_services(); + let mut metrics = RackMetrics::default(); + let mut db_writes = DbWriteBatch::default(); + let mut ctx = StateHandlerContext:: { + services: &mut services, + metrics: &mut metrics, + pending_db_writes: &mut db_writes, + }; + let fw_state = RackState::Maintenance { + maintenance_state: RackMaintenanceState::FirmwareUpgrade { + rack_firmware_upgrade: FirmwareUpgradeState::WaitForComplete, + }, + }; + + let outcome = handler + .handle_object_state(&rack_id, &mut rack, &fw_state, &mut ctx) + .await?; + + let StateHandlerOutcome::Transition { + next_state: RackState::Error { cause }, + .. + } = outcome + else { + panic!("power-blocked rack firmware job should transition to Error"); + }; + assert!(cause.contains(&blocked_host.host_snapshot.id.to_string())); + assert!(!cause.contains(&active_host.host_snapshot.id.to_string())); + + let rack = get_db_rack(env.db_reader().as_mut(), &rack_id).await; + assert!(rack.config.maintenance_requested.is_none()); + let job = rack + .firmware_upgrade_job + .expect("failed firmware job should be retained"); + assert_eq!(job.status.as_deref(), Some("failed")); + assert!(job.completed_at.is_some()); + + let blocked_machine = db::machine::find_one( + &pool, + &blocked_host.host_snapshot.id, + model::machine::machine_search_config::MachineSearchConfig::default(), + ) + .await? + .expect("blocked machine should exist"); + assert!(blocked_machine.host_reprovision_requested.is_none()); + let active_machine = db::machine::find_one( + &pool, + &active_host.host_snapshot.id, + model::machine::machine_search_config::MachineSearchConfig::default(), + ) + .await? + .expect("active machine should exist"); + assert!(active_machine.host_reprovision_requested.is_some()); + let token = env + .test_credential_manager + .get_credentials(&CredentialKey::RackMaintenanceAccessToken { + rack_id: rack_id.clone(), + }) + .await + .map_err(|error| eyre::eyre!("failed to get maintenance access token: {}", error))?; + assert!(token.is_none()); + + Ok(()) +} + /// test_firmware_upgrade_wait_for_complete_waits_while_jobs_running verifies /// that WaitForComplete remains in a wait state while machines are still in /// WaitingForRackFirmwareUpgrade and writes in-progress rack firmware status @@ -1957,10 +2238,11 @@ async fn test_firmware_upgrade_wait_for_complete_waits_while_jobs_running( txn.commit().await?; } - assert!( - matches!(outcome, StateHandlerOutcome::Wait { .. }), - "Expected Wait while machine controller is still WaitingForRackFirmwareUpgrade" - ); + let StateHandlerOutcome::Wait { reason, .. } = outcome else { + panic!("Expected Wait while machine controller is still WaitingForRackFirmwareUpgrade"); + }; + assert!(reason.contains(&host.host_snapshot.id.to_string())); + assert!(reason.contains("pending=1")); let machine = db::machine::find_one( &pool, diff --git a/crates/api-db/src/host_machine_update.rs b/crates/api-db/src/host_machine_update.rs index f32ff7c70b..8aec8ed8ac 100644 --- a/crates/api-db/src/host_machine_update.rs +++ b/crates/api-db/src/host_machine_update.rs @@ -123,6 +123,32 @@ pub async fn clear_host_reprovisioning_request( Ok(()) } +/// Clears a rack-owned host reprovisioning request if the machine has not left `Ready`. +/// +/// Returns whether the request was cleared. A concurrent transition out of `Ready`, or a +/// request from another initiator, leaves the request untouched so its owning controller can +/// unwind it. +pub async fn clear_ready_host_reprovisioning_request( + txn: &mut PgConnection, + machine_id: &MachineId, + initiator: &str, +) -> Result { + let query = r#"UPDATE machines + SET host_reprovisioning_requested = NULL + WHERE id = $1 + AND controller_state->>'state' = 'ready' + AND host_reprovisioning_requested->>'initiator' = $2 + RETURNING id"#; + let cleared = sqlx::query_as::<_, MachineId>(query) + .bind(machine_id) + .bind(initiator) + .fetch_optional(txn) + .await + .map_err(|e| DatabaseError::query(query, e))?; + + Ok(cleared.is_some()) +} + pub async fn reset_host_reprovisioning_request( txn: &mut PgConnection, machine_id: &MachineId, diff --git a/crates/rack-controller/src/maintenance.rs b/crates/rack-controller/src/maintenance.rs index a9478e9842..768a981b80 100644 --- a/crates/rack-controller/src/maintenance.rs +++ b/crates/rack-controller/src/maintenance.rs @@ -48,8 +48,8 @@ use component_manager::component_manager::ComponentManager; use component_manager::error::ComponentManagerError; use db::{ host_machine_update as db_host_machine_update, machine as db_machine, - machine_topology as db_machine_topology, power_shelf as db_power_shelf, rack as db_rack, - switch as db_switch, + machine_topology as db_machine_topology, power_options as db_power_options, + power_shelf as db_power_shelf, rack as db_rack, switch as db_switch, }; use librms::protos::{rack_manager as rms, rack_manager_v2 as rms_v2}; use model::rack::{ @@ -97,6 +97,10 @@ async fn clear_rv_labels( Ok(()) } +fn rack_maintenance_initiator(rack_id: &RackId) -> String { + format!("rack-{rack_id}") +} + async fn trigger_rack_firmware_reprovisioning_requests( txn: &mut sqlx::PgConnection, rack_id: &RackId, @@ -105,19 +109,16 @@ async fn trigger_rack_firmware_reprovisioning_requests( power_shelf_ids: &[carbide_uuid::power_shelf::PowerShelfId], activities: &[MaintenanceActivity], ) -> Result<(), StateHandlerError> { + let initiator = rack_maintenance_initiator(rack_id); for machine_id in machine_ids { - db_host_machine_update::trigger_host_reprovisioning_request( - txn, - &format!("rack-{}", rack_id), - machine_id, - ) - .await?; + db_host_machine_update::trigger_host_reprovisioning_request(txn, &initiator, machine_id) + .await?; } for switch_id in switch_ids { db_switch::set_switch_reprovisioning_requested( txn, *switch_id, - &format!("rack-{}", rack_id), + &initiator, activities.to_vec(), ) .await?; @@ -126,7 +127,7 @@ async fn trigger_rack_firmware_reprovisioning_requests( db_power_shelf::set_power_shelf_reprovisioning_requested( txn, *power_shelf_id, - &format!("rack-{}", rack_id), + &initiator, activities.to_vec(), ) .await?; @@ -162,9 +163,9 @@ async fn clear_nvos_update_statuses( Ok(()) } -/// Aggregated firmware progress for machines/switches participating in a rack -/// firmware job. Advancement out of `WaitForComplete` is based on machine and -/// switch controller states, not RMS job strings or `rack_fw_details`. +/// Aggregated firmware progress for machines, switches, and power shelves +/// participating in a rack firmware job. Advancement out of `WaitForComplete` +/// is based on device controller states, not RMS job strings or firmware status. #[derive(Debug, Clone, PartialEq, Eq)] enum DeviceFirmwareProgress { Waiting { @@ -190,6 +191,79 @@ enum DeviceFirmwareOutcome { Completed, } +async fn desired_off_machine_ids( + txn: &mut sqlx::PgConnection, + machine_ids: &[carbide_uuid::machine::MachineId], +) -> Result, StateHandlerError> { + if machine_ids.is_empty() { + return Ok(Vec::new()); + } + + let mut machine_ids = db_power_options::get_by_ids(machine_ids, txn) + .await? + .into_iter() + .filter(|options| options.desired_power_state == model::power_manager::PowerState::Off) + .map(|options| options.host_id) + .collect::>(); + machine_ids.sort_by_key(ToString::to_string); + Ok(machine_ids) +} + +fn format_machine_ids(machine_ids: &[carbide_uuid::machine::MachineId]) -> String { + machine_ids + .iter() + .map(ToString::to_string) + .collect::>() + .join(", ") +} + +async fn load_scoped_machines( + txn: &mut sqlx::PgConnection, + rack_id: &RackId, + scope: &MaintenanceScope, +) -> Result, StateHandlerError> { + let machine_ids = db_machine::find_machine_ids( + &mut *txn, + model::machine::machine_search_config::MachineSearchConfig { + rack_id: Some(rack_id.clone()), + ..Default::default() + }, + ) + .await?; + let machines = if machine_ids.is_empty() { + Vec::new() + } else { + db_machine::find( + &mut *txn, + db::ObjectFilter::List(&machine_ids), + model::machine::machine_search_config::MachineSearchConfig::default(), + ) + .await? + }; + Ok(filter_machines_by_scope(machines, scope)) +} + +async fn power_blocked_rack_firmware_machine_ids( + txn: &mut sqlx::PgConnection, + rack_id: &RackId, + scope: &MaintenanceScope, +) -> Result, StateHandlerError> { + let machines = load_scoped_machines(txn, rack_id, scope).await?; + let initiator = rack_maintenance_initiator(rack_id); + let ready_rack_requested_ids = machines + .iter() + .filter(|machine| matches!(machine.state.value, model::machine::ManagedHostState::Ready)) + .filter(|machine| { + machine + .host_reprovision_requested + .as_ref() + .is_some_and(|request| request.initiator == initiator) + }) + .map(|machine| machine.id) + .collect::>(); + desired_off_machine_ids(txn, &ready_rack_requested_ids).await +} + async fn resolve_machine_id_for_firmware_device( txn: &mut sqlx::PgConnection, device: &FirmwareUpgradeDeviceStatus, @@ -357,33 +431,15 @@ fn summarize_firmware_outcomes(outcomes: &[DeviceFirmwareOutcome]) -> DeviceFirm DeviceFirmwareProgress::Completed { completed, total } } -/// Reads machine and switch controller states for devices in `rack_id`, +/// Reads device controller states for devices in `rack_id`, /// filtered by `scope`, and decides whether firmware WaitForComplete can /// advance. Device membership comes from the DB + scope, not the firmware job. async fn evaluate_firmware_progress_from_devices( txn: &mut sqlx::PgConnection, rack_id: &RackId, scope: &MaintenanceScope, -) -> Result { - let machine_ids = db_machine::find_machine_ids( - &mut *txn, - model::machine::machine_search_config::MachineSearchConfig { - rack_id: Some(rack_id.clone()), - ..Default::default() - }, - ) - .await?; - let machines = if machine_ids.is_empty() { - Vec::new() - } else { - db_machine::find( - &mut *txn, - db::ObjectFilter::List(&machine_ids), - model::machine::machine_search_config::MachineSearchConfig::default(), - ) - .await? - }; - let machines = filter_machines_by_scope(machines, scope); +) -> Result<(DeviceFirmwareProgress, Vec), StateHandlerError> { + let machines = load_scoped_machines(txn, rack_id, scope).await?; let switch_ids = db_switch::find_ids( &mut *txn, @@ -426,10 +482,30 @@ async fn evaluate_firmware_progress_from_devices( let power_shelves = filter_power_shelves_by_scope(power_shelves, scope); let mut outcomes = Vec::with_capacity(machines.len() + switches.len() + power_shelves.len()); - outcomes.extend(machines.iter().map(machine_firmware_outcome)); - outcomes.extend(switches.iter().map(switch_firmware_outcome)); - outcomes.extend(power_shelves.iter().map(power_shelf_firmware_outcome)); - Ok(summarize_firmware_outcomes(&outcomes)) + let mut pending_device_ids = Vec::new(); + for machine in &machines { + let outcome = machine_firmware_outcome(machine); + if outcome == DeviceFirmwareOutcome::Waiting { + pending_device_ids.push(machine.id.to_string()); + } + outcomes.push(outcome); + } + for switch in &switches { + let outcome = switch_firmware_outcome(switch); + if outcome == DeviceFirmwareOutcome::Waiting { + pending_device_ids.push(switch.id.to_string()); + } + outcomes.push(outcome); + } + for power_shelf in &power_shelves { + let outcome = power_shelf_firmware_outcome(power_shelf); + if outcome == DeviceFirmwareOutcome::Waiting { + pending_device_ids.push(power_shelf.id.to_string()); + } + outcomes.push(outcome); + } + pending_device_ids.sort(); + Ok((summarize_firmware_outcomes(&outcomes), pending_device_ids)) } fn filter_machines_by_scope( @@ -2286,6 +2362,35 @@ pub async fn handle_maintenance( let nvos_json_pending = requested_nvos_config_json(scope).is_some(); + let desired_off_target_ids = { + let mut conn = ctx.services.db_pool.acquire().await?; + let machine_ids = load_scoped_machines(conn.as_mut(), id, scope) + .await? + .into_iter() + .map(|machine| machine.id) + .collect::>(); + desired_off_machine_ids(conn.as_mut(), &machine_ids).await? + }; + if !desired_off_target_ids.is_empty() { + if uses_stored_token { + delete_rack_maintenance_access_token( + ctx.services.credential_manager.as_ref(), + id, + ) + .await; + } + return transition_to_rack_error( + id, + state, + format!( + "rack firmware upgrade cannot target machines whose desired power state is Off: {}", + format_machine_ids(&desired_off_target_ids) + ), + ctx, + ) + .await; + } + let Some(rms_client) = ctx.services.rms_client.as_ref() else { if uses_stored_token { delete_rack_maintenance_access_token( @@ -2461,11 +2566,59 @@ pub async fn handle_maintenance( .with_txn(txn)) } FirmwareUpgradeState::WaitForComplete => { - if state.firmware_upgrade_job.is_none() { + let Some(current_job) = state.firmware_upgrade_job.clone() else { return Ok(StateHandlerOutcome::wait( "firmware upgrade: no job recorded yet".into(), )); + }; + + let power_blocked_machine_ids = { + let mut conn = ctx.services.db_pool.acquire().await?; + power_blocked_rack_firmware_machine_ids(conn.as_mut(), id, scope).await? + }; + if !power_blocked_machine_ids.is_empty() { + let mut recovery_txn = ctx.services.db_pool.begin().await?; + let now = chrono::Utc::now(); + let mut job = current_job.clone(); + job.status = Some("failed".into()); + if job.completed_at.is_none() { + job.completed_at = Some(now); + } + db_rack::update_firmware_upgrade_job(recovery_txn.as_mut(), id, Some(&job)) + .await?; + state.firmware_upgrade_job = Some(job); + + let initiator = rack_maintenance_initiator(id); + for machine_id in &power_blocked_machine_ids { + db_host_machine_update::clear_ready_host_reprovisioning_request( + recovery_txn.as_mut(), + machine_id, + &initiator, + ) + .await?; + } + + if state.config.maintenance_requested.is_some() { + state.config.maintenance_requested = None; + db_rack::update(recovery_txn.as_mut(), id, &state.config).await?; + } + let cause = format!( + "rack firmware upgrade cannot progress because target machines are Ready with desired power state Off: {}", + format_machine_ids(&power_blocked_machine_ids) + ); + + // Commit before credentials cleanup so the transaction is not held across + // that await. Controllers that already entered ReProvisioning retain their + // requests and use the rack Error state to unwind independently. + recovery_txn.commit().await?; + delete_rack_maintenance_access_token( + ctx.services.credential_manager.as_ref(), + id, + ) + .await; + return Ok(StateHandlerOutcome::transition(RackState::Error { cause })); } + let Some(rms_client) = ctx.services.rms_client.as_ref() else { if requested_nvos_config_json(scope).is_some() { delete_rack_maintenance_access_token( @@ -2477,9 +2630,8 @@ pub async fn handle_maintenance( return transition_to_rack_error(id, state, "RMS client not configured", ctx) .await; }; - let current_job = state.firmware_upgrade_job.as_ref().unwrap(); let mut job = - rms_get_firmware_upgrade_status(rms_client.as_ref(), current_job).await?; + rms_get_firmware_upgrade_status(rms_client.as_ref(), ¤t_job).await?; let mut txn = ctx.services.db_pool.begin().await?; @@ -2587,12 +2739,12 @@ pub async fn handle_maintenance( // Advancement is driven by machine/switch/power-shelf controller // states for devices in this rack that are selected by the // maintenance scope. - let progress = + let (progress, pending_device_ids) = evaluate_firmware_progress_from_devices(txn.as_mut(), id, scope).await?; match progress { DeviceFirmwareProgress::Waiting { - pending: _, + pending, total, completed, failed, @@ -2600,11 +2752,12 @@ pub async fn handle_maintenance( db_rack::update_firmware_upgrade_job(txn.as_mut(), id, Some(&job)).await?; state.firmware_upgrade_job = Some(job); Ok(StateHandlerOutcome::wait(format!( - "firmware upgrade: waiting on machine/switch/power-shelf controller state ({}/{} past firmware wait, completed={}, failed={})", - completed + failed, - total, + "firmware upgrade: waiting for machine/switch/power-shelf controllers (completed={}, failed={}, pending={}/{}; pending devices: [{}])", completed, - failed + failed, + pending, + total, + pending_device_ids.join(", ") )) .with_txn(txn)) }