From a5ee413b9a8fbad3b66b2d42ae916e637b95da5d Mon Sep 17 00:00:00 2001 From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:36:41 +0200 Subject: [PATCH 1/4] fix(session): serve stops accepting once its listener is closed (#353) --- aimdb-core/src/session/server.rs | 17 +++++++- aimdb-core/tests/session_engine.rs | 64 ++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs index 69c42ef8..68f09453 100644 --- a/aimdb-core/src/session/server.rs +++ b/aimdb-core/src/session/server.rs @@ -28,7 +28,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 +494,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(mut listener: L, codec: Arc, dispatch: Arc, config: SessionConfig) where L: Listener, @@ -535,6 +539,17 @@ 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. diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs index e58f180d..1d972425 100644 --- a/aimdb-core/tests/session_engine.rs +++ b/aimdb-core/tests/session_engine.rs @@ -1390,3 +1390,67 @@ 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>, + error: TransportError, +} + +impl Listener for ScriptedListener { + fn accept(&mut self) -> BoxFut<'_, TransportResult>> { + 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, 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"); +} From 86277faeac3b5c12b0716f05c327df78912477ef Mon Sep 17 00:00:00 2001 From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:44:49 +0200 Subject: [PATCH 2/4] fix(session): the accept after a failed one yields first, in serve itself (#354) --- aimdb-core/src/session/server.rs | 42 ++++++++++++++++++++++++++++-- aimdb-core/tests/session_engine.rs | 27 +++++++++++++++++++ aimdb-embassy-adapter/src/net.rs | 16 +++++------- 3 files changed, 74 insertions(+), 11 deletions(-) diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs index 68f09453..72c3a3a0 100644 --- a/aimdb-core/src/session/server.rs +++ b/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; @@ -506,12 +509,18 @@ where D: Dispatch + 'static + ?Sized, { let mut conns: FuturesUnordered> = FuturesUnordered::new(); + // Set by a failed accept: the next one yields once before it starts. + let mut after_error = false; 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. @@ -519,6 +528,7 @@ where } }; + after_error = accept.is_err(); match accept { Ok(conn) => { // Soft cap; `len()` counts finished-but-not-yet-reaped futures, so @@ -552,8 +562,36 @@ where } Err(_e) => { log_error!("accept failed: {:?}", _e); - // Keep serving existing connections despite a transient accept error. + // Keep serving existing connections despite a transient accept + // error; the next accept yields first (`after_error`). } } } } + +/// `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 { + inner: F, + yield_first: bool, +} + +impl PendingOnce { + fn new(inner: F, yield_first: bool) -> Self { + Self { inner, yield_first } + } +} + +impl Future for PendingOnce { + type Output = F::Output; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + if self.yield_first { + self.yield_first = false; + cx.waker().wake_by_ref(); + return Poll::Pending; + } + Pin::new(&mut self.inner).poll(cx) + } +} diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs index 1d972425..4f711677 100644 --- a/aimdb-core/tests/session_engine.rs +++ b/aimdb-core/tests/session_engine.rs @@ -1454,3 +1454,30 @@ async fn serve_returns_once_a_closed_listeners_connections_end() { .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()); +} diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index 6afb5e1a..7d07a4f3 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -402,8 +402,10 @@ impl StreamDialer for EmbassyTcpDialer { Err(_) => { // Dropping `guard` aborts the socket and returns it to the slot. drop(guard); - // A synchronously-failing `connect()` gives the caller no - // yield point before it retries; see `arm` below. + // `connect()` can fail synchronously, and a caller that + // retries at once (a reconnect loop) would then spin on the + // single-core cooperative executor, starving every other + // task. Yield so a misconfig retries instead of hanging. yield_now().await; Err(TransportError::Io) } @@ -459,14 +461,10 @@ impl EmbassyTcpListener { Ok(()) => Ok(guard.into_socket()), Err(_) => { // Dropping `guard` aborts the socket and returns it to the slot. + // A synchronous failure (e.g. port-0 `InvalidPort`) needs no + // yield here: core's `serve` yields before the accept after + // an error. drop(guard); - // `accept()` can fail synchronously (e.g. port-0 - // `InvalidPort`), and core's `serve` loop logs an accept - // error and re-enters `accept()` immediately. Without a - // yield point that spins forever on the single-core - // cooperative executor, starving every other task. Yield so - // a misconfig warn-loops instead of hanging the device. - yield_now().await; Err(TransportError::Io) } } From 5ab3597caba5d64375b01511f053011353ec470b Mon Sep 17 00:00:00 2001 From: "sounds.like.lx" <147444674+lxsaah@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:51:06 +0200 Subject: [PATCH 3/4] fix(session): summarize repeated accept failures instead of logging each (#355) --- aimdb-core/src/session/server.rs | 109 ++++++++++++++++++++++++++++++- 1 file changed, 108 insertions(+), 1 deletion(-) diff --git a/aimdb-core/src/session/server.rs b/aimdb-core/src/session/server.rs index 72c3a3a0..39c9ad86 100644 --- a/aimdb-core/src/session/server.rs +++ b/aimdb-core/src/session/server.rs @@ -511,6 +511,9 @@ where let mut conns: FuturesUnordered> = 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 @@ -531,6 +534,9 @@ where 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. @@ -561,7 +567,20 @@ where return; } Err(_e) => { - log_error!("accept failed: {:?}", _e); + 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`). } @@ -569,6 +588,50 @@ where } } +/// 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 { + 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. @@ -595,3 +658,47 @@ impl Future for PendingOnce { 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 = (0..=10_000).filter(|&n| is_power_of_ten(n)).collect(); + assert_eq!(found, [1, 10, 100, 1000, 10_000]); + } +} From 28c889b31d3281ecccfc15832accb22bb35e8ac4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sat, 24 Oct 2026 19:54:22 +0200 Subject: [PATCH 4/4] docs(#351): changelogs for the serve accept-error fixes `aimdb-core` (serve stops accepting on `Closed`, yields before the accept after any other error, summarizes repeated failures), `aimdb-embassy-adapter` (the listener's own yield is gone), and the root changelog. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YPDjQ947yKjLLAq3Xzgu3e --- CHANGELOG.md | 7 +++++++ aimdb-core/CHANGELOG.md | 20 ++++++++++++++++++++ aimdb-embassy-adapter/CHANGELOG.md | 8 ++++++++ 3 files changed, 35 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b3cf4f6..713ef07e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index d203fd98..e3995b21 100644 --- a/aimdb-core/CHANGELOG.md +++ b/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 diff --git a/aimdb-embassy-adapter/CHANGELOG.md b/aimdb-embassy-adapter/CHANGELOG.md index 8f931a96..5878ae6d 100644 --- a/aimdb-embassy-adapter/CHANGELOG.md +++ b/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