Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions lib/src/network/gateway.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Connection> {
let create_backoff = || {
ExponentialBackoffBuilder::new()
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -495,6 +509,10 @@ impl QuicStack {
self.listener_task.abort();
self.connector.close();
}

fn is_dead(&self) -> bool {
self.connector.is_dead()
}
}

struct TcpStack {
Expand Down
9 changes: 8 additions & 1 deletion lib/src/network/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand Down
48 changes: 44 additions & 4 deletions net/src/quic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,10 @@ use std::{
io,
net::SocketAddr,
pin::Pin,
sync::Arc,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
task::{Context, Poll},
};
use tokio::sync::{
Expand All @@ -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<AtomicBool>,
}

impl Connector {
Expand All @@ -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)
}
}

//------------------------------------------------------------------------------
Expand Down Expand Up @@ -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),
Expand All @@ -171,6 +188,7 @@ pub fn configure(

let connector = Connector {
endpoint: endpoint.clone(),
socket_died,
};
let acceptor = Acceptor {
endpoint,
Expand Down Expand Up @@ -332,6 +350,7 @@ struct CustomUdpSocket {
io: tokio::net::UdpSocket,
state: quinn::udp::UdpSocketState,
side_channel_tx: broadcast::Sender<Packet>,
died: Arc<AtomicBool>,
}

impl CustomUdpSocket {
Expand All @@ -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)),
})
}

Expand Down Expand Up @@ -374,8 +394,12 @@ impl quinn::AsyncUdpSocket for CustomUdpSocket {
metas: &mut [quinn::udp::RecvMeta],
) -> Poll<io::Result<usize>> {
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 {
Expand All @@ -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));
}
}
}
}
Expand Down