diff --git a/members/nullnet-client/src/control_channel.rs b/members/nullnet-client/src/control_channel.rs index 67e02db..4d080e2 100644 --- a/members/nullnet-client/src/control_channel.rs +++ b/members/nullnet-client/src/control_channel.rs @@ -34,6 +34,33 @@ fn fire_event(grpc: &NullnetGrpcInterface, kind: AgentEventKind) { }); } +/// Confirm a completed teardown so the server can return the net id to its +/// pool. Must be called only once the teardown has actually run — the whole +/// point of the ack is that the id stays out of circulation until this edge's +/// kernel state is gone. +/// +/// `msg_id` is absent when the server predates the ack field; there is then +/// nothing to confirm and the server falls back to its grace timer. +async fn ack_teardown( + outbound: &Sender, + msg_id: Option, + grpc: &NullnetGrpcInterface, + message_type: &str, +) { + let Some(msg_id) = msg_id else { + return; + }; + if outbound.send(msg_id.clone()).await.is_err() { + fire_event( + grpc, + AgentEventKind::ControlChannelAckFailed(AgentControlChannelAckFailed { + msg_id: msg_id.id, + message_type: message_type.to_string(), + }), + ); + } +} + #[allow(clippy::too_many_arguments)] pub(crate) async fn control_channel( server: NullnetGrpcInterface, @@ -87,6 +114,7 @@ pub(crate) async fn control_channel( vlan_teardown, rtnetlink_handle, peers, + outbound, host_mappings_state, server, firewall_peers, @@ -118,12 +146,14 @@ pub(crate) async fn control_channel( handle_vxlan_teardown( vxlan_teardown, triggers_state, + outbound, host_mappings_state, server, firewall_peers, firewall_vxlan_ports, egress_state, - ); + ) + .await; }); } Some(net_message::Message::ContainerSuspend(container_suspend)) => { @@ -290,10 +320,12 @@ async fn handle_vlan_teardown( message: VlanTeardown, rtnetlink_handle: RtNetLinkHandle, peers: Arc>, + outbound: Sender, host_mappings_state: Arc, grpc: NullnetGrpcInterface, firewall_peers: Arc, ) -> Result<(), Error> { + let ack_id = message.msg_id.clone(); let vlan_id = u16::try_from(message.vlan_id) .handle_err(location!()) .inspect_err(|e| { @@ -327,6 +359,10 @@ async fn handle_vlan_teardown( let _ = remove_host_mapping(&host_mapping, None); } + // Acked last: the server frees the net id on this, so everything above must + // already be undone. + ack_teardown(&outbound, ack_id, &grpc, "vlan_teardown").await; + Ok(()) } @@ -624,15 +660,18 @@ async fn handle_vxlan_setup( Ok(()) } -fn handle_vxlan_teardown( +#[allow(clippy::too_many_arguments)] +async fn handle_vxlan_teardown( message: VxlanTeardown, triggers_state: Arc, + outbound: Sender, host_mappings_state: Arc, grpc: NullnetGrpcInterface, firewall_peers: Arc, firewall_vxlan_ports: Arc, egress_state: Arc, ) { + let ack_id = message.msg_id.clone(); // reverse egress steering/interception if this was an egress edge if let Some(rec) = egress_state.take(message.vxlan_id) { match rec { @@ -719,6 +758,11 @@ fn handle_vxlan_teardown( "VXLAN teardown completed in {} ms", init_t.elapsed().as_millis() ); + + // Acked last: the server frees the net id on this, so every kernel object + // named after it — bridge, veth/macsec pair, XFRM SA, DNAT — must already + // be gone. + ack_teardown(&outbound, ack_id, &grpc, "vxlan_teardown").await; } /// Pause an idle container. Fire-and-forget: the server marks the replica diff --git a/members/nullnet-grpc-lib/proto/nullnet_grpc.proto b/members/nullnet-grpc-lib/proto/nullnet_grpc.proto index 70e2354..133a265 100644 --- a/members/nullnet-grpc-lib/proto/nullnet_grpc.proto +++ b/members/nullnet-grpc-lib/proto/nullnet_grpc.proto @@ -146,6 +146,11 @@ message VlanSetup { message VlanTeardown { uint32 vlan_id = 1; + // Acked once the teardown has actually run, so the server can hold the net id + // out of the pool until this edge is really gone. Optional: a client that + // predates this field simply never acks, and the server frees on its grace + // timer instead. + optional MsgId msg_id = 2; } message VxlanSetup { @@ -199,6 +204,11 @@ message VxlanTeardown { string local_ip = 5; string remote_ip = 6; uint32 dstport = 7; + // Acked once the teardown has actually run, so the server can hold the net id + // out of the pool until this edge is really gone. Optional: a client that + // predates this field simply never acks, and the server frees on its grace + // timer instead. + optional MsgId msg_id = 8; } message MsgId { diff --git a/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs b/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs index f913682..f40dc71 100644 --- a/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs +++ b/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs @@ -99,10 +99,16 @@ pub struct VlanSetup { #[prost(bool, tag = "9")] pub encrypted: bool, } -#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct VlanTeardown { #[prost(uint32, tag = "1")] pub vlan_id: u32, + /// Acked once the teardown has actually run, so the server can hold the net id + /// out of the pool until this edge is really gone. Optional: a client that + /// predates this field simply never acks, and the server frees on its grace + /// timer instead. + #[prost(message, optional, tag = "2")] + pub msg_id: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct VxlanSetup { @@ -179,6 +185,12 @@ pub struct VxlanTeardown { pub remote_ip: ::prost::alloc::string::String, #[prost(uint32, tag = "7")] pub dstport: u32, + /// Acked once the teardown has actually run, so the server can hold the net id + /// out of the pool until this edge is really gone. Optional: a client that + /// predates this field simply never acks, and the server frees on its grace + /// timer instead. + #[prost(message, optional, tag = "8")] + pub msg_id: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct MsgId { diff --git a/members/nullnet-server/src/events.rs b/members/nullnet-server/src/events.rs index 60e926d..9ce0fc1 100644 --- a/members/nullnet-server/src/events.rs +++ b/members/nullnet-server/src/events.rs @@ -85,6 +85,14 @@ pub(crate) enum Event { client_ip: String, timestamp: u64, }, + /// An endpoint never confirmed a teardown, so the net id went back to the + /// pool unverified. Its kernel state may still exist on that node, and a + /// later edge reusing the id would collide with it. + NetTeardownUnconfirmed { + net_id: u32, + node_ip: String, + timestamp: u64, + }, ConfigReloaded { stack: String, timestamp: u64, @@ -350,6 +358,7 @@ impl Event { Self::SetupTimeout { .. } => "setup_timeout", Self::SessionCreated { .. } => "session_created", Self::SessionTornDown { .. } => "session_torn_down", + Self::NetTeardownUnconfirmed { .. } => "net_teardown_unconfirmed", Self::ConfigReloaded { .. } => "config_reloaded", Self::ConfigStackRemoved { .. } => "config_stack_removed", Self::PortMappingConflict { .. } => "port_mapping_conflict", @@ -428,6 +437,7 @@ impl Event { | Self::MaxNetworksLimitEnforced { .. } | Self::BackendTriggerSetupBailed { .. } | Self::ControlChannelClosed { .. } + | Self::NetTeardownUnconfirmed { .. } | Self::CertificateRemoved { .. } => Severity::Warning, Self::SetupTimeout { .. } @@ -549,6 +559,14 @@ impl Event { } } + pub(crate) fn net_teardown_unconfirmed(net_id: u32, node_ip: String) -> Self { + Self::NetTeardownUnconfirmed { + net_id, + node_ip, + timestamp: now_secs(), + } + } + pub(crate) fn config_reloaded(stack: String) -> Self { Self::ConfigReloaded { stack, diff --git a/members/nullnet-server/src/net.rs b/members/nullnet-server/src/net.rs index 6c86306..560b736 100644 --- a/members/nullnet-server/src/net.rs +++ b/members/nullnet-server/src/net.rs @@ -48,6 +48,7 @@ pub(crate) trait NetExt { local_ip: IpAddr, remote_ip: IpAddr, dstport: Option, + msg_id: String, ) -> NetMessage; } @@ -100,11 +101,14 @@ impl NetExt for Net { local_ip: IpAddr, remote_ip: IpAddr, dstport: Option, + msg_id: String, ) -> NetMessage { + let msg_id = Some(MsgId { id: msg_id }); match self { Net::Vlan => NetMessage { message: Some(net_message::Message::VlanTeardown(VlanTeardown { vlan_id: net_id, + msg_id, })), }, Net::Vxlan => NetMessage { @@ -116,6 +120,7 @@ impl NetExt for Net { local_ip: local_ip.to_string(), remote_ip: remote_ip.to_string(), dstport: u32::from(dstport.unwrap_or(DEFAULT_VXLAN_DSTPORT)), + msg_id, })), }, } diff --git a/members/nullnet-server/src/net_id_pool.rs b/members/nullnet-server/src/net_id_pool.rs index 94d297f..f191251 100644 --- a/members/nullnet-server/src/net_id_pool.rs +++ b/members/nullnet-server/src/net_id_pool.rs @@ -1,6 +1,6 @@ use aes_gcm::aead::OsRng; use aes_gcm::aead::rand_core::RngCore; -use std::collections::BTreeSet; +use std::collections::{HashSet, VecDeque}; use std::sync::LazyLock; use crate::env::NET_TYPE; @@ -19,29 +19,41 @@ static MAX_NET_ID: LazyLock = LazyLock::new(|| match *NET_TYPE { /// Pool for VLAN/VXLAN network IDs. /// -/// Reuses freed IDs (lowest available first) before allocating new ones. +/// Reuses freed IDs oldest-first (FIFO) before allocating new ones. FIFO is +/// deliberate: every kernel-side name an edge owns — `br__*`, `veth--*`, +/// `macsec--*`, the derived MACs, `SPI +1000`, the flock path — derives +/// from the ID alone, so generation N and N+1 are indistinguishable to the +/// kernel. Handing an ID straight back out is what lets a late teardown for +/// generation N delete generation N+1's edge. Popping the *oldest* freed ID +/// maximizes the gap between free and reuse; the previous lowest-first +/// `BTreeSet` did the opposite. #[derive(Debug)] pub(crate) struct NetIdPool { /// The next fresh ID to allocate (when no freed IDs are available). next_fresh: u32, - /// Set of IDs that were freed and can be reused. - freed: BTreeSet, + /// IDs that were freed and can be reused, oldest first. + freed: VecDeque, + /// Membership index for `freed`. A `VecDeque` cannot dedupe on its own, and + /// a double `free` of the same ID would otherwise queue it twice and hand + /// it to two live edges at once — a corruption the old `BTreeSet` made + /// impossible for free. Keeps that invariant at O(1). + freed_set: HashSet, } impl NetIdPool { pub(crate) fn new() -> Self { Self { next_fresh: MIN_NET_ID, - freed: BTreeSet::new(), + freed: VecDeque::new(), + freed_set: HashSet::new(), } } - /// Allocate a network ID, reusing a previously freed one if available. - /// Returns `None` if the pool is exhausted. + /// Allocate a network ID, reusing the longest-freed one if any are + /// available. Returns `None` if the pool is exhausted. pub(crate) fn allocate(&mut self) -> Option { - // Prefer reusing the lowest freed ID - if let Some(&id) = self.freed.iter().next() { - self.freed.remove(&id); + if let Some(id) = self.freed.pop_front() { + self.freed_set.remove(&id); return Some(id); } @@ -55,10 +67,11 @@ impl NetIdPool { } } - /// Return a network ID to the pool for reuse. + /// Return a network ID to the pool for reuse. Freeing an ID that is already + /// queued is a no-op. pub(crate) fn free(&mut self, id: u32) { - if id >= MIN_NET_ID && id <= *MAX_NET_ID { - self.freed.insert(id); + if id >= MIN_NET_ID && id <= *MAX_NET_ID && self.freed_set.insert(id) { + self.freed.push_back(id); } } } @@ -86,23 +99,28 @@ const MAX_VXLAN_PORT: u16 = 60000; /// tunnels between the same physical host pair each get a distinct dstport. /// This is what lets an XFRM policy (which selects by IP + port, not VNI) /// tell those tunnels apart. Same allocate/free-with-reuse shape as `NetIdPool`. +/// Same allocate/free-with-reuse shape as `NetIdPool`, including its FIFO reuse +/// order — a reused dstport is half of what an XFRM policy selects on, so +/// recycling one promptly reintroduces the same cross-generation ambiguity. #[derive(Debug)] pub(crate) struct UdpPortPool { next_fresh: u16, - freed: BTreeSet, + freed: VecDeque, + freed_set: HashSet, } impl UdpPortPool { pub(crate) fn new() -> Self { Self { next_fresh: MIN_VXLAN_PORT, - freed: BTreeSet::new(), + freed: VecDeque::new(), + freed_set: HashSet::new(), } } pub(crate) fn allocate(&mut self) -> Option { - if let Some(&port) = self.freed.iter().next() { - self.freed.remove(&port); + if let Some(port) = self.freed.pop_front() { + self.freed_set.remove(&port); return Some(port); } @@ -116,8 +134,8 @@ impl UdpPortPool { } pub(crate) fn free(&mut self, port: u16) { - if (MIN_VXLAN_PORT..=MAX_VXLAN_PORT).contains(&port) { - self.freed.insert(port); + if (MIN_VXLAN_PORT..=MAX_VXLAN_PORT).contains(&port) && self.freed_set.insert(port) { + self.freed.push_back(port); } } } @@ -151,19 +169,22 @@ mod tests { assert_eq!(pool.allocate(), Some(103)); } + /// Reuse is oldest-freed-first, so the ID that has been out of service + /// longest comes back first — the widest possible gap between an edge being + /// torn down and its ID naming a different edge. #[test] - fn test_reuse_freed_net_ids() { + fn test_reuse_freed_net_ids_oldest_first() { let mut pool = NetIdPool::new(); let id1 = pool.allocate().unwrap(); let id2 = pool.allocate().unwrap(); let id3 = pool.allocate().unwrap(); - pool.free(id2); // free 102 - pool.free(id1); // free 101 + pool.free(id2); // 102 freed first + pool.free(id1); // 101 freed second - // Should reuse lowest freed ID first - assert_eq!(pool.allocate(), Some(101)); + // FIFO: 102 comes back before 101, even though 101 is numerically lower assert_eq!(pool.allocate(), Some(102)); + assert_eq!(pool.allocate(), Some(101)); // Then continue with fresh IDs assert_eq!(pool.allocate(), Some(104)); @@ -171,6 +192,38 @@ mod tests { assert_eq!(pool.allocate(), Some(103)); } + /// A just-freed ID must go to the back of the queue, never straight back + /// out — that immediate handback is the reuse race this ordering exists to + /// widen. + #[test] + fn test_freed_id_is_not_immediately_reallocated() { + let mut pool = NetIdPool::new(); + let a = pool.allocate().unwrap(); + let b = pool.allocate().unwrap(); + pool.free(a); + pool.free(b); + + // `a` was freed first, so it is handed out first; `b` waits behind it. + assert_eq!(pool.allocate(), Some(a)); + assert_eq!(pool.allocate(), Some(b)); + } + + /// Freeing the same ID twice must not queue it twice — otherwise two live + /// edges would be handed the same ID. + #[test] + fn test_double_free_does_not_duplicate_net_id() { + let mut pool = NetIdPool::new(); + let id = pool.allocate().unwrap(); + pool.free(id); + pool.free(id); + pool.free(id); + + assert_eq!(pool.allocate(), Some(id)); + // Next allocation must be a fresh ID, not `id` a second time. + assert_eq!(pool.allocate(), Some(102)); + assert!(pool.freed.is_empty()); + } + #[test] fn test_net_ids_exhaustion() { let mut pool = NetIdPool::new(); @@ -203,17 +256,30 @@ mod tests { } #[test] - fn test_udp_port_pool_reuse_freed() { + fn test_udp_port_pool_reuse_freed_oldest_first() { let mut pool = UdpPortPool::new(); let p1 = pool.allocate().unwrap(); let p2 = pool.allocate().unwrap(); pool.allocate(); - pool.free(p2); - pool.free(p1); + pool.free(p2); // freed first + pool.free(p1); // freed second - assert_eq!(pool.allocate(), Some(p1)); + // FIFO, so p2 comes back before the numerically lower p1 assert_eq!(pool.allocate(), Some(p2)); + assert_eq!(pool.allocate(), Some(p1)); + } + + #[test] + fn test_udp_port_pool_double_free_does_not_duplicate() { + let mut pool = UdpPortPool::new(); + let p = pool.allocate().unwrap(); + pool.free(p); + pool.free(p); + + assert_eq!(pool.allocate(), Some(p)); + assert_eq!(pool.allocate(), Some(MIN_VXLAN_PORT + 1)); + assert!(pool.freed.is_empty()); } #[test] diff --git a/members/nullnet-server/src/orchestrator.rs b/members/nullnet-server/src/orchestrator.rs index 30b773f..e7b45be 100644 --- a/members/nullnet-server/src/orchestrator.rs +++ b/members/nullnet-server/src/orchestrator.rs @@ -12,6 +12,7 @@ use nullnet_liberror::{Error, ErrorHandler, Location, location}; use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use tokio::sync::{Mutex, RwLock, mpsc, oneshot}; use tonic::{Request, Status, Streaming}; @@ -28,6 +29,12 @@ type EgressKey = (IpAddr, Option); /// that contacts a very large set of hosts (e.g. a crawler). const MAX_DESTS_PER_EDGE: usize = 256; +/// How long to wait for an endpoint to confirm a teardown before returning the +/// net id to the pool unconfirmed. Matches `send_container_resume`'s 30s ack +/// window; teardown is a short local operation, so reaching this means the +/// endpoint is wedged or gone rather than slow. +const TEARDOWN_ACK_GRACE: Duration = Duration::from_secs(30); + /// Per-destination stats on an egress edge, reported by the client (which owns /// the running count and latest-seen time; the server stores them verbatim). #[derive(Debug, Clone)] @@ -102,6 +109,10 @@ pub struct Orchestrator { egress_edges: Arc>>, /// IP → country/ASN cache enriching contacted egress destinations. geo: GeoCache, + /// Teardowns whose net id has not yet been returned to the pool because + /// their acks are still outstanding. Lets tests wait for the pool to reach + /// steady state instead of racing the detached free task. + inflight_teardowns: Arc, pub(crate) events: EventStore, } @@ -115,6 +126,7 @@ impl Orchestrator { net_id_ports: Arc::new(Mutex::new(HashMap::new())), egress_edges: Arc::new(RwLock::new(HashMap::new())), geo: GeoCache::from_env(), + inflight_teardowns: Arc::new(AtomicUsize::new(0)), events: EventStore::new(), } } @@ -645,6 +657,25 @@ impl Orchestrator { self.clients.read().await.keys().copied().collect() } + /// Tear an edge down on both endpoints and return the net id (and its + /// dstport) to their pools — but only **after** the endpoints confirm the + /// teardown actually ran. + /// + /// Freeing on enqueue, as this used to do, is the root of the net-id reuse + /// races: the id names every kernel object the edge owns, so handing it + /// straight back out lets the next generation collide with an edge that is + /// still being dismantled. The free is therefore deferred to a detached + /// task that waits on the endpoints' acks. + /// + /// Detached deliberately — callers `.await` this in loops (`decrement_chain` + /// per replica, `collect_dep_chain_edges` per edge), so blocking here would + /// turn a multi-edge chain teardown into a serial walk of ack timeouts. + /// Caller-visible latency is unchanged. + /// + /// The id is still freed if an ack never arrives (`TEARDOWN_ACK_GRACE`), + /// because the endpoint being gone is the *normal* case on this path — + /// `teardown_egress_edges_for_node` runs precisely when a node has + /// disconnected — and refusing to free would leak every id that node held. pub(crate) async fn send_net_teardown( &self, client: IpAddr, @@ -655,13 +686,15 @@ impl Orchestrator { ) { // Peeked (not removed yet) so both teardown messages can carry the // same dstport that was used to install this tunnel's XFRM state; - // the pool slot itself is freed below, after both sides are notified. + // the pool slot itself is freed by the task below. let dstport = self .net_id_ports .lock() .await .get(&net_id) .map(|(_pair, port)| *port); + + let mut acks = Vec::new(); for (dest, remote, side, docker) in [ (client, server, "c", client_docker), (server, client, "s", server_docker), @@ -670,11 +703,41 @@ impl Orchestrator { if let Some(outbound) = outbound { println!("Sending network {net_id} teardown to client {dest}"); - let message = NET_TYPE.teardown(net_id, side, docker, dest, remote, dstport); + let (tx, rx) = oneshot::channel(); + let msg_id = Uuid::new_v4().to_string(); + self.pending.lock().await.insert(msg_id.clone(), tx); - let _ = outbound.send(Ok(message)).await.handle_err(location!()); + let message = + NET_TYPE.teardown(net_id, side, docker, dest, remote, dstport, msg_id.clone()); + + if outbound + .send(Ok(message)) + .await + .handle_err(location!()) + .is_err() + { + // Nothing will ever ack a message that was never sent. + self.pending.lock().await.remove(&msg_id); + } else { + acks.push((dest, msg_id, rx)); + } } } + + if acks.is_empty() { + // Neither endpoint was reachable, so nothing was sent and there is + // nothing to wait for — deferring would hold the id without + // learning anything. Same behaviour as before this change; the + // edge's kernel state, if any survives, is reconciled by the + // client's startup purge when that node comes back. + self.free_net_id_and_port(net_id).await; + } else { + self.spawn_deferred_net_id_free(net_id, acks); + } + } + + /// Return a net id and its VXLAN dstport to their pools. + async fn free_net_id_and_port(&self, net_id: u32) { self.net_id_pool.lock().await.free(net_id); if let Some((pair, port)) = self.net_id_ports.lock().await.remove(&net_id) && let Some(pool) = self.udp_port_pools.lock().await.get_mut(&pair) @@ -682,6 +745,55 @@ impl Orchestrator { pool.free(port); } } + + /// Wait for every endpoint that was actually sent a teardown to ack it, + /// then return the net id and its dstport to the pools. See + /// `send_net_teardown` for why this is detached and why it frees anyway on + /// timeout. + fn spawn_deferred_net_id_free( + &self, + net_id: u32, + acks: Vec<(IpAddr, String, oneshot::Receiver<()>)>, + ) { + let net_id_pool = self.net_id_pool.clone(); + let net_id_ports = self.net_id_ports.clone(); + let udp_port_pools = self.udp_port_pools.clone(); + let pending = self.pending.clone(); + let events = self.events.clone(); + let inflight = self.inflight_teardowns.clone(); + + // Counted before the spawn so a caller that awaits `send_net_teardown` + // and then waits for quiescence can never observe zero prematurely. + inflight.fetch_add(1, Ordering::SeqCst); + + tokio::spawn(async move { + for (dest, msg_id, rx) in acks { + match tokio::time::timeout(TEARDOWN_ACK_GRACE, rx).await { + Ok(Ok(())) => {} + // Timed out, or the sender was dropped without acking. + _ => { + pending.lock().await.remove(&msg_id); + println!( + "Network {net_id} teardown was not acked by {dest} within {}s; \ + freeing the id anyway", + TEARDOWN_ACK_GRACE.as_secs() + ); + events + .emit(Event::net_teardown_unconfirmed(net_id, dest.to_string())) + .await; + } + } + } + + net_id_pool.lock().await.free(net_id); + if let Some((pair, port)) = net_id_ports.lock().await.remove(&net_id) + && let Some(pool) = udp_port_pools.lock().await.get_mut(&pair) + { + pool.free(port); + } + inflight.fetch_sub(1, Ordering::SeqCst); + }); + } } /// Normalize a host pair so both call orders (A, B) and (B, A) land on the @@ -700,6 +812,31 @@ impl Orchestrator { self.register_recording_client(ip).await; } + /// Wait until every in-flight teardown has returned its net id to the pool. + /// Production code never needs to observe the deferred free; assertions + /// about pool state do, and must not race the detached task. + pub(crate) async fn settle_teardowns(&self) { + // Bounded: a teardown whose ack never arrives resolves via + // TEARDOWN_ACK_GRACE, which no test should be waiting on. If this cap + // is ever hit, the assertion that follows will fail loudly rather than + // hang. + for _ in 0..10_000 { + if self.inflight_teardowns.load(Ordering::SeqCst) == 0 { + return; + } + tokio::task::yield_now().await; + } + } + + /// A connected client that receives messages but never acks any of them — + /// a node that is wedged rather than gone. Distinct from simply not being + /// registered, which means nothing is sent at all. + pub(crate) async fn register_silent_client(&self, ip: IpAddr) { + let (tx, mut rx) = mpsc::channel::>(64); + self.clients.write().await.insert(ip, tx); + tokio::spawn(async move { while rx.recv().await.is_some() {} }); + } + /// Like `register_fake_client`, but returns a log of every `NetMessage` sent /// to the client so tests can assert suspend/resume commands were issued. pub(crate) async fn register_recording_client( @@ -728,6 +865,14 @@ impl Orchestrator { | Some(net_message::Message::ContainerResume(ContainerResume { msg_id, .. })) => msg_id.clone(), + // Teardowns are ack'd too, mirroring the real client: the + // server holds the net id out of the pool until this lands. + Some(net_message::Message::VlanTeardown( + nullnet_grpc_lib::nullnet_grpc::VlanTeardown { msg_id, .. }, + )) + | Some(net_message::Message::VxlanTeardown( + nullnet_grpc_lib::nullnet_grpc::VxlanTeardown { msg_id, .. }, + )) => msg_id.clone(), _ => None, }; log_task.lock().await.push(msg); @@ -743,6 +888,76 @@ impl Orchestrator { } } +/// A net id names every kernel object its edge owns, so returning it to the +/// pool before the edge is actually gone is what lets the next generation +/// collide with the previous one. These cover when the id comes back. +#[cfg(test)] +mod teardown_ack_tests { + use super::*; + + fn ip(a: u8, b: u8, c: u8, d: u8) -> IpAddr { + IpAddr::V4(Ipv4Addr::new(a, b, c, d)) + } + + /// Both endpoints ack, so the id returns to the pool. + #[tokio::test] + async fn acked_teardown_returns_the_net_id() { + let orch = Orchestrator::new(); + let (a, b) = (ip(10, 0, 0, 1), ip(10, 0, 0, 2)); + orch.register_fake_client(a).await; + orch.register_fake_client(b).await; + + let id = orch.allocate_net_id().await.unwrap(); + assert_eq!(orch.net_ids_in_use().await, 1); + + orch.send_net_teardown(a, None, b, None, id).await; + orch.settle_teardowns().await; + + assert_eq!(orch.net_ids_in_use().await, 0); + } + + /// A connected but unresponsive endpoint must hold the id out of the pool. + /// Freeing here is precisely the race: the edge is still up on that node. + #[tokio::test] + async fn unacked_teardown_withholds_the_net_id() { + let orch = Orchestrator::new(); + let (a, b) = (ip(10, 0, 0, 1), ip(10, 0, 0, 2)); + orch.register_silent_client(a).await; + orch.register_silent_client(b).await; + + let id = orch.allocate_net_id().await.unwrap(); + orch.send_net_teardown(a, None, b, None, id).await; + + // Give the detached task every chance to run; it must still be parked + // on the ack rather than freeing (it frees only after the grace). + for _ in 0..256 { + tokio::task::yield_now().await; + } + + assert_eq!( + orch.net_ids_in_use().await, + 1, + "id must stay out of the pool until the teardown is confirmed" + ); + // And it must not be handed to the next edge. + assert_ne!(orch.allocate_net_id().await.unwrap(), id); + } + + /// Neither endpoint is connected, so nothing was sent and there is nothing + /// to wait for — the id comes back immediately, as before this change. + /// `teardown_egress_edges_for_node` runs in exactly this state. + #[tokio::test] + async fn teardown_to_absent_endpoints_frees_immediately() { + let orch = Orchestrator::new(); + let (a, b) = (ip(10, 0, 0, 1), ip(10, 0, 0, 2)); + + let id = orch.allocate_net_id().await.unwrap(); + orch.send_net_teardown(a, None, b, None, id).await; + + assert_eq!(orch.net_ids_in_use().await, 0); + } +} + #[cfg(test)] mod udp_port_pool_tests { use super::*; diff --git a/members/nullnet-server/src/tests.rs b/members/nullnet-server/src/tests.rs index 093a56c..1f9d78e 100644 --- a/members/nullnet-server/src/tests.rs +++ b/members/nullnet-server/src/tests.rs @@ -32,6 +32,9 @@ fn ip(a: u8, b: u8, c: u8, d: u8) -> IpAddr { } async fn assert_net_ids_in_use(server: &NullnetGrpcImpl, expected: u32) { + // A net id is returned to the pool only once both endpoints confirm the + // teardown, so let any in-flight teardown finish before sampling. + server.orchestrator().settle_teardowns().await; let in_use = server.orchestrator().net_ids_in_use().await; assert_eq!( in_use, expected, @@ -3141,6 +3144,7 @@ async fn concurrent_requests_same_client_tear_down_cleanly() { let residual = live_edges(&guard); drop(guard); + server.orchestrator().settle_teardowns().await; let in_use = server.orchestrator().net_ids_in_use().await; assert!( residual.is_empty() && in_use == 0, diff --git a/members/nullnet-server/ui/src/pages/Events.tsx b/members/nullnet-server/ui/src/pages/Events.tsx index e5fdd11..7fe0235 100644 --- a/members/nullnet-server/ui/src/pages/Events.tsx +++ b/members/nullnet-server/ui/src/pages/Events.tsx @@ -21,6 +21,7 @@ const KIND_LABELS: Record = { setup_timeout: 'setup_timeout', session_created: 'session_created', session_torn_down: 'session_torn_down', + net_teardown_unconfirmed: 'net_teardown_unconfirmed', config_reloaded: 'config_reloaded', config_stack_removed: 'config_stack_removed', all_replicas_removed: 'all_replicas_removed', @@ -91,6 +92,8 @@ function eventDetail(e: EventJson): string { return `net ${e.net_id} · ${e.service} ← ${e.client_ip}`; case 'session_torn_down': return `net ${e.net_id} · ${e.service} · ${e.client_ip}`; + case 'net_teardown_unconfirmed': + return `net ${e.net_id} · ${e.node_ip} never confirmed teardown`; case 'config_reloaded': case 'config_stack_removed': return e.stack; diff --git a/members/nullnet-server/ui/src/types.ts b/members/nullnet-server/ui/src/types.ts index f3b51e3..bc5b414 100644 --- a/members/nullnet-server/ui/src/types.ts +++ b/members/nullnet-server/ui/src/types.ts @@ -86,6 +86,7 @@ export type EventJson = | WithSeverity & { type: 'setup_timeout'; net_id: number; service: string } | WithSeverity & { type: 'session_created'; net_id: number; service: string; client_ip: string } | WithSeverity & { type: 'session_torn_down'; net_id: number; service: string; client_ip: string } + | WithSeverity & { type: 'net_teardown_unconfirmed'; net_id: number; node_ip: string } | WithSeverity & { type: 'config_reloaded'; stack: string } | WithSeverity & { type: 'config_stack_removed'; stack: string } | WithSeverity & { type: 'all_replicas_removed'; service: string; stack: string; ip: string }