diff --git a/crates/api-core/src/handlers/managed_host.rs b/crates/api-core/src/handlers/managed_host.rs index 9ae0e3dc91..ca9ac3c3ae 100644 --- a/crates/api-core/src/handlers/managed_host.rs +++ b/crates/api-core/src/handlers/managed_host.rs @@ -113,11 +113,13 @@ pub(crate) async fn set_primary_interface( } /// Moves the database primary to the selected interface and records that exact -/// row as the host's desired boot target. +/// interface as the host's desired boot target. /// -/// The transaction locks admin segments, host interfaces, and then the host -/// machine in the same order as Site Explorer. Once it commits, the machine -/// controller owns the Redfish write and any reboot needed to converge it. +/// The transaction locks the related admin segments and host interfaces, then +/// the host `Machine` and assigned `Instance`. This keeps deletion from +/// committing between the snapshot and the related writes. Once the transaction +/// commits, the machine controller owns the Redfish write and any reboot needed +/// to converge it. async fn set_primary_interface_core( api: &Api, host_machine_id: MachineId, @@ -231,7 +233,8 @@ async fn set_primary_interface_core( let primary_interface_mac_address = new_primary_interface.mac_address; let boot_interface_id = new_primary_interface.boot_interface_id.clone(); let boot_target = boot_target_for_interface(primary_interface_mac_address, boot_interface_id); - let instance = db::instance::find_by_machine_id(&mut txn, &host_machine_id).await?; + let instance = + db::instance::find_live_by_machine_id_for_update(&mut txn, &host_machine_id).await?; let should_enqueue = matches!(machine.current_state(), ManagedHostState::Ready) && instance.is_none(); diff --git a/crates/api-core/src/tests/set_primary_interface.rs b/crates/api-core/src/tests/set_primary_interface.rs index f123fa2fe7..7fb90d6812 100644 --- a/crates/api-core/src/tests/set_primary_interface.rs +++ b/crates/api-core/src/tests/set_primary_interface.rs @@ -18,7 +18,8 @@ use std::str::FromStr; use carbide_redfish::libredfish::test_support::RedfishSimAction; -use carbide_uuid::machine::MachineInterfaceId; +use carbide_uuid::instance::InstanceId; +use carbide_uuid::machine::{MachineId, MachineInterfaceId}; use ipnetwork::IpNetwork; use model::machine::{InstanceState, ManagedHostState}; use model::machine_boot_interface::MachineBootInterfaceTarget; @@ -36,6 +37,64 @@ use crate::tests::common::api_fixtures::network_segment::{ create_host_inband_network_segment, create_underlay_network_segment, }; +#[derive(Debug, PartialEq)] +struct SetPrimaryPersistenceState { + interface_primaries: Vec<(String, bool)>, + interface_addresses: Vec<(String, String, String)>, + machine_network_configs: Vec<(String, String, String)>, + instance_network_config: (String, String), + desired_boot_interface: Option<(Option, Option, Option)>, +} + +async fn load_set_primary_persistence_state( + pool: &sqlx::PgPool, + host_id: MachineId, + instance_id: InstanceId, +) -> Result { + Ok(SetPrimaryPersistenceState { + interface_primaries: sqlx::query_as( + "SELECT id::text, primary_interface FROM machine_interfaces \ + WHERE machine_id = $1 ORDER BY id", + ) + .bind(host_id) + .fetch_all(pool) + .await?, + interface_addresses: sqlx::query_as( + "SELECT address.interface_id::text, address.address::text, address.allocation_type \ + FROM machine_interface_addresses address \ + JOIN machine_interfaces interface ON interface.id = address.interface_id \ + WHERE interface.machine_id = $1 \ + ORDER BY address.interface_id, address.address, address.allocation_type", + ) + .bind(host_id) + .fetch_all(pool) + .await?, + machine_network_configs: sqlx::query_as( + "SELECT id::text, network_config::text, network_config_version::text \ + FROM machines \ + WHERE id IN (SELECT id FROM machine_group_member_ids($1)) \ + ORDER BY id", + ) + .bind(host_id) + .fetch_all(pool) + .await?, + instance_network_config: sqlx::query_as( + "SELECT network_config::text, network_config_version::text \ + FROM instances WHERE id = $1", + ) + .bind(instance_id) + .fetch_one(pool) + .await?, + desired_boot_interface: sqlx::query_as( + "SELECT desired_mac_address::text, desired_interface_id, desired_version::text \ + FROM machine_boot_interfaces WHERE machine_id = $1", + ) + .bind(host_id) + .fetch_optional(pool) + .await?, + }) +} + // Unlike `set_primary_dpu`, `set_primary_interface` has no zero-DPU guard -- a // zero-DPU host is a first-class target. So on a zero-DPU host the call must get // PAST the would-be guard: it can still fail (here, because the interface id @@ -395,6 +454,94 @@ async fn test_set_primary_interface_rolls_back_primary_and_desired_together( Ok(()) } +// A deleted `Instance` remains associated with its host while lifecycle cleanup +// is pending. The locked lookup must reject both a primary move and a forced +// reconciliation before either operation can persist changes. +#[crate::sqlx_test] +async fn test_set_primary_interface_rejects_deleted_instance_and_rolls_back_writes( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let env = api_fixtures::create_test_env(pool).await; + let segment_id = env.create_vpc_and_tenant_segment().await; + let host = api_fixtures::create_managed_host_multi_dpu(&env, 2).await; + let host_id = host.id; + let instance = host + .instance_builer(&env) + .single_interface_network_config(segment_id) + .build() + .await; + + let (current_primary_id, promote_id) = { + let mut txn = env.pool.begin().await?; + let interfaces = db::machine_interface::find_by_machine_ids(txn.as_mut(), &[host_id]) + .await? + .remove(&host_id) + .expect("host should have interface rows"); + let current_primary_id = interfaces + .iter() + .find(|interface| interface.primary_interface) + .expect("host should have a primary interface") + .id; + let promote_id = interfaces + .into_iter() + .find(|interface| { + !interface.primary_interface && interface.attached_dpu_machine_id.is_some() + }) + .expect("host should have a non-primary DPU-backed interface") + .id; + txn.commit().await?; + (current_primary_id, promote_id) + }; + + let state_before = load_set_primary_persistence_state(&env.pool, host_id, instance.id).await?; + + let mut deletion = env.pool.begin().await?; + db::instance::mark_as_deleted(instance.id, deletion.as_mut()).await?; + deletion.commit().await?; + + struct Case { + name: &'static str, + interface_id: MachineInterfaceId, + force_reconcile: bool, + } + let cases = [ + Case { + name: "primary move", + interface_id: promote_id, + force_reconcile: false, + }, + Case { + name: "forced reconciliation", + interface_id: current_primary_id, + force_reconcile: true, + }, + ]; + + for case in cases { + let error = env + .api + .set_primary_interface(tonic::Request::new(forge::SetPrimaryInterfaceRequest { + host_machine_id: Some(host_id), + interface_id: Some(case.interface_id), + force_reconcile: case.force_reconcile, + ..Default::default() + })) + .await + .expect_err(case.name); + assert_eq!(error.code(), tonic::Code::FailedPrecondition); + assert_eq!( + error.message(), + format!("instance {} is being deleted", instance.id), + ); + + let state_after = + load_set_primary_persistence_state(&env.pool, host_id, instance.id).await?; + assert_eq!(state_after, state_before, "{} persisted state", case.name); + } + + Ok(()) +} + // `set_primary_interface` wakes an unassigned Ready host only after its intent // commits. Assigned hosts keep the same durable pending intent, but their // current lifecycle owns when it is safe to act on it. diff --git a/crates/api-db/src/instance.rs b/crates/api-db/src/instance.rs index eaee9ccf42..3deca6949b 100644 --- a/crates/api-db/src/instance.rs +++ b/crates/api-db/src/instance.rs @@ -346,6 +346,40 @@ pub async fn find_by_machine_id( find_by_id(txn, instance_id).await } +/// Locks and returns the live `Instance` assigned to one `Machine`. +/// +/// A returned record remains locked until the caller's transaction ends. `None` +/// means the `Machine` has no assigned `Instance`. If the assigned `Instance` is +/// already marked for deletion, the lookup returns +/// [`DatabaseError::FailedPrecondition`] instead of a snapshot that a caller +/// could use for later writes. +pub async fn find_live_by_machine_id_for_update( + txn: &mut PgConnection, + machine_id: &MachineId, +) -> Result, DatabaseError> { + let query = "SELECT row_to_json(i.*) AS instance, row_to_json(o.*) AS operating_system + FROM instances i + LEFT JOIN operating_systems o ON i.operating_system_id = o.id AND o.deleted IS NULL + WHERE i.machine_id = $1 + FOR UPDATE OF i"; + let Some(instance_and_os_row) = sqlx::query_as::<_, InstanceAndOsRow>(query) + .bind(machine_id) + .fetch_optional(txn) + .await + .map_err(|error| DatabaseError::query(query, error))? + else { + return Ok(None); + }; + let instance: InstanceSnapshot = instance_and_os_row.try_into()?; + if instance.deleted.is_some() { + return Err(DatabaseError::FailedPrecondition(format!( + "instance {} is being deleted", + instance.id + ))); + } + Ok(Some(instance)) +} + pub async fn find_by_machine_ids( txn: &mut PgConnection, machine_ids: &[&MachineId], @@ -466,6 +500,29 @@ pub async fn update_network_config( .await } +/// Distinguishes a missing or deleted `Instance` from a version conflict after +/// an optimistic configuration update affects no records. +async fn ensure_live_for_config_update( + txn: &mut PgConnection, + instance_id: InstanceId, +) -> Result<(), DatabaseError> { + let query = "SELECT deleted IS NULL FROM instances WHERE id=$1 FOR UPDATE"; + let live: Option = sqlx::query_scalar(query) + .bind(instance_id) + .fetch_optional(txn) + .await + .map_err(|error| DatabaseError::query(query, error))?; + match live { + Some(true) => Ok(()), + Some(false) => Err(DatabaseError::FailedPrecondition(format!( + "instance {instance_id} is being deleted" + ))), + None => Err(DatabaseError::FailedPrecondition(format!( + "instance {instance_id} does not exist" + ))), + } +} + pub async fn update_phone_home_last_contact( txn: &mut PgConnection, instance_id: InstanceId, @@ -515,8 +572,10 @@ pub async fn clear_phone_home_last_contact( /// - instance network and infiniband configurations /// - tenant organization IDs /// -/// This method does not check if the instance still exists. -/// A previous `Instance::find` call should fulfill this purpose. +/// The update applies only while the instance exists, is not marked deleted, +/// and still has `expected_version`. A deleted or missing instance reports a +/// failed precondition; a live instance with another version reports a +/// concurrent modification. pub async fn update_config( txn: &mut PgConnection, instance_id: InstanceId, @@ -546,7 +605,7 @@ pub async fn update_config( os_image_id=$7, keyset_ids=$8, name=$9, description=$10, labels=$11::json, network_security_group_id=$14, power_profile=$15 - WHERE id=$12 AND config_version=$13 + WHERE id=$12 AND config_version=$13 AND deleted IS NULL RETURNING id"; let query_result: Result<(InstanceId,), _> = sqlx::query_as(query) .bind(next_version) @@ -564,24 +623,28 @@ pub async fn update_config( .bind(expected_version) .bind(config.network_security_group_id) .bind(config.power_profile) - .fetch_one(txn) + .fetch_one(&mut *txn) .await; match query_result { Ok((_instance_id,)) => Ok(()), - Err(e) => Err(match e { - sqlx::Error::RowNotFound => { - DatabaseError::ConcurrentModificationError("instance", expected_version.to_string()) - } - e => DatabaseError::query(query, e), - }), + Err(sqlx::Error::RowNotFound) => { + ensure_live_for_config_update(txn, instance_id).await?; + Err(DatabaseError::ConcurrentModificationError( + "instance", + expected_version.to_string(), + )) + } + Err(error) => Err(DatabaseError::query(query, error)), } } /// Updates the Operating System /// -/// This method does not check if the instance still exists. -/// A previous `Instance::find` call should fulfill this purpose. +/// The update applies only while the instance exists, is not marked deleted, +/// and still has `expected_version`. A deleted or missing instance reports a +/// failed precondition; a live instance with another version reports a +/// concurrent modification. pub async fn update_os( txn: &mut PgConnection, instance_id: InstanceId, @@ -607,7 +670,7 @@ pub async fn update_os( let query = "UPDATE instances SET config_version=$1, operating_system_id=$2, os_ipxe_script=$3, os_user_data=$4, os_always_boot_with_ipxe=$5, os_phone_home_enabled=$6, os_image_id=$7 - WHERE id=$8 AND config_version=$9 + WHERE id=$8 AND config_version=$9 AND deleted IS NULL RETURNING id"; let query_result: Result<(InstanceId,), _> = sqlx::query_as(query) .bind(next_version) @@ -619,17 +682,19 @@ pub async fn update_os( .bind(os_image_id) .bind(instance_id) .bind(expected_version) - .fetch_one(txn) + .fetch_one(&mut *txn) .await; match query_result { Ok((_instance_id,)) => Ok(()), - Err(e) => Err(match e { - sqlx::Error::RowNotFound => { - DatabaseError::ConcurrentModificationError("instance", expected_version.to_string()) - } - e => DatabaseError::query(query, e), - }), + Err(sqlx::Error::RowNotFound) => { + ensure_live_for_config_update(txn, instance_id).await?; + Err(DatabaseError::ConcurrentModificationError( + "instance", + expected_version.to_string(), + )) + } + Err(error) => Err(DatabaseError::query(query, error)), } } @@ -1317,6 +1382,231 @@ mod tests { .unwrap(); assert_eq!(snapshots.len(), 2); } + + /// General and OS updates distinguish missing and deleted `Instance`s from + /// live records whose version has changed. + #[crate::sqlx_test] + async fn config_writers_return_distinct_errors_for_missing_deleted_and_stale_records( + pool: sqlx::PgPool, + ) { + enum StaleUpdate { + Config, + OperatingSystem, + } + + enum RowState { + Deleted, + Missing, + LiveWithNewerVersion, + } + + let cases = [ + ( + "deleted general config", + 0x44, + StaleUpdate::Config, + RowState::Deleted, + ), + ( + "deleted operating system", + 0x45, + StaleUpdate::OperatingSystem, + RowState::Deleted, + ), + ( + "live stale general config", + 0x46, + StaleUpdate::Config, + RowState::LiveWithNewerVersion, + ), + ( + "live stale operating system", + 0x47, + StaleUpdate::OperatingSystem, + RowState::LiveWithNewerVersion, + ), + ( + "missing general config", + 0x48, + StaleUpdate::Config, + RowState::Missing, + ), + ( + "missing operating system", + 0x49, + StaleUpdate::OperatingSystem, + RowState::Missing, + ), + ]; + + for (case_name, machine_seed, stale_update, row_state) in cases { + let mut setup = pool.begin().await.unwrap(); + let instance_id = seed_instance(&mut setup, machine_seed, None).await; + setup.commit().await.unwrap(); + + let stale_snapshot = find_by_id(&pool, instance_id).await.unwrap().unwrap(); + let expected_version = stale_snapshot.config_version; + let mut prepare = pool.begin().await.unwrap(); + let persisted_version = match row_state { + RowState::Deleted => { + mark_as_deleted(instance_id, prepare.as_mut()) + .await + .unwrap(); + Some(expected_version) + } + RowState::Missing => { + delete(instance_id, prepare.as_mut()).await.unwrap(); + None + } + RowState::LiveWithNewerVersion => { + let newer_version = expected_version.increment(); + sqlx::query("UPDATE instances SET config_version = $1 WHERE id = $2") + .bind(newer_version) + .bind(instance_id) + .execute(prepare.as_mut()) + .await + .unwrap(); + Some(newer_version) + } + }; + prepare.commit().await.unwrap(); + + let mut update = pool.begin().await.unwrap(); + let error = match stale_update { + StaleUpdate::Config => { + update_config( + update.as_mut(), + instance_id, + expected_version, + stale_snapshot.config, + stale_snapshot.metadata, + ) + .await + } + StaleUpdate::OperatingSystem => { + update_os( + update.as_mut(), + instance_id, + expected_version, + stale_snapshot.config.os, + ) + .await + } + } + .expect_err("a stale writer must not update the instance"); + + match row_state { + RowState::Deleted => { + assert!(matches!(&error, DatabaseError::FailedPrecondition(_))); + assert_eq!( + error.to_string(), + format!("instance {instance_id} is being deleted"), + "unexpected error for {case_name}", + ); + } + RowState::Missing => { + assert!(matches!(&error, DatabaseError::FailedPrecondition(_))); + assert_eq!( + error.to_string(), + format!("instance {instance_id} does not exist"), + "unexpected error for {case_name}", + ); + } + RowState::LiveWithNewerVersion => assert!( + matches!( + &error, + DatabaseError::ConcurrentModificationError("instance", version) + if version == &expected_version.to_string() + ), + "unexpected error for {case_name}: {error:?}", + ), + } + update.rollback().await.unwrap(); + + let version_after_rejection: Option = + sqlx::query_scalar("SELECT config_version FROM instances WHERE id = $1") + .bind(instance_id) + .fetch_optional(&pool) + .await + .unwrap(); + assert_eq!( + version_after_rejection, persisted_version, + "the rejected {case_name} update changed the instance version", + ); + } + } + + #[crate::sqlx_test] + async fn live_machine_lookup_locks_the_instance_row(pool: sqlx::PgPool) { + let machine_id = MachineId::new( + MachineIdSource::ProductBoardChassisSerial, + [0x48; 32], + MachineType::Host, + ); + let unassigned_machine_id = MachineId::new( + MachineIdSource::ProductBoardChassisSerial, + [0x49; 32], + MachineType::Host, + ); + let mut setup = pool.begin().await.unwrap(); + let instance_id = seed_instance(&mut setup, 0x48, None).await; + sqlx::query("INSERT INTO machines (id, dpf) VALUES ($1, '{}'::jsonb)") + .bind(unassigned_machine_id) + .execute(setup.as_mut()) + .await + .unwrap(); + setup.commit().await.unwrap(); + + let mut reader = pool.begin().await.unwrap(); + assert!( + find_live_by_machine_id_for_update(reader.as_mut(), &unassigned_machine_id) + .await + .unwrap() + .is_none(), + "an unassigned machine must return no instance", + ); + let instance = find_live_by_machine_id_for_update(reader.as_mut(), &machine_id) + .await + .unwrap() + .expect("the machine should have an assigned instance"); + assert_eq!(instance.id, instance_id); + + let mut deletion = pool.begin().await.unwrap(); + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(deletion.as_mut()) + .await + .unwrap(); + let error = sqlx::query("UPDATE instances SET deleted = NOW() WHERE id = $1") + .bind(instance_id) + .execute(deletion.as_mut()) + .await + .expect_err("deletion must wait for the instance reader"); + assert_eq!( + error + .as_database_error() + .and_then(sqlx::error::DatabaseError::code) + .as_deref(), + Some("55P03"), + ); + deletion.rollback().await.unwrap(); + reader.rollback().await.unwrap(); + + let mut deletion = pool.begin().await.unwrap(); + mark_as_deleted(instance_id, deletion.as_mut()) + .await + .unwrap(); + deletion.commit().await.unwrap(); + + let mut reader = pool.begin().await.unwrap(); + let error = find_live_by_machine_id_for_update(reader.as_mut(), &machine_id) + .await + .expect_err("a deleted instance must not be returned for writes"); + assert_eq!( + error.to_string(), + format!("instance {instance_id} is being deleted"), + ); + reader.rollback().await.unwrap(); + } } #[cfg(test)]