diff --git a/lib/src/network/gateway.rs b/lib/src/network/gateway.rs index 4548a47b5..8e4693c72 100644 --- a/lib/src/network/gateway.rs +++ b/lib/src/network/gateway.rs @@ -102,6 +102,12 @@ impl Gateway { self.stacks.borrow().connectivity() } + /// Whether any currently bound stack has hit a fatal error and stopped working. See + /// `Stacks::any_stack_is_dead`. + pub fn any_stack_is_dead(&self) -> bool { + self.stacks.borrow().any_stack_is_dead() + } + pub async fn connect_with_retries(&self, peer: &SeenPeer) -> Option { let create_backoff = || { ExponentialBackoffBuilder::new() @@ -299,6 +305,14 @@ impl Stacks { Connectivity::infer(&self.addresses()) } + /// Whether any currently bound stack has hit a fatal error and stopped working (e.g. the + /// network interface going down while the app is backgrounded). Such a stack never recovers + /// on its own - it must be rebound, even if the address it was bound to hasn't changed. + fn any_stack_is_dead(&self) -> bool { + self.quic_v4.as_ref().is_some_and(QuicStack::is_dead) + || self.quic_v6.as_ref().is_some_and(QuicStack::is_dead) + } + fn allows_connection_to(&self, addr: &PeerAddr) -> bool { let has_connector = match addr { PeerAddr::Tcp(SocketAddr::V4(_)) => self.tcp_v4.is_some(), @@ -495,6 +509,10 @@ impl QuicStack { self.listener_task.abort(); self.connector.close(); } + + fn is_dead(&self) -> bool { + self.connector.is_dead() + } } struct TcpStack { diff --git a/lib/src/network/mod.rs b/lib/src/network/mod.rs index ec0212fbf..9870ea778 100644 --- a/lib/src/network/mod.rs +++ b/lib/src/network/mod.rs @@ -698,7 +698,14 @@ impl Inner { let bind = StackAddresses::from(bind); // TODO: Would be preferable to only rebind those stacks that actually need rebinding. - if !self.gateway.addresses().any_stack_needs_rebind(&bind) { + // + // Also force a rebind if a currently bound stack has died (e.g. its UDP socket hit a + // fatal error because the network interface went down while the app was backgrounded): + // such a stack never recovers on its own, and comparing addresses alone would otherwise + // treat "bind to the same address again" as a no-op forever. + if !self.gateway.addresses().any_stack_needs_rebind(&bind) + && !self.gateway.any_stack_is_dead() + { return; } diff --git a/net/src/quic.rs b/net/src/quic.rs index 1711ae3c5..24cca7c80 100644 --- a/net/src/quic.rs +++ b/net/src/quic.rs @@ -16,7 +16,10 @@ use std::{ io, net::SocketAddr, pin::Pin, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, task::{Context, Poll}, }; use tokio::sync::{ @@ -37,6 +40,12 @@ const DISABLE_SEGMENTATION_OFFLOAD: bool = false; //------------------------------------------------------------------------------ pub struct Connector { endpoint: quinn::Endpoint, + // Set by the underlying `CustomUdpSocket` when it hits a fatal (non-`WouldBlock`) recv + // error, e.g. the network interface going down while the app is backgrounded. Once that + // happens, quinn's endpoint driver treats the socket as unusable and nothing in quinn itself + // ever rebinds it, so callers must check this and recreate the whole stack (socket + + // endpoint) to recover - see `Gateway`/`Stacks` in ouisync-app's network layer. + socket_died: Arc, } impl Connector { @@ -59,6 +68,12 @@ impl Connector { pub fn close(&self) { self.endpoint.close(quinn::VarInt::from_u32(0), &[]); } + + /// Whether the underlying UDP socket has hit a fatal error and stopped working. The + /// endpoint/connector need to be recreated (via a fresh call to `configure`) to recover. + pub fn is_dead(&self) -> bool { + self.socket_died.load(Ordering::Acquire) + } } //------------------------------------------------------------------------------ @@ -158,6 +173,8 @@ pub fn configure( let custom_socket = Arc::new(CustomUdpSocket::bind(bind_addr, options)?); let side_channel_maker = custom_socket.clone().side_channel_maker(); + let socket_died = custom_socket.died.clone(); + let mut endpoint = quinn::Endpoint::new_with_abstract_socket( quinn::EndpointConfig::default(), Some(server_config), @@ -171,6 +188,7 @@ pub fn configure( let connector = Connector { endpoint: endpoint.clone(), + socket_died, }; let acceptor = Acceptor { endpoint, @@ -332,6 +350,7 @@ struct CustomUdpSocket { io: tokio::net::UdpSocket, state: quinn::udp::UdpSocketState, side_channel_tx: broadcast::Sender, + died: Arc, } impl CustomUdpSocket { @@ -345,6 +364,7 @@ impl CustomUdpSocket { io: tokio::net::UdpSocket::from_std(socket)?, state, side_channel_tx: broadcast::channel(MAX_SIDE_CHANNEL_PENDING_PACKETS).0, + died: Arc::new(AtomicBool::new(false)), }) } @@ -374,8 +394,12 @@ impl quinn::AsyncUdpSocket for CustomUdpSocket { metas: &mut [quinn::udp::RecvMeta], ) -> Poll> { loop { - ready!(self.io.poll_recv_ready(cx))?; - if let Ok(res) = self.io.try_io(Interest::READABLE, || { + if let Err(e) = ready!(self.io.poll_recv_ready(cx)) { + self.died.store(true, Ordering::Release); + return Poll::Ready(Err(e)); + } + + match self.io.try_io(Interest::READABLE, || { let res = self.state.recv((&self.io).into(), bufs, metas); if let Ok(msg_count) = res { @@ -384,7 +408,23 @@ impl quinn::AsyncUdpSocket for CustomUdpSocket { res }) { - return Poll::Ready(Ok(res)); + Ok(res) => return Poll::Ready(Ok(res)), + // `WouldBlock` means the readiness we just observed was stale (the socket isn't + // actually readable yet) - `try_io` already cleared it, so looping back to + // `poll_recv_ready` is correct: it'll wait for the next real readiness event. + Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue, + // Any other error (e.g. the network interface going down while the app is + // backgrounded) is a real failure, not a spurious wakeup. `try_io` doesn't clear + // readiness for these, so looping back to `poll_recv_ready` would immediately + // observe the socket as "ready" again and re-enter `recv` synchronously, + // spinning this task at 100% CPU forever instead of ever yielding. Propagate it + // instead so the caller can handle/recreate the connection. Mark the socket dead + // so the owner knows this endpoint needs to be rebuilt - quinn never does this on + // its own once its driver sees a fatal error. + Err(e) => { + self.died.store(true, Ordering::Release); + return Poll::Ready(Err(e)); + } } } }