refactor: collapse the four AcceptWaiters copies onto kio::Fan (#359)

kio 0.5.4 released `Fan`, so the shared accept list, its waker, and the hold that
keeps an inline wake from landing under the accept lock all come from upstream.
That deletes 1,264 lines of near-identical `waiters.rs` across the four backends.

The `arm`/`disarm` pair becomes a `Hold` guard, which is harder to misuse: there
is no count to forget to bring down. It is still dropped explicitly after the
accept lock, because where it drops is where a held-back wake is delivered.

`Arc<AcceptWaiters>` becomes a plain `Fan` — it is already a shared handle — and
the cached wakers come from `Fan::waker()`.

iroh and noq carried nothing else in that module, so the file goes. quinn and
quiche keep `Parked`, the release-on-`Ready` bridge that kio deliberately does
not do, at 56 lines each.

Behaviour is unchanged and the accept tests say so: every accepter still gets
woken, a departing one still does not strand the others, and an abandoned one
still releases its slot.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: kixelated

rs/web-transport-iroh/Cargo.toml

@@ -22,7 +22,7 @@ qlog = ["iroh/qlog"]
 bytes = "1"
 http = "1"
 iroh = { version = "1", default-features = false, features = ["fast-apple-datapath"] }
-kio = "0.5.3"
+kio = "0.5.4"
 n0-error = "1"
 n0-future = "0.3.1"
 tokio = { version = "1", default-features = false, features = [

rs/web-transport-iroh/src/lib.rs

@@ -32,7 +32,6 @@ mod session;
 mod settings;
 #[cfg(test)]
 mod tests;
-mod waiters;
 
 pub use client::*;
 pub use connect::*;

rs/web-transport-iroh/src/session.rs

@@ -10,7 +10,7 @@ use std::{
 
 use bytes::{Bytes, BytesMut};
 use iroh::endpoint::{self, Connection, PathStats};
-use kio::Waiter;
+use kio::{Fan, Waiter};
 use n0_future::{
     FuturesUnordered,
     stream::{Stream, StreamExt},
@@ -19,7 +19,6 @@ use web_transport_proto::{ConnectRequest, ConnectResponse, Frame, StreamUni, Var
 
 use crate::{
     ClientError, Connected, RecvStream, SendStream, SessionError, Settings, WebTransportError,
-    waiters::AcceptWaiters,
 };
 
 /// An established WebTransport session, acting like a full QUIC connection. See [`iroh::endpoint::Connection`].
@@ -378,21 +377,24 @@ fn poll_accept_uni_shared(
     accept: &Mutex<H3SessionAccept>,
     waiter: &Waiter,
 ) -> Poll<Result<RecvStream, SessionError>> {
-    let (result, waiters) = {
+    let (result, waiters, hold) = {
         let mut accept = accept.lock().unwrap();
         let waiters = accept.uni_waiters.clone();
 
-        // The poll below drives the shared accept futures with this list's waker, and
-        // one of them may wake it inline. Hold those back until the lock is gone.
-        waiters.arm();
+        // The poll below drives the shared accept futures with this list's waker, and one
+        // of them may wake it inline. `Fan` holds those back while this guard is alive.
+        let hold = waiters.hold();
         let result = accept.poll_accept_uni(waiter);
 
-        (result, waiters)
+        (result, waiters, hold)
     };
 
-    // `disarm` runs first either way: the count has to come down even on a `Ready`.
-    if waiters.disarm() || result.is_ready() {
-        waiters.wake_all();
+    // Dropped after the accept lock, never before: that is where a held-back wake is
+    // delivered, and delivering it under the lock is the hazard the hold exists for.
+    drop(hold);
+
+    if result.is_ready() {
+        waiters.wake();
     }
 
     result
@@ -402,21 +404,24 @@ fn poll_accept_bi_shared(
     accept: &Mutex<H3SessionAccept>,
     waiter: &Waiter,
 ) -> Poll<Result<(SendStream, RecvStream), SessionError>> {
-    let (result, waiters) = {
+    let (result, waiters, hold) = {
         let mut accept = accept.lock().unwrap();
         let waiters = accept.bi_waiters.clone();
 
-        // The poll below drives the shared accept futures with this list's waker, and
-        // one of them may wake it inline. Hold those back until the lock is gone.
-        waiters.arm();
+        // The poll below drives the shared accept futures with this list's waker, and one
+        // of them may wake it inline. `Fan` holds those back while this guard is alive.
+        let hold = waiters.hold();
         let result = accept.poll_accept_bi(waiter);
 
-        (result, waiters)
+        (result, waiters, hold)
     };
 
-    // `disarm` runs first either way: the count has to come down even on a `Ready`.
-    if waiters.disarm() || result.is_ready() {
-        waiters.wake_all();
+    // Dropped after the accept lock, never before: that is where a held-back wake is
+    // delivered, and delivering it under the lock is the hazard the hold exists for.
+    drop(hold);
+
+    if result.is_ready() {
+        waiters.wake();
     }
 
     result
@@ -442,8 +447,8 @@ struct H3SessionAccept {
     // Every clone of the session polls this one struct, so an arrival has to be fanned
     // out: each caller registers here and all of them are woken when a stream lands —
     // by the caller that saw it, once it has released the lock on this struct.
-    bi_waiters: Arc<AcceptWaiters>,
-    uni_waiters: Arc<AcceptWaiters>,
+    bi_waiters: Fan,
+    uni_waiters: Fan,
 
     // `Waker::from(waiters.clone())`, cached so the inner accept futures are polled with
     // the same waker every time. That waker outlives every caller, so an accepter that
@@ -464,10 +469,10 @@ impl H3SessionAccept {
             Some((conn.accept_bi().await, conn))
         }));
 
-        let bi_waiters = Arc::new(AcceptWaiters::default());
-        let uni_waiters = Arc::new(AcceptWaiters::default());
-        let bi_waker = Waker::from(bi_waiters.clone());
-        let uni_waker = Waker::from(uni_waiters.clone());
+        let bi_waiters = Fan::new();
+        let uni_waiters = Fan::new();
+        let bi_waker = bi_waiters.waker();
+        let uni_waker = uni_waiters.waker();
 
         Self {
             session_id,