From 017af4339e4cfd777051f76acc1fc23db0bec85a Mon Sep 17 00:00:00 2001 From: Chet Nichols III Date: Fri, 31 Jul 2026 07:59:53 -0700 Subject: [PATCH] fix: prevent duplicate machine interface addresses Enforce one database owner per machine-interface IP and stop the migration with owner details if existing rows conflict. Return typed fixed-address conflicts while dynamic allocation retries another candidate. Keep address changes atomic and claim addresses before publishing DNS names so concurrent writers cannot leave partial rows or invert constraint lock order. Stage admin-network remaps before applying them so overlapping updates remain valid under the new unique key. This supports https://github.com/NVIDIA/infra-controller/issues/4150 Signed-off-by: Chet Nichols III --- crates/api-core/src/dhcp/expire.rs | 6 +- crates/api-core/src/dhcp/v6.rs | 16 +- .../api-core/src/tests/machine_discovery.rs | 17 +- ...152_machine_interface_addresses_unique.sql | 312 +++++ crates/api-db/src/machine_interface.rs | 360 ++--- crates/api-db/src/machine_interface/tests.rs | 1174 ++++++++++++++++- .../api-db/src/machine_interface_address.rs | 436 +++++- crates/api-db/src/network_segment.rs | 12 +- 8 files changed, 2116 insertions(+), 217 deletions(-) create mode 100644 crates/api-db/migrations/20260730222152_machine_interface_addresses_unique.sql diff --git a/crates/api-core/src/dhcp/expire.rs b/crates/api-core/src/dhcp/expire.rs index 2f18e8e31f..23af22941b 100644 --- a/crates/api-core/src/dhcp/expire.rs +++ b/crates/api-core/src/dhcp/expire.rs @@ -60,9 +60,9 @@ pub async fn expire_dhcp_lease( // When the caller provides the MAC, scope the delete to the (ip, mac) pair. // Otherwise use the address-only variant, which is what an admin-cli call // deleting a specific IP allocation would hit. Either way, both variants - // return the interfaces whose rows were actually deleted, so we resync those - // authoritative owners rather than a separately looked-up interface (which - // could differ if ownership changed or multiple rows share the address). + // return the interfaces whose rows were actually deleted, so we resync + // from those authoritative rows rather than a separate ownership lookup + // (which could differ if ownership changed between the lookup and delete). let resync_targets = match mac_address { Some(mac) => { db::machine_interface_address::delete_by_address_and_mac( diff --git a/crates/api-core/src/dhcp/v6.rs b/crates/api-core/src/dhcp/v6.rs index 4d74fdf242..14e862f79f 100644 --- a/crates/api-core/src/dhcp/v6.rs +++ b/crates/api-core/src/dhcp/v6.rs @@ -95,13 +95,15 @@ pub async fn observe_slaac_address( if let Some(address) = slaac_gua_from_eui64(prefix, mac) { let address = IpAddr::V6(address); - // TODO: This is a best-effort ownership check, not a complete - // concurrency boundary. Static assignment and preallocation do not yet - // share a segment lock with SLAAC observation, so they can still race - // between this read and insert. A future PR should route DHCP, SLAAC, - // and static address writes through one DB helper that locks the owning - // segment, checks global address ownership, applies the replacement - // policy, and writes the row. + // This lookup keeps sequential replays and existing-owner errors + // specific to SLAAC. The shared insert below closes an exact-address + // race if another interface claims this address after the lookup. + // + // TODO: A different static or DHCPv6 address can still land on this + // interface after the family check above. The existing per-interface + // family constraint rejects that race, but the losing request receives + // a database error. Same-interface family replacement needs one shared + // transaction policy across SLAAC, DHCP, and static writers. if let Some(existing) = db::machine_interface_address::find_by_address(&mut *txn, address).await? { diff --git a/crates/api-core/src/tests/machine_discovery.rs b/crates/api-core/src/tests/machine_discovery.rs index d59d319efd..b5a7b212d8 100644 --- a/crates/api-core/src/tests/machine_discovery.rs +++ b/crates/api-core/src/tests/machine_discovery.rs @@ -888,31 +888,16 @@ async fn test_insecure_discovery_requires_interface_id( } #[crate::sqlx_test] -async fn test_discovery_ip_lookup_rejects_missing_and_ambiguous_mappings( +async fn test_discovery_ip_lookup_rejects_missing_mapping( pool: sqlx::PgPool, ) -> Result<(), Box> { let env = create_test_env(pool).await; - let first_interface_id = - common::api_fixtures::dpu::dpu_discover_dhcp(&env, "02:00:00:00:10:01").await; - let second_interface_id = - common::api_fixtures::dpu::dpu_discover_dhcp(&env, "02:00:00:00:10:02").await; let mut txn = env.pool.begin().await?; let missing = db::machine_interface::find_for_update_by_ip(&mut txn, "203.0.113.253".parse().unwrap()) .await; assert!(missing.is_err()); - - let first_interface = db::machine_interface::find_one(&mut *txn, first_interface_id).await?; - let first_address = first_interface.addresses[0]; - sqlx::query("UPDATE machine_interface_addresses SET address = $1 WHERE interface_id = $2") - .bind(first_address) - .bind(second_interface_id) - .execute(&mut *txn) - .await?; - - let ambiguous = db::machine_interface::find_for_update_by_ip(&mut txn, first_address).await; - assert!(ambiguous.is_err()); Ok(()) } diff --git a/crates/api-db/migrations/20260730222152_machine_interface_addresses_unique.sql b/crates/api-db/migrations/20260730222152_machine_interface_addresses_unique.sql new file mode 100644 index 0000000000..5e3e47de08 --- /dev/null +++ b/crates/api-db/migrations/20260730222152_machine_interface_addresses_unique.sql @@ -0,0 +1,312 @@ +-- A machine-interface address identifies one interface across the site. The +-- existing `(interface_id, address)` constraint only protects one interface +-- from storing the same address twice, so concurrent writers can still give +-- the same IP to different interfaces. +-- +-- `inet` equality includes the mask. Group by `host(address)` before changing +-- anything so a legacy `/24` and the normal `/32` form of the same IPv4 host +-- cannot slip past the audit (and likewise for IPv6). +-- +-- Take the strongest lock this migration will need before the audit starts. +-- That avoids a lock-upgrade deadlock with an older API transaction that reads +-- the table and then waits to write. API reads and writes resume after this +-- migration commits, at which point the database applies the same invariant. +LOCK TABLE machine_interface_addresses IN ACCESS EXCLUSIVE MODE; + +DO $$ +DECLARE + duplicate_addresses text; + duplicate_address_count bigint; +BEGIN + SELECT count(*) + INTO duplicate_address_count + FROM ( + SELECT host(address) + FROM machine_interface_addresses + GROUP BY host(address) + HAVING count(*) > 1 + ) duplicate_groups; + + SELECT string_agg( + format('%s: %s', host_address, owners), + E'\n' ORDER BY host_address + ) + INTO duplicate_addresses + FROM ( + SELECT + host(mia.address) AS host_address, + string_agg( + format( + 'interface %s (MAC %s, segment %s)', + mi.id, + mi.mac_address, + mi.segment_id + ), + ', ' ORDER BY mi.id + ) AS owners + FROM machine_interface_addresses mia + JOIN machine_interfaces mi ON mi.id = mia.interface_id + GROUP BY host(mia.address) + HAVING count(*) > 1 + ORDER BY host(mia.address) + LIMIT 20 + ) duplicates; + + IF duplicate_addresses IS NOT NULL THEN + RAISE EXCEPTION + 'machine_interface_addresses contains duplicate host addresses (% total; showing at most 20): %', + duplicate_address_count, + duplicate_addresses + USING HINT = 'Resolve each duplicate owner before retrying this migration.'; + END IF; +END +$$; + +-- Rust writes `IpAddr`, and the legacy admin-network procedure already uses +-- `host(address)::inet`, so normal rows are `/32` or `/128`. Normalize any +-- single-owner legacy rows before requiring that form from every future writer. +UPDATE machine_interface_addresses +SET address = host(address)::inet +WHERE address != host(address)::inet; + +-- This is a storage-representation invariant: the unique key and prefix-range +-- lookups both require host-form `inet` values even when rows arrive outside +-- the Rust model boundary. +ALTER TABLE machine_interface_addresses + ADD CONSTRAINT machine_interface_addresses_host_address_check + CHECK (address = host(address)::inet); + +-- The unique index created by this constraint covers the address lookups that +-- used the older non-unique index. +DROP INDEX IF EXISTS machine_interface_addresses_address_idx; + +ALTER TABLE machine_interface_addresses + ADD CONSTRAINT machine_interface_addresses_address_key UNIQUE (address); + +-- `update_admin_network` used to rewrite addresses one interface at a time. +-- An overlapping range can temporarily place one interface on an address that +-- a later interface still owns. Stage the complete mapping, remove the old +-- rows, and then insert the final addresses under the global constraint. +CREATE OR REPLACE PROCEDURE update_admin_network( + admin_network uuid, + new_range inet, + new_gw inet +) +LANGUAGE plpgsql +AS $$ +DECLARE + v_prefix_id uuid; + v_previous_prefix inet; + v_num_reserved integer; +BEGIN + -- A dual-stack segment selects its matching family. The legacy procedure + -- could also replace the only prefix with a different family, so keep that + -- behavior when there is exactly one unambiguous prefix. + SELECT candidate.id, candidate.prefix, candidate.num_reserved + INTO v_prefix_id, v_previous_prefix, v_num_reserved + FROM ( + SELECT + p.id, + p.prefix, + p.num_reserved, + count(*) OVER () AS prefix_count, + count(*) FILTER ( + WHERE family(p.prefix) = family(new_range) + ) OVER () AS matching_family_count + FROM network_segments s + JOIN network_prefixes p ON p.segment_id = s.id + WHERE s.network_segment_type = 'admin' + AND s.id = admin_network + ) candidate + WHERE ( + family(candidate.prefix) = family(new_range) + AND candidate.matching_family_count = 1 + ) OR ( + candidate.matching_family_count = 0 + AND candidate.prefix_count = 1 + ) + LIMIT 1; + + IF v_prefix_id IS NULL THEN + RAISE NOTICE 'Could not select one admin network prefix for the requested range'; + RETURN; + END IF; + + IF NOT masklen(new_range) <= masklen(v_previous_prefix) THEN + RAISE NOTICE 'New range is smaller than old range, aborting because it cannot be guaranteed to have enough space'; + RETURN; + END IF; + + -- Keep the legacy NULL gateway behavior, which IPv6 prefixes require. + IF new_gw IS NOT NULL AND NOT host(new_range + 1) = host(new_gw) THEN + RAISE NOTICE 'new range first IP is not the gateway, exiting'; + RETURN; + END IF; + + IF v_num_reserved = 0 THEN + RAISE NOTICE 'num_reserved is zero but we need to at least reserve the network and gateway, so making it 2'; + v_num_reserved := 2; + END IF; + + CREATE TEMP TABLE IF NOT EXISTS tmp_admin_network_address_mapping ( + address_id uuid PRIMARY KEY, + interface_id uuid NOT NULL, + address inet NOT NULL, + allocation_type text NOT NULL, + domain_id uuid, + hostname varchar(63) NOT NULL + ) ON COMMIT DROP; + TRUNCATE tmp_admin_network_address_mapping; + + INSERT INTO tmp_admin_network_address_mapping ( + address_id, + interface_id, + address, + allocation_type, + domain_id, + hostname + ) + SELECT + address_id, + interface_id, + address, + allocation_type, + domain_id, + CASE + -- The procedure cannot read the process-wide naming strategy. A + -- name matching the old address is IP-derived, so move it with the + -- address. `fun`, `serial_number`, `mac_address`, and operator-set + -- names stay as-is. + WHEN previous_hostname IN ( + previous_ip_hostname, + previous_legacy_ip_hostname + ) THEN ip_hostname + ELSE previous_hostname + END + FROM ( + SELECT + preferred_addresses.*, + -- Keep these equivalent to + -- `host_naming::address_to_hostname`: dotted IPv4 becomes dashed, + -- while IPv6 uses eight zero-padded groups. + CASE + WHEN family(previous_address) = 4 THEN + replace(host(previous_address), '.', '-') + ELSE + rtrim( + regexp_replace( + encode( + substring(inet_send(previous_address) FROM 5), + 'hex' + ), + '(.{4})', + '\1-', + 'g' + ), + '-' + ) + END AS previous_ip_hostname, + -- The older procedure only replaced dots. Accept that stored form + -- too, then write the canonical name on this remap. + replace( + host(previous_address), + '.', + '-' + ) AS previous_legacy_ip_hostname, + CASE + WHEN family(hostname_address) = 4 THEN + replace(host(hostname_address), '.', '-') + ELSE + rtrim( + regexp_replace( + encode( + substring(inet_send(hostname_address) FROM 5), + 'hex' + ), + '(.{4})', + '\1-', + 'g' + ), + '-' + ) + END AS ip_hostname + FROM ( + SELECT + mapped_addresses.*, + CASE + -- Rust naming prefers an interface's IPv4 address. Keep + -- that name when only its IPv6 half is being remapped. + WHEN family(address) = 6 THEN + COALESCE(surviving_ipv4_address, address) + ELSE address + END AS hostname_address + FROM ( + SELECT + mia.id AS address_id, + mia.interface_id, + mi.domain_id, + mia.address AS previous_address, + mi.hostname AS previous_hostname, + ( + SELECT existing.address + FROM machine_interface_addresses existing + WHERE existing.interface_id = mia.interface_id + AND existing.id != mia.id + AND family(existing.address) = 4 + -- Reuse only an address that survives this remap. + AND family(existing.address) != family(v_previous_prefix) + LIMIT 1 + ) AS surviving_ipv4_address, + host( + new_range + + (v_num_reserved + row_number() OVER (ORDER BY mi.id, mia.id) - 1) + )::inet AS address, + mia.allocation_type + FROM machine_interface_addresses mia + JOIN machine_interfaces mi ON mi.id = mia.interface_id + WHERE mi.segment_id = admin_network + AND family(mia.address) = family(v_previous_prefix) + ) mapped_addresses + ) preferred_addresses + ) named_addresses; + + DELETE FROM machine_interface_addresses mia + USING tmp_admin_network_address_mapping mapped + WHERE mia.id = mapped.address_id; + + INSERT INTO machine_interface_addresses ( + id, + interface_id, + address, + allocation_type + ) + SELECT + address_id, + interface_id, + address, + allocation_type + FROM tmp_admin_network_address_mapping + ORDER BY address; + + -- Remove affected rows from the active FQDN namespace while applying the + -- final names. This lets two interfaces swap IP-derived hostnames without + -- tripping the immediate `(domain_id, hostname)` constraint halfway + -- through the update. Restoring each saved domain below still rejects a + -- real conflict with an interface outside this remap. + UPDATE machine_interfaces mi + SET domain_id = NULL, + hostname = mapped.hostname + FROM tmp_admin_network_address_mapping mapped + WHERE mi.id = mapped.interface_id; + + UPDATE machine_interfaces mi + SET domain_id = mapped.domain_id + FROM tmp_admin_network_address_mapping mapped + WHERE mi.id = mapped.interface_id; + + UPDATE network_prefixes + SET gateway = new_gw, + prefix = new_range + WHERE id = v_prefix_id; +END +$$; diff --git a/crates/api-db/src/machine_interface.rs b/crates/api-db/src/machine_interface.rs index 6cdc7e5114..ac0081480b 100644 --- a/crates/api-db/src/machine_interface.rs +++ b/crates/api-db/src/machine_interface.rs @@ -52,9 +52,17 @@ const SQL_VIOLATION_DUPLICATE_MAC: &str = "machine_interfaces_segment_id_mac_add const SQL_VIOLATION_ONE_PRIMARY_INTERFACE: &str = "one_primary_interface_per_machine"; const SQL_VIOLATION_MAX_ONE_ASSOCIATION: &str = "chk_max_one_association"; const FAST_PATH_MAX_RETRIES: usize = 128; +// A static writer can bypass segment locks and win a candidate after a dynamic +// allocator reads it. Bound the new whole-allocation retries because each one +// rebuilds the candidate set instead of trying another small SQL batch. +const ALLOCATION_CONFLICT_MAX_RETRIES: usize = 8; const FAST_PATH_CANDIDATE_BATCH: i64 = 32; -pub struct UsedAdminNetworkIpResolver { +/// Resolve addresses that are unavailable to machine-interface allocators. +/// +/// Address ownership is global even though each allocator works within one +/// segment. This includes reservations parked on `static-assignments`. +pub struct UsedMachineInterfaceIpResolver { pub segment_id: NetworkSegmentId, // All the IPs which can not be allocated, e.g. SVI IP. pub busy_ips: Vec, @@ -1732,9 +1740,10 @@ async fn create_fast_path( // Another simultaneous create got the same FQDN, try again. false } - Err(DatabaseError::TryAgain) => { - // All the IP's in the batch we grabbed from the database got taken by other - // concurrent calls to create_fast_path. Try again. + Err(DatabaseError::TryAgain | DatabaseError::AddressAlreadyInUse(_)) => { + // Another dynamic attempt may hold a candidate lock, while + // a static writer only meets us at the database constraint. + // Either way, roll back this attempt and select again. false } Err(DatabaseError::ResourceExhausted(_)) if segments_idx < segments.len() - 1 => { @@ -1770,7 +1779,7 @@ async fn create_fast_path( } /// Create a machine interface with a specific static IP address. -/// A perfect compliment to create_fast_path and create_slow_path. +/// The fixed-address counterpart to `create_fast_path` and `create_slow_path`. /// /// If the target IP is already allocated to an interface with /// same MAC, just return the existing interface snapshot. @@ -1785,7 +1794,7 @@ async fn create_static_path( address: IpAddr, interface_type: InterfaceType, ) -> DatabaseResult { - // For the staic path, we need to be a little forgiving since + // For the static path, we need to be a little forgiving since // we expect to allow static assignment even if the requested // assignment is outside any network segment as long as // there is a "static assignment segment". @@ -1816,8 +1825,13 @@ async fn create_static_path( .into()); } - let interface_id = create_inner( - txn, + // The availability read is only a fast path. Keep both inserts in one + // savepoint so the database constraint can decide a race and roll the + // losing interface row back before callers handle the conflict. + let mut static_txn = Transaction::begin_inner(txn).await?; + + let interface_id = match create_inner( + static_txn.as_pgconn(), segment, macaddr, segment.config.subdomain_id, @@ -1826,7 +1840,15 @@ async fn create_static_path( AllocationType::Static, interface_type, ) - .await?; + .await + { + Ok(interface_id) => interface_id, + Err(error) => { + static_txn.rollback().await?; + return Err(error); + } + }; + static_txn.commit().await?; Ok( find_by(txn, ObjectColumnFilter::One(IdColumn, &interface_id)) @@ -1839,9 +1861,10 @@ async fn create_static_path( /// /// This uses [`crate::IpAllocator`], which requires: /// -/// - Locking the machine_interfaces_lock table -/// - Reading all used IP's from the database for the given segment -/// - Selecting a batch of IP's according to the selection strategy +/// - Locking the network segment against concurrent address allocation. +/// - Reading every machine-interface address covered by this segment's +/// prefixes, even when another segment owns the row. +/// - Selecting a batch of IPs according to the selection strategy. pub async fn create_slow_path( txn: &mut PgConnection, segment: &NetworkSegment, @@ -1850,36 +1873,73 @@ pub async fn create_slow_path( address_strategy: AddressSelectionStrategy, interface_type: InterfaceType, ) -> DatabaseResult { - // We're potentially about to insert a couple rows, so create a savepoint. - let mut inner_txn = Transaction::begin_inner(txn).await?; - - // If either requested addresses are auto-generated, we lock the entire table - // by way of the inner_txn. - lock_network_segment_exclusive(&mut inner_txn, segment).await?; - - // Collect SVI IPs so the allocator knows they're already reserved. - let mut reserved_ips = vec![]; - for prefix in &segment.prefixes { - if let Some(svi_ip) = prefix.svi_ip { - reserved_ips.push(svi_ip); + for _ in 0..ALLOCATION_CONFLICT_MAX_RETRIES { + let mut inner_txn = Transaction::begin_inner(txn).await?; + match try_create_slow_path( + &mut inner_txn, + segment, + macaddr, + primary_interface, + address_strategy, + interface_type, + ) + .await + { + Ok(interface_id) => { + inner_txn.commit().await?; + return Ok( + find_by(txn, ObjectColumnFilter::One(IdColumn, &interface_id)) + .await? + .remove(0), + ); + } + Err(DatabaseError::AddressAlreadyInUse(_)) => { + // A static writer claimed the candidate address. Roll back the + // complete prefix attempt before asking the allocator for a + // fresh set. + inner_txn.rollback().await?; + } + Err(error) => { + inner_txn.rollback().await?; + return Err(error); + } } } + Err(DatabaseError::internal(format!( + "unable to create machine interface in slow path on segment {} after {} retries", + segment.id, ALLOCATION_CONFLICT_MAX_RETRIES, + ))) +} + +/// Try one whole-prefix allocation inside the caller's savepoint. +/// +/// The caller retries this complete operation if a static writer claims one of +/// the selected addresses before insertion. +async fn try_create_slow_path( + txn: &mut PgTransaction<'_>, + segment: &NetworkSegment, + macaddr: &MacAddress, + primary_interface: bool, + address_strategy: AddressSelectionStrategy, + interface_type: InterfaceType, +) -> DatabaseResult { + lock_network_segment_exclusive(txn, segment).await?; + + let reserved_ips = segment + .prefixes + .iter() + .filter_map(|prefix| prefix.svi_ip) + .collect(); let dhcp_handler: Box + Send> = - Box::new(UsedAdminNetworkIpResolver { + Box::new(UsedMachineInterfaceIpResolver { segment_id: segment.id, busy_ips: reserved_ips, }); // Allocate an address from each prefix in the segment. For dual-stack // segments this means one IPv4 address and one IPv6 address. - let allocator = IpAllocator::new( - inner_txn.as_pgconn(), - segment, - dhcp_handler, - address_strategy, - ) - .await?; + let allocator = IpAllocator::new(txn.as_mut(), segment, dhcp_handler, address_strategy).await?; let mut allocated_addresses = Vec::new(); for (_, maybe_address) in allocator { @@ -1887,8 +1947,8 @@ pub async fn create_slow_path( allocated_addresses.push(address.ip()); } - let interface_id = create_inner( - inner_txn.as_pgconn(), + create_inner( + txn.as_mut(), segment, macaddr, segment.config.subdomain_id, @@ -1897,14 +1957,7 @@ pub async fn create_slow_path( AllocationType::Dhcp, interface_type, ) - .await?; - inner_txn.commit().await?; - - Ok( - find_by(txn, ObjectColumnFilter::One(IdColumn, &interface_id)) - .await? - .remove(0), - ) + .await } /// Fast path for single-IP allocation. @@ -1981,7 +2034,7 @@ async fn allocate_v6_addresses_via_ip_allocator( .collect(); let dhcp_handler: Box + Send> = - Box::new(UsedAdminNetworkIpResolver { + Box::new(UsedMachineInterfaceIpResolver { segment_id: segment.id, busy_ips: reserved_ips, }); @@ -2058,7 +2111,7 @@ async fn create_without_addresses( txn, &segment.id, macaddr, - hostname, + &hostname, segment.config.subdomain_id, primary_interface, interface_type, @@ -2121,20 +2174,28 @@ async fn create_inner( }; let hostname = host_naming::hostname_for(txn, &ctx).await?; + // Claim addresses before placing this hostname in the configured DNS + // domain. Existing-interface updates use that same order, which keeps the + // address and FQDN uniqueness constraints from waiting on each other in + // opposite directions. let interface_id = insert_machine_interface( txn, &segment.id, macaddr, - hostname, - domain_id, + &hostname, + None, primary_interface, interface_type, ) .await?; - for address in allocated_addresses { - insert_machine_interface_address(txn, &interface_id, address, allocation_type).await?; + let mut address_insert_order = allocated_addresses.to_vec(); + address_insert_order.sort_unstable(); + for address in address_insert_order { + crate::machine_interface_address::insert(txn, interface_id, address, allocation_type) + .await?; } + update_hostname_and_domain(txn, interface_id, &hostname, domain_id).await?; Ok(interface_id) } @@ -2214,8 +2275,9 @@ LIMIT $6; /// Attempts to acquire a transaction-scoped advisory lock for one IP candidate. /// -/// A successful lock means this transaction "owns" that candidate for the current attempt, which -/// avoids same-IP races across concurrent allocations. +/// A successful lock reserves the candidate against other dynamic allocators +/// for this transaction. Static writers do not take this lock, so the address +/// constraint remains the final ownership boundary. async fn try_lock_ip_candidate( // Note: Must be a transaction since we're doing locks txn: &mut PgTransaction<'_>, @@ -2293,7 +2355,7 @@ pub async fn allocate_svi_ip( segment: &NetworkSegment, ) -> DatabaseResult<(NetworkPrefixId, IpAddr)> { let dhcp_handler: Box + Send> = - Box::new(UsedAdminNetworkIpResolver { + Box::new(UsedMachineInterfaceIpResolver { segment_id: segment.id, busy_ips: vec![], }); @@ -2331,18 +2393,15 @@ pub async fn find_optional_for_update_by_ip( WHERE mia.address = $1::inet FOR UPDATE OF mia, mi "#; - let interface_ids: Vec<(MachineInterfaceId,)> = sqlx::query_as(query) + let interface_id: Option<(MachineInterfaceId,)> = sqlx::query_as(query) .bind(remote_ip) - .fetch_all(&mut *txn) + .fetch_optional(&mut *txn) .await .map_err(|e| DatabaseError::query(query, e))?; - match interface_ids.as_slice() { - [] => Ok(None), - [(interface_id,)] => find_one(txn, *interface_id).await.map(Some), - _ => Err(DatabaseError::internal(format!( - "multiple machine interfaces map to discovery IP {remote_ip}" - ))), + match interface_id { + Some((interface_id,)) => find_one(txn, interface_id).await.map(Some), + None => Ok(None), } } @@ -2402,7 +2461,7 @@ async fn insert_machine_interface( txn: &mut PgConnection, segment_id: &NetworkSegmentId, mac_address: &MacAddress, - hostname: String, + hostname: &str, domain_id: Option, is_primary_interface: bool, interface_type: InterfaceType, @@ -2436,28 +2495,6 @@ async fn insert_machine_interface( Ok(interface_id) } -/// insert_machine_interface_address inserts a new machine interface -/// address entry into the database. In the case of machine interfaces, -/// this explicitly takes an `IpAddr`, since machine interfaces are -/// always going to be a /32. It is up to the caller to ensure a possible -/// IpNetwork returned from the IpAllocator is of the correct size. -async fn insert_machine_interface_address( - txn: &mut PgConnection, - interface_id: &MachineInterfaceId, - address: &IpAddr, - allocation_type: model::allocation_type::AllocationType, -) -> DatabaseResult<()> { - let query = "INSERT INTO machine_interface_addresses (interface_id, address, allocation_type) VALUES ($1::uuid, $2::inet, $3)"; - sqlx::query(query) - .bind(interface_id) - .bind(address) - .bind(allocation_type) - .execute(txn) - .await - .map_err(|e| DatabaseError::query(query, e))?; - Ok(()) -} - async fn find_by<'a, C: ColumnInfo<'a, TableType = MachineInterfaceSnapshot>>( txn: impl DbReader<'_>, filter: ObjectColumnFilter<'a, C>, @@ -3451,65 +3488,81 @@ pub async fn allocate_address_for_family( segment: &NetworkSegment, family: carbide_network::ip::IpAddressFamily, ) -> DatabaseResult> { - let mut fast_txn = Transaction::begin_inner(txn).await?; - if family == IpAddressFamily::Ipv6 { - lock_network_segment_exclusive(&mut fast_txn, segment).await?; - } else { - lock_network_segment_shared(&mut fast_txn, segment).await?; - } - - let mut allocated_addresses = Vec::new(); - if family == IpAddressFamily::Ipv6 { - // Use a family-only segment view so lease recovery allocates exactly one - // address from each IPv6 prefix and does not disturb IPv4 ordering. - let ipv6_segment = NetworkSegment { - prefixes: segment - .prefixes - .iter() - .filter(|prefix| prefix.prefix.is_ipv6()) - .cloned() - .collect(), - ..segment.clone() - }; - allocated_addresses = - allocate_v6_addresses_via_ip_allocator(&mut fast_txn, &ipv6_segment).await?; - for address in &allocated_addresses { - insert_machine_interface_address( - fast_txn.as_pgconn(), - &interface_id, - address, - AllocationType::Dhcp, - ) - .await?; - } - } else { - for prefix in segment + let family_segment = NetworkSegment { + prefixes: segment .prefixes .iter() - .filter(|p| p.prefix.is_address_family(family)) - { - let address = allocate_next_ip_with_retry(&mut fast_txn, segment, prefix).await?; - allocated_addresses.push(address); - insert_machine_interface_address( - fast_txn.as_pgconn(), - &interface_id, - &address, - AllocationType::Dhcp, - ) - .await?; - } + .filter(|prefix| prefix.prefix.is_address_family(family)) + .cloned() + .collect(), + ..segment.clone() + }; + if family_segment.prefixes.is_empty() { + return Ok(Vec::new()); } - fast_txn.commit().await?; + for _ in 0..ALLOCATION_CONFLICT_MAX_RETRIES { + let mut fast_txn = Transaction::begin_inner(txn).await?; + if family == IpAddressFamily::Ipv6 { + lock_network_segment_exclusive(&mut fast_txn, segment).await?; + } else { + lock_network_segment_shared(&mut fast_txn, segment).await?; + } - // Nothing allocated (no prefix for the requested family): leave the - // hostname and domain exactly as they were. - if allocated_addresses.is_empty() { - return Ok(allocated_addresses); + let attempt = + try_allocate_addresses_for_family(&mut fast_txn, interface_id, &family_segment).await; + + match attempt { + Ok(allocated_addresses) => { + fast_txn.commit().await?; + sync_hostname_after_address_assignment( + txn, + interface_id, + segment.config.subdomain_id, + ) + .await?; + return Ok(allocated_addresses); + } + Err(DatabaseError::AddressAlreadyInUse(_)) => { + // A concurrent allocator or static writer took the selected + // address. The savepoint also removes any earlier address rows + // from this attempt before we select again. + fast_txn.rollback().await?; + } + Err(error) => { + fast_txn.rollback().await?; + return Err(error); + } + } } - sync_hostname_after_address_assignment(txn, interface_id, segment.config.subdomain_id).await?; + Err(DatabaseError::internal(format!( + "unable to allocate a {family:?} address for interface {interface_id} on segment {} after {} retries", + segment.id, ALLOCATION_CONFLICT_MAX_RETRIES, + ))) +} +/// Try one address-family allocation inside the caller's savepoint. +/// +/// The caller rolls the savepoint back and retries if one of these candidates +/// is claimed before its address row is inserted. +async fn try_allocate_addresses_for_family( + txn: &mut PgTransaction<'_>, + interface_id: MachineInterfaceId, + family_segment: &NetworkSegment, +) -> DatabaseResult> { + let allocated_addresses = allocate_addresses_from_segment(txn, family_segment).await?; + let mut address_insert_order = allocated_addresses.clone(); + address_insert_order.sort_unstable(); + for address in address_insert_order { + crate::machine_interface_address::insert( + txn.as_mut(), + interface_id, + address, + AllocationType::Dhcp, + ) + .await?; + } Ok(allocated_addresses) } @@ -3604,25 +3657,15 @@ pub async fn find_ids_by_power_shelf_id( } #[async_trait::async_trait] -impl UsedIpResolver for UsedAdminNetworkIpResolver +impl UsedIpResolver for UsedMachineInterfaceIpResolver where for<'db> &'db mut DB: DbReader<'db>, { - // DEPRECATED - // With the introduction of `used_prefixes()` this is no - // longer an accurate approach for finding all allocated - // IPs in a segment, since used_ips() completely ignores - // the fact wider prefixes may have been allocated, even - // though in the case of machine interfaces, its probably - // always going to just be a /32. - // - // used_ips returns the used (or allocated) IPs for machine - // interfaces in a given network segment. - // - // More specifically, this is intended to specifically - // target the `address` column of the `machine_interface_addresses` - // table, in which a single /32 is stored (although, as an - // `inet`, it could techincally also have a prefix length). + // `used_ips` returns every machine-interface IP covered by this allocator's + // prefixes, regardless of which segment owns it. That keeps a reservation + // on `static-assignments` from being selected without loading unrelated + // addresses from the rest of the site. The migration's host-address check + // keeps every `inet` row inside these inclusive prefix bounds. async fn used_ips(&self, txn: &mut DB) -> Result, DatabaseError> { // IpAddrContainer is a small private struct used // for binding the result of the subsequent SQL @@ -3634,10 +3677,12 @@ where } let query = " -SELECT address FROM machine_interface_addresses -INNER JOIN machine_interfaces ON machine_interfaces.id = machine_interface_addresses.interface_id -INNER JOIN network_segments ON machine_interfaces.segment_id = network_segments.id -WHERE network_segments.id = $1::uuid"; +SELECT DISTINCT mia.address +FROM network_prefixes np +JOIN machine_interface_addresses mia + ON mia.address >= host(network(np.prefix::inet))::inet + AND mia.address <= host(broadcast(np.prefix::inet))::inet +WHERE np.segment_id = $1"; let containers: Vec = sqlx::query_as(query) .bind(self.segment_id) @@ -3653,11 +3698,10 @@ WHERE network_segments.id = $1::uuid"; // used_prefixes returns the used (or allocated) prefixes // for machine interfaces in a given network segment. // - // NOTE(Chet): This is kind of a hack! Machine interfaces - // aren't allocated prefixes other than a /32, and I think - // it might be confusing if we added a `prefix` column to the - // machine_interface_addresses table (since it's always - // just going to be a /32 anyway). + // NOTE(Chet): This is kind of a hack! Machine interfaces store host + // addresses (`/32` or `/128`) rather than allocated prefix rows, and I + // think it might be confusing if we added another `prefix` column to + // machine_interface_addresses. // // So, instead of database schema changes, this just gets all // of the used IPs and turns them into IpNetworks. diff --git a/crates/api-db/src/machine_interface/tests.rs b/crates/api-db/src/machine_interface/tests.rs index 64d2c3b867..7772630cab 100644 --- a/crates/api-db/src/machine_interface/tests.rs +++ b/crates/api-db/src/machine_interface/tests.rs @@ -15,6 +15,7 @@ * limitations under the License. */ +use carbide_uuid::domain::DomainId; use carbide_uuid::machine::{MachineId, MachineIdSource, MachineType}; use carbide_uuid::network::NetworkSegmentId; use model::allocation_type::AllocationType; @@ -97,9 +98,32 @@ async fn create_managed_segment( prefix: &str, segment_type: NetworkSegmentType, allocation_strategy: AllocationStrategy, +) -> Result> { + create_managed_segment_with_prefixes(pool, name, &[prefix], segment_type, allocation_strategy) + .await +} + +/// Create a managed segment with one or more address-family prefixes. +async fn create_managed_segment_with_prefixes( + pool: &sqlx::PgPool, + name: &str, + prefixes: &[&str], + segment_type: NetworkSegmentType, + allocation_strategy: AllocationStrategy, ) -> Result> { let segment_id = NetworkSegmentId::new(); let mut txn = db::Transaction::begin(pool).await?; + let prefixes = prefixes + .iter() + .map(|prefix| { + Ok(NewNetworkPrefix { + prefix: prefix.parse()?, + gateway: None, + dhcpv6_link_address: None, + num_reserved: 0, + }) + }) + .collect::, Box>>()?; db::network_segment::persist( NewNetworkSegment { id: segment_id, @@ -107,12 +131,7 @@ async fn create_managed_segment( subdomain_id: None, vpc_id: None, mtu: 1500, - prefixes: vec![NewNetworkPrefix { - prefix: prefix.parse()?, - gateway: None, - dhcpv6_link_address: None, - num_reserved: 0, - }], + prefixes, vlan_id: None, vni: None, segment_type, @@ -128,6 +147,1149 @@ async fn create_managed_segment( Ok(segment_id) } +/// Load a test segment after its setup transaction commits. +async fn load_test_segment( + pool: &sqlx::PgPool, + name: &str, +) -> Result> { + let mut txn = db::Transaction::begin(pool).await?; + let segment = db::network_segment::find_by_name(txn.as_pgconn(), name).await?; + txn.rollback().await?; + Ok(segment) +} + +/// Create an addressless interface whose hostname cannot collide with the +/// IP-derived name used by a competing allocation. +async fn create_addressless_race_interface( + pool: &sqlx::PgPool, + segment_id: NetworkSegmentId, + mac_address: MacAddress, + hostname: &str, +) -> Result> { + let mut txn = db::Transaction::begin(pool).await?; + let interface_id = sqlx::query_scalar( + "INSERT INTO machine_interfaces + (segment_id, mac_address, primary_interface, hostname) + VALUES ($1, $2, false, $3) + RETURNING id", + ) + .bind(segment_id) + .bind(mac_address) + .bind(hostname) + .fetch_one(txn.as_pgconn()) + .await?; + txn.commit().await?; + Ok(interface_id) +} + +/// Wait until `blocked_pid` is actually waiting on `blocker_pid`. +/// +/// This proves the competing write reached the PostgreSQL uniqueness boundary +/// under test before the winner commits. +async fn wait_for_transaction_conflict( + pool: &sqlx::PgPool, + blocker_pid: i32, + blocked_pid: i32, +) -> Result<(), Box> { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let blocked: bool = sqlx::query_scalar("SELECT $1 = ANY(pg_blocking_pids($2))") + .bind(blocker_pid) + .bind(blocked_pid) + .fetch_one(pool) + .await?; + if blocked { + return Ok::<(), sqlx::Error>(()); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .map_err(|_| std::io::Error::other("transaction did not reach the expected database wait"))??; + Ok(()) +} + +/// Run one fixed/fixed race and verify the committed owner wins cleanly. +/// +/// The losing create commits its outer transaction after handling the typed +/// conflict. That proves the create path's savepoint removed its partial +/// interface row without making the caller discard unrelated work. +#[allow(txn_held_across_await)] // The uncommitted winner is the concurrency boundary under test. +async fn run_fixed_address_race( + pool: &sqlx::PgPool, + segment: NetworkSegment, + address: IpAddr, + owner_mac: MacAddress, + contender_mac: MacAddress, +) -> Result<(), Box> { + let segment_id = segment.id; + let owner_id = create_addressless_race_interface( + pool, + segment_id, + owner_mac, + &format!("fixed-owner-{}", owner_mac.to_string().replace(':', "-")), + ) + .await?; + + let mut owner_txn = db::Transaction::begin(pool).await?; + let owner_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(owner_txn.as_pgconn()) + .await?; + crate::machine_interface_address::assign_static(owner_txn.as_pgconn(), owner_id, address) + .await?; + + let (contender_pid_tx, contender_pid_rx) = tokio::sync::oneshot::channel(); + let contender_pool = pool.clone(); + let mut contender_task = tokio::spawn(async move { + let mut txn = db::Transaction::begin(&contender_pool) + .await + .map_err(|error| error.to_string())?; + let contender_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(txn.as_pgconn()) + .await + .map_err(|error| error.to_string())?; + contender_pid_tx + .send(contender_pid) + .map_err(|_| "could not report contender backend PID".to_string())?; + + let result = create( + txn.as_pgconn(), + std::slice::from_ref(&segment), + &contender_mac, + false, + AddressSelectionStrategy::StaticAddress(address), + None, + ) + .await; + match result { + Ok(snapshot) => { + txn.commit().await.map_err(|error| error.to_string())?; + Ok::<_, String>(Ok(snapshot)) + } + Err(error @ DatabaseError::AddressAlreadyInUse(_)) => { + txn.commit().await.map_err(|error| error.to_string())?; + Ok(Err(error)) + } + Err(error) => { + txn.rollback().await.map_err(|error| error.to_string())?; + Ok(Err(error)) + } + } + }); + + let contender_pid = contender_pid_rx.await?; + if let Err(error) = wait_for_transaction_conflict(pool, owner_pid, contender_pid).await { + contender_task.abort(); + let _ = contender_task.await; + owner_txn.rollback().await?; + return Err(error); + } + owner_txn.commit().await?; + + let contender_result = + match tokio::time::timeout(std::time::Duration::from_secs(5), &mut contender_task).await { + Ok(result) => result + .map_err(|error| std::io::Error::other(error.to_string()))? + .map_err(std::io::Error::other)?, + Err(_) => { + contender_task.abort(); + let _ = contender_task.await; + return Err(std::io::Error::other("fixed-address contender did not finish").into()); + } + }; + + match contender_result { + Err(DatabaseError::AddressAlreadyInUse(AddressAlreadyInUseError( + conflict_address, + conflict_mac, + conflict_segment_id, + conflict_interface_id, + ))) => { + assert_eq!(conflict_address, address); + assert_eq!(conflict_mac, owner_mac); + assert_eq!(conflict_segment_id, segment_id); + assert_eq!(conflict_interface_id, owner_id); + } + other => panic!("expected the fixed owner to win, got {other:?}"), + } + + let owner_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM machine_interface_addresses WHERE address = $1::inet", + ) + .bind(address) + .fetch_one(pool) + .await?; + assert_eq!(owner_count, 1); + let contender_count: i64 = + sqlx::query_scalar("SELECT count(*) FROM machine_interfaces WHERE mac_address = $1") + .bind(contender_mac) + .fetch_one(pool) + .await?; + assert_eq!( + contender_count, 0, + "the failed fixed-address create must roll back its interface row" + ); + Ok(()) +} + +/// Run one fixed/first-DHCP race and verify DHCP retries another address. +#[allow(txn_held_across_await)] // The uncommitted winner is the concurrency boundary under test. +async fn run_fixed_dynamic_create_race( + pool: &sqlx::PgPool, + segment: NetworkSegment, + owner_segment: NetworkSegment, + fixed_address: IpAddr, + address_strategy: AddressSelectionStrategy, + owner_mac: MacAddress, + dhcp_mac: MacAddress, +) -> Result> { + let mut owner_txn = db::Transaction::begin(pool).await?; + let owner_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(owner_txn.as_pgconn()) + .await?; + create( + owner_txn.as_pgconn(), + std::slice::from_ref(&owner_segment), + &owner_mac, + false, + AddressSelectionStrategy::StaticAddress(fixed_address), + None, + ) + .await?; + + let (dhcp_pid_tx, dhcp_pid_rx) = tokio::sync::oneshot::channel(); + let dhcp_pool = pool.clone(); + let mut dhcp_task = tokio::spawn(async move { + let mut txn = db::Transaction::begin(&dhcp_pool) + .await + .map_err(|error| error.to_string())?; + let dhcp_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(txn.as_pgconn()) + .await + .map_err(|error| error.to_string())?; + dhcp_pid_tx + .send(dhcp_pid) + .map_err(|_| "could not report DHCP backend PID".to_string())?; + + let result = create( + txn.as_pgconn(), + std::slice::from_ref(&segment), + &dhcp_mac, + false, + address_strategy, + None, + ) + .await; + match result { + Ok(snapshot) => { + txn.commit().await.map_err(|error| error.to_string())?; + Ok::<_, String>(snapshot) + } + Err(error) => { + txn.rollback() + .await + .map_err(|rollback| rollback.to_string())?; + Err(error.to_string()) + } + } + }); + + let dhcp_pid = dhcp_pid_rx.await?; + if let Err(error) = wait_for_transaction_conflict(pool, owner_pid, dhcp_pid).await { + dhcp_task.abort(); + let _ = dhcp_task.await; + owner_txn.rollback().await?; + return Err(error); + } + owner_txn.commit().await?; + + let dhcp_interface = + match tokio::time::timeout(std::time::Duration::from_secs(5), &mut dhcp_task).await { + Ok(result) => result + .map_err(|error| std::io::Error::other(error.to_string()))? + .map_err(std::io::Error::other)?, + Err(_) => { + dhcp_task.abort(); + let _ = dhcp_task.await; + return Err(std::io::Error::other("DHCP allocation did not retry").into()); + } + }; + assert!( + !dhcp_interface.addresses.contains(&fixed_address), + "DHCP must retry instead of sharing {fixed_address}", + ); + assert!( + dhcp_interface + .addresses + .iter() + .all(|address| address.is_address_family(fixed_address.address_family())), + ); + Ok(dhcp_interface) +} + +/// Run one fixed/DHCP-recovery race and verify family reallocation retries. +#[allow(txn_held_across_await)] // The uncommitted winner is the concurrency boundary under test. +async fn run_fixed_dynamic_reallocation_race( + pool: &sqlx::PgPool, + segment: NetworkSegment, + owner_segment_id: NetworkSegmentId, + fixed_address: IpAddr, + family: IpAddressFamily, + owner_mac: MacAddress, + dhcp_mac: MacAddress, +) -> Result<(), Box> { + let owner_id = create_addressless_race_interface( + pool, + owner_segment_id, + owner_mac, + &format!( + "reallocation-owner-{}", + owner_mac.to_string().replace(':', "-") + ), + ) + .await?; + let dhcp_interface_id = create_addressless_race_interface( + pool, + segment.id, + dhcp_mac, + &format!( + "reallocation-contender-{}", + dhcp_mac.to_string().replace(':', "-") + ), + ) + .await?; + + let mut owner_txn = db::Transaction::begin(pool).await?; + let owner_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(owner_txn.as_pgconn()) + .await?; + crate::machine_interface_address::assign_static(owner_txn.as_pgconn(), owner_id, fixed_address) + .await?; + + let (dhcp_pid_tx, dhcp_pid_rx) = tokio::sync::oneshot::channel(); + let dhcp_pool = pool.clone(); + let mut dhcp_task = tokio::spawn(async move { + let mut txn = db::Transaction::begin(&dhcp_pool) + .await + .map_err(|error| error.to_string())?; + let dhcp_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(txn.as_pgconn()) + .await + .map_err(|error| error.to_string())?; + dhcp_pid_tx + .send(dhcp_pid) + .map_err(|_| "could not report DHCP backend PID".to_string())?; + + let result = + allocate_address_for_family(txn.as_pgconn(), dhcp_interface_id, &segment, family).await; + match result { + Ok(addresses) => { + txn.commit().await.map_err(|error| error.to_string())?; + Ok::<_, String>(addresses) + } + Err(error) => { + txn.rollback() + .await + .map_err(|rollback| rollback.to_string())?; + Err(error.to_string()) + } + } + }); + + let dhcp_pid = dhcp_pid_rx.await?; + if let Err(error) = wait_for_transaction_conflict(pool, owner_pid, dhcp_pid).await { + dhcp_task.abort(); + let _ = dhcp_task.await; + owner_txn.rollback().await?; + return Err(error); + } + owner_txn.commit().await?; + + let addresses = + match tokio::time::timeout(std::time::Duration::from_secs(5), &mut dhcp_task).await { + Ok(result) => result + .map_err(|error| std::io::Error::other(error.to_string()))? + .map_err(std::io::Error::other)?, + Err(_) => { + dhcp_task.abort(); + let _ = dhcp_task.await; + return Err(std::io::Error::other("DHCP reallocation did not retry").into()); + } + }; + assert!( + !addresses.is_empty(), + "DHCP reallocation must return a {family:?} address", + ); + assert!(!addresses.contains(&fixed_address)); + assert!( + addresses + .iter() + .all(|address| address.is_address_family(family)), + ); + Ok(()) +} + +/// Fixed requests use the same database ownership boundary for managed and +/// external addresses, in both families. +#[crate::sqlx_test] +async fn concurrent_fixed_requests_keep_one_owner( + pool: sqlx::PgPool, +) -> Result<(), Box> { + create_static_assignments_segment(&pool).await?; + create_managed_segment( + &pool, + "fixed-race-managed-ipv4", + "192.0.2.0/29", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + create_managed_segment( + &pool, + "fixed-race-managed-ipv6", + "2001:db8:100::/125", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + + let managed_ipv4 = load_test_segment(&pool, "fixed-race-managed-ipv4").await?; + let managed_ipv6 = load_test_segment(&pool, "fixed-race-managed-ipv6").await?; + let static_assignments = + load_test_segment(&pool, db::network_segment::STATIC_ASSIGNMENTS_SEGMENT_NAME).await?; + let cases = [ + ( + managed_ipv4, + "192.0.2.3".parse::()?, + "02:00:00:10:01:01".parse::()?, + "02:00:00:10:01:02".parse::()?, + ), + ( + managed_ipv6, + "2001:db8:100::3".parse::()?, + "02:00:00:10:02:01".parse::()?, + "02:00:00:10:02:02".parse::()?, + ), + ( + static_assignments.clone(), + "203.0.113.30".parse::()?, + "02:00:00:10:03:01".parse::()?, + "02:00:00:10:03:02".parse::()?, + ), + ( + static_assignments, + "2001:db8:ffff::30".parse::()?, + "02:00:00:10:04:01".parse::()?, + "02:00:00:10:04:02".parse::()?, + ), + ]; + + for (segment, address, owner_mac, contender_mac) in cases { + run_fixed_address_race(&pool, segment, address, owner_mac, contender_mac).await?; + } + + Ok(()) +} + +/// A first DHCP allocation retries when a fixed writer on another segment +/// commits the candidate it selected, for IPv4 and IPv6. +#[crate::sqlx_test] +async fn initial_dhcp_retries_a_concurrent_fixed_address( + pool: sqlx::PgPool, +) -> Result<(), Box> { + create_static_assignments_segment(&pool).await?; + create_managed_segment( + &pool, + "dynamic-race-ipv4", + "198.51.100.0/29", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let fixed_owner_segment = + load_test_segment(&pool, db::network_segment::STATIC_ASSIGNMENTS_SEGMENT_NAME).await?; + create_managed_segment( + &pool, + "dynamic-race-ipv6", + "2001:db8:200::/125", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + + let cases = [ + ( + load_test_segment(&pool, "dynamic-race-ipv4").await?, + "198.51.100.2".parse::()?, + "02:00:00:20:01:01".parse::()?, + "02:00:00:20:01:02".parse::()?, + ), + ( + load_test_segment(&pool, "dynamic-race-ipv6").await?, + "2001:db8:200::1".parse::()?, + "02:00:00:20:02:01".parse::()?, + "02:00:00:20:02:02".parse::()?, + ), + ]; + + for (segment, fixed_address, owner_mac, dhcp_mac) in cases { + run_fixed_dynamic_create_race( + &pool, + segment, + fixed_owner_segment.clone(), + fixed_address, + AddressSelectionStrategy::NextAvailableIp, + owner_mac, + dhcp_mac, + ) + .await?; + } + + Ok(()) +} + +/// Whole-prefix allocation rolls its failed interface row back before trying +/// the next prefix after a fixed writer wins the first candidate. +#[crate::sqlx_test] +async fn prefix_allocation_retries_a_concurrent_fixed_address( + pool: sqlx::PgPool, +) -> Result<(), Box> { + create_static_assignments_segment(&pool).await?; + create_managed_segment( + &pool, + "prefix-allocation-race", + "192.0.2.0/28", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + + let fixed_owner_segment = + load_test_segment(&pool, db::network_segment::STATIC_ASSIGNMENTS_SEGMENT_NAME).await?; + let segment = load_test_segment(&pool, "prefix-allocation-race").await?; + let interface = run_fixed_dynamic_create_race( + &pool, + segment, + fixed_owner_segment, + "192.0.2.4".parse()?, + AddressSelectionStrategy::NextAvailablePrefix(30), + "02:00:00:20:03:01".parse()?, + "02:00:00:20:03:02".parse()?, + ) + .await?; + + assert_eq!(interface.addresses, vec!["192.0.2.8".parse::()?]); + Ok(()) +} + +/// New interface creation claims its address before entering the configured +/// DNS domain. Pause that address insert so an existing interface can claim +/// both values without the two transactions waiting on each other. +#[crate::sqlx_test] +#[allow(txn_held_across_await)] // The gate transaction creates the exact ordering under test. +async fn new_interface_defers_fqdn_until_its_address_is_owned( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let segment_id = create_managed_segment( + &pool, + "address-fqdn-order", + "192.0.2.0/29", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let contender_mac: MacAddress = "02:00:00:20:03:11".parse()?; + let owner_mac: MacAddress = "02:00:00:20:03:12".parse()?; + let contender_id = create_addressless_race_interface( + &pool, + segment_id, + contender_mac, + "address-fqdn-contender", + ) + .await?; + + let mut setup_txn = db::Transaction::begin(&pool).await?; + let domain = db::dns::domain::persist( + model::dns::NewDomain::new("address-order.example"), + setup_txn.as_pgconn(), + ) + .await?; + sqlx::query("UPDATE network_segments SET subdomain_id = $1 WHERE id = $2") + .bind(domain.id) + .bind(segment_id) + .execute(setup_txn.as_pgconn()) + .await?; + sqlx::query("UPDATE machine_interfaces SET domain_id = $1 WHERE id = $2") + .bind(domain.id) + .bind(contender_id) + .execute(setup_txn.as_pgconn()) + .await?; + // MacAddress formatting is constrained to a normalized MAC literal, so it + // is safe to reuse owner_mac in this test-only trigger definition. + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION test_block_fixed_address_insert() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + IF EXISTS ( + SELECT 1 + FROM machine_interfaces + WHERE id = NEW.interface_id + AND mac_address = '{owner_mac}'::macaddr + ) THEN + PERFORM pg_advisory_xact_lock( + hashtextextended(current_database() || ':address-fqdn-order', 0) + ); + END IF; + RETURN NEW; + END + $$; + CREATE TRIGGER test_block_fixed_address_insert + BEFORE INSERT ON machine_interface_addresses + FOR EACH ROW + EXECUTE FUNCTION test_block_fixed_address_insert();" + ))) + .execute(setup_txn.as_pgconn()) + .await?; + setup_txn.commit().await?; + let segment = load_test_segment(&pool, "address-fqdn-order").await?; + let fixed_address: IpAddr = "192.0.2.2".parse()?; + + let mut gate_txn = db::Transaction::begin(&pool).await?; + let gate_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(gate_txn.as_pgconn()) + .await?; + sqlx::query( + "SELECT pg_advisory_xact_lock( + hashtextextended(current_database() || ':address-fqdn-order', 0) + )", + ) + .execute(gate_txn.as_pgconn()) + .await?; + + let (owner_pid_tx, owner_pid_rx) = tokio::sync::oneshot::channel(); + let owner_pool = pool.clone(); + let owner_segment = segment.clone(); + let mut owner_task = tokio::spawn(async move { + let mut txn = db::Transaction::begin(&owner_pool) + .await + .map_err(|error| error.to_string())?; + let owner_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(txn.as_pgconn()) + .await + .map_err(|error| error.to_string())?; + owner_pid_tx + .send(owner_pid) + .map_err(|_| "could not report fixed-owner backend PID".to_string())?; + + let result = create( + txn.as_pgconn(), + std::slice::from_ref(&owner_segment), + &owner_mac, + false, + AddressSelectionStrategy::StaticAddress(fixed_address), + None, + ) + .await; + match result { + Ok(snapshot) => { + txn.commit().await.map_err(|error| error.to_string())?; + Ok::<_, String>(Ok(snapshot)) + } + Err(error) => { + txn.rollback().await.map_err(|error| error.to_string())?; + Ok(Err(error)) + } + } + }); + + let owner_pid = owner_pid_rx.await?; + if let Err(error) = wait_for_transaction_conflict(&pool, gate_pid, owner_pid).await { + gate_txn.rollback().await?; + owner_task.abort(); + let _ = owner_task.await; + return Err(error); + } + + let mut contender_txn = db::Transaction::begin(&pool).await?; + let allocated = match tokio::time::timeout( + std::time::Duration::from_secs(5), + allocate_address_for_family( + contender_txn.as_pgconn(), + contender_id, + &segment, + IpAddressFamily::Ipv4, + ), + ) + .await + { + Ok(Ok(allocated)) => allocated, + Ok(Err(error)) => { + contender_txn.rollback().await?; + gate_txn.commit().await?; + owner_task.abort(); + let _ = owner_task.await; + return Err(error.into()); + } + Err(_) => { + contender_txn.rollback().await?; + gate_txn.commit().await?; + owner_task.abort(); + let _ = owner_task.await; + return Err(std::io::Error::other( + "address recovery blocked on the fixed owner's uncommitted FQDN", + ) + .into()); + } + }; + contender_txn.commit().await?; + assert_eq!(allocated, vec![fixed_address]); + + let (contender_hostname, contender_domain): (String, Option) = + sqlx::query_as("SELECT hostname, domain_id FROM machine_interfaces WHERE id = $1") + .bind(contender_id) + .fetch_one(&pool) + .await?; + assert_eq!(contender_hostname, "192-0-2-2"); + assert_eq!(contender_domain, Some(domain.id)); + + gate_txn.commit().await?; + let owner_result = + match tokio::time::timeout(std::time::Duration::from_secs(5), &mut owner_task).await { + Ok(result) => result + .map_err(|error| std::io::Error::other(error.to_string()))? + .map_err(std::io::Error::other)?, + Err(_) => { + owner_task.abort(); + let _ = owner_task.await; + return Err(std::io::Error::other("fixed owner did not finish").into()); + } + }; + assert!(matches!( + owner_result, + Err(DatabaseError::AddressAlreadyInUse(AddressAlreadyInUseError( + address, + mac_address, + conflict_segment_id, + conflict_interface_id, + ))) if address == fixed_address + && mac_address == contender_mac + && conflict_segment_id == segment_id + && conflict_interface_id == contender_id + )); + + let owner_rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM machine_interfaces WHERE mac_address = $1") + .bind(owner_mac) + .fetch_one(&pool) + .await?; + assert_eq!(owner_rows, 0); + Ok(()) +} + +/// Reapplying an admin range can permute existing addresses. The stored +/// procedure stages the complete mapping so the global constraint never sees +/// a temporary duplicate, updates IP-derived hostnames, and leaves names from +/// the other host naming strategies alone. +#[crate::sqlx_test] +async fn update_admin_network_replaces_overlapping_addresses( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let segment_id = create_managed_segment( + &pool, + "overlapping-admin-update", + "198.18.0.0/29", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let mut interface_ids = [ + create_addressless_race_interface( + &pool, + segment_id, + "02:00:00:20:04:01".parse()?, + "overlapping-admin-one", + ) + .await?, + create_addressless_race_interface( + &pool, + segment_id, + "02:00:00:20:04:02".parse()?, + "overlapping-admin-two", + ) + .await?, + ]; + interface_ids.sort_unstable(); + + let mut txn = db::Transaction::begin(&pool).await?; + let domain = db::dns::domain::persist( + model::dns::NewDomain::new("overlapping-admin.example"), + txn.as_pgconn(), + ) + .await?; + sqlx::query("UPDATE network_segments SET subdomain_id = $1 WHERE id = $2") + .bind(domain.id) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + for (interface_id, hostname) in [ + (interface_ids[0], "198-18-0-3"), + (interface_ids[1], "wholesale-walrus"), + ] { + sqlx::query( + "UPDATE machine_interfaces + SET domain_id = $1, hostname = $2 + WHERE id = $3", + ) + .bind(domain.id) + .bind(hostname) + .bind(interface_id) + .execute(txn.as_pgconn()) + .await?; + } + crate::machine_interface_address::insert( + txn.as_pgconn(), + interface_ids[0], + "198.18.0.3".parse()?, + AllocationType::Static, + ) + .await?; + crate::machine_interface_address::insert( + txn.as_pgconn(), + interface_ids[1], + "198.18.0.2".parse()?, + AllocationType::Static, + ) + .await?; + let address_ids: Vec = sqlx::query_scalar( + "SELECT mia.id + FROM machine_interfaces mi + JOIN machine_interface_addresses mia ON mia.interface_id = mi.id + WHERE mi.segment_id = $1 + ORDER BY mi.id", + ) + .bind(segment_id) + .fetch_all(txn.as_pgconn()) + .await?; + + sqlx::query( + "CALL update_admin_network( + $1, + '198.18.0.0/29'::inet, + '198.18.0.1'::inet + )", + ) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + + let rows: Vec<( + uuid::Uuid, + MachineInterfaceId, + IpAddr, + AllocationType, + String, + )> = sqlx::query_as( + "SELECT mia.id, mi.id, mia.address, mia.allocation_type, mi.hostname + FROM machine_interfaces mi + JOIN machine_interface_addresses mia ON mia.interface_id = mi.id + WHERE mi.segment_id = $1 + ORDER BY mi.id", + ) + .bind(segment_id) + .fetch_all(txn.as_pgconn()) + .await?; + + assert_eq!( + rows, + vec![ + ( + address_ids[0], + interface_ids[0], + "198.18.0.2".parse()?, + AllocationType::Static, + "198-18-0-2".to_string(), + ), + ( + address_ids[1], + interface_ids[1], + "198.18.0.3".parse()?, + AllocationType::Static, + "wholesale-walrus".to_string(), + ), + ] + ); + let preserved_domains: i64 = sqlx::query_scalar( + "SELECT count(*) + FROM machine_interfaces + WHERE segment_id = $1 AND domain_id = $2", + ) + .bind(segment_id) + .bind(domain.id) + .fetch_one(txn.as_pgconn()) + .await?; + assert_eq!(preserved_domains, 2); + Ok(()) +} + +/// A single-prefix admin segment keeps the legacy ability to change address +/// families while the procedure replaces its stored addresses. The first +/// remap also recognizes and normalizes the older compressed IPv6 hostname. +#[crate::sqlx_test] +async fn update_admin_network_replaces_a_single_prefix_with_another_family( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let segment_id = create_managed_segment( + &pool, + "cross-family-admin-update", + "2001:db8:400::/125", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let interface_id = create_addressless_race_interface( + &pool, + segment_id, + "02:00:00:20:05:01".parse()?, + "2001:db8:400::2", + ) + .await?; + + let mut txn = db::Transaction::begin(&pool).await?; + crate::machine_interface_address::insert( + txn.as_pgconn(), + interface_id, + "2001:db8:400::2".parse()?, + AllocationType::Static, + ) + .await?; + sqlx::query( + "CALL update_admin_network( + $1, + '2001:db8:401::/125'::inet, + NULL::inet + )", + ) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + let (ipv6_address, ipv6_hostname): (IpAddr, String) = sqlx::query_as( + "SELECT mia.address, mi.hostname + FROM machine_interfaces mi + JOIN machine_interface_addresses mia ON mia.interface_id = mi.id + WHERE mi.id = $1", + ) + .bind(interface_id) + .fetch_one(txn.as_pgconn()) + .await?; + assert_eq!(ipv6_address, "2001:db8:401::2".parse::()?); + assert_eq!(ipv6_hostname, "2001-0db8-0401-0000-0000-0000-0000-0002"); + + sqlx::query( + "CALL update_admin_network( + $1, + '198.19.0.0/29'::inet, + '198.19.0.1'::inet + )", + ) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + + let rows: Vec<(IpNetwork, Option, IpAddr, String)> = sqlx::query_as( + "SELECT np.prefix, np.gateway, mia.address, mi.hostname + FROM network_prefixes np + JOIN machine_interfaces mi ON mi.segment_id = np.segment_id + JOIN machine_interface_addresses mia ON mia.interface_id = mi.id + WHERE np.segment_id = $1", + ) + .bind(segment_id) + .fetch_all(txn.as_pgconn()) + .await?; + + assert_eq!( + rows, + vec![( + "198.19.0.0/29".parse()?, + Some("198.19.0.1".parse()?), + "198.19.0.2".parse::()?, + "198-19-0-2".to_string(), + )], + ); + Ok(()) +} + +/// A dual-stack admin update selects the matching prefix and leaves the other +/// address family untouched. +#[crate::sqlx_test] +async fn update_admin_network_changes_only_the_matching_family( + pool: sqlx::PgPool, +) -> Result<(), Box> { + let segment_id = create_managed_segment_with_prefixes( + &pool, + "dual-stack-admin-update", + &["198.20.0.0/29", "2001:db8:500::/125"], + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let interface_id = create_addressless_race_interface( + &pool, + segment_id, + "02:00:00:20:06:01".parse()?, + "198-20-0-2", + ) + .await?; + + let mut txn = db::Transaction::begin(&pool).await?; + for address in [ + "198.20.0.2".parse::()?, + "2001:db8:500::2".parse::()?, + ] { + crate::machine_interface_address::insert( + txn.as_pgconn(), + interface_id, + address, + AllocationType::Static, + ) + .await?; + } + sqlx::query( + "CALL update_admin_network( + $1, + '198.20.1.0/29'::inet, + '198.20.1.1'::inet + )", + ) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + + let prefixes: Vec = sqlx::query_scalar( + "SELECT prefix + FROM network_prefixes + WHERE segment_id = $1 + ORDER BY family(prefix)", + ) + .bind(segment_id) + .fetch_all(txn.as_pgconn()) + .await?; + let addresses: Vec = sqlx::query_scalar( + "SELECT address + FROM machine_interface_addresses + WHERE interface_id = $1 + ORDER BY family(address)", + ) + .bind(interface_id) + .fetch_all(txn.as_pgconn()) + .await?; + + assert_eq!( + prefixes, + vec!["198.20.1.0/29".parse()?, "2001:db8:500::/125".parse()?,] + ); + assert_eq!( + addresses, + vec![ + "198.20.1.2".parse::()?, + "2001:db8:500::2".parse::()?, + ] + ); + + sqlx::query( + "CALL update_admin_network( + $1, + '2001:db8:501::/125'::inet, + NULL::inet + )", + ) + .bind(segment_id) + .execute(txn.as_pgconn()) + .await?; + let addresses: Vec = sqlx::query_scalar( + "SELECT address + FROM machine_interface_addresses + WHERE interface_id = $1 + ORDER BY family(address)", + ) + .bind(interface_id) + .fetch_all(txn.as_pgconn()) + .await?; + let hostname: String = + sqlx::query_scalar("SELECT hostname FROM machine_interfaces WHERE id = $1") + .bind(interface_id) + .fetch_one(txn.as_pgconn()) + .await?; + assert_eq!( + addresses, + vec![ + "198.20.1.2".parse::()?, + "2001:db8:501::2".parse::()?, + ] + ); + assert_eq!(hostname, "198-20-1-2"); + Ok(()) +} + +/// DHCP family recovery also sees fixed owners on another segment. +#[crate::sqlx_test] +async fn dhcp_family_reallocation_retries_a_concurrent_fixed_address( + pool: sqlx::PgPool, +) -> Result<(), Box> { + create_static_assignments_segment(&pool).await?; + create_managed_segment( + &pool, + "reallocation-race-ipv4", + "203.0.113.0/29", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + let fixed_owner_segment = + load_test_segment(&pool, db::network_segment::STATIC_ASSIGNMENTS_SEGMENT_NAME).await?; + create_managed_segment( + &pool, + "reallocation-race-ipv6", + "2001:db8:300::/125", + NetworkSegmentType::Admin, + AllocationStrategy::Dynamic, + ) + .await?; + + let cases = [ + ( + load_test_segment(&pool, "reallocation-race-ipv4").await?, + "203.0.113.2".parse::()?, + IpAddressFamily::Ipv4, + "02:00:00:30:01:01".parse::()?, + "02:00:00:30:01:02".parse::()?, + ), + ( + load_test_segment(&pool, "reallocation-race-ipv6").await?, + "2001:db8:300::1".parse::()?, + IpAddressFamily::Ipv6, + "02:00:00:30:02:01".parse::()?, + "02:00:00:30:02:02".parse::()?, + ), + ]; + + for (segment, fixed_address, family, owner_mac, dhcp_mac) in cases { + run_fixed_dynamic_reallocation_race( + &pool, + segment, + fixed_owner_segment.id, + fixed_address, + family, + owner_mac, + dhcp_mac, + ) + .await?; + } + + Ok(()) +} + /// A MAC identifies one physical interface even when stale or transitional /// rows represent it on more than one segment. Site Explorer learns one /// vendor-native Redfish id for that interface, so `set_boot_interface_id` diff --git a/crates/api-db/src/machine_interface_address.rs b/crates/api-db/src/machine_interface_address.rs index 48cac96373..66127495ba 100644 --- a/crates/api-db/src/machine_interface_address.rs +++ b/crates/api-db/src/machine_interface_address.rs @@ -26,14 +26,18 @@ use model::machine_interface::InterfaceType; use model::network_segment::NetworkSegmentType; use sqlx::{FromRow, PgConnection}; -use super::DatabaseError; +use super::{DatabaseError, Transaction}; use crate::db_read::DbReader; #[cfg(test)] mod test_find_by_address; -/// Returned by allocation paths with `AddressSelectionStrategy::StaticAddress` -/// when the target IP is already held by some other interface. +const ADDRESS_INSERT_MAX_RETRIES: usize = 3; + +/// Returned when a writer tries to give an address to a second interface. +/// +/// Fixed and manual assignment paths return this conflict to the caller. +/// Dynamic allocation paths use it to select another candidate. #[derive(thiserror::Error, Debug)] #[error("address already in use: {0} by {1} in network segment {2} (interface: {3})")] pub struct AddressAlreadyInUseError( @@ -175,21 +179,77 @@ pub async fn delete_by_interface_and_address( } /// Insert a new address for an interface with the given allocation type. +/// +/// `machine_interface_addresses_address_key` is the final ownership boundary: +/// at most one machine interface may own a given IP across the site. +/// `ON CONFLICT` keeps the caller's transaction usable when that address +/// constraint wins, so a replay by the current owner stays idempotent without +/// changing its stored allocation type. A different owner returns +/// [`AddressAlreadyInUseError`] instead of a generic database error. +/// +/// This conflict handling applies only to exact-address ownership. The +/// existing one-address-per-family constraint on each interface remains a +/// separate write boundary. pub async fn insert( txn: &mut PgConnection, interface_id: MachineInterfaceId, address: IpAddr, allocation_type: AllocationType, ) -> Result<(), DatabaseError> { - let query = "INSERT INTO machine_interface_addresses (interface_id, address, allocation_type) VALUES ($1::uuid, $2::inet, $3)"; - sqlx::query(query) - .bind(interface_id) - .bind(address) - .bind(allocation_type) - .execute(txn) - .await - .map(|_| ()) - .map_err(|e| DatabaseError::query(query, e)) + let insert_query = "INSERT INTO machine_interface_addresses + (interface_id, address, allocation_type) + VALUES ($1::uuid, $2::inet, $3) + ON CONFLICT ON CONSTRAINT machine_interface_addresses_address_key + DO NOTHING"; + let owner_query = "SELECT mi.mac_address, mi.segment_id, mi.id + FROM machine_interface_addresses mia + JOIN machine_interfaces mi ON mi.id = mia.interface_id + WHERE mia.address = $1::inet"; + + for _ in 0..ADDRESS_INSERT_MAX_RETRIES { + let inserted = sqlx::query(insert_query) + .bind(interface_id) + .bind(address) + .bind(allocation_type) + .execute(&mut *txn) + .await + .map_err(|error| DatabaseError::query(insert_query, error))?; + if inserted.rows_affected() == 1 { + return Ok(()); + } + + // Under `READ COMMITTED`, the conflicting row may have committed after + // the INSERT statement took its snapshot. A second statement can see + // that winner and gives callers the same typed error as the sequential + // fixed-address check. + let owner: Option<(MacAddress, NetworkSegmentId, MachineInterfaceId)> = + sqlx::query_as(owner_query) + .bind(address) + .fetch_optional(&mut *txn) + .await + .map_err(|error| DatabaseError::query(owner_query, error))?; + if let Some((mac_address, segment_id, owner_interface_id)) = owner { + if owner_interface_id == interface_id { + // The existing allocation type remains authoritative. In + // particular, a DHCP replay must not downgrade a static row. + return Ok(()); + } + return Err(AddressAlreadyInUseError( + address, + mac_address, + segment_id, + owner_interface_id, + ) + .into()); + } + + // The winner disappeared between conflict detection and lookup. Try + // the insert again rather than returning an ownerless conflict. + } + + Err(DatabaseError::internal(format!( + "address {address} repeatedly changed owners while assigning it to interface {interface_id}", + ))) } /// Assign a static address to an interface. If the interface already @@ -199,10 +259,35 @@ pub async fn insert( /// - `Static`: the old static address is replaced. /// - `Dhcp` or `Slaac`: the managed allocation is removed and /// replaced with the static assignment. +/// +/// The replacement runs in a savepoint so a conflicting owner cannot leave the +/// interface addressless when the caller handles the typed error and continues +/// its outer transaction. pub async fn assign_static( txn: &mut PgConnection, interface_id: MachineInterfaceId, address: IpAddr, +) -> Result { + let mut assign_txn = Transaction::begin_inner(txn).await?; + let result = assign_static_inner(assign_txn.as_pgconn(), interface_id, address).await; + match result { + Ok(result) => { + assign_txn.commit().await?; + Ok(result) + } + Err(error) => { + assign_txn.rollback().await?; + Err(error) + } + } +} + +/// Perform the delete-and-insert portion of [`assign_static`] inside its +/// savepoint. +async fn assign_static_inner( + txn: &mut PgConnection, + interface_id: MachineInterfaceId, + address: IpAddr, ) -> Result { let family = address.address_family(); @@ -226,11 +311,12 @@ pub async fn assign_static( Ok(result) } -/// Delete an address allocation of the given type. Returns the interfaces that -/// owned the deleted addresses (normally one, empty if nothing matched) so -/// callers can resync each one's hostname rather than guessing the owner. The -/// delete removes every matching row, so all owners are returned — not just the -/// first — since `(address, allocation_type)` is not unique on its own. +/// Delete an address allocation of the given type. Returns the interface that +/// owned the deleted address (or an empty vector if nothing matched) so callers +/// can resync its hostname without a separate ownership lookup. +/// +/// The vector return stays in place for existing callers, although the global +/// address constraint now limits it to one interface. pub async fn delete_by_address( txn: &mut PgConnection, address: IpAddr, @@ -248,9 +334,10 @@ pub async fn delete_by_address( /// Delete an address allocation for a given (ip, mac) pair, which /// of course only actually deletes when the pair matches. /// -/// Returns the interfaces that owned the deleted allocations (normally one, -/// empty if the pair matched nothing) so callers can resync each one's hostname -/// against the authoritative deleted rows rather than a separate lookup. +/// Returns the interface that owned the deleted allocation (or an empty vector +/// if the pair matched nothing) so callers can resync its hostname against the +/// authoritative deleted row rather than a separate lookup. The vector return +/// stays in place for existing callers. pub async fn delete_by_address_and_mac( txn: &mut PgConnection, address: IpAddr, @@ -311,6 +398,44 @@ pub struct MachineInterfaceSearchResult { mod tests { use super::*; + const ADDRESS_UNIQUENESS_MIGRATION: &str = + include_str!("../migrations/20260730222152_machine_interface_addresses_unique.sql"); + + /// Create the minimal segment used by address ownership tests. + async fn create_test_segment( + txn: &mut PgConnection, + name: &str, + ) -> Result { + sqlx::query_scalar( + "INSERT INTO network_segments (name, version) + VALUES ($1, 'V1-T0') + RETURNING id", + ) + .bind(name) + .fetch_one(txn) + .await + } + + /// Create an addressless interface with a stable hostname. + async fn create_test_interface( + txn: &mut PgConnection, + segment_id: NetworkSegmentId, + mac_address: &str, + hostname: &str, + ) -> Result { + sqlx::query_scalar( + "INSERT INTO machine_interfaces + (segment_id, mac_address, primary_interface, hostname) + VALUES ($1, $2::macaddr, false, $3) + RETURNING id", + ) + .bind(segment_id) + .bind(mac_address) + .bind(hostname) + .fetch_one(txn) + .await + } + /// Verifies the new SLAAC allocation type survives a database round trip. #[crate::sqlx_test] async fn slaac_allocation_type_round_trips( @@ -351,4 +476,275 @@ mod tests { txn.rollback().await?; Ok(()) } + + /// The shared insert accepts an exact owner replay and returns a typed + /// conflict for another owner without poisoning the caller's transaction. + #[crate::sqlx_test] + async fn insert_rejects_an_address_owned_by_another_interface( + pool: sqlx::PgPool, + ) -> Result<(), Box> { + let mut txn = pool.begin().await?; + let segment_id = create_test_segment(txn.as_mut(), "unique-address-owner").await?; + let owner_mac: MacAddress = "02:00:00:00:01:01".parse()?; + let owner_id = create_test_interface( + txn.as_mut(), + segment_id, + &owner_mac.to_string(), + "address-owner", + ) + .await?; + let contender_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:01:02", + "address-contender", + ) + .await?; + + for address in [ + "192.0.2.10".parse::()?, + "2001:db8::10".parse::()?, + ] { + insert(txn.as_mut(), owner_id, address, AllocationType::Static).await?; + insert(txn.as_mut(), owner_id, address, AllocationType::Static).await?; + let owner_rows: i64 = sqlx::query_scalar( + "SELECT count(*) + FROM machine_interface_addresses + WHERE interface_id = $1 AND address = $2", + ) + .bind(owner_id) + .bind(address) + .fetch_one(txn.as_mut()) + .await?; + assert_eq!(owner_rows, 1, "an exact owner replay stays idempotent"); + + let error = insert(txn.as_mut(), contender_id, address, AllocationType::Static) + .await + .expect_err("a second interface must not own the same address"); + match error { + DatabaseError::AddressAlreadyInUse(AddressAlreadyInUseError( + conflict_address, + conflict_mac, + conflict_segment_id, + conflict_interface_id, + )) => { + assert_eq!(conflict_address, address); + assert_eq!(conflict_mac, owner_mac); + assert_eq!(conflict_segment_id, segment_id); + assert_eq!(conflict_interface_id, owner_id); + } + other => panic!("expected AddressAlreadyInUse, got {other:?}"), + } + + // `ON CONFLICT` leaves the outer transaction usable. + let one: i32 = sqlx::query_scalar("SELECT 1") + .fetch_one(txn.as_mut()) + .await?; + assert_eq!(one, 1); + } + + Ok(()) + } + + /// A failed static replacement rolls its delete back before returning the + /// conflicting owner. + #[crate::sqlx_test] + async fn assign_static_conflict_keeps_the_previous_address( + pool: sqlx::PgPool, + ) -> Result<(), Box> { + let mut txn = pool.begin().await?; + let segment_id = create_test_segment(txn.as_mut(), "static-replacement-conflict").await?; + let owner_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:02:01", + "replacement-owner", + ) + .await?; + let contender_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:02:02", + "replacement-contender", + ) + .await?; + + let cases = [ + ( + "192.0.2.20".parse::()?, + "192.0.2.21".parse::()?, + ), + ( + "2001:db8::20".parse::()?, + "2001:db8::21".parse::()?, + ), + ]; + for (owned_address, previous_address) in cases { + insert( + txn.as_mut(), + owner_id, + owned_address, + AllocationType::Static, + ) + .await?; + insert( + txn.as_mut(), + contender_id, + previous_address, + AllocationType::Static, + ) + .await?; + + let error = assign_static(txn.as_mut(), contender_id, owned_address) + .await + .expect_err("the replacement must reject another interface's address"); + assert!(matches!(error, DatabaseError::AddressAlreadyInUse(_))); + + let contender_addresses = find_for_interface(txn.as_mut(), contender_id).await?; + assert!( + contender_addresses + .iter() + .any(|address| address.address == previous_address), + "the failed replacement must restore {previous_address}", + ); + } + + Ok(()) + } + + /// The migration reports historical duplicate owners rather than choosing + /// one and deleting the other. + #[crate::sqlx_test] + async fn migration_rejects_duplicate_hosts_with_different_masks( + pool: sqlx::PgPool, + ) -> Result<(), Box> { + sqlx::raw_sql( + "ALTER TABLE machine_interface_addresses + DROP CONSTRAINT machine_interface_addresses_address_key; + ALTER TABLE machine_interface_addresses + DROP CONSTRAINT machine_interface_addresses_host_address_check;", + ) + .execute(&pool) + .await?; + + let mut txn = pool.begin().await?; + let segment_id = create_test_segment(txn.as_mut(), "migration-duplicate-address").await?; + let first_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:03:01", + "migration-owner-one", + ) + .await?; + let second_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:03:02", + "migration-owner-two", + ) + .await?; + sqlx::query( + "INSERT INTO machine_interface_addresses (interface_id, address) + VALUES + ($1, '192.0.2.30/24'::inet), + ($2, '192.0.2.30/32'::inet)", + ) + .bind(first_id) + .bind(second_id) + .execute(txn.as_mut()) + .await?; + txn.commit().await?; + + let error = sqlx::raw_sql(ADDRESS_UNIQUENESS_MIGRATION) + .execute(&pool) + .await + .expect_err("the migration must stop on ambiguous ownership"); + let message = error.to_string(); + assert!(message.contains("duplicate host addresses"), "{message}"); + assert!(message.contains("1 total"), "{message}"); + assert!(message.contains(&first_id.to_string()), "{message}"); + assert!(message.contains(&second_id.to_string()), "{message}"); + + let row_count: i64 = sqlx::query_scalar("SELECT count(*) FROM machine_interface_addresses") + .fetch_one(&pool) + .await?; + assert_eq!( + row_count, 2, + "the failed audit must not delete either owner" + ); + + Ok(()) + } + + /// The migration converts lone legacy masks to host addresses before + /// requiring `/32` and `/128` from future writers. + #[crate::sqlx_test] + async fn migration_normalizes_non_host_masks( + pool: sqlx::PgPool, + ) -> Result<(), Box> { + sqlx::raw_sql( + "ALTER TABLE machine_interface_addresses + DROP CONSTRAINT machine_interface_addresses_address_key; + ALTER TABLE machine_interface_addresses + DROP CONSTRAINT machine_interface_addresses_host_address_check;", + ) + .execute(&pool) + .await?; + + let mut txn = pool.begin().await?; + let segment_id = create_test_segment(txn.as_mut(), "migration-normalize-address").await?; + let ipv4_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:04:01", + "normalize-ipv4", + ) + .await?; + let ipv6_id = create_test_interface( + txn.as_mut(), + segment_id, + "02:00:00:00:04:02", + "normalize-ipv6", + ) + .await?; + sqlx::query( + "INSERT INTO machine_interface_addresses (interface_id, address) + VALUES + ($1, '192.0.2.40/24'::inet), + ($2, '2001:db8::40/64'::inet)", + ) + .bind(ipv4_id) + .bind(ipv6_id) + .execute(txn.as_mut()) + .await?; + txn.commit().await?; + + sqlx::raw_sql(ADDRESS_UNIQUENESS_MIGRATION) + .execute(&pool) + .await?; + + let masks: Vec = sqlx::query_scalar( + "SELECT masklen(address) + FROM machine_interface_addresses + ORDER BY family(address)", + ) + .fetch_all(&pool) + .await?; + assert_eq!(masks, vec![32, 128]); + + let duplicate = sqlx::query( + "INSERT INTO machine_interface_addresses (interface_id, address) + VALUES ($1, '192.0.2.40'::inet)", + ) + .bind(ipv6_id) + .execute(&pool) + .await + .expect_err("the replayed migration must enforce global ownership"); + let constraint = duplicate + .as_database_error() + .and_then(|error| error.constraint()); + assert_eq!(constraint, Some("machine_interface_addresses_address_key")); + + Ok(()) + } } diff --git a/crates/api-db/src/network_segment.rs b/crates/api-db/src/network_segment.rs index df1f511361..b636337aa8 100644 --- a/crates/api-db/src/network_segment.rs +++ b/crates/api-db/src/network_segment.rs @@ -34,7 +34,7 @@ use sqlx::{PgConnection, PgTransaction}; use crate::db_read::DbReader; use crate::instance_address::UsedOverlayNetworkIpResolver; use crate::ip_allocator::{IpAllocator, UsedIpResolver}; -use crate::machine_interface::UsedAdminNetworkIpResolver; +use crate::machine_interface::UsedMachineInterfaceIpResolver; use crate::{ ColumnInfo, DatabaseError, DatabaseResult, FilterableQueryBuilder, ObjectColumnFilter, }; @@ -739,15 +739,13 @@ where busy_ips, }) } else { - // Note on UsedAdminNetworkIpResolver: + // Note on UsedMachineInterfaceIpResolver: // In this case, the IpAllocator isn't being used to iterate to get // the next available prefix_length allocation -- it's actually just // being used to get the number of free IPs left in a given admin - // network segment, so just hard-code a /32 prefix_length. Unlike the - // tenant segments, the admin segments are always (at least for the - // foreseeable future) just going to allocate a /32 for the machine - // interface. - Box::new(UsedAdminNetworkIpResolver { + // network segment. Machine interfaces consume one host address at a + // time for either family. + Box::new(UsedMachineInterfaceIpResolver { segment_id: record.id, busy_ips, })