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, })