From 0a783c8e04561d1fee4e3e922e9576402d5bfea3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=88=9A=E5=88=9A?= <225284228a@gmail.com>
Date: Wed, 23 Sep 2026 03:38:59 +1200
Subject: [PATCH] fix(listener): reap completed handlers during continuous
 accepts (#2598)

Continuous incoming connections recreate the one-second sleep in run_accepted_task_runner, so completed handlers are only reaped after a one-second gap in accepts. Busy public nodes can therefore accumulate completed Tokio tasks and JoinSet entries even when their established connection count stays stable.

Keep a persistent interval outside the select loop so incoming accepts cannot reset the reaping deadline. Add a paused-clock regression test that accepts an immediately completing handler every 100 ms, checks that completed entries remain bounded, and verifies shutdown drains the set. Enable Tokio's test clock utilities only for development builds.

Co-authored-by: 225284228a-droid <239500008+225284228a-droid@users.noreply.github.com>
---
 easytier-core/Cargo.toml          |  2 +-
 easytier-core/src/listener/mod.rs | 38 ++++++++++++++++++++++++++++++-
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/easytier-core/Cargo.toml b/easytier-core/Cargo.toml
index 64daee5c..19cf1705 100644
--- a/easytier-core/Cargo.toml
+++ b/easytier-core/Cargo.toml
@@ -153,7 +153,7 @@ tracing-log = ["tracing/log"]
 zstd = ["dep:zstd"]
 
 [target.'cfg(not(target_os = "wasi"))'.dev-dependencies]
-tokio = { workspace = true, features = ["rt-multi-thread"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "test-util"] }
 
 [package.metadata.cargo-machete]
 ignored = [
diff --git a/easytier-core/src/listener/mod.rs b/easytier-core/src/listener/mod.rs
index ec118e56..e59247bf 100644
--- a/easytier-core/src/listener/mod.rs
+++ b/easytier-core/src/listener/mod.rs
@@ -430,6 +430,8 @@ async fn run_accepted_task_runner(
     handler_tasks: Arc<Mutex<JoinSet<()>>>,
     cancel: CancellationToken,
 ) {
+    // Preserve the reaping deadline when new accepts win the select.
+    let mut reap_interval = crate::foundation::time::interval(Duration::from_secs(1));
     loop {
         tokio::select! {
             _ = cancel.cancelled() => break,
@@ -441,7 +443,7 @@ async fn run_accepted_task_runner(
                     None => break,
                 }
             }
-            _ = crate::foundation::time::sleep(Duration::from_secs(1)) => {
+            _ = reap_interval.tick() => {
                 let mut handler_tasks = handler_tasks.lock().await;
                 while let Some(task) = handler_tasks.try_join_next() {
                     if let Err(error) = task {
@@ -889,6 +891,40 @@ mod tests {
         }
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn continuous_accepts_do_not_prevent_completed_handler_reaping() {
+        let (spawner, rx) = AcceptedTaskSpawner::new();
+        let handler_tasks = Arc::new(tokio::sync::Mutex::new(JoinSet::new()));
+        let cancel = CancellationToken::new();
+        let runner = tokio::spawn(run_accepted_task_runner(
+            rx,
+            handler_tasks.clone(),
+            cancel.clone(),
+        ));
+
+        // Keep accepting more frequently than the one-second reaping period.
+        // Waiting for completion ensures that retained entries are finished tasks.
+        for _ in 0..100 {
+            let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
+            spawner.spawn(async move {
+                completed_tx.send(()).unwrap();
+            });
+            completed_rx.await.unwrap();
+            tokio::time::advance(Duration::from_millis(100)).await;
+            tokio::task::yield_now().await;
+        }
+
+        let retained = handler_tasks.lock().await.len();
+        cancel.cancel();
+        runner.await.unwrap();
+
+        assert!(
+            retained <= 10,
+            "continuous accepts retained {retained} completed handlers"
+        );
+        assert!(handler_tasks.lock().await.is_empty());
+    }
+
     #[tokio::test]
     async fn stop_joins_in_flight_handler_tasks_and_is_one_shot() {
         let (drop_tx, drop_rx) = tokio::sync::oneshot::channel();