diff --git a/docs/adr/ADR-013-relay-tunnel-teardown-ordering.md b/docs/adr/ADR-013-relay-tunnel-teardown-ordering.md new file mode 100644 index 00000000..69122892 --- /dev/null +++ b/docs/adr/ADR-013-relay-tunnel-teardown-ordering.md @@ -0,0 +1,197 @@ +# ADR-013: Relay Tunnel Teardown Ordering and Failure Classification + +## Status + +Proposed (2026-08-19) + +## Context + +A proactive relay allocation owns two coupled resources: a MASQUE tunnel +(`MasqueRelaySocket` and its reader, writer and keepalive tasks) and a second +Quinn endpoint that uses that tunnel as its `AsyncUdpSocket`. The endpoint's +inbound path exists only for as long as the tunnel does. + +saorsa-core gates relay publication behind a canary quorum, so an allocation may +be prepared and then discarded without ever being published, and PR #131 made +that discard deterministic by aborting and joining the tunnel tasks. Between +them, relay teardown went from a rare event to a routine one: 1,773 teardowns +across 73 hosts in the 48 h after the 0.36.0 rollout, tracking node restarts. + +That exposed two defects in the teardown path (V2-986), both invisible while +teardown was rare. + +### 1. Ordering + +`teardown` closed the endpoint, then destroyed the tunnel, then waited for the +endpoint to drain. `Endpoint::close` only *queues* a `ConnectionEvent::Close` per +connection; the frames still have to travel out through the tunnel. Destroying +the tunnel first removed the transport those frames needed. The connection driver +then handled the queued `Close`, immediately hit channel EOF because +`EndpointDriver::drop` had cleared `connections.senders`, and exited with +"endpoint driver future was dropped" *before* `drive_transmit`. The close frame +was built and never sent, and the peer was left to time out. The connection's +local close reason was overwritten by the internal transport error for the same +reason. + +The `wait_idle` that followed was measuring nothing: it defines idle as an empty +sender map, which is exactly the map the dying driver had just cleared. + +### 2. Classification + +`MasqueRelaySocket::poll_recv` reported the recv channel closing as +`io::ErrorKind::BrokenPipe`, with no way to tell "the relay stream broke" from +"we aborted the reader task on purpose". Quinn treats a `poll_recv` error as a +socket failure: `EndpointDriver` resolves to `Err`, which is logged at ERROR and +whose `Drop` sets `driver_lost` and clears the connection senders instead of +letting the endpoint retire through its refcount-reaches-zero path. Every +requested teardown therefore produced +`ERROR ... I/O error: relay recv stream closed`: 2,338 occurrences in 48 h across +73 hosts, roughly 100x the previous rate. + +Unexpected tunnel loss produces the identical line, so the message alone cannot +classify the fleet's occurrences; correlation with relay lifecycle logs puts at +least 56% of them on the requested-teardown path. + +## Decision + +**A teardown we requested is not a transport failure, and the endpoint drains +before the tunnel is dismantled.** + +`ProactiveRelay::teardown` becomes `close` → bounded 1 s `wait_idle` → +`tunnel.shutdown()` → release the relay session. The reorder is what lets the +queued CONNECTION_CLOSE frames leave, and it makes `wait_idle` measure a real +drain. + +`RelayTunnelControl` carries a `TunnelCause` of `Live`, `Failed` or +`ShutdownRequested` in an `AtomicU8` shared with the `MasqueRelaySocket` it owns, +settled by compare-exchange so the **first** transition out of `Live` wins. +`poll_recv`, on a closed recv channel with nothing buffered, returns +`Poll::Pending` for `ShutdownRequested` and the existing `BrokenPipe` otherwise. +The tunnel-death watcher likewise stays quiet for a requested teardown. + +First-cause rather than a "shutdown was called" flag is load-bearing: cleanup +routinely arrives *after* a tunnel has broken, so last-writer-wins would let it +reclassify the failure that triggered it as intentional. For the same reason the +relay health monitor gets its own abort path +(`abort_unhealthy_proactive_relay`), which marks the tunnel failed before tearing +it down. Without that, `is_relay_healthy` can condemn a relay from the state of +the *outer* relay session while the tunnel's own cause is still `Live`, and the +teardown would settle `ShutdownRequested` first. + +That verdict now names the relay it was reached about +(`unhealthy_published_relay`). Asking whether the relay is healthy and then +asking separately which relay is published are two awaits, and a replacement can +publish between them, in which case the monitor would tear down the healthy +replacement on the strength of a verdict about its predecessor. + +Parking `poll_recv` does not strand the endpoint driver. Nothing can arrive on a +torn-down tunnel, and the driver keeps its other wake sources: the endpoint-event +channel, a connection's `Drained` event, and the explicit wake +`EndpointRef::drop` issues at refcount zero. So it still completes with `Ok(())` +once the endpoint is dropped. + +One consequence has to be handled. The driver crashing was what released a +connection parked waiting for send capacity; it no longer crashes, so +`abort_tasks` wakes `send_capacity_freed` itself, and `TunnelPoller` consults a +`writer_stopped` flag alongside the channel state. Both halves are needed: +`JoinHandle::abort` is asynchronous, so the channel is typically still open when +that wake arrives, and a poller that looked only at the channel would see it full +and open, consume the notification, re-park, and never be woken again. Whenever +the poller answers "writable", `enqueue_outbound` must not answer `WouldBlock`, +because Quinn retries that immediately and without yielding. So a full queue +whose writer has stopped drops the datagram instead. + +`writer_stopped` is tracked independently of the cause because the relay's two +stream halves are independent: a peer that resets only its server-to-client half +ends the tunnel while the writer remains able to flush what is queued. It is +recorded by a `WriterExit` guard held by the writer future, so it is set on every +path that ends the writer, including an abort that lands before the future's +first poll. + +The cause, the writer state and both `Notify`s live in one `TunnelState` that the +control, the socket and the tunnel tasks all hold strongly. Reaching back through +a `Weak` would not do: the dial-through path in +`p2p_endpoint` drops its control as soon as the dial completes, while the socket +and its tasks live on, and a writer that could not record its exit there would +leave a parked poller waiting forever. + +## Consequences + +### Benefits + +- A requested teardown no longer produces an endpoint-driver I/O ERROR; + unrequested tunnel loss still does. +- Peers reached through a discarded relay address get a real opportunity to + receive CONNECTION_CLOSE and fail fast, instead of waiting out an idle timeout. +- The relay endpoint retires through the same path as any other endpoint, so + `wait_idle` measures a real drain. +- Relay churn stays observable on the `info!` lifecycle logs that describe it + accurately ("Proactive relay allocation prepared" / "Proactive relay torn + down"). + +### Trade-offs + +- Teardown may spend up to its 1 s drain budget before it begins releasing the + relay server's capacity slot, since it is `shutdown()` closing the tunnel + streams that lets the server observe the release. `NatTraversalEndpoint::shutdown` + inherits the same bound before it closes ordinary connections. +- CONNECTION_CLOSE delivery is improved, not guaranteed. `Endpoint::close` uses + `try_send` and drops the close event if a connection mailbox is full; + `wait_idle` waits for Quinn's connection map to empty rather than for the + MASQUE send queue to flush; `shutdown()` aborts the writer regardless of what + is still queued; and Quinn does not consider a locally closed connection + drained until `3 × PTO`, which 1 s need not cover. +- A connection whose close event was dropped is now left to its 30 s idle timeout + rather than killed outright by the driver crash. +- A requested teardown no longer produces a transport-level receive signal. This + is deliberate, and is why the change is paired with tests asserting that + unrequested tunnel loss still fails `poll_recv`, is still logged, and is not + reclassified by cleanup arriving afterwards. +- `ProactiveRelay::drop`, the forced-cleanup fallback that already logs a + warning, cannot await. It still closes and calls `shutdown_now()` without + draining. That path is reached only when an allocation is dropped without going + through `teardown`, which is what the existing warning is for. +- A tunnel that breaks on its own between `close()` and `shutdown()` settles as + `Failed`, so `poll_recv` reports it. Correct, but it means the log line can + still appear during an otherwise orderly teardown. + +### Risks + +- If some future caller shuts a tunnel down and keeps using its endpoint, that + endpoint goes quiet instead of erroring. `TunnelCause` is reachable only + through `RelayTunnelControl`, whose callers are the relay lifecycle paths in + `nat_traversal_api`, so the blast radius is bounded to code that already + intends the socket to be dead. + +## Alternatives Considered + +- **Downgrade the log line at the driver.** Rejected: the log site is generic + code shared by every endpoint, and it would suppress genuine `BrokenPipe` + failures from ordinary UDP sockets too. +- **Return `Poll::Ready(Ok(0))` instead of parking.** Rejected: `poll_socket` + re-polls immediately, records no work, and self-schedules, so it spins. +- **Reuse `is_transient_recv_error`.** Rejected for the same reason: that + classifier makes `poll_socket` `continue`. +- **Fix only the ordering.** Rejected: the endpoint still holds live references + (the accept loop's, and `ProactiveRelay`'s own) when the tunnel is finally + destroyed, so the driver would still observe the socket disappear and still + fail. +- **A boolean "shutdown was requested" flag.** Rejected: cleanup arriving after a + failure would silence it. Hence first-cause `TunnelCause`. +- **Classify `try_send` errors too.** Rejected: `ConnectionDriver` already treats + every non-`WouldBlock` send error as ordinary packet loss (rate-limited + `warn!`, datagram dropped, driver untouched), so there is no fatal path to + spare. + +## References + +- Linear V2-986: "relay recv stream closed" transport errors up ~100x fleet-wide +- PR #131: identity-scoped relay publication. Introduced `RelayTunnelControl` and + deterministic teardown, first released in 0.36.0. +- saorsa-core ADR-016: canary-gated proactive relays +- `src/masque/relay_socket.rs`: `RelayTunnelControl`, `TunnelCause`, + `WriterExit`, `MasqueRelaySocket::poll_recv` +- `src/nat_traversal_api.rs`: `ProactiveRelay::teardown`, + `abort_unhealthy_proactive_relay` +- `src/high_level/endpoint.rs`: `EndpointDriver::poll`, `State::drive_recv`, + `EndpointRef::drop` diff --git a/docs/adr/README.md b/docs/adr/README.md index 15353676..20368276 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -22,6 +22,7 @@ ADRs document significant architectural decisions made in the project. Each reco | [ADR-010](ADR-010-repository-ownership.md) | Repository Ownership under WithAutonomi | Accepted | 2026-05-29 | | [ADR-011](ADR-011-stable-relay-port-reservations.md) | Stable Relay Port Reservations (Authenticated, Leased) | Proposed | 2026-06-24 | | [ADR-012](ADR-012-keep-alive-dial-accept-split.md) | Keep-Alive on the Dialling Side Only | Proposed | 2026-08-14 | +| [ADR-013](ADR-013-relay-tunnel-teardown-ordering.md) | Relay Tunnel Teardown Ordering and Failure Classification | Proposed | 2026-08-19 | | [ADR-014](ADR-014-accept-side-keep-alive-backstop.md) | Accept-Side Keep-Alive Backstop | Proposed | 2026-08-20 | ## ADR Template diff --git a/src/masque/relay_socket.rs b/src/masque/relay_socket.rs index 721c6abb..c1edde07 100644 --- a/src/masque/relay_socket.rs +++ b/src/masque/relay_socket.rs @@ -45,7 +45,7 @@ use std::future::Future; use std::io::{self, IoSliceMut}; use std::net::SocketAddr; use std::pin::Pin; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, Weak}; use std::task::{Context, Poll}; use std::time::Duration; @@ -107,25 +107,139 @@ pub struct RawRelayStreams { /// Aborting the reader and writer tasks drops their QUIC stream halves, which /// promptly tells the relay server to close the associated MASQUE session and /// release its capacity slot. Shutdown is idempotent. +/// +/// A tunnel can end because it broke or because we dismantled it, and the layers +/// above need to tell those apart: the first is a transport failure the backing +/// endpoint must hear about, the second is routine. [`is_closed`](Self::is_closed) +/// is true either way; [`shutdown_requested`](Self::shutdown_requested) only for +/// the second. #[derive(Debug)] pub(crate) struct RelayTunnelControl { tasks: PlMutex>>, + state: Arc, +} + +/// The tunnel facts that outlive any one owner. +/// +/// Held by the control, by the [`MasqueRelaySocket`] it owns, and by the tunnel +/// tasks. Sharing it strongly rather than reaching back through a +/// `Weak` matters: the dial-through path in `p2p_endpoint` +/// drops its control as soon as the dial completes while the socket and its +/// tasks live on, and a task that could not record its exit there would leave a +/// parked poller waiting forever. +#[derive(Debug)] +struct TunnelState { + /// Why the tunnel stopped carrying traffic. Holds a [`TunnelCause`]. + cause: AtomicU8, + /// Woken once `cause` settles. closed: Notify, - is_closed: AtomicBool, + /// Whether the writer has stopped, so nothing more can leave the tunnel. + /// Distinct from the tunnel being closed: the relay's two stream halves are + /// independent, so a peer that resets only its server-to-client half leaves + /// the writer able to flush a queued CONNECTION_CLOSE. + writer_stopped: AtomicBool, + /// Woken when the outbound queue frees a slot, and when nothing will ever + /// free one again. A [`TunnelPoller`] parked on a full send queue is + /// normally released by the writer draining a slot; when the writer is + /// aborted instead, shutdown and the writer's own guard wake it here. + send_capacity_freed: Notify, } -impl RelayTunnelControl { +/// Why a tunnel stopped carrying traffic. +/// +/// Settled by compare-exchange, so the **first** transition out of +/// [`TunnelCause::Live`] wins. A tunnel that broke and was then cleaned up stays +/// classified as a failure; otherwise cleanup arriving a moment later would +/// silence the fault that triggered it. +/// +/// A `u8` so it can live in an `AtomicU8` alongside the rest of [`TunnelState`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +enum TunnelCause { + Live = 0, + Failed = 1, + ShutdownRequested = 2, +} + +impl TunnelState { fn new() -> Arc { Arc::new(Self { - tasks: PlMutex::new(Vec::new()), + cause: AtomicU8::new(TunnelCause::Live as u8), closed: Notify::new(), - is_closed: AtomicBool::new(false), + writer_stopped: AtomicBool::new(false), + send_capacity_freed: Notify::new(), + }) + } + + /// Settle the terminal cause, if it has not already settled. + /// + /// Losing the race is normal and not an error: it means something else ended + /// the tunnel first, and that first cause is the true one. + fn settle(&self, cause: TunnelCause) { + if self + .cause + .compare_exchange( + TunnelCause::Live as u8, + cause as u8, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + self.closed.notify_waiters(); + } + } + + fn is_closed(&self) -> bool { + self.cause.load(Ordering::Acquire) != TunnelCause::Live as u8 + } + + fn shutdown_requested(&self) -> bool { + self.cause.load(Ordering::Acquire) == TunnelCause::ShutdownRequested as u8 + } + + fn writer_stopped(&self) -> bool { + self.writer_stopped.load(Ordering::Acquire) + } + + /// Record that the writer has stopped and release anything waiting on it. + /// + /// The cause settles first so that a writer which broke on its own is filed + /// as a failure before a concurrent teardown can claim the tunnel as + /// intentionally dismantled. + fn mark_writer_stopped(&self, cause: TunnelCause) { + self.settle(cause); + self.writer_stopped.store(true, Ordering::Release); + self.send_capacity_freed.notify_waiters(); + } +} + +/// Records the writer's exit exactly once, on whatever path ends it. +/// +/// Held by the writer future, so it also runs when the task is aborted, even +/// before its first poll, since the guard is created before the spawn. +struct WriterExit(Arc); + +impl Drop for WriterExit { + fn drop(&mut self) { + // `Failed` loses to a cause already settled, which is the point: a + // writer aborted by shutdown must not overwrite `ShutdownRequested`, and + // one aborted after a reader failure must not overwrite `Failed`. + self.0.mark_writer_stopped(TunnelCause::Failed); + } +} + +impl RelayTunnelControl { + fn new(state: Arc) -> Arc { + Arc::new(Self { + tasks: PlMutex::new(Vec::new()), + state, }) } #[cfg(test)] pub(crate) fn detached() -> Arc { - Self::new() + Self::new(TunnelState::new()) } fn register(&self, handle: tokio::task::JoinHandle<()>) { @@ -142,15 +256,23 @@ impl RelayTunnelControl { } } - fn mark_closed(&self) { - if !self.is_closed.swap(true, Ordering::AcqRel) { - self.closed.notify_waiters(); - } + /// Record that the tunnel broke rather than being dismantled on request. + /// + /// Called by the tunnel tasks when their stream fails, and by the relay + /// health monitor before it tears down a relay it has found dead. + pub(crate) fn mark_failed(&self) { + self.state.settle(TunnelCause::Failed); } /// Returns whether the tunnel has failed or has been explicitly shut down. pub(crate) fn is_closed(&self) -> bool { - self.is_closed.load(Ordering::Acquire) + self.state.is_closed() + } + + /// Returns whether this tunnel was dismantled by a local shutdown request + /// rather than by a transport failure. + pub(crate) fn shutdown_requested(&self) -> bool { + self.state.shutdown_requested() } /// Wait until the tunnel reader exits or shutdown is requested. @@ -159,7 +281,7 @@ impl RelayTunnelControl { if self.is_closed() { return; } - let notified = self.closed.notified(); + let notified = self.state.closed.notified(); if self.is_closed() { return; } @@ -185,7 +307,18 @@ impl RelayTunnelControl { } fn abort_tasks(&self) -> Vec> { - self.mark_closed(); + // Record everything before aborting anything. `settle` releases the + // tunnel-death watcher and unblocks `poll_recv`, and the writer state + // releases a parked `TunnelPoller`; all three must already be able to + // see that this teardown was asked for rather than suffered. + // + // The writer's own `WriterExit` guard does the same, but only once the + // abort lands, and `JoinHandle::abort` is asynchronous. Doing it here + // means a parked poller is not waiting on a cancellation to complete: it + // re-checks the writer state, not the send channel, which is typically + // still open at this point. + self.state + .mark_writer_stopped(TunnelCause::ShutdownRequested); let handles = { let mut tasks = self.tasks.lock(); std::mem::take(&mut *tasks) @@ -197,12 +330,6 @@ impl RelayTunnelControl { } } -fn mark_writer_exit(control: &Weak) { - if let Some(control) = control.upgrade() { - control.mark_closed(); - } -} - /// A virtual UDP socket backed entirely by a MASQUE relay tunnel. /// /// All traffic — both outgoing and incoming — flows through the relay @@ -222,12 +349,6 @@ pub struct MasqueRelaySocket { /// Bounded channel for outbound packets (drained by the background /// writer task into the relay send stream). send_tx: mpsc::Sender, - /// Notified once after every item the writer task drains from - /// `send_tx`. Pollers parked on a full queue re-check capacity - /// after each notification. `notify_one` is used (not - /// `notify_waiters`) so a drain that races with a poller entering - /// the wait state stores a permit, avoiding lost wakeups. - send_capacity_freed: Arc, /// Per-target maximum payload size enforced by [`Self::try_send`], /// populated by [`TunnelControlFrame::PmtuUpdate`] frames decoded by /// the reader task. When a destination has an entry, any @@ -237,6 +358,11 @@ pub struct MasqueRelaySocket { /// converges to the true egress path MTU. Targets without an /// entry are unconstrained by this layer (Quinn governs sizing). target_mtu: Arc>, + /// The tunnel facts, shared with the [`RelayTunnelControl`] that owns the + /// tunnel tasks. `poll_recv` reads the cause to tell "we dismantled this" + /// from "this broke", and the poller reads the writer state to know whether + /// waiting for send capacity is still worth anything. + state: Arc, /// The original socket is kept alive so the relay connection's own /// QUIC traffic (keepalives, ACKs, stream data) continues to flow /// directly. Without this reference the OS may reclaim the socket. @@ -286,8 +412,8 @@ impl MasqueRelaySocket { ) -> (Arc, Arc) { let (send_tx, mut send_rx) = mpsc::channel::(SEND_QUEUE_CAPACITY); let (recv_tx, recv_rx) = mpsc::channel::<(Bytes, SocketAddr)>(RECV_QUEUE_CAPACITY); - let control = RelayTunnelControl::new(); - let send_capacity_freed = Arc::new(Notify::new()); + let state = TunnelState::new(); + let control = RelayTunnelControl::new(Arc::clone(&state)); let target_mtu: Arc> = Arc::new(DashMap::new()); let target_mtu_reader = Arc::clone(&target_mtu); @@ -296,7 +422,7 @@ impl MasqueRelaySocket { relay_public_addr, recv_rx: PlMutex::new(recv_rx), send_tx: send_tx.clone(), - send_capacity_freed: Arc::clone(&send_capacity_freed), + state: Arc::clone(&state), target_mtu, _original_socket: original_socket, }); @@ -403,15 +529,18 @@ impl MasqueRelaySocket { // Signal the owner before the endpoint driver is dropped so it // can gracefully close connections accepted through the tunnel. if let Some(control) = weak_control.upgrade() { - control.mark_closed(); + control.mark_failed(); } }); control.register(reader_handle); // Background task: write queued outbound packets to relay stream. - let writer_capacity = Arc::clone(&send_capacity_freed); - let writer_control = Arc::downgrade(&control); + let writer_capacity = Arc::clone(&state); + // Created before the spawn so an abort that lands before the future's + // first poll still records the exit. + let writer_exit = WriterExit(Arc::clone(&state)); let writer_handle = tokio::spawn(async move { + let _writer_exit = writer_exit; while let Some(encoded) = send_rx.recv().await { // `recv` completing means the channel just freed a // slot. Wake any poller parked on full-queue @@ -422,7 +551,7 @@ impl MasqueRelaySocket { let mut batch = Vec::with_capacity(encoded.len().saturating_add(std::mem::size_of::())); append_relay_frame(&mut batch, &encoded); - writer_capacity.notify_one(); + writer_capacity.send_capacity_freed.notify_one(); let mut frames = 1usize; while frames < RELAY_STREAM_BATCH_MAX_FRAMES @@ -431,7 +560,7 @@ impl MasqueRelaySocket { match send_rx.try_recv() { Ok(next) => { append_relay_frame(&mut batch, &next); - writer_capacity.notify_one(); + writer_capacity.send_capacity_freed.notify_one(); frames += 1; } Err(mpsc::error::TryRecvError::Empty) => break, @@ -445,15 +574,10 @@ impl MasqueRelaySocket { } } // Writer exited (stream error or receiver dropped). Dropping - // `send_rx` closes the channel so subsequent `try_send` - // calls fail fast with `Closed` instead of filling a queue - // that nobody will drain. Wake any poller currently parked - // on `send_capacity_freed`: it will re-check, observe the closed - // channel via `is_closed`, and surface the failure instead - // of waiting forever. + // `send_rx` closes the channel so subsequent `try_send` calls fail + // fast with `Closed` instead of filling a queue nobody will drain. + // `_writer_exit` then records the exit and wakes parked pollers. drop(send_rx); - writer_capacity.notify_waiters(); - mark_writer_exit(&writer_control); }); control.register(writer_handle); @@ -480,6 +604,30 @@ impl MasqueRelaySocket { (socket, control) } + /// Whether a [`TunnelPoller`] should stop waiting for send capacity. + /// + /// The writer state is checked alongside the channel state because shutdown + /// publishes it *before* aborting the writer, whereas the channel's closure + /// trails the abort and arrives with no notification of its own. A poller + /// that consulted only the channel could wake to one still full and open, + /// re-park, and never be woken again. + /// + /// Every `true` here must be matched by a non-`WouldBlock` result from + /// [`enqueue_outbound`](Self::enqueue_outbound): Quinn retries a `WouldBlock` + /// immediately and without yielding, so claiming writability and then + /// refusing the datagram spins the connection driver instead of parking it. + fn writable_or_finished(&self) -> bool { + self.send_tx.capacity() > 0 || self.send_tx.is_closed() || self.state.writer_stopped() + } + + /// Whether the outbound send channel is still open. Lets the teardown test + /// assert that a poller was released by the writer state rather than by the + /// channel closing. + #[cfg(test)] + pub(crate) fn send_channel_open(&self) -> bool { + !self.send_tx.is_closed() + } + /// Remaining capacity in the outbound send channel. Exposed for /// tests and metrics — a sustained value of 0 means the tunnel /// stream can't keep up with Quinn's offered load and the poller @@ -500,10 +648,20 @@ impl MasqueRelaySocket { fn enqueue_outbound(&self, encoded: Bytes) -> io::Result<()> { match self.send_tx.try_send(encoded) { Ok(()) => Ok(()), - Err(mpsc::error::TrySendError::Full(_)) => Err(io::Error::new( - io::ErrorKind::WouldBlock, - "relay send queue full", - )), + Err(mpsc::error::TrySendError::Full(_)) => { + if self.state.writer_stopped() { + // This queue will never drain, and `writable_or_finished` + // has already told Quinn the socket is writable, so + // `WouldBlock` here would put it into an immediate, + // unyielding retry. Drop the datagram instead — which is + // what an undeliverable packet is. + return Ok(()); + } + Err(io::Error::new( + io::ErrorKind::WouldBlock, + "relay send queue full", + )) + } Err(mpsc::error::TrySendError::Closed(_)) => Err(io::Error::new( io::ErrorKind::ConnectionAborted, "relay stream closed", @@ -618,6 +776,21 @@ impl AsyncUdpSocket for MasqueRelaySocket { // packets in this poll; otherwise deliver what we // have and let the next poll see the closed state. if filled == 0 { + if self.state.shutdown_requested() { + // We dismantled this tunnel ourselves, so nothing + // failed. An I/O error here would end the endpoint + // driver through its failure path: logged at ERROR, + // and its `Drop` clears the connection senders + // instead of letting the endpoint retire once its + // last handle goes away. + // + // Park instead. Nothing can arrive on a torn-down + // tunnel, and the driver keeps its other wakers — + // the endpoint-event channel and the explicit wake + // `EndpointRef::drop` issues at refcount zero — so + // it still retires with `Ok(())`. + return Poll::Pending; + } return Poll::Ready(Err(io::Error::new( io::ErrorKind::BrokenPipe, "relay recv stream closed", @@ -681,11 +854,11 @@ impl UdpPoller for TunnelPoller { // freely take `&mut self` out of the `Pin`. let this = self.get_mut(); - // Fast path: capacity is available right now, or the channel - // is closed (writer task exited — return Ready so Quinn - // attempts a `try_send`, which surfaces the failure as - // `ConnectionAborted`). - if this.socket.send_tx.capacity() > 0 || this.socket.send_tx.is_closed() { + // Fast path: capacity is available right now, the channel is closed + // (writer task exited — return Ready so Quinn attempts a `try_send`, + // which surfaces the failure as `ConnectionAborted`), or the tunnel has + // ended and no drain is ever coming. + if this.socket.writable_or_finished() { this.wait = None; return Poll::Ready(Ok(())); } @@ -705,15 +878,15 @@ impl UdpPoller for TunnelPoller { // last check and `enable`, `enable` stashes the // permit and the subsequent `.await` returns // immediately. - let notified = socket.send_capacity_freed.notified(); + let notified = socket.state.send_capacity_freed.notified(); tokio::pin!(notified); notified.as_mut().enable(); - if socket.send_tx.capacity() > 0 || socket.send_tx.is_closed() { + if socket.writable_or_finished() { return; } notified.await; - if socket.send_tx.capacity() > 0 || socket.send_tx.is_closed() { + if socket.writable_or_finished() { return; } // Spurious wake (e.g., another poller consumed the @@ -735,7 +908,7 @@ impl UdpPoller for TunnelPoller { #[cfg(test)] mod relay_tunnel_control_tests { - use super::{RelayTunnelControl, mark_writer_exit}; + use super::{RelayTunnelControl, TunnelState, WriterExit}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -749,7 +922,7 @@ mod relay_tunnel_control_tests { #[tokio::test] async fn shutdown_aborts_registered_tasks_and_wakes_waiters() { - let control = RelayTunnelControl::new(); + let control = RelayTunnelControl::detached(); let dropped = Arc::new(AtomicBool::new(false)); let marker = DropMarker(Arc::clone(&dropped)); control.register(tokio::spawn(async move { @@ -772,7 +945,7 @@ mod relay_tunnel_control_tests { #[tokio::test] async fn shutdown_is_idempotent() { - let control = RelayTunnelControl::new(); + let control = RelayTunnelControl::detached(); control.shutdown().await; control.shutdown().await; @@ -782,7 +955,7 @@ mod relay_tunnel_control_tests { #[tokio::test] async fn shutdown_now_aborts_registered_tasks_without_an_await() { - let control = RelayTunnelControl::new(); + let control = RelayTunnelControl::detached(); let dropped = Arc::new(AtomicBool::new(false)); let marker = DropMarker(Arc::clone(&dropped)); control.register(tokio::spawn(async move { @@ -804,11 +977,69 @@ mod relay_tunnel_control_tests { } #[test] - fn writer_exit_marks_tunnel_closed() { - let control = RelayTunnelControl::new(); + fn writer_exit_marks_the_tunnel_failed_not_shut_down() { + let control = RelayTunnelControl::detached(); + + drop(WriterExit(Arc::clone(&control.state))); + + assert!(control.is_closed(), "a writer exit ends the tunnel"); + assert!( + !control.shutdown_requested(), + "a tunnel that broke on its own was not dismantled by us" + ); + } - mark_writer_exit(&Arc::downgrade(&control)); + #[tokio::test] + async fn shutdown_records_that_the_teardown_was_requested() { + let control = RelayTunnelControl::detached(); + + control.shutdown().await; assert!(control.is_closed()); + assert!(control.shutdown_requested()); + } + + #[tokio::test] + async fn a_writer_exit_is_recorded_even_after_its_control_is_dropped() { + // The dial-through path in `p2p_endpoint` discards its control as soon + // as the dial completes, while the socket and the tunnel tasks live on. + // A writer that could not record its exit there would leave a poller + // parked on a full send queue waiting forever. + let control = RelayTunnelControl::detached(); + let state: Arc = Arc::clone(&control.state); + let exit = WriterExit(Arc::clone(&state)); + + let waiter = tokio::spawn({ + let state = Arc::clone(&state); + async move { state.send_capacity_freed.notified().await } + }); + tokio::task::yield_now().await; + + drop(control); + drop(exit); + + assert!(state.writer_stopped(), "the writer's exit is recorded"); + assert!(state.is_closed(), "and it ends the tunnel"); + tokio::time::timeout(std::time::Duration::from_secs(5), waiter) + .await + .expect("a parked capacity waiter must be released by the writer's exit") + .expect("waiter task"); + } + + #[tokio::test] + async fn cleanup_after_a_failure_does_not_reclassify_it_as_requested() { + // Cleanup routinely arrives after a tunnel has already broken — the + // health monitor calls `shutdown` on tunnels it finds dead. That must + // not rewrite the record, or it silences the fault that triggered it. + let control = RelayTunnelControl::detached(); + + control.mark_failed(); + control.shutdown().await; + + assert!(control.is_closed()); + assert!( + !control.shutdown_requested(), + "cleanup arriving after a failure must not silence the failure" + ); } } diff --git a/src/nat_traversal_api.rs b/src/nat_traversal_api.rs index 3e34a075..2eaf30f1 100644 --- a/src/nat_traversal_api.rs +++ b/src/nat_traversal_api.rs @@ -508,9 +508,12 @@ struct ProactiveRelay { impl ProactiveRelay { async fn teardown(mut self, reason: &'static [u8]) { + // Order matters. `close` only queues CONNECTION_CLOSE on each + // connection; the frames still have to travel out through the tunnel. + // Drain first and dismantle second, or the transport is torn out from + // under the endpoint mid-drain and the peer is left to time out. self.endpoint .close(crate::VarInt::from_u32(RELAY_TUNNEL_LOST_CODE), reason); - self.tunnel.shutdown().await; if tokio::time::timeout(Duration::from_secs(1), self.endpoint.wait_idle()) .await @@ -522,6 +525,8 @@ impl ProactiveRelay { ); } + self.tunnel.shutdown().await; + self.relay_session_owner.remove_owned_session(reason); info!( @@ -577,6 +582,7 @@ impl RelayLifecycleState { } struct RelayHealthSnapshot { + handle: PreparedRelay, relay_addr: SocketAddr, tunnel: Arc, } @@ -738,6 +744,7 @@ impl RelayLifecycleHandle { RelayLifecycleCommand::Health { reply } => { let health = match &state { RelayLifecycleState::Published(relay) => Some(RelayHealthSnapshot { + handle: relay.handle, relay_addr: relay.handle.public_addr(), tunnel: Arc::clone(&relay.tunnel), }), @@ -4467,11 +4474,20 @@ impl NatTraversalEndpoint { /// Returns false if a relay was established but the underlying QUIC /// connection has closed. pub async fn is_relay_healthy(&self) -> bool { - let Some(relay) = self.relay_lifecycle.health().await else { - return true; - }; + self.unhealthy_published_relay().await.is_none() + } + + /// The published relay, if there is one and it is dead. + /// + /// Returns the exact allocation the verdict was reached about. Asking + /// whether the relay is healthy and then asking separately which relay is + /// published leaves a window in which a replacement can publish between the + /// two, and the caller would then tear down the healthy replacement on the + /// strength of a verdict about its predecessor. + pub(crate) async fn unhealthy_published_relay(&self) -> Option { + let relay = self.relay_lifecycle.health().await?; if relay.tunnel.is_closed() { - return false; + return Some(relay.handle); } // Check the specific session for the advertised relay address. @@ -4479,7 +4495,7 @@ impl NatTraversalEndpoint { // using relay_addr, so that's the one that must be healthy. for entry in self.relay_sessions.iter() { if entry.value().public_address == Some(relay.relay_addr) { - return entry.value().is_active(); + return (!entry.value().is_active()).then_some(relay.handle); } } @@ -4487,7 +4503,7 @@ impl NatTraversalEndpoint { "Relay session for {} is dead — re-establishment required", relay.relay_addr ); - false + Some(relay.handle) } pub(crate) async fn published_relay_handle(&self) -> Option { @@ -5752,6 +5768,11 @@ impl NatTraversalEndpoint { let tunnel = Arc::clone(&tunnel); tokio::spawn(async move { tunnel.closed().await; + if tunnel.shutdown_requested() { + // `teardown` is already closing and draining this endpoint + // in the right order. Nothing died. + return; + } info!( "MASQUE tunnel for relay {} died — closing relay endpoint gracefully", relay_public_addr @@ -5843,6 +5864,31 @@ impl NatTraversalEndpoint { Ok(()) } + /// Abort a proactive relay that the health monitor has found dead. + /// + /// The same teardown, but the tunnel is recorded as failed first. + /// `is_relay_healthy` can condemn a relay from the state of the *outer* + /// relay session while the tunnel's own cause is still `Live`; without this, + /// the teardown's `shutdown()` would settle first and file a genuine failure + /// as an intentional one, silencing the transport error and the tunnel-death + /// log that report it. + pub(crate) async fn abort_unhealthy_proactive_relay( + &self, + prepared: PreparedRelay, + ) -> Result<(), NatTraversalError> { + if let Some(relay) = self + .relay_lifecycle + .take_matching(prepared) + .await + .map_err(NatTraversalError::ConnectionFailed)? + { + relay.tunnel.mark_failed(); + self.teardown_proactive_relay(relay, b"relay tunnel unhealthy") + .await; + } + Ok(()) + } + /// Legacy eager setup: prepare and immediately publish. /// /// Canary-gated callers should use [`prepare_proactive_relay`](Self::prepare_proactive_relay) @@ -8600,6 +8646,613 @@ mod tests { use super::*; + use std::task::{Context, Poll, Wake, Waker}; + + use crate::high_level::AsyncUdpSocket; + + /// A waker that records whether it was woken, for driving a poller by hand + /// and then checking that the code under test actually notified it. + #[derive(Default)] + struct RecordingWake(AtomicBool); + + impl RecordingWake { + fn was_woken(&self) -> bool { + self.0.load(Ordering::Acquire) + } + } + + impl Wake for RecordingWake { + fn wake(self: Arc) { + self.wake_by_ref(); + } + + fn wake_by_ref(self: &Arc) { + self.0.store(true, Ordering::Release); + } + } + + /// Collects formatted `tracing` output so a test can assert on what an + /// operator would have seen in the node log. + #[derive(Clone, Default)] + struct CapturedLogs(Arc>>); + + impl CapturedLogs { + fn text(&self) -> String { + match self.0.lock() { + Ok(buffer) => String::from_utf8_lossy(&buffer).into_owned(), + Err(poisoned) => String::from_utf8_lossy(&poisoned.into_inner()).into_owned(), + } + } + } + + impl std::io::Write for CapturedLogs { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if let Ok(mut buffer) = self.0.lock() { + buffer.extend_from_slice(buf); + } + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + /// A proactive relay backed by a real MASQUE tunnel over an in-process QUIC + /// connection, plus the peers holding the tunnel up. + struct LiveProactiveRelay { + relay: ProactiveRelay, + socket: Arc, + relay_peer: NatTraversalEndpoint, + local: NatTraversalEndpoint, + } + + async fn live_proactive_relay() -> LiveProactiveRelay { + let endpoint_config = || NatTraversalConfig { + bind_addr: Some("127.0.0.1:0".parse().expect("test bind address")), + ..Default::default() + }; + let relay_peer = NatTraversalEndpoint::new(endpoint_config(), None, None) + .await + .expect("relay peer endpoint"); + let local = NatTraversalEndpoint::new(endpoint_config(), None, None) + .await + .expect("local endpoint"); + + let relay_server_addr = relay_peer + .get_endpoint() + .expect("relay peer transport endpoint") + .local_addr() + .expect("relay peer address"); + let server_name = relay_server_addr.ip().to_string(); + + let connection = local + .connect_to(&server_name, relay_server_addr) + .await + .expect("relay control connection"); + let (send_stream, recv_stream) = connection.open_bi().await.expect("relay tunnel stream"); + + let original_socket = local + .get_endpoint() + .expect("local transport endpoint") + .current_socket() + .expect("original socket"); + + let relay_public_addr = "203.0.113.7:9000".parse().expect("relay public address"); + let (socket, tunnel) = crate::masque::MasqueRelaySocket::new( + send_stream, + recv_stream, + relay_public_addr, + relay_server_addr, + original_socket, + ); + + let runtime = crate::high_level::default_runtime().expect("async runtime"); + let relay_endpoint = InnerEndpoint::new_with_abstract_socket( + EndpointConfig::default(), + None, + Arc::clone(&socket) as Arc, + runtime, + ) + .expect("relay endpoint"); + + LiveProactiveRelay { + relay: ProactiveRelay { + handle: PreparedRelay::new(relay_public_addr), + endpoint: Arc::new(relay_endpoint), + tunnel, + relay_session_owner: RelaySessionOwner { + relay_server_addr, + public_address: Some(relay_public_addr), + relay_sessions: Arc::new(dashmap::DashMap::new()), + stable_id: usize::MAX, + cleanup_armed: false, + }, + cleanup_armed: true, + }, + socket, + relay_peer, + local, + } + } + + /// V2-986: an orderly relay teardown used to end the backing endpoint's + /// driver through its I/O-failure path, logging + /// `ERROR ... I/O error: relay recv stream closed` once per teardown. The + /// fleet upgrade that made teardown deterministic turned that into a ~100x + /// rise in transport ERROR volume. + #[tokio::test(flavor = "current_thread")] + async fn proactive_relay_teardown_reports_no_transport_error() { + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::ERROR) + .with_writer(logs.clone()) + .with_ansi(false) + .finish(); + let _log_guard = tracing::subscriber::set_default(subscriber); + + let LiveProactiveRelay { + relay, + socket, + relay_peer, + local, + } = live_proactive_relay().await; + // This handle and the relay endpoint's own are the only two; once the + // endpoint retires, this test holds the socket alone. + assert_eq!(Arc::strong_count(&socket), 2); + + relay.teardown(b"relay allocation aborted").await; + + // Parking `poll_recv` must not strand the driver: it still has to retire + // once the endpoint is dropped, releasing the tunnel socket. Waiting for + // that first also means the driver has finished whatever it was going to + // log, so the assertions below cannot pass by reading too early. + tokio::time::timeout(Duration::from_secs(5), async { + while Arc::strong_count(&socket) > 1 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("relay endpoint driver should retire and release the tunnel socket"); + tokio::time::sleep(Duration::from_millis(50)).await; + + let captured = logs.text(); + assert!( + !captured.contains("relay recv stream closed"), + "a teardown we initiated must not be logged as a transport failure; got:\n{captured}" + ); + assert!( + !captured.contains("I/O error"), + "the relay endpoint driver must not end through its I/O-failure path; got:\n{captured}" + ); + + let _ = local.shutdown().await; + let _ = relay_peer.shutdown().await; + } + + /// Both halves of the fix on the real production path: a real CONNECT-UDP + /// session, the real `run_stream_forwarding_loop` data plane, the real + /// `prepare_proactive_relay` / `abort_proactive_relay` lifecycle, and a real + /// peer dialling the relay-allocated address. + /// + /// The peer must be *told* the relay is going away (the ordering half) with + /// no transport error logged (the classification half), and the relay server + /// must get its capacity slot back. + #[tokio::test] + async fn real_relay_teardown_closes_the_peer_and_releases_capacity() { + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::ERROR) + .with_writer(logs.clone()) + .with_ansi(false) + .finish(); + let _log_guard = tracing::subscriber::set_default(subscriber); + + let relay_node = NatTraversalEndpoint::new( + NatTraversalConfig { + bind_addr: Some("127.0.0.1:0".parse().expect("test bind address")), + enable_relay_service: true, + ..Default::default() + }, + None, + None, + ) + .await + .expect("relay node"); + let relay_node_addr = relay_node + .get_endpoint() + .expect("relay transport endpoint") + .local_addr() + .expect("relay node address"); + let relay_server = relay_node + .relay_server + .clone() + .expect("relay service should be enabled"); + + let client = NatTraversalEndpoint::new( + NatTraversalConfig { + bind_addr: Some("127.0.0.1:0".parse().expect("test bind address")), + enable_relay_service: false, + ..Default::default() + }, + None, + None, + ) + .await + .expect("client node"); + + // The real lifecycle: establish a CONNECT-UDP session and stand a relay + // endpoint on the resulting tunnel. + let prepared = tokio::time::timeout( + Duration::from_secs(30), + client.prepare_proactive_relay(relay_node_addr), + ) + .await + .expect("relay preparation timed out") + .expect("prepare a proactive relay through the real relay server"); + let relay_public_addr = prepared.public_addr(); + + let sessions_while_live = relay_server.stats().current_active_sessions(); + assert!( + sessions_while_live >= 1, + "the relay server should hold a session while the allocation is live" + ); + + // A real peer dials the relay-allocated address. + let peer = NatTraversalEndpoint::new( + NatTraversalConfig { + bind_addr: Some("127.0.0.1:0".parse().expect("test bind address")), + enable_relay_service: false, + ..Default::default() + }, + None, + None, + ) + .await + .expect("peer node"); + let peer_connection = tokio::time::timeout( + Duration::from_secs(30), + peer.connect_to(&relay_public_addr.ip().to_string(), relay_public_addr), + ) + .await + .expect("peer dial through the real relay timed out") + .expect("peer connection through the real relay"); + assert!(peer_connection.close_reason().is_none()); + + // Wait for the relay side to finish accepting. `connect_to` returns as + // soon as the *peer* has its keys; tearing down before the relay + // endpoint has completed its half means the close goes out as the + // transport-level frame RFC 9000 requires during a handshake, rather + // than the application close this test is checking for. + tokio::time::timeout(Duration::from_secs(10), async { + while client.connections.is_empty() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("the relay endpoint should register the accepted peer connection"); + + // The real teardown. + client + .abort_proactive_relay(prepared) + .await + .expect("abort the proactive relay"); + + let close = tokio::time::timeout(Duration::from_secs(10), peer_connection.closed()) + .await + .expect("a peer relayed through a real relay must be told, not left to time out"); + assert!( + matches!( + close, + crate::ConnectionError::ApplicationClosed(ref closed) + if closed.error_code == crate::VarInt::from_u32(RELAY_TUNNEL_LOST_CODE) + ), + "expected the teardown's application close, got {close:?}" + ); + + // The relay server must get its capacity slot back. + tokio::time::timeout(Duration::from_secs(10), async { + while relay_server.stats().current_active_sessions() >= sessions_while_live { + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("teardown must release the relay server's capacity slot"); + + let captured = logs.text(); + assert!( + !captured.contains("relay recv stream closed"), + "the real teardown path must not report a transport failure; got:\n{captured}" + ); + + let _ = peer.shutdown().await; + let _ = client.shutdown().await; + let _ = relay_node.shutdown().await; + } + + /// A live tunnel whose far side accepts the stream and never reads it, so + /// QUIC flow control stalls the writer and the outbound queue fills. + /// + /// This is the state the lost-wakeup race needs: a `TunnelPoller` parked on + /// a full send queue while the channel is still open. + #[allow(clippy::type_complexity)] + async fn stalled_tunnel() -> ( + Arc, + Arc, + Vec, + InnerEndpoint, + Box, + ) { + let config_source = NatTraversalEndpoint::new( + NatTraversalConfig { + bind_addr: Some("127.0.0.1:0".parse().expect("test bind address")), + ..Default::default() + }, + None, + None, + ) + .await + .expect("config source endpoint"); + let server_config = config_source + .relay_server_config + .lock() + .expect("relay server config") + .clone() + .expect("relay server config present"); + let client_config = config_source + .get_endpoint() + .expect("transport endpoint") + .default_client_config + .clone() + .expect("client config present"); + + let runtime = crate::high_level::default_runtime().expect("async runtime"); + let tunnel_server = InnerEndpoint::new( + EndpointConfig::default(), + Some(server_config), + std::net::UdpSocket::bind("127.0.0.1:0").expect("tunnel server socket"), + Arc::clone(&runtime), + ) + .expect("tunnel server endpoint"); + let tunnel_server_addr = tunnel_server.local_addr().expect("tunnel server addr"); + + let mut tunnel_client = InnerEndpoint::new( + EndpointConfig::default(), + None, + std::net::UdpSocket::bind("127.0.0.1:0").expect("tunnel client socket"), + Arc::clone(&runtime), + ) + .expect("tunnel client endpoint"); + tunnel_client.set_default_client_config(client_config); + + // Accept the tunnel stream and then deliberately never read it. + let accept = tokio::spawn(async move { + let incoming = tunnel_server.accept().await.expect("tunnel incoming"); + let connection = incoming.await.expect("tunnel accepted"); + let streams = connection.accept_bi().await.expect("tunnel stream"); + (tunnel_server, connection, streams) + }); + + let tunnel_conn = tunnel_client + .connect(tunnel_server_addr, &tunnel_server_addr.ip().to_string()) + .expect("dial tunnel server") + .await + .expect("tunnel connection"); + let (mut client_send, client_recv) = + tunnel_conn.open_bi().await.expect("open tunnel stream"); + client_send + .write_all(&0u32.to_be_bytes()) + .await + .expect("open the tunnel stream on the wire"); + let held = tokio::time::timeout(Duration::from_secs(10), accept) + .await + .expect("tunnel accept timed out") + .expect("tunnel accept task"); + + let original_socket = tunnel_client + .current_socket() + .expect("tunnel client socket"); + let (socket, tunnel) = crate::masque::MasqueRelaySocket::new( + client_send, + client_recv, + "203.0.113.9:9000".parse().expect("relay public address"), + tunnel_server_addr, + original_socket, + ); + + // The far side is returned, not forgotten: it must stay alive and unread + // for the duration of the test, and the caller drops it at the end. + ( + socket, + tunnel, + vec![config_source], + tunnel_client, + Box::new(held), + ) + } + + /// The driver crashing used to be what freed a connection parked waiting for + /// send capacity. It no longer crashes, so shutdown has to free it. + /// + /// `shutdown_now` records the writer as stopped, aborts it, and wakes + /// capacity waiters — but `JoinHandle::abort` is asynchronous, so the send + /// channel is still open when the poller wakes. A poller that consulted only + /// the channel would see it full and open, consume the notification, re-park, + /// and never be woken again, because closing an mpsc does not touch the + /// `Notify`. + /// + /// The tunnel is marked failed first, so this also pins that the writer + /// state is tracked independently of which cause was settled first. + #[tokio::test] + async fn shutdown_releases_a_poller_parked_on_a_full_send_queue() { + let (socket, tunnel, _keepalive, _tunnel_client, _far_side) = stalled_tunnel().await; + + let transmit = |contents: &'static [u8]| quinn_udp::Transmit { + destination: "198.51.100.4:4433".parse().expect("destination address"), + ecn: None, + contents, + segment_size: None, + src_ip: None, + }; + + let filled = tokio::time::timeout(Duration::from_secs(30), async { + loop { + match socket.try_send(&transmit(&[0u8; 1024])) { + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => return Ok(()), + Err(error) => return Err(error), + Ok(()) => tokio::task::yield_now().await, + } + } + }) + .await + .expect("filling the send queue should not take 30s"); + filled.expect("the queue should fill, not fail"); + + let wake_record = Arc::new(RecordingWake::default()); + let waker = Waker::from(Arc::clone(&wake_record)); + let mut cx = Context::from_waker(&waker); + let mut poller = Arc::clone(&socket).create_io_poller(); + assert!( + matches!(poller.as_mut().poll_writable(&mut cx), Poll::Pending), + "a full send queue should park the poller" + ); + assert!( + !wake_record.was_woken(), + "nothing should have woken the poller yet" + ); + + // Synchronous on purpose: the aborted writer has not run yet, so the + // send channel is still open. Only the recorded writer state can release + // the poller here — and the cause settles as `Failed`, not + // `ShutdownRequested`, so the two really are independent. + tunnel.mark_failed(); + tunnel.shutdown_now(); + assert!( + wake_record.was_woken(), + "shutdown must wake the parked poller, not leave it for the abort to land" + ); + assert!( + !tunnel.shutdown_requested(), + "the first cause wins, so this teardown stays classified as a failure" + ); + assert!( + socket.send_channel_open(), + "this test is only meaningful while the channel is still open" + ); + + assert!( + matches!(poller.as_mut().poll_writable(&mut cx), Poll::Ready(Ok(()))), + "shutdown must release a parked poller without waiting for the channel to close" + ); + + // Readiness has to be matched by a send that completes. Quinn retries a + // `WouldBlock` immediately and without yielding, so answering "writable" + // and then refusing the datagram would spin the connection driver — on a + // current-thread runtime, possibly forever, because the aborted writer + // never gets to run and close the channel. + assert!( + socket.send_channel_open(), + "the channel must still be open for this to test what it claims" + ); + let sent = socket.try_send(&transmit(&[0u8; 1024])); + assert!( + !matches!(&sent, Err(error) if error.kind() == std::io::ErrorKind::WouldBlock), + "a writable answer must not be followed by WouldBlock; got {sent:?}" + ); + } + + /// The quiet path must stay scoped to teardowns we asked for: a tunnel that + /// breaks on its own is still a transport failure and must still be logged. + #[tokio::test(flavor = "current_thread")] + async fn unexpected_tunnel_loss_still_reports_a_transport_error() { + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::ERROR) + .with_writer(logs.clone()) + .with_ansi(false) + .finish(); + let _log_guard = tracing::subscriber::set_default(subscriber); + + let live = live_proactive_relay().await; + + // The relay peer goes away; the tunnel reader's stream read fails. + live.relay_peer.shutdown().await.expect("relay peer down"); + tokio::time::timeout(Duration::from_secs(5), live.relay.tunnel.closed()) + .await + .expect("reader task should observe the broken relay stream"); + + // `closed()` fires when the cause settles, which is before the reader + // drops `recv_tx` and before the driver has necessarily logged. Wait for + // the line itself rather than guessing at a sleep. + let captured = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let captured = logs.text(); + if captured.contains("relay recv stream closed") { + return captured; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap_or_else(|_| logs.text()); + + assert!( + captured.contains("relay recv stream closed"), + "a tunnel that broke on its own must still surface as a transport failure; got:\n{captured}" + ); + + let _ = live.local.shutdown().await; + } + + /// `is_relay_healthy` can condemn a relay from the outer session's state + /// while the tunnel's own cause is still `Live`. The monitor's abort has to + /// record that as a failure, or the teardown's `shutdown()` settles first + /// and files it as something we chose to do. + #[tokio::test] + async fn the_health_monitors_abort_is_recorded_as_a_failure() { + let live = live_proactive_relay().await; + let tunnel = Arc::clone(&live.relay.tunnel); + let prepared = live.relay.handle; + assert!(!tunnel.is_closed(), "the tunnel starts out live"); + + let (generation, previous) = live + .local + .relay_lifecycle + .begin_prepare() + .await + .expect("begin prepare"); + assert!(previous.is_none()); + live.local + .relay_lifecycle + .complete_prepare(generation, live.relay) + .await + .expect("lifecycle actor") + .map_err(|_| "the relay must install") + .expect("relay installed"); + + live.local + .abort_unhealthy_proactive_relay(prepared) + .await + .expect("the health monitor's abort"); + + assert!(tunnel.is_closed()); + assert!( + !tunnel.shutdown_requested(), + "a relay torn down because it was found dead is a failure, not a request" + ); + + let _ = live.local.shutdown().await; + let _ = live.relay_peer.shutdown().await; + } + async fn detached_proactive_relay(public_addr: SocketAddr) -> ProactiveRelay { let endpoint = NatTraversalEndpoint::new( NatTraversalConfig { diff --git a/src/p2p_endpoint.rs b/src/p2p_endpoint.rs index 07b7c18c..68a652cc 100644 --- a/src/p2p_endpoint.rs +++ b/src/p2p_endpoint.rs @@ -4028,26 +4028,33 @@ impl P2pEndpoint { // Monitor relay health. If the relay session died (connection // closed, server restarted, etc.), tear down its one lifecycle // state so the upper layer can acquire a replacement. - if relay_event_sent && !inner.is_relay_healthy().await { - let dead = inner.published_relay_handle().await; - if let Some(prepared) = dead { - if let Err(error) = inner.abort_proactive_relay(prepared).await { - warn!( - relay_addr = %prepared.public_addr(), - %error, - "Failed to tear down unhealthy proactive relay" - ); - } - } - relay_event_sent = false; - if let Some(prepared) = dead { - let relay_addr = prepared.public_addr(); - info!( - "Relay tunnel at {} is unhealthy — emitting RelayLost event", - relay_addr + // One verdict that names the relay it is about. Asking whether + // the relay is healthy and then asking separately which relay is + // published would let a replacement publish in between, and this + // would tear down the healthy replacement. + let dead = if relay_event_sent { + inner.unhealthy_published_relay().await + } else { + None + }; + if let Some(prepared) = dead { + // The unhealthy path, so the teardown does not file this + // failure as an intentional shutdown. + if let Err(error) = inner.abort_unhealthy_proactive_relay(prepared).await { + warn!( + relay_addr = %prepared.public_addr(), + %error, + "Failed to tear down unhealthy proactive relay" ); - let _ = event_tx_for_nat.send(P2pEvent::RelayLost { relay_addr }); } + relay_event_sent = false; + + let relay_addr = prepared.public_addr(); + info!( + "Relay tunnel at {} is unhealthy — emitting RelayLost event", + relay_addr + ); + let _ = event_tx_for_nat.send(P2pEvent::RelayLost { relay_addr }); } } });