From e3f16d4e73365d972bdb6af2d0fb7e45390669bd Mon Sep 17 00:00:00 2001 From: grumbach Date: Wed, 19 Aug 2026 16:28:01 +0900 Subject: [PATCH 1/2] fix(relay): drain the relay endpoint before dismantling its tunnel A proactive relay allocation owns two coupled resources: a MASQUE tunnel to the relay server, and a Quinn endpoint that uses that tunnel as its socket. `ProactiveRelay::teardown` destroyed the tunnel while the endpoint was still running on it, and two things followed. The endpoint driver's next `poll_recv` returned `BrokenPipe`, so a teardown we asked for was reported as a transport failure and logged at ERROR. Fleet-wide this was 2,338 occurrences of `relay recv stream closed` in 48 h across 73 hosts after the 0.36.0 rollout, about 100x the previous rate. Ending the driver that way also discarded the CONNECTION_CLOSE frames `close()` had just queued. `EndpointDriver::drop` clears `connections.senders`, so the connection driver handled the queued close, hit channel EOF, and exited before `drive_transmit`. Peers reached through the relay were told nothing and waited out their own idle timeout. The connection's local close reason was overwritten by the internal transport error for the same reason. Teardown now closes the endpoint, drains it for up to one second, and only then shuts the tunnel down, so the close frames leave through a tunnel that is still alive and `wait_idle` measures a real drain rather than the sender map the dying driver had just cleared. A `TunnelState`, shared strongly by the control, the socket and the tunnel tasks, records why the tunnel ended: `Live`, `Failed` or `ShutdownRequested`, settled by compare-exchange so the first cause wins. `poll_recv` parks for a teardown we requested and still reports `BrokenPipe` for a tunnel that broke, and the tunnel-death watcher stays quiet for the former. First cause matters because cleanup routinely arrives after a tunnel has already broken; last writer wins would let it silence the fault that triggered it. For the same reason the relay health monitor gets its own abort path, which marks the tunnel failed before tearing it down and names the exact allocation its verdict was about. Because the driver no longer crashes, shutdown has to release what the crash used to release. A `WriterExit` guard held by the writer future records the writer's exit on every path that ends it, including an abort that lands before the future's first poll, and wakes anything parked on send capacity. Whenever the poller answers "writable", `enqueue_outbound` must not answer `WouldBlock`, which Quinn retries immediately without yielding, so a full queue whose writer has stopped drops the datagram. Parking `poll_recv` does not strand the driver: nothing can arrive on a torn-down tunnel, and it keeps its other wakers, so it still retires with `Ok(())` once the endpoint is dropped. Documented in ADR-013. --- .../ADR-013-relay-tunnel-teardown-ordering.md | 197 ++++++ docs/adr/README.md | 1 + src/masque/relay_socket.rs | 351 +++++++-- src/nat_traversal_api.rs | 667 +++++++++++++++++- src/p2p_endpoint.rs | 43 +- 5 files changed, 1174 insertions(+), 85 deletions(-) create mode 100644 docs/adr/ADR-013-relay-tunnel-teardown-ordering.md 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 4860333a..27aff45e 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 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 150d9f90..974c3cfc 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), }), @@ -4434,11 +4441,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. @@ -4446,7 +4462,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); } } @@ -4454,7 +4470,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 { @@ -5723,6 +5739,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 @@ -5814,6 +5835,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) @@ -8418,6 +8464,613 @@ impl crate::TokenStore for DefaultTokenStore { 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 }); } } }); From f6d30aee00c4a926c293bebb6b86df539ca69198 Mon Sep 17 00:00:00 2001 From: grumbach Date: Wed, 19 Aug 2026 17:57:17 +0900 Subject: [PATCH 2/2] chore(deps): bump h2 to 0.4.16 for RUSTSEC-2026-0258 `cargo audit` fails on `h2` 0.4.15, which accepts unbounded empty HTTP/2 DATA frames. The advisory landed in the RustSec database on 2026-08-19 and fails on `main` as well, so it blocks every open PR in the repo until the lockfile moves. Lockfile only. No manifest or source file changes, and the h2 stanza's version and checksum are the only lines touched, so no other dependency edge is re-resolved. `h2` is reachable here through reqwest and hyper on the UPnP path; the node's own protocol is QUIC. `cargo audit` now exits clean with the three warnings already allowed on `main`. --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f02087bd..ad4a9efa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1341,9 +1341,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.15" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes",