fix(session): a failing listener no longer stalls serve (#351)

CHANGELOG.md

@@ -76,6 +76,13 @@ is unchanged. ([aimdb-core](aimdb-core/CHANGELOG.md))
 
 ### Fixed
 
+- **A failing listener no longer stalls a session server.** `session::serve`
+  stops accepting once its listener reports `Closed`, yields before the
+  accept after any other error so open connections keep being served, and
+  summarizes repeated failures in the log instead of one line per retry
+  (#351). ([aimdb-core](aimdb-core/CHANGELOG.md),
+  [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md))
+
 - **AimX over serial: every reconnect gets its events.** A client's `hello`
   on a live connection starts a new session, an ended subscription no longer
   cancels a newer one with the same id, and a subscription whose source ended

aimdb-core/CHANGELOG.md

@@ -131,6 +131,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **`session::serve` no longer spins on a failing listener (#351).** It
+  logged a failed `accept` and called `accept` again at once, and its select
+  is biased toward `accept`: a listener whose `accept` failed at once, every
+  time, kept it spinning (millions of retries a second) without polling the
+  open connections, and the task could not be aborted. Now:
+  - **`TransportError::Closed` ends accepting.** `serve` serves its open
+    connections to their end and then returns; a listener that never closes
+    keeps it running, as before.
+  - **The `accept` after any other error yields once first**, so the open
+    connections get their turn and the task stays cancellable. Listeners no
+    longer need their own guard (`aimdb-embassy-adapter` dropped its yield).
+  - **Repeated failures are summarized.** The first failure of a series is
+    logged with its error, then a summary at 10, 100, 1000, … attempts in a
+    row, and the next success logs how long the series was.
+
+  A listener whose `accept` keeps failing still retries without a pause:
+  `serve` is timer-free. Tokio's UDS and TCP listeners do not spin on a
+  persistent error such as `EMFILE`: their readiness is edge-triggered, so
+  they wait for the next connection attempt.
+
 - **Every client on a serial line gets its events, not every other one (#328).**
   A UART never closes, so the AimX session engine saw every client behind it
   as one session: a departed client's subscriptions kept streaming and held

aimdb-core/src/session/server.rs

@@ -19,6 +19,9 @@ use alloc::boxed::Box;
 use alloc::string::{String, ToString};
 use alloc::sync::Arc;
 use alloc::vec::Vec;
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll};
 
 use async_channel::Sender;
 use futures_channel::oneshot;
@@ -28,7 +31,7 @@ use hashbrown::HashMap;
 
 use super::{
     BoxFut, BoxStream, Connection, Dispatch, EnvelopeCodec, Inbound, Listener, Outbound, Payload,
-    RpcError, SessionLimits, SubUpdate,
+    RpcError, SessionLimits, SubUpdate, TransportError,
 };
 
 /// Per-session engine knobs.
@@ -494,6 +497,10 @@ async fn pump_subscription(
 /// Accept connections from `listener` and serve each with [`run_session`],
 /// bounded by [`SessionLimits::max_connections`]. The accept loop and all
 /// per-connection futures share one [`FuturesUnordered`] — spawn-free.
+///
+/// Runs until the listener reports [`TransportError::Closed`] (no more
+/// connections will come) and every open connection has ended, then returns.
+/// A listener that never closes keeps it running.
 pub async fn serve<L, C, D>(mut listener: L, codec: Arc<C>, dispatch: Arc<D>, config: SessionConfig)
 where
     L: Listener,
@@ -502,21 +509,34 @@ where
     D: Dispatch + 'static + ?Sized,
 {
     let mut conns: FuturesUnordered<BoxFut<'static, ()>> = FuturesUnordered::new();
+    // Set by a failed accept: the next one yields once before it starts.
+    let mut after_error = false;
+    // Consecutive failed accepts, so a listener that keeps failing logs a
+    // summary now and then instead of one line per attempt.
+    let mut failures = AcceptFailures::default();
 
     loop {
         // Extract the accept result first, then act (keeps the `listener` and
         // `conns` borrows apart).
         let accept = {
-            let mut accept = listener.accept().fuse();
+            // After an error, the next accept first returns `Pending` once, so
+            // this select falls through to the connections: a listener that
+            // fails at once, every time, cannot starve them, and the task
+            // stays cancellable.
+            let mut accept = PendingOnce::new(listener.accept(), after_error).fuse();
             select_biased! {
                 a = accept => a,
                 // Parks on the empty set, so the accept arm keeps the select alive.
                 () = conns.select_next_some() => continue,
             }
         };
 
+        after_error = accept.is_err();
         match accept {
             Ok(conn) => {
+                if let Some(_failed) = failures.on_success() {
+                    log_info!("accept recovered after {} failed attempt(s)", _failed);
+                }
                 // Soft cap; `len()` counts finished-but-not-yet-reaped futures, so
                 // under an accept flood (the biased `accept` arm starves the reap
                 // arm) it may read high transiently — acceptable for a soft cap.
@@ -535,10 +555,150 @@ where
                     run_session(conn, codec.as_ref(), dispatch.as_ref(), &cfg).await;
                 }));
             }
+            // The listener hands out no more connections: stop accepting, and
+            // serve the open ones to their end. Calling `accept` again would
+            // only fail again, at once — a spin that starves those connections.
+            Err(TransportError::Closed) => {
+                log_info!(
+                    "listener closed; serving {} open connection(s) to their end",
+                    conns.len()
+                );
+                while conns.next().await.is_some() {}
+                return;
+            }
             Err(_e) => {
-                log_error!("accept failed: {:?}", _e);
-                // Keep serving existing connections despite a transient accept error.
+                match failures.on_error() {
+                    AcceptReport::First => log_error!(
+                        "accept failed: {:?}; retrying, further failures are summarized",
+                        _e
+                    ),
+                    AcceptReport::Summary(_n) => {
+                        log_warn!(
+                            "accept still failing: {} attempts in a row, last: {:?}",
+                            _n,
+                            _e
+                        )
+                    }
+                    AcceptReport::Quiet => {}
+                }
+                // Keep serving existing connections despite a transient accept
+                // error; the next accept yields first (`after_error`).
             }
         }
     }
 }
+
+/// Counts consecutive failed accepts and decides which of them to log: the
+/// first of a series, then a summary at 10, 100, 1000, … failures, so a
+/// listener that keeps failing grows the log by one line per decade instead
+/// of one per attempt.
+#[derive(Default)]
+struct AcceptFailures {
+    consecutive: u64,
+}
+
+/// What [`AcceptFailures::on_error`] asks the caller to log.
+#[derive(Debug, PartialEq, Eq)]
+enum AcceptReport {
+    /// The first failure of a series: log the error.
+    First,
+    /// A summary: this many failures in a row.
+    Summary(u64),
+    /// Nothing.
+    Quiet,
+}
+
+impl AcceptFailures {
+    fn on_error(&mut self) -> AcceptReport {
+        self.consecutive = self.consecutive.saturating_add(1);
+        match self.consecutive {
+            1 => AcceptReport::First,
+            n if is_power_of_ten(n) => AcceptReport::Summary(n),
+            _ => AcceptReport::Quiet,
+        }
+    }
+
+    /// The length of the series a success ends, if there was one.
+    fn on_success(&mut self) -> Option<u64> {
+        let failed = core::mem::take(&mut self.consecutive);
+        (failed > 0).then_some(failed)
+    }
+}
+
+fn is_power_of_ten(mut n: u64) -> bool {
+    while n >= 10 && n.is_multiple_of(10) {
+        n /= 10;
+    }
+    n == 1
+}
+
+/// `inner`, but when `yield_first` is set, its first poll returns `Pending`
+/// (waking itself) without polling `inner`. Runtime-neutral: one turn back to
+/// the executor and to the sibling arms of a select.
+struct PendingOnce<F> {
+    inner: F,
+    yield_first: bool,
+}
+
+impl<F> PendingOnce<F> {
+    fn new(inner: F, yield_first: bool) -> Self {
+        Self { inner, yield_first }
+    }
+}
+
+impl<F: Future + Unpin> Future for PendingOnce<F> {
+    type Output = F::Output;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.yield_first {
+            self.yield_first = false;
+            cx.waker().wake_by_ref();
+            return Poll::Pending;
+        }
+        Pin::new(&mut self.inner).poll(cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_failing_series_logs_its_first_failure_then_once_per_decade() {
+        let mut failures = AcceptFailures::default();
+        let logged: Vec<(u64, AcceptReport)> = (1..=1000)
+            .map(|n| (n, failures.on_error()))
+            .filter(|(_, r)| *r != AcceptReport::Quiet)
+            .collect();
+        assert_eq!(
+            logged,
+            [
+                (1, AcceptReport::First),
+                (10, AcceptReport::Summary(10)),
+                (100, AcceptReport::Summary(100)),
+                (1000, AcceptReport::Summary(1000)),
+            ]
+        );
+    }
+
+    #[test]
+    fn a_success_ends_the_series_and_reports_its_length() {
+        let mut failures = AcceptFailures::default();
+        assert_eq!(failures.on_success(), None, "no series, nothing to report");
+        for _ in 0..12 {
+            failures.on_error();
+        }
+        assert_eq!(failures.on_success(), Some(12));
+        assert_eq!(
+            failures.on_error(),
+            AcceptReport::First,
+            "a new series starts"
+        );
+    }
+
+    #[test]
+    fn powers_of_ten() {
+        let found: Vec<u64> = (0..=10_000).filter(|&n| is_power_of_ten(n)).collect();
+        assert_eq!(found, [1, 10, 100, 1000, 10_000]);
+    }
+}

aimdb-core/tests/session_engine.rs

@@ -1390,3 +1390,94 @@ async fn an_ended_subscription_the_client_dropped_is_not_reported() {
     assert_eq!(frames, ["REPLY\n1\nOK\nA", "REPLY\n4\nOK\nB"]);
     server.abort();
 }
+
+// ===========================================================================
+// Accept errors (#351) — a listener that hands out one connection, then fails
+// every further accept.
+// ===========================================================================
+
+/// Hands out `first`, then answers every accept with `error`, at once.
+struct ScriptedListener {
+    first: Option<Box<dyn Connection>>,
+    error: TransportError,
+}
+
+impl Listener for ScriptedListener {
+    fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
+        Box::pin(async move {
+            match self.first.take() {
+                Some(conn) => Ok(conn),
+                None => Err(self.error.clone()),
+            }
+        })
+    }
+}
+
+/// `serve` over a [`ScriptedListener`] failing with `error`, and the peer of
+/// its one connection.
+fn scripted_serve(error: TransportError) -> (Box<dyn Connection>, tokio::task::JoinHandle<()>) {
+    let (server_side, client) = conn_pair();
+    let (feed, _) = tokio::sync::broadcast::channel(16);
+    let server = tokio::spawn(serve(
+        ScriptedListener {
+            first: Some(Box::new(server_side)),
+            error,
+        },
+        Arc::new(LineCodec),
+        Arc::new(FeedDispatch { feed }),
+        SessionConfig::default(),
+    ));
+    (Box::new(client), server)
+}
+
+/// A listener that reports `Closed` hands out no more connections; `serve`
+/// keeps serving the open one. Before, it called `accept` again at once,
+/// forever, and the open connection never got a turn.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_closed_listener_still_serves_its_open_connections() {
+    let (mut conn, server) = scripted_serve(TransportError::Closed);
+
+    conn.send(b"PING").await.unwrap();
+    let frames = frames_within(&mut conn, Duration::from_secs(1)).await;
+    assert_eq!(frames, ["PONG"]);
+    server.abort();
+}
+
+/// Once the listener is closed and its last connection ends, `serve` returns.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn serve_returns_once_a_closed_listeners_connections_end() {
+    let (conn, server) = scripted_serve(TransportError::Closed);
+
+    drop(conn);
+    tokio::time::timeout(Duration::from_secs(1), server)
+        .await
+        .expect("serve returns after its last connection ends")
+        .expect("serve does not panic");
+}
+
+/// A listener whose accept fails at once, every time, no longer starves the
+/// open connections: the accept after an error yields first. Before, `serve`
+/// retried millions of times a second and never polled them.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn an_accept_that_keeps_failing_does_not_starve_open_connections() {
+    let (mut conn, server) = scripted_serve(TransportError::Io);
+
+    conn.send(b"PING").await.unwrap();
+    let frames = frames_within(&mut conn, Duration::from_secs(1)).await;
+    assert_eq!(frames, ["PONG"]);
+    server.abort();
+}
+
+/// ... and `serve` stays cancellable: it returns to the executor between
+/// accepts, so `abort` takes effect.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn serve_over_a_failing_listener_can_be_aborted() {
+    let (_conn, server) = scripted_serve(TransportError::Io);
+
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    server.abort();
+    let joined = tokio::time::timeout(Duration::from_secs(1), server)
+        .await
+        .expect("an aborted serve ends");
+    assert!(joined.unwrap_err().is_cancelled());
+}

aimdb-embassy-adapter/CHANGELOG.md

@@ -15,6 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   runs its own task and pulls from `OutboundRoutes`. The `into_box_future`
   helper and `NetStack` remain.
 
+### Changed
+
+- **`EmbassyTcpListener` no longer yields before returning an accept error
+  (#351).** It did so because core's `serve` retried a failed accept at once,
+  which spun the single-core executor; `serve` now yields before the accept
+  after an error itself, for every listener. The dial-side yield (a
+  synchronously failing `connect`) stays, for the client's reconnect loop.
+
 ## [0.7.0] - 2026-09-18
 
 ### Added