From dde3d7c20fd790799488044f79ea292326e0310c Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Fri, 4 Sep 2026 14:40:58 +0100 Subject: [PATCH 1/4] experiment(network): support a caller-owned poll Separate the TCP network state from Poll by storing a Registry in NetworkState. TcpNetwork continues to own its Poll and Events, while shared network operations move to TcpNetworkRaw. Add TcpNetworkWithExternalPoll so a caller can register TCP sockets with its own poll. The caller drives reconnect processing, passes relevant events to the network, and drains pending disconnect notifications through pre_poll, epoll_event, and post_poll. The external-poll constructor accepts a half-open token range. Listener and connection tokens are assigned from that range in ascending order, and an operation requiring another token panics when the range is exhausted. The caller remains responsible for avoiding collisions with other tokens registered with the same poll. TcpNetwork and TcpNetworkWithExternalPoll expose the shared operations through Deref. Ordinary method-call syntax remains available, but fully qualified calls to methods previously defined directly on TcpNetwork will need updating. Co-authored-by: vladimir-ea Assisted-by: Claude:claude-fable-5 Assisted-by: Codex:gpt-5.6-sol --- crates/flux-network/src/tcp/network.rs | 197 ++++++++++++++++++------- 1 file changed, 141 insertions(+), 56 deletions(-) diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index c5dcb20..4c0d10f 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -1,7 +1,7 @@ use std::{ io::{self, IoSlice, Read, Write}, net::{Shutdown, SocketAddr}, - ops::{Deref, DerefMut}, + ops::{Deref, DerefMut, Range}, }; use flux_communication::Timer; @@ -332,11 +332,14 @@ struct PendingDisconnect { } struct NetworkState { - poll: Poll, + registry: Registry, groups: Vec, listeners: Vec, connections: Vec, pending_disconnects: Vec, + /// Range used to assign tokens to new listeners and connections. + /// Assigned tokens are not reused. + token_range: Range, next_token: usize, /// Frames staged for the next socket write, each as a contiguous /// `[header][payload]` for length-prefixed groups or bare bytes for raw @@ -344,24 +347,29 @@ struct NetworkState { send_buffer: Vec, } -impl Default for NetworkState { - fn default() -> Self { +impl NetworkState { + fn new(registry: Registry, tokens: Range) -> Self { Self { - poll: Poll::new().expect("couldn't set up a poll for tcp network"), + registry, groups: Vec::with_capacity(INITIAL_GROUP_CAPACITY), listeners: Vec::with_capacity(INITIAL_LISTENER_CAPACITY), connections: Vec::with_capacity(INITIAL_CONNECTION_CAPACITY), pending_disconnects: Vec::with_capacity(INITIAL_CONNECTION_CAPACITY), - next_token: 0, + next_token: tokens.start, + token_range: tokens, send_buffer: Vec::with_capacity(INITIAL_SEND_BUFFER_SIZE), } } -} -impl NetworkState { fn next_token(&mut self) -> Token { + // Reject allocations beyond the caller-provided range. + assert!( + self.next_token < self.token_range.end, + "tcp token range {:?} exhausted", + self.token_range + ); let token = Token(self.next_token); - self.next_token = self.next_token.checked_add(1).expect("tcp token space exhausted"); + self.next_token += 1; token } @@ -375,7 +383,7 @@ impl NetworkState { } let mut socket = TcpListener::bind(addr)?; let token = self.next_token(); - self.poll.registry().register(&mut socket, token, Interest::READABLE)?; + self.registry.register(&mut socket, token, Interest::READABLE)?; self.listeners.push(Listener { token, group, socket }); Ok(()) } @@ -420,7 +428,7 @@ impl NetworkState { if let Some(size) = socket_buf_size { set_socket_buf_size(&socket, size); } - if let Err(err) = self.poll.registry().register(&mut socket, token, Interest::WRITABLE) { + if let Err(err) = self.registry.register(&mut socket, token, Interest::WRITABLE) { warn!(?err, %peer_addr, "couldn't register connecting tcp stream"); let _ = socket.shutdown(Shutdown::Both); return; @@ -498,7 +506,7 @@ impl NetworkState { let Err(err) = socket.set_nodelay(true) { warn!(?err, %peer_addr, "couldn't set nodelay on tcp stream"); - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); return false; } @@ -506,12 +514,12 @@ impl NetworkState { let Err(err) = set_keepalive(&socket) { warn!(?err, %peer_addr, "couldn't set keepalive on tcp stream"); - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); return false; } set_user_timeout(&socket, config.user_timeout_ms); - if let Err(err) = self.poll.registry().reregister(&mut socket, token, Interest::READABLE) { + if let Err(err) = self.registry.reregister(&mut socket, token, Interest::READABLE) { warn!(?err, %peer_addr, "couldn't register connected tcp stream"); let _ = socket.shutdown(Shutdown::Both); return false; @@ -524,15 +532,10 @@ impl NetworkState { write_frame_header(&mut header, message.len(), Nanos::now()); header }); - if stream.write_frame( - self.poll.registry(), - header.as_ref(), - message, - config, - &mut timers, - ) == StreamState::Disconnected + if stream.write_frame(&self.registry, header.as_ref(), message, config, &mut timers) == + StreamState::Disconnected { - stream.close(self.poll.registry()); + stream.close(&self.registry); self.connections[index].timers = timers; return false; } @@ -581,9 +584,7 @@ impl NetworkState { continue; } set_user_timeout(&socket, config.user_timeout_ms); - if let Err(err) = - self.poll.registry().register(&mut socket, token, Interest::READABLE) - { + if let Err(err) = self.registry.register(&mut socket, token, Interest::READABLE) { warn!(?err, %peer_addr, "couldn't register accepted tcp stream"); let _ = socket.shutdown(Shutdown::Both); continue; @@ -604,14 +605,14 @@ impl NetworkState { header }); if stream.write_frame( - self.poll.registry(), + &self.registry, header.as_ref(), message, config, &mut timers, ) == StreamState::Disconnected { - stream.close(self.poll.registry()); + stream.close(&self.registry); continue; } } @@ -663,7 +664,7 @@ impl NetworkState { let connection = &mut self.connections[index]; let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; let state = stream.poll_with( - self.poll.registry(), + &self.registry, event, config, &mut connection.timers, @@ -687,12 +688,12 @@ impl NetworkState { match old_state { ConnectionState::Disconnected => false, ConnectionState::Connecting(mut socket) => { - let _ = self.poll.registry().deregister(&mut socket); + let _ = self.registry.deregister(&mut socket); let _ = socket.shutdown(Shutdown::Both); false } ConnectionState::Connected(mut stream) => { - stream.close(self.poll.registry()); + stream.close(&self.registry); true } } @@ -799,7 +800,7 @@ impl NetworkState { let connection = &mut self.connections[index]; let ConnectionState::Connected(stream) = &mut connection.state else { unreachable!() }; let state = stream.write_frame( - self.poll.registry(), + &self.registry, None, &self.send_buffer, config, @@ -952,24 +953,74 @@ impl NetworkState { } } +pub struct TcpNetwork { + events: Events, + poll: Poll, + raw: TcpNetworkRaw, +} + +impl Default for TcpNetwork { + fn default() -> Self { + let poll = Poll::new().expect("failed to create poll"); + let registry = poll.registry().try_clone().expect("failed to clone poll registry"); + // This network owns the poll, so its token range starts at zero. + Self { + events: Events::with_capacity(EVENTS_CAPACITY), + poll, + raw: TcpNetworkRaw::new(registry, 0..usize::MAX), + } + } +} + +impl Deref for TcpNetwork { + type Target = TcpNetworkRaw; + + fn deref(&self) -> &Self::Target { + &self.raw + } +} + +impl DerefMut for TcpNetwork { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.raw + } +} + +impl TcpNetwork { + pub fn poll_with(&mut self, mut handler: F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.raw.state.drain_pending_disconnects(&mut handler); + self.raw.state.maybe_reconnect(); + if let Err(err) = self.poll.poll(&mut self.events, Some(std::time::Duration::ZERO)) { + if err.kind() != io::ErrorKind::Interrupted { + flux_utils::safe_panic!("couldn't poll tcp network: {err}"); + } + return; + } + for event in &self.events { + self.raw.state.handle_event(event, &mut handler); + } + self.raw.state.drain_pending_disconnects(&mut handler); + } +} + /// A grouped collection of TCP listeners and persistent outbound endpoints /// driven by one nonblocking poll. /// /// Unlike [`super::TcpConnector`], queued bytes are never retained across a /// disconnected socket. Use `TcpConnector` when reconnect backlog replay is /// required. -pub struct TcpNetwork { - events: Events, +pub struct TcpNetworkRaw { state: NetworkState, } -impl Default for TcpNetwork { - fn default() -> Self { - Self { events: Events::with_capacity(EVENTS_CAPACITY), state: NetworkState::default() } +impl TcpNetworkRaw { + fn new(registry: Registry, tokens: Range) -> Self { + Self { state: NetworkState::new(registry, tokens) } } -} -impl TcpNetwork { /// Adds a protocol group and returns its handle. #[must_use = "the group handle identifies listeners and outbound endpoints"] pub fn add_group(&mut self, config: TcpGroupConfig) -> TcpGroup { @@ -1011,24 +1062,6 @@ impl TcpNetwork { self.state.connect(group, peer_addr) } - pub fn poll_with(&mut self, mut handler: F) - where - F: for<'a> FnMut(TcpEvent<'a>), - { - self.state.drain_pending_disconnects(&mut handler); - self.state.maybe_reconnect(); - if let Err(err) = self.state.poll.poll(&mut self.events, Some(std::time::Duration::ZERO)) { - if err.kind() != io::ErrorKind::Interrupted { - flux_utils::safe_panic!("couldn't poll tcp network: {err}"); - } - return; - } - for event in &self.events { - self.state.handle_event(event, &mut handler); - } - self.state.drain_pending_disconnects(&mut handler); - } - /// Serializes and sends one payload to a connected token. Length-prefixed /// groups add a frame header; raw groups send the payload unchanged. The /// closure is not called when the token is unknown or currently @@ -1102,6 +1135,58 @@ impl TcpNetwork { } } +pub struct TcpNetworkWithExternalPoll { + raw: TcpNetworkRaw, +} + +impl Deref for TcpNetworkWithExternalPoll { + type Target = TcpNetworkRaw; + + fn deref(&self) -> &Self::Target { + &self.raw + } +} + +impl DerefMut for TcpNetworkWithExternalPoll { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.raw + } +} + +impl TcpNetworkWithExternalPoll { + /// Creates a network that assigns listener and connection tokens from + /// `tokens` in ascending order without reuse. + /// + /// An operation that needs a new token panics if the range is exhausted. + /// The caller is responsible for avoiding collisions with other tokens + /// registered with the same poll. + pub fn new(registry: Registry, tokens: Range) -> Self { + Self { raw: TcpNetworkRaw::new(registry, tokens) } + } + + pub fn pre_poll(&mut self, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.raw.state.drain_pending_disconnects(handler); + self.raw.state.maybe_reconnect(); + } + + pub fn post_poll(&mut self, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.raw.state.drain_pending_disconnects(handler); + } + + pub fn epoll_event(&mut self, event: &Event, handler: &mut F) + where + F: for<'a> FnMut(TcpEvent<'a>), + { + self.raw.state.handle_event(event, handler); + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum StreamState { Alive, From a57833d92084c2fc5913a49c2ef4294a04363fd6 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Fri, 4 Sep 2026 14:57:39 +0100 Subject: [PATCH 2/4] feat(network): expose caller-owned poll integration Export TcpNetworkCore and TcpNetworkWithExternalPoll. Rename the shared type from TcpNetworkRaw to TcpNetworkCore and rename epoll_event to the platform-neutral handle_event. Document the internally polled and caller-polled variants, including the pre_poll, poll, handle_event, and post_poll sequence. Add a debug assertion to detect events routed outside the network's configured token range. Add integration coverage for two TCP networks driven by one Poll with disjoint token ranges. The test exercises bidirectional messages, reconnection with a stable outbound token, and disconnect notification delivery. Additional tests cover token-range exhaustion and the owned network's shared listener/connection token counter. Assisted-by: Claude:claude-fable-5 Assisted-by: Codex:gpt-5.6-sol --- crates/flux-network/src/tcp/mod.rs | 5 +- crates/flux-network/src/tcp/network.rs | 77 ++++-- crates/flux-network/tests/tcp_network.rs | 11 + .../tests/tcp_network_external_poll.rs | 251 ++++++++++++++++++ 4 files changed, 320 insertions(+), 24 deletions(-) create mode 100644 crates/flux-network/tests/tcp_network_external_poll.rs diff --git a/crates/flux-network/src/tcp/mod.rs b/crates/flux-network/src/tcp/mod.rs index 3d5599b..cad2054 100644 --- a/crates/flux-network/src/tcp/mod.rs +++ b/crates/flux-network/src/tcp/mod.rs @@ -3,6 +3,9 @@ mod network; mod stream; pub use connector::{PollEvent, SendBehavior, TcpConnector}; -pub use network::{Framing, PayloadBuf, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork}; +pub use network::{ + Framing, PayloadBuf, TcpEvent, TcpGroup, TcpGroupConfig, TcpNetwork, TcpNetworkCore, + TcpNetworkWithExternalPoll, +}; pub(crate) use stream::set_socket_buf_size; pub use stream::{ConnState, TcpStream, TcpTelemetry}; diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index 4c0d10f..6142a92 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -953,10 +953,16 @@ impl NetworkState { } } +/// A grouped collection of TCP listeners and persistent outbound endpoints +/// with an internal nonblocking poll. +/// +/// Shared network operations are provided by [`TcpNetworkCore`] through +/// `Deref`. Use [`TcpNetworkWithExternalPoll`] to register the sockets with a +/// caller-owned poll instead. pub struct TcpNetwork { events: Events, poll: Poll, - raw: TcpNetworkRaw, + core: TcpNetworkCore, } impl Default for TcpNetwork { @@ -967,22 +973,22 @@ impl Default for TcpNetwork { Self { events: Events::with_capacity(EVENTS_CAPACITY), poll, - raw: TcpNetworkRaw::new(registry, 0..usize::MAX), + core: TcpNetworkCore::new(registry, 0..usize::MAX), } } } impl Deref for TcpNetwork { - type Target = TcpNetworkRaw; + type Target = TcpNetworkCore; fn deref(&self) -> &Self::Target { - &self.raw + &self.core } } impl DerefMut for TcpNetwork { fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.raw + &mut self.core } } @@ -991,8 +997,8 @@ impl TcpNetwork { where F: for<'a> FnMut(TcpEvent<'a>), { - self.raw.state.drain_pending_disconnects(&mut handler); - self.raw.state.maybe_reconnect(); + self.core.state.drain_pending_disconnects(&mut handler); + self.core.state.maybe_reconnect(); if let Err(err) = self.poll.poll(&mut self.events, Some(std::time::Duration::ZERO)) { if err.kind() != io::ErrorKind::Interrupted { flux_utils::safe_panic!("couldn't poll tcp network: {err}"); @@ -1000,23 +1006,26 @@ impl TcpNetwork { return; } for event in &self.events { - self.raw.state.handle_event(event, &mut handler); + self.core.state.handle_event(event, &mut handler); } - self.raw.state.drain_pending_disconnects(&mut handler); + self.core.state.drain_pending_disconnects(&mut handler); } } -/// A grouped collection of TCP listeners and persistent outbound endpoints -/// driven by one nonblocking poll. +/// Shared state and network operations used by [`TcpNetwork`] and +/// [`TcpNetworkWithExternalPoll`]. +/// +/// This type stores protocol groups, listeners, and connections, but does not +/// own or poll a [`Poll`]. /// /// Unlike [`super::TcpConnector`], queued bytes are never retained across a /// disconnected socket. Use `TcpConnector` when reconnect backlog replay is /// required. -pub struct TcpNetworkRaw { +pub struct TcpNetworkCore { state: NetworkState, } -impl TcpNetworkRaw { +impl TcpNetworkCore { fn new(registry: Registry, tokens: Range) -> Self { Self { state: NetworkState::new(registry, tokens) } } @@ -1135,21 +1144,28 @@ impl TcpNetworkRaw { } } +/// A TCP network whose sockets are registered with a poll owned by the caller. +/// +/// A polling pass consists of [`Self::pre_poll`], polling, calling +/// [`Self::handle_event`] for each event in this network's token range, and +/// [`Self::post_poll`]. The caller owns the poll, event buffer, and timeout. +/// Dropping this value drops its sockets but does not explicitly deregister +/// them. pub struct TcpNetworkWithExternalPoll { - raw: TcpNetworkRaw, + core: TcpNetworkCore, } impl Deref for TcpNetworkWithExternalPoll { - type Target = TcpNetworkRaw; + type Target = TcpNetworkCore; fn deref(&self) -> &Self::Target { - &self.raw + &self.core } } impl DerefMut for TcpNetworkWithExternalPoll { fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.raw + &mut self.core } } @@ -1161,29 +1177,44 @@ impl TcpNetworkWithExternalPoll { /// The caller is responsible for avoiding collisions with other tokens /// registered with the same poll. pub fn new(registry: Registry, tokens: Range) -> Self { - Self { raw: TcpNetworkRaw::new(registry, tokens) } + Self { core: TcpNetworkCore::new(registry, tokens) } } + /// Delivers pending disconnect notifications and attempts due reconnects. pub fn pre_poll(&mut self, handler: &mut F) where F: for<'a> FnMut(TcpEvent<'a>), { - self.raw.state.drain_pending_disconnects(handler); - self.raw.state.maybe_reconnect(); + self.core.state.drain_pending_disconnects(handler); + self.core.state.maybe_reconnect(); } + /// Delivers pending disconnect notifications. + /// + /// If this step is skipped, notifications remain queued for a later call + /// to this method or [`Self::pre_poll`]. pub fn post_poll(&mut self, handler: &mut F) where F: for<'a> FnMut(TcpEvent<'a>), { - self.raw.state.drain_pending_disconnects(handler); + self.core.state.drain_pending_disconnects(handler); } - pub fn epoll_event(&mut self, event: &Event, handler: &mut F) + /// Processes one readiness event routed to this network. + /// + /// The event token must be in the range supplied to [`Self::new`]. This is + /// checked by an assertion in debug builds. + pub fn handle_event(&mut self, event: &Event, handler: &mut F) where F: for<'a> FnMut(TcpEvent<'a>), { - self.raw.state.handle_event(event, handler); + debug_assert!( + self.core.state.token_range.contains(&event.token().0), + "event token {:?} lies outside this network's token range {:?}", + event.token(), + self.core.state.token_range + ); + self.core.state.handle_event(event, handler); } } diff --git a/crates/flux-network/tests/tcp_network.rs b/crates/flux-network/tests/tcp_network.rs index 67470ee..9f5b509 100644 --- a/crates/flux-network/tests/tcp_network.rs +++ b/crates/flux-network/tests/tcp_network.rs @@ -728,3 +728,14 @@ fn tcp_network_client_is_wire_compatible_with_tcp_connector_server() { } assert!(contains(&network_messages, RESPONSE)); } + +#[test] +fn owned_poll_assigns_tokens_from_zero() { + let mut network = TcpNetwork::default(); + let group = network.add_group(TcpGroupConfig::default()); + let first = network.connect(group, unused_addr()); + assert_eq!(first, mio::Token(0)); + network.listen(group, unused_addr()).unwrap(); + let second = network.connect(group, unused_addr()); + assert_eq!(second, mio::Token(2), "the listener consumed token 1 from the same counter"); +} diff --git a/crates/flux-network/tests/tcp_network_external_poll.rs b/crates/flux-network/tests/tcp_network_external_poll.rs new file mode 100644 index 0000000..c1d1670 --- /dev/null +++ b/crates/flux-network/tests/tcp_network_external_poll.rs @@ -0,0 +1,251 @@ +use std::{ + cell::Cell, + net::{Ipv4Addr, SocketAddr}, + ops::Range, + time::{Duration, Instant}, +}; + +use flux_network::tcp::{TcpEvent, TcpGroupConfig, TcpNetworkWithExternalPoll}; +use mio::{Events, Poll, Token}; + +const SERVER_TOKENS: Range = 100..200; +const CLIENT_TOKENS: Range = 200..300; + +const SERVER_HELLO: &[u8] = b"server-hello"; +const CLIENT_HELLO: &[u8] = b"client-hello"; +const REQUEST: &[u8] = b"request-payload"; +const RESPONSE: &[u8] = b"response-payload"; +const AFTER_RECONNECT: &[u8] = b"after-reconnect"; + +/// The external-poll phase that invoked a handler. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Phase { + Pre, + Event, + Post, +} + +#[derive(Debug)] +enum Ev { + Accepted(Token), + Connected(Token), + Message(Token, Vec), + Disconnected(Token), +} + +#[derive(Debug)] +struct Record { + phase: Phase, + event: Ev, +} + +fn unused_addr() -> SocketAddr { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + addr +} + +fn accepts(log: &[Record]) -> Vec { + log.iter() + .filter_map(|r| if let Ev::Accepted(token) = r.event { Some(token) } else { None }) + .collect() +} + +fn connects(log: &[Record]) -> usize { + log.iter().filter(|r| matches!(r.event, Ev::Connected(_))).count() +} + +fn has_message(log: &[Record], payload: &[u8]) -> bool { + log.iter().any(|r| matches!(&r.event, Ev::Message(_, p) if p == payload)) +} + +fn message_token(log: &[Record], payload: &[u8]) -> Option { + log.iter().find_map(|r| match &r.event { + Ev::Message(token, p) if p == payload => Some(*token), + _ => None, + }) +} + +fn disconnect_phase(log: &[Record], token: Token) -> Option { + log.iter().find_map(|r| match r.event { + Ev::Disconnected(t) if t == token => Some(r.phase), + _ => None, + }) +} + +#[test] +#[allow(clippy::too_many_lines)] +fn one_poll_drives_two_networks_with_disjoint_ranges() { + let addr = unused_addr(); + let mut poll = Poll::new().unwrap(); + let mut events = Events::with_capacity(128); + + let mut server = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), SERVER_TOKENS); + let mut client = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), CLIENT_TOKENS); + + let server_group = server.add_group(TcpGroupConfig { + name: "server", + on_connect_msg: Some(SERVER_HELLO.to_vec()), + ..TcpGroupConfig::default() + }); + let client_group = client.add_group(TcpGroupConfig { + name: "client", + on_connect_msg: Some(CLIENT_HELLO.to_vec()), + reconnect_interval: flux_timing::Duration::from_millis(1), + ..TcpGroupConfig::default() + }); + + server.listen(server_group, addr).unwrap(); + let client_token = client.connect(client_group, addr); + assert!(CLIENT_TOKENS.contains(&client_token.0)); + + let phase = Cell::new(Phase::Pre); + let mut server_log: Vec = Vec::new(); + let mut client_log: Vec = Vec::new(); + + let mut request_sent = false; + let mut response_sent = false; + let mut disconnect_issued = false; + let mut resent = false; + let mut done = false; + + let deadline = Instant::now() + Duration::from_secs(10); + while !done && Instant::now() < deadline { + { + let mut on_server = |event: TcpEvent<'_>| { + let ev = match event { + TcpEvent::Accepted { group, token, .. } => { + assert_eq!(group, server_group); + Ev::Accepted(token) + } + TcpEvent::Message { group, token, payload, .. } => { + assert_eq!(group, server_group); + Ev::Message(token, payload.to_vec()) + } + TcpEvent::Disconnected { group, token, .. } => { + assert_eq!(group, server_group); + Ev::Disconnected(token) + } + TcpEvent::Connected { .. } => { + panic!("server network has no outbound endpoints") + } + }; + let token = match &ev { + Ev::Accepted(t) | + Ev::Message(t, _) | + Ev::Disconnected(t) | + Ev::Connected(t) => *t, + }; + assert!( + SERVER_TOKENS.contains(&token.0), + "server event token {token:?} out of range" + ); + server_log.push(Record { phase: phase.get(), event: ev }); + }; + let mut on_client = |event: TcpEvent<'_>| { + let ev = match event { + TcpEvent::Connected { group, token, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token, "reconnect must keep the endpoint token"); + Ev::Connected(token) + } + TcpEvent::Message { group, token, payload, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token); + Ev::Message(token, payload.to_vec()) + } + TcpEvent::Disconnected { group, token, .. } => { + assert_eq!(group, client_group); + assert_eq!(token, client_token); + Ev::Disconnected(token) + } + TcpEvent::Accepted { .. } => panic!("client network has no listeners"), + }; + client_log.push(Record { phase: phase.get(), event: ev }); + }; + + phase.set(Phase::Pre); + server.pre_poll(&mut on_server); + client.pre_poll(&mut on_client); + + poll.poll(&mut events, Some(Duration::from_millis(1))).unwrap(); + phase.set(Phase::Event); + for event in &events { + let token = event.token().0; + if SERVER_TOKENS.contains(&token) { + server.handle_event(event, &mut on_server); + } else if CLIENT_TOKENS.contains(&token) { + client.handle_event(event, &mut on_client); + } else { + panic!("event token {token} lies outside every network's range"); + } + } + + phase.set(Phase::Post); + server.post_poll(&mut on_server); + client.post_poll(&mut on_client); + } + + if !request_sent && connects(&client_log) >= 1 && has_message(&client_log, SERVER_HELLO) { + assert!(client.send_with(client_token, |buf| buf.extend_from_slice(REQUEST))); + request_sent = true; + } + if !response_sent && has_message(&server_log, REQUEST) { + let token = accepts(&server_log)[0]; + assert!(server.send_with(token, |buf| buf.extend_from_slice(RESPONSE))); + response_sent = true; + } + if !disconnect_issued && has_message(&client_log, RESPONSE) { + assert!(server.disconnect(accepts(&server_log)[0])); + disconnect_issued = true; + } + if disconnect_issued && + !resent && + accepts(&server_log).len() >= 2 && + connects(&client_log) >= 2 + { + assert!(client.send_with(client_token, |buf| buf.extend_from_slice(AFTER_RECONNECT))); + resent = true; + } + if resent && has_message(&server_log, AFTER_RECONNECT) { + done = true; + } + } + assert!(done, "the two-network exchange did not complete before the deadline"); + + // Verify the initial bidirectional exchange. + assert!(has_message(&server_log, CLIENT_HELLO)); + assert!(has_message(&client_log, SERVER_HELLO)); + assert!(has_message(&server_log, REQUEST)); + assert!(has_message(&client_log, RESPONSE)); + + // The reconnect uses the next server-side token and preserves the client + // token, which is checked in the handler. + let accepted = accepts(&server_log); + assert_eq!(accepted.len(), 2); + assert_eq!(accepted[1].0, accepted[0].0 + 1); + assert_eq!(message_token(&server_log, AFTER_RECONNECT), Some(accepted[1])); + + // A requested disconnect is reported by the next pre_poll; the peer close + // is reported by handle_event. + assert_eq!(disconnect_phase(&server_log, accepted[0]), Some(Phase::Pre)); + assert_eq!(disconnect_phase(&client_log, client_token), Some(Phase::Event)); +} + +#[test] +#[should_panic(expected = "tcp token range 100..102 exhausted")] +fn exhausted_token_range_panics_naming_the_range() { + let poll = Poll::new().unwrap(); + let mut network = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), 100..102); + let group = network.add_group(TcpGroupConfig::default()); + + network.listen(group, unused_addr()).unwrap(); + let second = network.connect(group, unused_addr()); + assert_eq!(second, Token(101), "tokens are assigned in ascending order from the range start"); + + let _ = network.connect(group, unused_addr()); +} From 46dc1af3d2e8ce3a552f44ce2aecaf728e362ba7 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Fri, 4 Sep 2026 15:31:06 +0100 Subject: [PATCH 3/4] test(network): cover external-poll dispatch Add a compiling example for TcpNetworkWithExternalPoll showing the pre_poll, poll, token dispatch, handle_event, and post_poll sequence. Add coverage for dispatching an event outside the network's configured token range. Debug builds reject it through the containment assertion; release builds emit no TcpEvent for the unknown token. Assisted-by: Claude:claude-fable-5 Assisted-by: Codex:gpt-5.6-sol --- Cargo.lock | 30 +++++++------- Cargo.toml | 2 +- crates/flux-network/src/tcp/network.rs | 41 ++++++++++++++++++- .../tests/tcp_network_external_poll.rs | 24 +++++++++++ 4 files changed, 80 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3049be1..06324c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -585,7 +585,7 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" [[package]] name = "flux" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bitcode", "core_affinity", @@ -609,7 +609,7 @@ dependencies = [ [[package]] name = "flux-communication" -version = "0.2.0" +version = "0.2.1" dependencies = [ "directories", "flux-timing", @@ -624,7 +624,7 @@ dependencies = [ [[package]] name = "flux-ctl" -version = "0.2.0" +version = "0.2.1" dependencies = [ "clap", "crossterm", @@ -643,7 +643,7 @@ dependencies = [ [[package]] name = "flux-disk" -version = "0.2.0" +version = "0.2.1" dependencies = [ "flux-timing", "flux-utils", @@ -655,7 +655,7 @@ dependencies = [ [[package]] name = "flux-network" -version = "0.2.0" +version = "0.2.1" dependencies = [ "flux", "flux-communication", @@ -674,7 +674,7 @@ dependencies = [ [[package]] name = "flux-profiler" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bytesize", "clap", @@ -692,7 +692,7 @@ dependencies = [ [[package]] name = "flux-profiler-macros" -version = "0.2.0" +version = "0.2.1" dependencies = [ "quote", "syn 2.0.117", @@ -700,7 +700,7 @@ dependencies = [ [[package]] name = "flux-timekeeper" -version = "0.2.0" +version = "0.2.1" dependencies = [ "auto_impl", "bitflags 2.10.0", @@ -715,7 +715,7 @@ dependencies = [ [[package]] name = "flux-timing" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bitcode", "chrono", @@ -733,7 +733,7 @@ dependencies = [ [[package]] name = "flux-utils" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bytes", "core_affinity", @@ -755,7 +755,7 @@ dependencies = [ [[package]] name = "flux-versioned-types" -version = "0.2.0" +version = "0.2.1" dependencies = [ "bincode", "flux", @@ -765,7 +765,7 @@ dependencies = [ [[package]] name = "flux-versioned-types-macros" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro2", "quote", @@ -1791,7 +1791,7 @@ dependencies = [ [[package]] name = "spine-derive" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro2", "quote", @@ -2034,11 +2034,11 @@ dependencies = [ [[package]] name = "type-hash" -version = "0.2.0" +version = "0.2.1" [[package]] name = "type-hash-derive" -version = "0.2.0" +version = "0.2.1" dependencies = [ "proc-macro-crate", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 1b30149..9d31d54 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ license = "Apache-2.0 AND MIT" publish = false repository = "https://github.com/gattaca-com/flux" rust-version = "1.91.0" -version = "0.2.0" +version = "0.2.1" [workspace.dependencies] diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index 6142a92..a783d0a 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -1110,7 +1110,7 @@ impl TcpNetworkCore { /// Serializes multiple payloads once and sends the batch to every /// connected member of `group`. Framing, size limits, and skipping of - /// invalid payloads follow [`TcpNetwork::send_many_with`]; each member + /// invalid payloads follow [`Self::send_many_with`]; each member /// receives the batch in one socket write when it has no backlog. The /// closure is not called when the group has no connected member. Returns /// the number of recipients attempted. @@ -1151,6 +1151,45 @@ impl TcpNetworkCore { /// [`Self::post_poll`]. The caller owns the poll, event buffer, and timeout. /// Dropping this value drops its sockets but does not explicitly deregister /// them. +/// +/// # Example +/// +/// ```no_run +/// use std::time::Duration; +/// +/// use flux_network::tcp::{TcpEvent, TcpGroupConfig, TcpNetworkWithExternalPoll}; +/// use mio::{Events, Poll}; +/// +/// // Every source sharing the poll must use non-overlapping token range. +/// const NETWORK_TOKENS: std::ops::Range = (1 << 48)..(2 << 48); +/// +/// let mut poll = Poll::new().unwrap(); +/// let mut events = Events::with_capacity(128); +/// let mut network = TcpNetworkWithExternalPoll::new( +/// poll.registry().try_clone().unwrap(), +/// NETWORK_TOKENS, +/// ); +/// let group = network.add_group(TcpGroupConfig::default()); +/// network.listen(group, "127.0.0.1:9099".parse().unwrap()).unwrap(); +/// +/// let mut handle_tcp_event = |_event: TcpEvent<'_>| { +/// // Process the event here. Copy payload bytes if they must outlive the +/// // callback. +/// }; +/// +/// loop { +/// network.pre_poll(&mut handle_tcp_event); +/// poll.poll(&mut events, Some(Duration::from_millis(1))).unwrap(); +/// for event in &events { +/// if NETWORK_TOKENS.contains(&event.token().0) { +/// network.handle_event(event, &mut handle_tcp_event); +/// } else { +/// // Route the event to another source registered with this poll. +/// } +/// } +/// network.post_poll(&mut handle_tcp_event); +/// } +/// ``` pub struct TcpNetworkWithExternalPoll { core: TcpNetworkCore, } diff --git a/crates/flux-network/tests/tcp_network_external_poll.rs b/crates/flux-network/tests/tcp_network_external_poll.rs index c1d1670..a94b899 100644 --- a/crates/flux-network/tests/tcp_network_external_poll.rs +++ b/crates/flux-network/tests/tcp_network_external_poll.rs @@ -249,3 +249,27 @@ fn exhausted_token_range_panics_naming_the_range() { let _ = network.connect(group, unused_addr()); } + +#[test] +#[cfg_attr(debug_assertions, should_panic(expected = "lies outside this network's token range"))] +fn foreign_token_trips_the_containment_assert() { + let mut poll = Poll::new().unwrap(); + let mut network = + TcpNetworkWithExternalPoll::new(poll.registry().try_clone().unwrap(), 100..200); + + // Register a readiness source outside the network's token range. + let mut listener = mio::net::TcpListener::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + poll.registry().register(&mut listener, Token(7), mio::Interest::READABLE).unwrap(); + let _client = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + + let mut events = Events::with_capacity(4); + let deadline = Instant::now() + Duration::from_secs(5); + while events.is_empty() && Instant::now() < deadline { + poll.poll(&mut events, Some(Duration::from_millis(10))).unwrap(); + } + let event = events.iter().next().expect("listener readiness did not arrive"); + assert_eq!(event.token(), Token(7)); + + // Debug builds reject the token; release builds emit no TcpEvent for it. + network.handle_event(event, &mut |_| panic!("no TcpEvent expected for a foreign token")); +} From 505759c781669ae01710f79dd681a52dd6649f5b Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Fri, 4 Sep 2026 16:26:39 +0100 Subject: [PATCH 4/4] fix(network): deregister sockets when network state drops Attempt to deregister registered connection sockets and listeners when NetworkState is dropped. This matters for a caller-owned poll because, on Linux, a duplicated descriptor can keep an epoll registration alive after the original descriptor closes. Order TcpNetwork's core before its poll so network cleanup runs while the owned poll is still alive. Add a Linux-only regression test that keeps a duplicated listener descriptor open and verifies that dropping the state removes its listener and endpoint registrations. Assisted-by: Claude:claude-fable-5 Assisted-by: Codex:gpt-5.6-sol --- crates/flux-network/src/tcp/network.rs | 63 ++++++++++++++++++++++++-- 1 file changed, 59 insertions(+), 4 deletions(-) diff --git a/crates/flux-network/src/tcp/network.rs b/crates/flux-network/src/tcp/network.rs index a783d0a..cb9175f 100644 --- a/crates/flux-network/src/tcp/network.rs +++ b/crates/flux-network/src/tcp/network.rs @@ -953,6 +953,19 @@ impl NetworkState { } } +impl Drop for NetworkState { + fn drop(&mut self) { + // Deregister before closing. On Linux, a duplicated descriptor can + // keep an epoll registration alive after the original descriptor closes. + for index in 0..self.connections.len() { + self.close_connection_socket(index); + } + for listener in &mut self.listeners { + let _ = self.registry.deregister(&mut listener.socket); + } + } +} + /// A grouped collection of TCP listeners and persistent outbound endpoints /// with an internal nonblocking poll. /// @@ -961,8 +974,8 @@ impl NetworkState { /// caller-owned poll instead. pub struct TcpNetwork { events: Events, - poll: Poll, core: TcpNetworkCore, + poll: Poll, } impl Default for TcpNetwork { @@ -972,8 +985,8 @@ impl Default for TcpNetwork { // This network owns the poll, so its token range starts at zero. Self { events: Events::with_capacity(EVENTS_CAPACITY), - poll, core: TcpNetworkCore::new(registry, 0..usize::MAX), + poll, } } } @@ -1149,8 +1162,7 @@ impl TcpNetworkCore { /// A polling pass consists of [`Self::pre_poll`], polling, calling /// [`Self::handle_event`] for each event in this network's token range, and /// [`Self::post_poll`]. The caller owns the poll, event buffer, and timeout. -/// Dropping this value drops its sockets but does not explicitly deregister -/// them. +/// Dropping this value attempts to deregister its sockets before closing them. /// /// # Example /// @@ -1815,4 +1827,47 @@ mod tests { let mut payload = PayloadBuf::new(&mut bytes); payload.resize(usize::MAX, 0); } + + #[cfg(target_os = "linux")] + #[test] + fn drop_removes_listener_and_endpoint_registrations() { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + + use flux_timing::{Duration, Repeater}; + + use super::{GroupState, NetworkState, TcpGroup}; + + fn registered_fds(epoll_fd: i32) -> usize { + std::fs::read_to_string(format!("/proc/self/fdinfo/{epoll_fd}")) + .unwrap() + .lines() + .filter(|line| line.starts_with("tfd:")) + .count() + } + + let poll = Poll::new().unwrap(); + let epoll_fd = poll.as_raw_fd(); + let registry = poll.registry().try_clone().unwrap(); + let mut state = NetworkState::new(registry, 100..200); + state.groups.push(GroupState { + config: TcpGroupConfig::default(), + reconnector: Repeater::every(Duration::from_secs(2)), + }); + let group = TcpGroup(0); + state.listen(group, (Ipv4Addr::LOCALHOST, 0).into()).unwrap(); + let addr = state.listeners[0].socket.local_addr().unwrap(); + let _endpoint = state.connect(group, addr); + assert_eq!(registered_fds(epoll_fd), 2); + + // Keep the listener's file description alive after its mio socket drops. + // SAFETY: the listener owns this descriptor for the duration of the call. + let dup_fd = unsafe { libc::dup(state.listeners[0].socket.as_raw_fd()) }; + assert!(dup_fd >= 0); + // SAFETY: dup_fd is valid after the check above, and ownership is + // transferred exactly once. + let _dup = unsafe { OwnedFd::from_raw_fd(dup_fd) }; + + drop(state); + assert_eq!(registered_fds(epoll_fd), 0); + } }