From 142f71904a5e91bc5a5fb5d6e1bbb4007b83f3a1 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Wed, 15 Jul 2026 15:52:58 +0800
Subject: [PATCH 01/24] test(liveliness): candidate-a self-declare self-query
 repro attempt

---
 zenoh/tests/liveliness_self_deadlock.rs | 110 ++++++++++++++++++++++++
 1 file changed, 110 insertions(+)
 create mode 100644 zenoh/tests/liveliness_self_deadlock.rs

diff --git a/zenoh/tests/liveliness_self_deadlock.rs b/zenoh/tests/liveliness_self_deadlock.rs
new file mode 100644
index 0000000000..27c39607c0
--- /dev/null
+++ b/zenoh/tests/liveliness_self_deadlock.rs
@@ -0,0 +1,110 @@
+//! Minimal single-process repro attempt for the liveliness_query synchronous-replay deadlock.
+//!
+//! Hypothesis (Candidate A): a single Session (peer or router mode) that declares >256
+//! liveliness tokens ON ITSELF, then calls session.liveliness().get(pattern).wait() against
+//! its OWN local table, might trigger the synchronous fifo-channel-full self-deadlock
+//! without any network/gossip/multi-process machinery.
+//!
+//! Each sub-test runs the risky call on a background thread and joins with a bounded
+//! timeout so a real hang doesn't wedge the test runner.
+
+use std::sync::mpsc;
+use std::time::Duration;
+use zenoh::config::{Config, WhatAmI};
+use zenoh::Wait;
+
+const N_TOKENS: usize = 300;
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+fn run_candidate_a(whatami: WhatAmI) -> &'static str {
+    let (tx, rx) = mpsc::channel();
+
+    let handle = std::thread::spawn(move || {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        let result = rt.block_on(async {
+            let mut config = Config::default();
+            config.set_mode(Some(whatami)).unwrap();
+            // No network: no listen/connect endpoints configured.
+            let session = zenoh::open(config).await.unwrap();
+
+            let mut tokens = Vec::with_capacity(N_TOKENS);
+            for i in 0..N_TOKENS {
+                let ke = format!("test/liveliness/selfdeadlock/{i}");
+                let tok = session.liveliness().declare_token(ke).await.unwrap();
+                tokens.push(tok);
+            }
+            eprintln!("[candidate-a] declared {} tokens", tokens.len());
+
+            // This is the synchronous, blocking call under test. It happens on the tokio
+            // worker thread here since we called it inside block_on — but the hypothesis
+            // is that it self-deadlocks regardless of executor because it's the *same*
+            // calling thread doing both the declare-replay writes and (would-be) the
+            // channel drain, and replay happens before the function returns.
+            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
+            let replies = session
+                .liveliness()
+                .get("test/liveliness/selfdeadlock/**")
+                .wait();
+
+            match replies {
+                Ok(replies) => {
+                    let mut count = 0;
+                    // Drain with a short per-recv timeout too, in case get() itself
+                    // returned but the channel is wedged some other way.
+                    loop {
+                        match replies.recv_timeout(Duration::from_secs(5)) {
+                            Ok(_reply) => count += 1,
+                            Err(_) => break,
+                        }
+                    }
+                    eprintln!("[candidate-a] drained {count} replies");
+                    count
+                }
+                Err(e) => {
+                    eprintln!("[candidate-a] get() errored: {e:?}");
+                    0
+                }
+            }
+        });
+
+        let _ = tx.send(result);
+    });
+
+    match rx.recv_timeout(TIMEOUT) {
+        Ok(count) => {
+            eprintln!("[candidate-a] completed normally, {count} replies, no hang");
+            // best-effort join, don't block test exit
+            let _ = handle.join();
+            "no-hang"
+        }
+        Err(_) => {
+            eprintln!(
+                "[candidate-a] TIMED OUT after {:?} waiting for liveliness_query to return -- \
+                 background thread still parked, NOT joining (would wedge the test process)",
+                TIMEOUT
+            );
+            // Deliberately do not join() -- it would hang the test process too.
+            // Leak the thread; process exit will reap it.
+            std::mem::forget(handle);
+            "HUNG"
+        }
+    }
+}
+
+#[test]
+fn candidate_a_peer_mode() {
+    let outcome = run_candidate_a(WhatAmI::Peer);
+    eprintln!("=== candidate_a_peer_mode result: {outcome} ===");
+    assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
+}
+
+#[test]
+fn candidate_a_router_mode() {
+    let outcome = run_candidate_a(WhatAmI::Router);
+    eprintln!("=== candidate_a_router_mode result: {outcome} ===");
+    assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
+}

From c50a8dd4fcbfd654bbca9cf4757941ff3fc9e451 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Wed, 15 Jul 2026 15:53:08 +0800
Subject: [PATCH 02/24] test(liveliness): candidate-b direct-link two-session
 repro attempt

---
 zenoh/tests/liveliness_deadlock_repro.rs | 201 +++++++++++++++++++++++
 1 file changed, 201 insertions(+)
 create mode 100644 zenoh/tests/liveliness_deadlock_repro.rs

diff --git a/zenoh/tests/liveliness_deadlock_repro.rs b/zenoh/tests/liveliness_deadlock_repro.rs
new file mode 100644
index 0000000000..10f829f197
--- /dev/null
+++ b/zenoh/tests/liveliness_deadlock_repro.rs
@@ -0,0 +1,201 @@
+//
+// Candidate B repro: two sessions in the same process, connected via an
+// EXPLICIT direct TCP link (connect.endpoints), gossip and multicast
+// scouting fully disabled on both sides.
+//
+// Session A (Peer, listener) declares N liveliness tokens on itself.
+// Session B (Peer, connector, direct link only, no discovery) issues
+// `liveliness().get(pattern).wait()` with the DEFAULT handler (256-slot
+// bounded FIFO) against A's tokens, which must propagate to B purely via
+// the direct TCP link's interest/declare mechanism (no gossip).
+//
+// Two sub-variants:
+//   - `immediate`: B queries right after connecting, racing propagation.
+//   - `settled`: B first declares its own liveliness subscriber and spins
+//     until it has independently observed all N tokens locally, THEN
+//     issues the query -- controlling for the timing variable.
+//
+// Everything runs on a background OS thread with a bounded wall-clock
+// timeout so a genuine deadlock fails the test cleanly instead of wedging
+// the whole `cargo test` process.
+
+#![cfg(feature = "unstable")]
+
+use std::{
+    sync::mpsc,
+    thread,
+    time::Duration,
+};
+
+use zenoh::{config::WhatAmI, Config, Wait};
+
+const N_TOKENS: usize = 300;
+const LIVELINESS_PREFIX: &str = "test/liveliness/deadlock/repro";
+const TEST_TIMEOUT: Duration = Duration::from_secs(15);
+
+fn make_a_config(port: u16) -> Config {
+    let mut config = Config::default();
+    config.set_mode(Some(WhatAmI::Peer)).unwrap();
+    config
+        .listen
+        .endpoints
+        .set(vec![format!("tcp/127.0.0.1:{port}").parse().unwrap()])
+        .unwrap();
+    config.scouting.multicast.set_enabled(Some(false)).unwrap();
+    config.scouting.gossip.set_enabled(Some(false)).unwrap();
+    config
+}
+
+fn make_b_config(port: u16) -> Config {
+    let mut config = Config::default();
+    config.set_mode(Some(WhatAmI::Peer)).unwrap();
+    // Explicit direct link ONLY -- no listen endpoints, no discovery.
+    config
+        .connect
+        .endpoints
+        .set(vec![format!("tcp/127.0.0.1:{port}")
+            .parse::<zenoh_protocol::core::EndPoint>()
+            .unwrap()
+            .into()])
+        .unwrap();
+    config.scouting.multicast.set_enabled(Some(false)).unwrap();
+    config.scouting.gossip.set_enabled(Some(false)).unwrap();
+    config
+}
+
+/// Runs the core repro body on the calling thread. `settle_before_query`
+/// selects between the "immediate" and "settled" sub-variants.
+fn run_repro(settle_before_query: bool, handler_capacity: Option<usize>) {
+    let port = zenoh_test::get_free_tcp_port();
+
+    let session_a = zenoh::open(make_a_config(port)).wait().unwrap();
+    println!("[A] zid = {}", session_a.zid());
+
+    // Declare N liveliness tokens on session A, on itself.
+    let mut tokens = Vec::with_capacity(N_TOKENS);
+    for i in 0..N_TOKENS {
+        let ke = format!("{LIVELINESS_PREFIX}/{i}");
+        let token = session_a.liveliness().declare_token(ke).wait().unwrap();
+        tokens.push(token);
+    }
+    println!("[A] declared {N_TOKENS} liveliness tokens");
+
+    let session_b = zenoh::open(make_b_config(port)).wait().unwrap();
+    println!("[B] zid = {}", session_b.zid());
+
+    if settle_before_query {
+        // Independently confirm B has already received all N tokens
+        // locally (e.g. via its own subscriber) BEFORE issuing the query,
+        // to control for the timing/settle variable.
+        let sub = session_b
+            .liveliness()
+            .declare_subscriber(format!("{LIVELINESS_PREFIX}/**"))
+            .wait()
+            .unwrap();
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        let mut seen = 0usize;
+        while seen < N_TOKENS && std::time::Instant::now() < deadline {
+            if let Ok(Some(_sample)) = sub.recv_timeout(Duration::from_millis(200)) {
+                seen += 1;
+            }
+        }
+        println!("[B] settled: observed {seen}/{N_TOKENS} tokens via subscriber before query");
+        sub.undeclare().wait().unwrap();
+        assert_eq!(seen, N_TOKENS, "did not observe all tokens before querying");
+    }
+
+    println!(
+        "[B] issuing liveliness().get() with {} handler, settle_before_query={settle_before_query}",
+        match handler_capacity {
+            Some(c) => format!("bounded({c})"),
+            None => "DEFAULT (256)".to_string(),
+        }
+    );
+
+    let query = session_b.liveliness().get(format!("{LIVELINESS_PREFIX}/**"));
+    let replies = if let Some(cap) = handler_capacity {
+        query
+            .with(zenoh::handlers::FifoChannel::new(cap))
+            .wait()
+            .unwrap()
+    } else {
+        // Default handler = bounded FIFO, capacity 256.
+        query.wait().unwrap()
+    };
+
+    let mut count = 0;
+    while let Ok(reply) = replies.recv() {
+        if reply.result().is_ok() {
+            count += 1;
+        }
+    }
+    println!("[B] drained {count} replies");
+    assert_eq!(count, N_TOKENS, "did not receive all tokens in reply");
+
+    for token in tokens {
+        token.undeclare().wait().unwrap();
+    }
+    session_b.close().wait().unwrap();
+    session_a.close().wait().unwrap();
+}
+
+/// Spawns `run_repro` on a background thread and enforces a wall-clock
+/// timeout, reporting hang vs completion vs panic.
+fn run_bounded(name: &str, settle_before_query: bool, handler_capacity: Option<usize>) {
+    let (tx, rx) = mpsc::channel();
+    let handle = thread::spawn(move || {
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            run_repro(settle_before_query, handler_capacity)
+        }));
+        let _ = tx.send(result);
+    });
+
+    match rx.recv_timeout(TEST_TIMEOUT) {
+        Ok(Ok(())) => {
+            println!("[{name}] completed within timeout -- NO HANG");
+        }
+        Ok(Err(e)) => {
+            // Don't join; thread panicked, that's fine, propagate as test failure.
+            std::panic::resume_unwind(e);
+        }
+        Err(mpsc::RecvTimeoutError::Timeout) => {
+            panic!(
+                "[{name}] TIMED OUT after {:?} -- deadlock reproduced (thread still alive: {})",
+                TEST_TIMEOUT,
+                !handle.is_finished()
+            );
+        }
+        Err(mpsc::RecvTimeoutError::Disconnected) => {
+            panic!("[{name}] worker thread died without sending a result");
+        }
+    }
+}
+
+/// Candidate B, immediate variant, default (256) bounded handler.
+/// Expected (per prior hiroz findings on direct-link / non-gossip paths):
+/// TBD -- this is the very case under test.
+#[test]
+fn candidate_b_immediate_default_handler() {
+    zenoh_util::init_log_from_env_or("error");
+    run_bounded("candidate_b_immediate_default_handler", false, None);
+}
+
+/// Candidate B, settled variant (subscriber-confirmed all tokens observed
+/// locally before query), default (256) bounded handler.
+#[test]
+fn candidate_b_settled_default_handler() {
+    zenoh_util::init_log_from_env_or("error");
+    run_bounded("candidate_b_settled_default_handler", true, None);
+}
+
+/// Control: oversized handler capacity should never hang, immediate variant.
+#[test]
+fn candidate_b_immediate_oversized_handler_control() {
+    zenoh_util::init_log_from_env_or("error");
+    run_bounded(
+        "candidate_b_immediate_oversized_handler_control",
+        false,
+        Some(N_TOKENS + 100),
+    );
+}

From 186a624a7ba15d3e8f98d6dcd60dd9382f4cab78 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Wed, 15 Jul 2026 16:17:28 +0800
Subject: [PATCH 03/24] test(liveliness): add oversized-handler companion to
 candidate-a repro

Confirms the same self-declare/self-query topology completes cleanly
once the FifoChannel capacity exceeds the token count, isolating the
root cause to synchronous-replay-size vs. channel-capacity rather than
anything inherent to the self-declare/self-query shape.
---
 zenoh/tests/liveliness_self_deadlock.rs | 97 +++++++++++++++++++------
 1 file changed, 75 insertions(+), 22 deletions(-)

diff --git a/zenoh/tests/liveliness_self_deadlock.rs b/zenoh/tests/liveliness_self_deadlock.rs
index 27c39607c0..50d5e78474 100644
--- a/zenoh/tests/liveliness_self_deadlock.rs
+++ b/zenoh/tests/liveliness_self_deadlock.rs
@@ -1,22 +1,41 @@
-//! Minimal single-process repro attempt for the liveliness_query synchronous-replay deadlock.
+//! Minimal single-process repro for the liveliness_query synchronous-replay
+//! self-deadlock (Candidate A).
 //!
-//! Hypothesis (Candidate A): a single Session (peer or router mode) that declares >256
-//! liveliness tokens ON ITSELF, then calls session.liveliness().get(pattern).wait() against
-//! its OWN local table, might trigger the synchronous fifo-channel-full self-deadlock
-//! without any network/gossip/multi-process machinery.
+//! A single `Session` (peer or router mode, fully offline -- no listen /
+//! connect endpoints) declares many liveliness tokens ON ITSELF, then calls
+//! `session.liveliness().get(pattern).wait()` against its OWN local table.
+//! Tokens declared by a session are visible in that same session's local
+//! interest/routing table synchronously, with no settle-time or network
+//! delay required, so `send_interest`'s synchronous replay walk in
+//! `dispatcher::face::send_interest` fires the full token fan-out into the
+//! query's reply channel on the calling thread before
+//! `liveliness_query()`/`.wait()` can return. With the default bounded FIFO
+//! handler (256 slots, `API_DATA_RECEPTION_CHANNEL_SIZE` in
+//! `zenoh/src/api/session.rs`) and a token count above that capacity, the
+//! synchronous `flume::Sender::send` blocks forever waiting for a receiver
+//! that cannot run until the very call that would start it returns --
+//! a deterministic self-deadlock.
 //!
-//! Each sub-test runs the risky call on a background thread and joins with a bounded
-//! timeout so a real hang doesn't wedge the test runner.
+//! Each sub-test runs the risky call on a background thread and joins with a
+//! bounded wall-clock timeout so a real hang fails the test cleanly instead
+//! of wedging the test process. Companion tests using an oversized
+//! `FifoChannel` capacity prove the same topology completes quickly once the
+//! channel can never fill during the synchronous replay -- the same
+//! mitigation shape as the hiroz-side fix.
 
 use std::sync::mpsc;
 use std::time::Duration;
 use zenoh::config::{Config, WhatAmI};
+use zenoh::handlers::FifoChannel;
 use zenoh::Wait;
 
 const N_TOKENS: usize = 300;
 const TIMEOUT: Duration = Duration::from_secs(15);
 
-fn run_candidate_a(whatami: WhatAmI) -> &'static str {
+/// `handler_capacity = None` uses the default (256-slot) bounded FIFO
+/// handler -- the vulnerable path. `Some(cap)` uses an explicit oversized
+/// `FifoChannel`, the companion "fix shape" case that must never hang.
+fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'static str {
     let (tx, rx) = mpsc::channel();
 
     let handle = std::thread::spawn(move || {
@@ -39,22 +58,27 @@ fn run_candidate_a(whatami: WhatAmI) -> &'static str {
             }
             eprintln!("[candidate-a] declared {} tokens", tokens.len());
 
-            // This is the synchronous, blocking call under test. It happens on the tokio
-            // worker thread here since we called it inside block_on — but the hypothesis
-            // is that it self-deadlocks regardless of executor because it's the *same*
-            // calling thread doing both the declare-replay writes and (would-be) the
-            // channel drain, and replay happens before the function returns.
-            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
-            let replies = session
-                .liveliness()
-                .get("test/liveliness/selfdeadlock/**")
-                .wait();
+            // This is the synchronous, blocking call under test. The
+            // hypothesis is that it self-deadlocks regardless of executor
+            // because it's the *same* calling thread doing both the
+            // declare-replay writes and (would-be) the channel drain, and
+            // replay happens before the function returns.
+            eprintln!(
+                "[candidate-a] issuing liveliness_query (blocking wait), handler={:?}...",
+                handler_capacity
+            );
+            let query = session.liveliness().get("test/liveliness/selfdeadlock/**");
+            let replies = match handler_capacity {
+                Some(cap) => query.with(FifoChannel::new(cap)).wait(),
+                None => query.wait(),
+            };
 
             match replies {
                 Ok(replies) => {
                     let mut count = 0;
-                    // Drain with a short per-recv timeout too, in case get() itself
-                    // returned but the channel is wedged some other way.
+                    // Drain with a short per-recv timeout too, in case get()
+                    // itself returned but the channel is wedged some other
+                    // way.
                     loop {
                         match replies.recv_timeout(Duration::from_secs(5)) {
                             Ok(_reply) => count += 1,
@@ -95,16 +119,45 @@ fn run_candidate_a(whatami: WhatAmI) -> &'static str {
     }
 }
 
+/// Vulnerable path: default (256-slot) bounded FIFO handler, peer mode.
 #[test]
 fn candidate_a_peer_mode() {
-    let outcome = run_candidate_a(WhatAmI::Peer);
+    let outcome = run_candidate_a(WhatAmI::Peer, None);
     eprintln!("=== candidate_a_peer_mode result: {outcome} ===");
     assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
 }
 
+/// Vulnerable path: default (256-slot) bounded FIFO handler, router mode.
 #[test]
 fn candidate_a_router_mode() {
-    let outcome = run_candidate_a(WhatAmI::Router);
+    let outcome = run_candidate_a(WhatAmI::Router, None);
     eprintln!("=== candidate_a_router_mode result: {outcome} ===");
     assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
 }
+
+/// Companion "fix shape" case: same topology (peer mode, N_TOKENS = 300),
+/// but with an oversized `FifoChannel` (capacity > N_TOKENS) so the
+/// synchronous replay can never fill the channel. Must complete quickly and
+/// must not hang -- proves the root cause is specifically channel capacity
+/// vs. synchronous replay size, not something inherent to self-declare
+/// self-query topology.
+#[test]
+fn candidate_a_peer_mode_oversized_handler_control() {
+    let outcome = run_candidate_a(WhatAmI::Peer, Some(N_TOKENS + 100));
+    eprintln!("=== candidate_a_peer_mode_oversized_handler_control result: {outcome} ===");
+    assert_eq!(
+        outcome, "no-hang",
+        "peer mode self-declare/self-query hung even with oversized handler"
+    );
+}
+
+/// Companion "fix shape" case, router mode.
+#[test]
+fn candidate_a_router_mode_oversized_handler_control() {
+    let outcome = run_candidate_a(WhatAmI::Router, Some(N_TOKENS + 100));
+    eprintln!("=== candidate_a_router_mode_oversized_handler_control result: {outcome} ===");
+    assert_eq!(
+        outcome, "no-hang",
+        "router mode self-declare/self-query hung even with oversized handler"
+    );
+}

From c5458e672abb9c1a0abd2916113306287191f297 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Wed, 15 Jul 2026 17:12:35 +0800
Subject: [PATCH 04/24] fix(test): decouple query outcome from thread/runtime
 teardown

Dropping tokens/session and joining the background thread happened as
part of detecting the query's outcome, so their own (possibly slow or
hanging) cleanup was silently conflated with the query hang under
test. Leak session/tokens/thread once the query result is known so
teardown cost can never masquerade as (or mask) a real deadlock.
---
 zenoh/tests/liveliness_self_deadlock.rs | 42 +++++++++++++++++++------
 1 file changed, 32 insertions(+), 10 deletions(-)

diff --git a/zenoh/tests/liveliness_self_deadlock.rs b/zenoh/tests/liveliness_self_deadlock.rs
index 50d5e78474..aea64d665a 100644
--- a/zenoh/tests/liveliness_self_deadlock.rs
+++ b/zenoh/tests/liveliness_self_deadlock.rs
@@ -73,7 +73,7 @@ fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'stati
                 None => query.wait(),
             };
 
-            match replies {
+            let count = match replies {
                 Ok(replies) => {
                     let mut count = 0;
                     // Drain with a short per-recv timeout too, in case get()
@@ -92,31 +92,53 @@ fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'stati
                     eprintln!("[candidate-a] get() errored: {e:?}");
                     0
                 }
-            }
+            };
+
+            // Leak `tokens` and `session` deliberately. Dropping 300
+            // LivelinessTokens (and the Session itself) triggers a second
+            // round of synchronous undeclare/cleanup work that can itself
+            // take a long time (or hang) -- and since Drop runs as part of
+            // this async block's scope exit, it would run *before*
+            // `rt.block_on` returns to the outer thread, confounding
+            // "did the query hang" with "did cleanup hang". Leaking here
+            // decouples the two: the moment the query genuinely completes
+            // (or doesn't), that's what determines the test's timing, not
+            // teardown cost. This is a short-lived test thread/process --
+            // leaking is harmless.
+            std::mem::forget(tokens);
+            std::mem::forget(session);
+
+            count
         });
 
         let _ = tx.send(result);
     });
 
-    match rx.recv_timeout(TIMEOUT) {
+    let outcome = match rx.recv_timeout(TIMEOUT) {
         Ok(count) => {
             eprintln!("[candidate-a] completed normally, {count} replies, no hang");
-            // best-effort join, don't block test exit
-            let _ = handle.join();
             "no-hang"
         }
         Err(_) => {
             eprintln!(
                 "[candidate-a] TIMED OUT after {:?} waiting for liveliness_query to return -- \
-                 background thread still parked, NOT joining (would wedge the test process)",
+                 background thread still parked",
                 TIMEOUT
             );
-            // Deliberately do not join() -- it would hang the test process too.
-            // Leak the thread; process exit will reap it.
-            std::mem::forget(handle);
             "HUNG"
         }
-    }
+    };
+
+    // Deliberately never join() the background thread, in either branch.
+    // The query outcome is already known from the channel recv above; the
+    // background thread's tokio Runtime, when it eventually drops (in the
+    // no-hang case) or never does (in the HUNG case), blocks until all of
+    // the session's own spawned background tasks terminate -- which is
+    // teardown cost unrelated to the query itself and must not be allowed
+    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
+    // process exit reaps it.
+    std::mem::forget(handle);
+    outcome
 }
 
 /// Vulnerable path: default (256-slot) bounded FIFO handler, peer mode.

From 015d2c661b8d78036b5eeb7ae12766c2cf14d79d Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Wed, 15 Jul 2026 17:20:19 +0800
Subject: [PATCH 05/24] test(liveliness): split candidate-a repro into per-test
 binaries

Each candidate_a test deliberately leaks its session/tokens/thread to
keep query-hang detection uncontaminated by teardown cost. Running all
four as #[test]s in one process meant leaked state from earlier tests
could bleed into later ones -- confirmed directly: the router-mode
oversized-handler control hung when run after 3 other tests in the
same process, but passed cleanly (0.53s) in isolation.

Cargo treats each direct tests/*.rs file as its own binary/process, so
splitting into four single-test files gives real process isolation
without needing nextest or manual subprocess forking. Shared setup
lives in tests/liveliness_self_deadlock_support/mod.rs, which cargo's
default tests/*.rs discovery does not pick up as its own binary.
---
 zenoh/tests/liveliness_self_deadlock_peer.rs  | 21 +++++
 .../liveliness_self_deadlock_peer_control.rs  | 26 ++++++
 .../tests/liveliness_self_deadlock_router.rs  | 21 +++++
 ...liveliness_self_deadlock_router_control.rs | 23 +++++
 .../mod.rs}                                   | 84 +++----------------
 5 files changed, 103 insertions(+), 72 deletions(-)
 create mode 100644 zenoh/tests/liveliness_self_deadlock_peer.rs
 create mode 100644 zenoh/tests/liveliness_self_deadlock_peer_control.rs
 create mode 100644 zenoh/tests/liveliness_self_deadlock_router.rs
 create mode 100644 zenoh/tests/liveliness_self_deadlock_router_control.rs
 rename zenoh/tests/{liveliness_self_deadlock.rs => liveliness_self_deadlock_support/mod.rs} (56%)

diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
new file mode 100644
index 0000000000..9807ef090b
--- /dev/null
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -0,0 +1,21 @@
+//! Candidate A, vulnerable path: single `Session` in peer mode declares 300
+//! liveliness tokens on itself, then queries its own local table with the
+//! default (256-slot) bounded FIFO handler. Expected: HUNG.
+//!
+//! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
+//! Split into its own single-test file (its own cargo test binary / OS
+//! process) so this test's deliberately-leaked session/tokens/thread can
+//! never contaminate any other candidate_a test.
+
+#[path = "liveliness_self_deadlock_support/mod.rs"]
+mod support;
+
+use support::run_candidate_a;
+use zenoh::config::WhatAmI;
+
+#[test]
+fn candidate_a_peer_mode() {
+    let outcome = run_candidate_a(WhatAmI::Peer, None);
+    eprintln!("=== candidate_a_peer_mode result: {outcome} ===");
+    assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
+}
diff --git a/zenoh/tests/liveliness_self_deadlock_peer_control.rs b/zenoh/tests/liveliness_self_deadlock_peer_control.rs
new file mode 100644
index 0000000000..1698039170
--- /dev/null
+++ b/zenoh/tests/liveliness_self_deadlock_peer_control.rs
@@ -0,0 +1,26 @@
+//! Candidate A, "fix shape" control: same topology as
+//! `liveliness_self_deadlock_peer.rs` (peer mode, 300 tokens), but the
+//! query uses an oversized `FifoChannel` (capacity > token count) instead
+//! of the default handler. Must complete quickly with no hang -- proves the
+//! root cause is specifically synchronous-replay-size vs. channel-capacity,
+//! and that a larger handler capacity at the call site (the same mitigation
+//! shape hiroz applied) resolves it at the zenoh-API level too.
+//!
+//! Own single-test file (own cargo test binary / OS process) so leaked
+//! state from this test never contaminates another candidate_a test.
+
+#[path = "liveliness_self_deadlock_support/mod.rs"]
+mod support;
+
+use support::{run_candidate_a, N_TOKENS};
+use zenoh::config::WhatAmI;
+
+#[test]
+fn candidate_a_peer_mode_oversized_handler_control() {
+    let outcome = run_candidate_a(WhatAmI::Peer, Some(N_TOKENS + 100));
+    eprintln!("=== candidate_a_peer_mode_oversized_handler_control result: {outcome} ===");
+    assert_eq!(
+        outcome, "no-hang",
+        "peer mode self-declare/self-query hung even with oversized handler"
+    );
+}
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
new file mode 100644
index 0000000000..f2720337ea
--- /dev/null
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -0,0 +1,21 @@
+//! Candidate A, vulnerable path: single `Session` in router mode declares
+//! 300 liveliness tokens on itself, then queries its own local table with
+//! the default (256-slot) bounded FIFO handler. Expected: HUNG.
+//!
+//! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
+//! Split into its own single-test file (its own cargo test binary / OS
+//! process) so this test's deliberately-leaked session/tokens/thread can
+//! never contaminate any other candidate_a test.
+
+#[path = "liveliness_self_deadlock_support/mod.rs"]
+mod support;
+
+use support::run_candidate_a;
+use zenoh::config::WhatAmI;
+
+#[test]
+fn candidate_a_router_mode() {
+    let outcome = run_candidate_a(WhatAmI::Router, None);
+    eprintln!("=== candidate_a_router_mode result: {outcome} ===");
+    assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
+}
diff --git a/zenoh/tests/liveliness_self_deadlock_router_control.rs b/zenoh/tests/liveliness_self_deadlock_router_control.rs
new file mode 100644
index 0000000000..3bf00a65e5
--- /dev/null
+++ b/zenoh/tests/liveliness_self_deadlock_router_control.rs
@@ -0,0 +1,23 @@
+//! Candidate A, "fix shape" control: same topology as
+//! `liveliness_self_deadlock_router.rs` (router mode, 300 tokens), but the
+//! query uses an oversized `FifoChannel` (capacity > token count) instead
+//! of the default handler. Must complete quickly with no hang.
+//!
+//! Own single-test file (own cargo test binary / OS process) so leaked
+//! state from this test never contaminates another candidate_a test.
+
+#[path = "liveliness_self_deadlock_support/mod.rs"]
+mod support;
+
+use support::{run_candidate_a, N_TOKENS};
+use zenoh::config::WhatAmI;
+
+#[test]
+fn candidate_a_router_mode_oversized_handler_control() {
+    let outcome = run_candidate_a(WhatAmI::Router, Some(N_TOKENS + 100));
+    eprintln!("=== candidate_a_router_mode_oversized_handler_control result: {outcome} ===");
+    assert_eq!(
+        outcome, "no-hang",
+        "router mode self-declare/self-query hung even with oversized handler"
+    );
+}
diff --git a/zenoh/tests/liveliness_self_deadlock.rs b/zenoh/tests/liveliness_self_deadlock_support/mod.rs
similarity index 56%
rename from zenoh/tests/liveliness_self_deadlock.rs
rename to zenoh/tests/liveliness_self_deadlock_support/mod.rs
index aea64d665a..d30643bcc5 100644
--- a/zenoh/tests/liveliness_self_deadlock.rs
+++ b/zenoh/tests/liveliness_self_deadlock_support/mod.rs
@@ -1,27 +1,9 @@
-//! Minimal single-process repro for the liveliness_query synchronous-replay
-//! self-deadlock (Candidate A).
-//!
-//! A single `Session` (peer or router mode, fully offline -- no listen /
-//! connect endpoints) declares many liveliness tokens ON ITSELF, then calls
-//! `session.liveliness().get(pattern).wait()` against its OWN local table.
-//! Tokens declared by a session are visible in that same session's local
-//! interest/routing table synchronously, with no settle-time or network
-//! delay required, so `send_interest`'s synchronous replay walk in
-//! `dispatcher::face::send_interest` fires the full token fan-out into the
-//! query's reply channel on the calling thread before
-//! `liveliness_query()`/`.wait()` can return. With the default bounded FIFO
-//! handler (256 slots, `API_DATA_RECEPTION_CHANNEL_SIZE` in
-//! `zenoh/src/api/session.rs`) and a token count above that capacity, the
-//! synchronous `flume::Sender::send` blocks forever waiting for a receiver
-//! that cannot run until the very call that would start it returns --
-//! a deterministic self-deadlock.
-//!
-//! Each sub-test runs the risky call on a background thread and joins with a
-//! bounded wall-clock timeout so a real hang fails the test cleanly instead
-//! of wedging the test process. Companion tests using an oversized
-//! `FifoChannel` capacity prove the same topology completes quickly once the
-//! channel can never fill during the synchronous replay -- the same
-//! mitigation shape as the hiroz-side fix.
+//! Shared helper for the Candidate A liveliness self-deadlock repro tests.
+//! Not a test binary itself (lives in a subdirectory, so cargo's default
+//! `tests/*.rs` auto-discovery skips it) -- included via `#[path = ...] mod
+//! support;` from each of the four single-test files that ARE separate
+//! cargo test binaries (their own OS processes), so that leaked
+//! sessions/tokens/threads from one test can never contaminate another.
 
 use std::sync::mpsc;
 use std::time::Duration;
@@ -29,13 +11,13 @@ use zenoh::config::{Config, WhatAmI};
 use zenoh::handlers::FifoChannel;
 use zenoh::Wait;
 
-const N_TOKENS: usize = 300;
-const TIMEOUT: Duration = Duration::from_secs(15);
+pub const N_TOKENS: usize = 300;
+pub const TIMEOUT: Duration = Duration::from_secs(15);
 
 /// `handler_capacity = None` uses the default (256-slot) bounded FIFO
 /// handler -- the vulnerable path. `Some(cap)` uses an explicit oversized
 /// `FifoChannel`, the companion "fix shape" case that must never hang.
-fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'static str {
+pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'static str {
     let (tx, rx) = mpsc::channel();
 
     let handle = std::thread::spawn(move || {
@@ -103,8 +85,9 @@ fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'stati
             // "did the query hang" with "did cleanup hang". Leaking here
             // decouples the two: the moment the query genuinely completes
             // (or doesn't), that's what determines the test's timing, not
-            // teardown cost. This is a short-lived test thread/process --
-            // leaking is harmless.
+            // teardown cost. Each candidate_a_* test is its own cargo test
+            // binary (own OS process), so leaking here cannot contaminate
+            // any other test.
             std::mem::forget(tokens);
             std::mem::forget(session);
 
@@ -140,46 +123,3 @@ fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'stati
     std::mem::forget(handle);
     outcome
 }
-
-/// Vulnerable path: default (256-slot) bounded FIFO handler, peer mode.
-#[test]
-fn candidate_a_peer_mode() {
-    let outcome = run_candidate_a(WhatAmI::Peer, None);
-    eprintln!("=== candidate_a_peer_mode result: {outcome} ===");
-    assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
-}
-
-/// Vulnerable path: default (256-slot) bounded FIFO handler, router mode.
-#[test]
-fn candidate_a_router_mode() {
-    let outcome = run_candidate_a(WhatAmI::Router, None);
-    eprintln!("=== candidate_a_router_mode result: {outcome} ===");
-    assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
-}
-
-/// Companion "fix shape" case: same topology (peer mode, N_TOKENS = 300),
-/// but with an oversized `FifoChannel` (capacity > N_TOKENS) so the
-/// synchronous replay can never fill the channel. Must complete quickly and
-/// must not hang -- proves the root cause is specifically channel capacity
-/// vs. synchronous replay size, not something inherent to self-declare
-/// self-query topology.
-#[test]
-fn candidate_a_peer_mode_oversized_handler_control() {
-    let outcome = run_candidate_a(WhatAmI::Peer, Some(N_TOKENS + 100));
-    eprintln!("=== candidate_a_peer_mode_oversized_handler_control result: {outcome} ===");
-    assert_eq!(
-        outcome, "no-hang",
-        "peer mode self-declare/self-query hung even with oversized handler"
-    );
-}
-
-/// Companion "fix shape" case, router mode.
-#[test]
-fn candidate_a_router_mode_oversized_handler_control() {
-    let outcome = run_candidate_a(WhatAmI::Router, Some(N_TOKENS + 100));
-    eprintln!("=== candidate_a_router_mode_oversized_handler_control result: {outcome} ===");
-    assert_eq!(
-        outcome, "no-hang",
-        "router mode self-declare/self-query hung even with oversized handler"
-    );
-}

From 3c84ce62ef64754ce257d071c6a97aedb65ad8ca Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 02:11:30 +0800
Subject: [PATCH 06/24] fix(liveliness): decouple send_interest replay from
 calling thread

Session::liveliness_query synchronously replayed every matching token
onto the query's reply channel on the calling thread before returning,
so a bounded FIFO handler self-deadlocked once the token count exceeded
its capacity. Offload the replay via spawn_abortable_with_rt +
spawn_blocking so the caller can return and start draining concurrently.

spawn_blocking (rather than running inline on the async task) avoids a
second regression: send_declare can invoke arbitrary blocking user
callbacks, which would otherwise starve ZRuntime::Net's small worker
pool and stall Session::close_inner's own teardown.
---
 zenoh/src/net/routing/dispatcher/face.rs      | 28 ++++++++++++--
 zenoh/tests/liveliness_self_deadlock_peer.rs  | 19 +++++++---
 .../liveliness_self_deadlock_peer_control.rs  | 26 -------------
 .../tests/liveliness_self_deadlock_router.rs  | 19 +++++++---
 ...liveliness_self_deadlock_router_control.rs | 23 ------------
 .../liveliness_self_deadlock_support/mod.rs   | 37 ++++++++++---------
 6 files changed, 70 insertions(+), 82 deletions(-)
 delete mode 100644 zenoh/tests/liveliness_self_deadlock_peer_control.rs
 delete mode 100644 zenoh/tests/liveliness_self_deadlock_router_control.rs

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index 206e658fad..db0735e35b 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -548,9 +548,31 @@ impl Primitives for Face {
             let mut declares = vec![];
             self.interest(msg, &mut |p, m| declares.push((p.clone(), m)));
             drop(ctrl_lock);
-            for (p, m) in declares {
-                m.with_mut(|m| p.send_declare(m));
-            }
+            self.state.task_controller.spawn_abortable_with_rt(
+                zenoh_runtime::ZRuntime::Net,
+                async move {
+                    // NOTE: `send_declare` ultimately invokes arbitrary user
+                    // callbacks (FIFO handler sends, or fully user-supplied
+                    // closures) that may block synchronously for an
+                    // unbounded time (see tests/callback_drop_on_undeclare.rs).
+                    // Running that directly in this async task would occupy
+                    // one of ZRuntime::Net's small async-worker threads for
+                    // that entire duration, starving other Net-scheduled
+                    // control-plane work (e.g. Session::close_inner's
+                    // teardown, the liveliness-query timeout task) that
+                    // shares the same worker pool. Offload the actual replay
+                    // onto Net's dedicated blocking-thread pool via
+                    // `spawn_blocking`, which is a separate pool from the
+                    // async workers and therefore can never starve them.
+                    let _ = zenoh_runtime::ZRuntime::Net
+                        .spawn_blocking(move || {
+                            for (p, m) in declares {
+                                m.with_mut(|m| p.send_declare(m));
+                            }
+                        })
+                        .await;
+                },
+            );
         } else {
             self.interest_final(msg);
         }
diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index 9807ef090b..df543cd568 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -1,6 +1,9 @@
-//! Candidate A, vulnerable path: single `Session` in peer mode declares 300
-//! liveliness tokens on itself, then queries its own local table with the
-//! default (256-slot) bounded FIFO handler. Expected: HUNG.
+//! Candidate A, vulnerable path (regression test post-fix): single `Session`
+//! in peer mode declares 300 liveliness tokens on itself, then queries its
+//! own local table with the default (256-slot) bounded FIFO handler.
+//! Pre-fix, this self-deadlocked (see KNOWLEDGE.md); post-fix (async
+//! replay in `Face::send_interest`), it must complete promptly and deliver
+//! every reply.
 //!
 //! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
 //! Split into its own single-test file (its own cargo test binary / OS
@@ -10,12 +13,16 @@
 #[path = "liveliness_self_deadlock_support/mod.rs"]
 mod support;
 
-use support::run_candidate_a;
+use support::{run_candidate_a, N_TOKENS};
 use zenoh::config::WhatAmI;
 
 #[test]
 fn candidate_a_peer_mode() {
-    let outcome = run_candidate_a(WhatAmI::Peer, None);
-    eprintln!("=== candidate_a_peer_mode result: {outcome} ===");
+    let (outcome, reply_count) = run_candidate_a(WhatAmI::Peer);
+    eprintln!("=== candidate_a_peer_mode result: {outcome}, {reply_count} replies ===");
     assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
+    assert_eq!(
+        reply_count, N_TOKENS,
+        "expected all {N_TOKENS} tokens to be replayed as replies, got {reply_count}"
+    );
 }
diff --git a/zenoh/tests/liveliness_self_deadlock_peer_control.rs b/zenoh/tests/liveliness_self_deadlock_peer_control.rs
deleted file mode 100644
index 1698039170..0000000000
--- a/zenoh/tests/liveliness_self_deadlock_peer_control.rs
+++ /dev/null
@@ -1,26 +0,0 @@
-//! Candidate A, "fix shape" control: same topology as
-//! `liveliness_self_deadlock_peer.rs` (peer mode, 300 tokens), but the
-//! query uses an oversized `FifoChannel` (capacity > token count) instead
-//! of the default handler. Must complete quickly with no hang -- proves the
-//! root cause is specifically synchronous-replay-size vs. channel-capacity,
-//! and that a larger handler capacity at the call site (the same mitigation
-//! shape hiroz applied) resolves it at the zenoh-API level too.
-//!
-//! Own single-test file (own cargo test binary / OS process) so leaked
-//! state from this test never contaminates another candidate_a test.
-
-#[path = "liveliness_self_deadlock_support/mod.rs"]
-mod support;
-
-use support::{run_candidate_a, N_TOKENS};
-use zenoh::config::WhatAmI;
-
-#[test]
-fn candidate_a_peer_mode_oversized_handler_control() {
-    let outcome = run_candidate_a(WhatAmI::Peer, Some(N_TOKENS + 100));
-    eprintln!("=== candidate_a_peer_mode_oversized_handler_control result: {outcome} ===");
-    assert_eq!(
-        outcome, "no-hang",
-        "peer mode self-declare/self-query hung even with oversized handler"
-    );
-}
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index f2720337ea..717e2e0c50 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -1,6 +1,9 @@
-//! Candidate A, vulnerable path: single `Session` in router mode declares
-//! 300 liveliness tokens on itself, then queries its own local table with
-//! the default (256-slot) bounded FIFO handler. Expected: HUNG.
+//! Candidate A, vulnerable path (regression test post-fix): single `Session`
+//! in router mode declares 300 liveliness tokens on itself, then queries
+//! its own local table with the default (256-slot) bounded FIFO handler.
+//! Pre-fix, this self-deadlocked (see KNOWLEDGE.md); post-fix (async
+//! replay in `Face::send_interest`), it must complete promptly and deliver
+//! every reply.
 //!
 //! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
 //! Split into its own single-test file (its own cargo test binary / OS
@@ -10,12 +13,16 @@
 #[path = "liveliness_self_deadlock_support/mod.rs"]
 mod support;
 
-use support::run_candidate_a;
+use support::{run_candidate_a, N_TOKENS};
 use zenoh::config::WhatAmI;
 
 #[test]
 fn candidate_a_router_mode() {
-    let outcome = run_candidate_a(WhatAmI::Router, None);
-    eprintln!("=== candidate_a_router_mode result: {outcome} ===");
+    let (outcome, reply_count) = run_candidate_a(WhatAmI::Router);
+    eprintln!("=== candidate_a_router_mode result: {outcome}, {reply_count} replies ===");
     assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
+    assert_eq!(
+        reply_count, N_TOKENS,
+        "expected all {N_TOKENS} tokens to be replayed as replies, got {reply_count}"
+    );
 }
diff --git a/zenoh/tests/liveliness_self_deadlock_router_control.rs b/zenoh/tests/liveliness_self_deadlock_router_control.rs
deleted file mode 100644
index 3bf00a65e5..0000000000
--- a/zenoh/tests/liveliness_self_deadlock_router_control.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-//! Candidate A, "fix shape" control: same topology as
-//! `liveliness_self_deadlock_router.rs` (router mode, 300 tokens), but the
-//! query uses an oversized `FifoChannel` (capacity > token count) instead
-//! of the default handler. Must complete quickly with no hang.
-//!
-//! Own single-test file (own cargo test binary / OS process) so leaked
-//! state from this test never contaminates another candidate_a test.
-
-#[path = "liveliness_self_deadlock_support/mod.rs"]
-mod support;
-
-use support::{run_candidate_a, N_TOKENS};
-use zenoh::config::WhatAmI;
-
-#[test]
-fn candidate_a_router_mode_oversized_handler_control() {
-    let outcome = run_candidate_a(WhatAmI::Router, Some(N_TOKENS + 100));
-    eprintln!("=== candidate_a_router_mode_oversized_handler_control result: {outcome} ===");
-    assert_eq!(
-        outcome, "no-hang",
-        "router mode self-declare/self-query hung even with oversized handler"
-    );
-}
diff --git a/zenoh/tests/liveliness_self_deadlock_support/mod.rs b/zenoh/tests/liveliness_self_deadlock_support/mod.rs
index d30643bcc5..94992ea7f8 100644
--- a/zenoh/tests/liveliness_self_deadlock_support/mod.rs
+++ b/zenoh/tests/liveliness_self_deadlock_support/mod.rs
@@ -8,16 +8,21 @@
 use std::sync::mpsc;
 use std::time::Duration;
 use zenoh::config::{Config, WhatAmI};
-use zenoh::handlers::FifoChannel;
 use zenoh::Wait;
 
 pub const N_TOKENS: usize = 300;
 pub const TIMEOUT: Duration = Duration::from_secs(15);
 
-/// `handler_capacity = None` uses the default (256-slot) bounded FIFO
-/// handler -- the vulnerable path. `Some(cap)` uses an explicit oversized
-/// `FifoChannel`, the companion "fix shape" case that must never hang.
-pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'static str {
+/// Declares `N_TOKENS` liveliness tokens on a session, then issues a
+/// liveliness query against its own local table using the default
+/// (256-slot) bounded FIFO handler -- the vulnerable path.
+///
+/// Returns `(outcome, reply_count)`: `outcome` is `"no-hang"` if the query
+/// returned and was drained within `TIMEOUT`, `"HUNG"` otherwise.
+/// `reply_count` is the number of replies actually drained (`0` if the
+/// query hung, since the background thread's result is never observed in
+/// that case).
+pub fn run_candidate_a(whatami: WhatAmI) -> (&'static str, usize) {
     let (tx, rx) = mpsc::channel();
 
     let handle = std::thread::spawn(move || {
@@ -45,15 +50,11 @@ pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'s
             // because it's the *same* calling thread doing both the
             // declare-replay writes and (would-be) the channel drain, and
             // replay happens before the function returns.
-            eprintln!(
-                "[candidate-a] issuing liveliness_query (blocking wait), handler={:?}...",
-                handler_capacity
-            );
-            let query = session.liveliness().get("test/liveliness/selfdeadlock/**");
-            let replies = match handler_capacity {
-                Some(cap) => query.with(FifoChannel::new(cap)).wait(),
-                None => query.wait(),
-            };
+            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
+            let replies = session
+                .liveliness()
+                .get("test/liveliness/selfdeadlock/**")
+                .wait();
 
             let count = match replies {
                 Ok(replies) => {
@@ -97,10 +98,10 @@ pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'s
         let _ = tx.send(result);
     });
 
-    let outcome = match rx.recv_timeout(TIMEOUT) {
+    let (outcome, reply_count) = match rx.recv_timeout(TIMEOUT) {
         Ok(count) => {
             eprintln!("[candidate-a] completed normally, {count} replies, no hang");
-            "no-hang"
+            ("no-hang", count)
         }
         Err(_) => {
             eprintln!(
@@ -108,7 +109,7 @@ pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'s
                  background thread still parked",
                 TIMEOUT
             );
-            "HUNG"
+            ("HUNG", 0)
         }
     };
 
@@ -121,5 +122,5 @@ pub fn run_candidate_a(whatami: WhatAmI, handler_capacity: Option<usize>) -> &'s
     // to confound (or wedge) this test's pass/fail signal. Leak the thread;
     // process exit reaps it.
     std::mem::forget(handle);
-    outcome
+    (outcome, reply_count)
 }

From e82b8fdb5f8df0c9c230daf6312ff5076fa110b7 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 02:24:15 +0800
Subject: [PATCH 07/24] fix(liveliness): log panics from declare-replay instead
 of discarding

The spawn_blocking result was silently dropped, so a panic inside
send_declare would vanish instead of being visible anywhere. Also trim
the comment explaining the ZRuntime::Net-vs-blocking-pool rationale
down to what's load-bearing.
---
 zenoh/src/net/routing/dispatcher/face.rs | 24 +++++++++---------------
 1 file changed, 9 insertions(+), 15 deletions(-)

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index db0735e35b..c51c06d137 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -551,26 +551,20 @@ impl Primitives for Face {
             self.state.task_controller.spawn_abortable_with_rt(
                 zenoh_runtime::ZRuntime::Net,
                 async move {
-                    // NOTE: `send_declare` ultimately invokes arbitrary user
-                    // callbacks (FIFO handler sends, or fully user-supplied
-                    // closures) that may block synchronously for an
-                    // unbounded time (see tests/callback_drop_on_undeclare.rs).
-                    // Running that directly in this async task would occupy
-                    // one of ZRuntime::Net's small async-worker threads for
-                    // that entire duration, starving other Net-scheduled
-                    // control-plane work (e.g. Session::close_inner's
-                    // teardown, the liveliness-query timeout task) that
-                    // shares the same worker pool. Offload the actual replay
-                    // onto Net's dedicated blocking-thread pool via
-                    // `spawn_blocking`, which is a separate pool from the
-                    // async workers and therefore can never starve them.
-                    let _ = zenoh_runtime::ZRuntime::Net
+                    // `send_declare` may invoke a blocking user callback, which
+                    // would otherwise starve ZRuntime::Net's small async-worker
+                    // pool (shared with Session::close_inner's teardown). Run it
+                    // on Net's separate blocking-thread pool instead.
+                    if let Err(e) = zenoh_runtime::ZRuntime::Net
                         .spawn_blocking(move || {
                             for (p, m) in declares {
                                 m.with_mut(|m| p.send_declare(m));
                             }
                         })
-                        .await;
+                        .await
+                    {
+                        tracing::error!(?e, "panic while replaying declares for send_interest");
+                    }
                 },
             );
         } else {

From 51d326dfe1e145d397c29cbbdfff20b6559e2137 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 02:41:19 +0800
Subject: [PATCH 08/24] test(liveliness): inline shared helper, drop support
 module

Per review: the shared support/mod.rs added an extra file for a helper
only two tests used, and running them in the same process (rather than
separate cargo test binaries) previously caused real cross-test
contamination. Inline the ~90-line helper into each test file directly
instead, keeping the two files self-contained and process-isolated
without a third module file.
---
 zenoh/tests/liveliness_self_deadlock_peer.rs  | 121 +++++++++++++++--
 .../tests/liveliness_self_deadlock_router.rs  | 121 +++++++++++++++--
 .../liveliness_self_deadlock_support/mod.rs   | 126 ------------------
 3 files changed, 224 insertions(+), 144 deletions(-)
 delete mode 100644 zenoh/tests/liveliness_self_deadlock_support/mod.rs

diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index df543cd568..1a9e7120e1 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -5,20 +5,123 @@
 //! replay in `Face::send_interest`), it must complete promptly and deliver
 //! every reply.
 //!
-//! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
-//! Split into its own single-test file (its own cargo test binary / OS
-//! process) so this test's deliberately-leaked session/tokens/thread can
-//! never contaminate any other candidate_a test.
+//! Deliberately its own single-test file (its own cargo test binary / OS
+//! process, not merged with `liveliness_self_deadlock_router.rs`) so this
+//! test's deliberately-leaked session/tokens/thread can never contaminate
+//! another test -- running both in the same process previously caused real
+//! cross-test interference (leaked sessions autoconnecting to each other
+//! via scouting/multicast, which is enabled by default).
 
-#[path = "liveliness_self_deadlock_support/mod.rs"]
-mod support;
+use std::sync::mpsc;
+use std::time::Duration;
+use zenoh::config::{Config, WhatAmI};
+use zenoh::Wait;
 
-use support::{run_candidate_a, N_TOKENS};
-use zenoh::config::WhatAmI;
+const N_TOKENS: usize = 300;
+const TIMEOUT: Duration = Duration::from_secs(15);
 
 #[test]
 fn candidate_a_peer_mode() {
-    let (outcome, reply_count) = run_candidate_a(WhatAmI::Peer);
+    let (tx, rx) = mpsc::channel();
+
+    let handle = std::thread::spawn(move || {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        let result = rt.block_on(async {
+            let mut config = Config::default();
+            config.set_mode(Some(WhatAmI::Peer)).unwrap();
+            // No network: no listen/connect endpoints configured.
+            let session = zenoh::open(config).await.unwrap();
+
+            let mut tokens = Vec::with_capacity(N_TOKENS);
+            for i in 0..N_TOKENS {
+                let ke = format!("test/liveliness/selfdeadlock/{i}");
+                let tok = session.liveliness().declare_token(ke).await.unwrap();
+                tokens.push(tok);
+            }
+            eprintln!("[candidate-a] declared {} tokens", tokens.len());
+
+            // This is the synchronous, blocking call under test. The
+            // hypothesis is that it self-deadlocks regardless of executor
+            // because it's the *same* calling thread doing both the
+            // declare-replay writes and (would-be) the channel drain, and
+            // replay happens before the function returns.
+            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
+            let replies = session
+                .liveliness()
+                .get("test/liveliness/selfdeadlock/**")
+                .wait();
+
+            let count = match replies {
+                Ok(replies) => {
+                    let mut count = 0;
+                    // Drain with a short per-recv timeout too, in case get()
+                    // itself returned but the channel is wedged some other
+                    // way.
+                    loop {
+                        match replies.recv_timeout(Duration::from_secs(5)) {
+                            Ok(_reply) => count += 1,
+                            Err(_) => break,
+                        }
+                    }
+                    eprintln!("[candidate-a] drained {count} replies");
+                    count
+                }
+                Err(e) => {
+                    eprintln!("[candidate-a] get() errored: {e:?}");
+                    0
+                }
+            };
+
+            // Leak `tokens` and `session` deliberately. Dropping 300
+            // LivelinessTokens (and the Session itself) triggers a second
+            // round of synchronous undeclare/cleanup work that can itself
+            // take a long time (or hang) -- and since Drop runs as part of
+            // this async block's scope exit, it would run *before*
+            // `rt.block_on` returns to the outer thread, confounding
+            // "did the query hang" with "did cleanup hang". Leaking here
+            // decouples the two: the moment the query genuinely completes
+            // (or doesn't), that's what determines the test's timing, not
+            // teardown cost. This test is its own cargo test binary (own
+            // OS process), so leaking here cannot contaminate any other
+            // test.
+            std::mem::forget(tokens);
+            std::mem::forget(session);
+
+            count
+        });
+
+        let _ = tx.send(result);
+    });
+
+    let (outcome, reply_count) = match rx.recv_timeout(TIMEOUT) {
+        Ok(count) => {
+            eprintln!("[candidate-a] completed normally, {count} replies, no hang");
+            ("no-hang", count)
+        }
+        Err(_) => {
+            eprintln!(
+                "[candidate-a] TIMED OUT after {:?} waiting for liveliness_query to return -- \
+                 background thread still parked",
+                TIMEOUT
+            );
+            ("HUNG", 0)
+        }
+    };
+
+    // Deliberately never join() the background thread, in either branch.
+    // The query outcome is already known from the channel recv above; the
+    // background thread's tokio Runtime, when it eventually drops (in the
+    // no-hang case) or never does (in the HUNG case), blocks until all of
+    // the session's own spawned background tasks terminate -- which is
+    // teardown cost unrelated to the query itself and must not be allowed
+    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
+    // process exit reaps it.
+    std::mem::forget(handle);
+
     eprintln!("=== candidate_a_peer_mode result: {outcome}, {reply_count} replies ===");
     assert_eq!(outcome, "no-hang", "peer mode self-declare/self-query hung");
     assert_eq!(
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index 717e2e0c50..91b7584f6b 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -5,20 +5,123 @@
 //! replay in `Face::send_interest`), it must complete promptly and deliver
 //! every reply.
 //!
-//! See `liveliness_self_deadlock_support/mod.rs` for the full mechanism.
-//! Split into its own single-test file (its own cargo test binary / OS
-//! process) so this test's deliberately-leaked session/tokens/thread can
-//! never contaminate any other candidate_a test.
+//! Deliberately its own single-test file (its own cargo test binary / OS
+//! process, not merged with `liveliness_self_deadlock_peer.rs`) so this
+//! test's deliberately-leaked session/tokens/thread can never contaminate
+//! another test -- running both in the same process previously caused real
+//! cross-test interference (leaked sessions autoconnecting to each other
+//! via scouting/multicast, which is enabled by default).
 
-#[path = "liveliness_self_deadlock_support/mod.rs"]
-mod support;
+use std::sync::mpsc;
+use std::time::Duration;
+use zenoh::config::{Config, WhatAmI};
+use zenoh::Wait;
 
-use support::{run_candidate_a, N_TOKENS};
-use zenoh::config::WhatAmI;
+const N_TOKENS: usize = 300;
+const TIMEOUT: Duration = Duration::from_secs(15);
 
 #[test]
 fn candidate_a_router_mode() {
-    let (outcome, reply_count) = run_candidate_a(WhatAmI::Router);
+    let (tx, rx) = mpsc::channel();
+
+    let handle = std::thread::spawn(move || {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        let result = rt.block_on(async {
+            let mut config = Config::default();
+            config.set_mode(Some(WhatAmI::Router)).unwrap();
+            // No network: no listen/connect endpoints configured.
+            let session = zenoh::open(config).await.unwrap();
+
+            let mut tokens = Vec::with_capacity(N_TOKENS);
+            for i in 0..N_TOKENS {
+                let ke = format!("test/liveliness/selfdeadlock/{i}");
+                let tok = session.liveliness().declare_token(ke).await.unwrap();
+                tokens.push(tok);
+            }
+            eprintln!("[candidate-a] declared {} tokens", tokens.len());
+
+            // This is the synchronous, blocking call under test. The
+            // hypothesis is that it self-deadlocks regardless of executor
+            // because it's the *same* calling thread doing both the
+            // declare-replay writes and (would-be) the channel drain, and
+            // replay happens before the function returns.
+            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
+            let replies = session
+                .liveliness()
+                .get("test/liveliness/selfdeadlock/**")
+                .wait();
+
+            let count = match replies {
+                Ok(replies) => {
+                    let mut count = 0;
+                    // Drain with a short per-recv timeout too, in case get()
+                    // itself returned but the channel is wedged some other
+                    // way.
+                    loop {
+                        match replies.recv_timeout(Duration::from_secs(5)) {
+                            Ok(_reply) => count += 1,
+                            Err(_) => break,
+                        }
+                    }
+                    eprintln!("[candidate-a] drained {count} replies");
+                    count
+                }
+                Err(e) => {
+                    eprintln!("[candidate-a] get() errored: {e:?}");
+                    0
+                }
+            };
+
+            // Leak `tokens` and `session` deliberately. Dropping 300
+            // LivelinessTokens (and the Session itself) triggers a second
+            // round of synchronous undeclare/cleanup work that can itself
+            // take a long time (or hang) -- and since Drop runs as part of
+            // this async block's scope exit, it would run *before*
+            // `rt.block_on` returns to the outer thread, confounding
+            // "did the query hang" with "did cleanup hang". Leaking here
+            // decouples the two: the moment the query genuinely completes
+            // (or doesn't), that's what determines the test's timing, not
+            // teardown cost. This test is its own cargo test binary (own
+            // OS process), so leaking here cannot contaminate any other
+            // test.
+            std::mem::forget(tokens);
+            std::mem::forget(session);
+
+            count
+        });
+
+        let _ = tx.send(result);
+    });
+
+    let (outcome, reply_count) = match rx.recv_timeout(TIMEOUT) {
+        Ok(count) => {
+            eprintln!("[candidate-a] completed normally, {count} replies, no hang");
+            ("no-hang", count)
+        }
+        Err(_) => {
+            eprintln!(
+                "[candidate-a] TIMED OUT after {:?} waiting for liveliness_query to return -- \
+                 background thread still parked",
+                TIMEOUT
+            );
+            ("HUNG", 0)
+        }
+    };
+
+    // Deliberately never join() the background thread, in either branch.
+    // The query outcome is already known from the channel recv above; the
+    // background thread's tokio Runtime, when it eventually drops (in the
+    // no-hang case) or never does (in the HUNG case), blocks until all of
+    // the session's own spawned background tasks terminate -- which is
+    // teardown cost unrelated to the query itself and must not be allowed
+    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
+    // process exit reaps it.
+    std::mem::forget(handle);
+
     eprintln!("=== candidate_a_router_mode result: {outcome}, {reply_count} replies ===");
     assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
     assert_eq!(
diff --git a/zenoh/tests/liveliness_self_deadlock_support/mod.rs b/zenoh/tests/liveliness_self_deadlock_support/mod.rs
deleted file mode 100644
index 94992ea7f8..0000000000
--- a/zenoh/tests/liveliness_self_deadlock_support/mod.rs
+++ /dev/null
@@ -1,126 +0,0 @@
-//! Shared helper for the Candidate A liveliness self-deadlock repro tests.
-//! Not a test binary itself (lives in a subdirectory, so cargo's default
-//! `tests/*.rs` auto-discovery skips it) -- included via `#[path = ...] mod
-//! support;` from each of the four single-test files that ARE separate
-//! cargo test binaries (their own OS processes), so that leaked
-//! sessions/tokens/threads from one test can never contaminate another.
-
-use std::sync::mpsc;
-use std::time::Duration;
-use zenoh::config::{Config, WhatAmI};
-use zenoh::Wait;
-
-pub const N_TOKENS: usize = 300;
-pub const TIMEOUT: Duration = Duration::from_secs(15);
-
-/// Declares `N_TOKENS` liveliness tokens on a session, then issues a
-/// liveliness query against its own local table using the default
-/// (256-slot) bounded FIFO handler -- the vulnerable path.
-///
-/// Returns `(outcome, reply_count)`: `outcome` is `"no-hang"` if the query
-/// returned and was drained within `TIMEOUT`, `"HUNG"` otherwise.
-/// `reply_count` is the number of replies actually drained (`0` if the
-/// query hung, since the background thread's result is never observed in
-/// that case).
-pub fn run_candidate_a(whatami: WhatAmI) -> (&'static str, usize) {
-    let (tx, rx) = mpsc::channel();
-
-    let handle = std::thread::spawn(move || {
-        let rt = tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-
-        let result = rt.block_on(async {
-            let mut config = Config::default();
-            config.set_mode(Some(whatami)).unwrap();
-            // No network: no listen/connect endpoints configured.
-            let session = zenoh::open(config).await.unwrap();
-
-            let mut tokens = Vec::with_capacity(N_TOKENS);
-            for i in 0..N_TOKENS {
-                let ke = format!("test/liveliness/selfdeadlock/{i}");
-                let tok = session.liveliness().declare_token(ke).await.unwrap();
-                tokens.push(tok);
-            }
-            eprintln!("[candidate-a] declared {} tokens", tokens.len());
-
-            // This is the synchronous, blocking call under test. The
-            // hypothesis is that it self-deadlocks regardless of executor
-            // because it's the *same* calling thread doing both the
-            // declare-replay writes and (would-be) the channel drain, and
-            // replay happens before the function returns.
-            eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
-            let replies = session
-                .liveliness()
-                .get("test/liveliness/selfdeadlock/**")
-                .wait();
-
-            let count = match replies {
-                Ok(replies) => {
-                    let mut count = 0;
-                    // Drain with a short per-recv timeout too, in case get()
-                    // itself returned but the channel is wedged some other
-                    // way.
-                    loop {
-                        match replies.recv_timeout(Duration::from_secs(5)) {
-                            Ok(_reply) => count += 1,
-                            Err(_) => break,
-                        }
-                    }
-                    eprintln!("[candidate-a] drained {count} replies");
-                    count
-                }
-                Err(e) => {
-                    eprintln!("[candidate-a] get() errored: {e:?}");
-                    0
-                }
-            };
-
-            // Leak `tokens` and `session` deliberately. Dropping 300
-            // LivelinessTokens (and the Session itself) triggers a second
-            // round of synchronous undeclare/cleanup work that can itself
-            // take a long time (or hang) -- and since Drop runs as part of
-            // this async block's scope exit, it would run *before*
-            // `rt.block_on` returns to the outer thread, confounding
-            // "did the query hang" with "did cleanup hang". Leaking here
-            // decouples the two: the moment the query genuinely completes
-            // (or doesn't), that's what determines the test's timing, not
-            // teardown cost. Each candidate_a_* test is its own cargo test
-            // binary (own OS process), so leaking here cannot contaminate
-            // any other test.
-            std::mem::forget(tokens);
-            std::mem::forget(session);
-
-            count
-        });
-
-        let _ = tx.send(result);
-    });
-
-    let (outcome, reply_count) = match rx.recv_timeout(TIMEOUT) {
-        Ok(count) => {
-            eprintln!("[candidate-a] completed normally, {count} replies, no hang");
-            ("no-hang", count)
-        }
-        Err(_) => {
-            eprintln!(
-                "[candidate-a] TIMED OUT after {:?} waiting for liveliness_query to return -- \
-                 background thread still parked",
-                TIMEOUT
-            );
-            ("HUNG", 0)
-        }
-    };
-
-    // Deliberately never join() the background thread, in either branch.
-    // The query outcome is already known from the channel recv above; the
-    // background thread's tokio Runtime, when it eventually drops (in the
-    // no-hang case) or never does (in the HUNG case), blocks until all of
-    // the session's own spawned background tasks terminate -- which is
-    // teardown cost unrelated to the query itself and must not be allowed
-    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
-    // process exit reaps it.
-    std::mem::forget(handle);
-    (outcome, reply_count)
-}

From 8161d4c5bca5e42267ab3dea1594da3874b6c5bc Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 03:52:25 +0800
Subject: [PATCH 09/24] fix(net): wait for async interest replay in region test
 harness

Face::send_interest's local-face replay now runs on a spawned task
(the liveliness-deadlock fix) rather than synchronously, but tests in
src/net/tests/regions use MockFace::interest[_wildcard] and then
immediately assert on the recorder synchronously. Give the replay a
moment to land before returning to the caller.

Also fixes rustfmt (import grouping, assert_eq wrapping) flagged by
CI on the two liveliness_self_deadlock test files.
---
 zenoh/src/net/tests/regions/mod.rs             |  8 ++++++++
 zenoh/tests/liveliness_self_deadlock_peer.rs   | 10 ++++++----
 zenoh/tests/liveliness_self_deadlock_router.rs | 15 ++++++++++-----
 3 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs
index 0acceb456a..be8862bdcd 100644
--- a/zenoh/src/net/tests/regions/mod.rs
+++ b/zenoh/src/net/tests/regions/mod.rs
@@ -32,6 +32,8 @@ use std::{
     any::Any,
     fmt::Debug,
     sync::{Arc, Mutex},
+    thread,
+    time::Duration,
 };
 
 use futures::executor::block_on;
@@ -680,6 +682,11 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
+        // `send_interest`'s local-face replay (declares + DeclareFinal) runs on a
+        // spawned task now, not synchronously on this thread -- give it a moment
+        // to land before returning, since callers immediately inspect the
+        // recorder synchronously.
+        thread::sleep(Duration::from_millis(50));
     }
 
     /// Declare an interest without a key expression.
@@ -698,6 +705,7 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
+        thread::sleep(Duration::from_millis(50));
     }
 
     /// Declare a liveliness token for `key_expr` from this face's perspective.
diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index 1a9e7120e1..2a0976a5b6 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -12,10 +12,12 @@
 //! cross-test interference (leaked sessions autoconnecting to each other
 //! via scouting/multicast, which is enabled by default).
 
-use std::sync::mpsc;
-use std::time::Duration;
-use zenoh::config::{Config, WhatAmI};
-use zenoh::Wait;
+use std::{sync::mpsc, time::Duration};
+
+use zenoh::{
+    config::{Config, WhatAmI},
+    Wait,
+};
 
 const N_TOKENS: usize = 300;
 const TIMEOUT: Duration = Duration::from_secs(15);
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index 91b7584f6b..3afb2564ee 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -12,10 +12,12 @@
 //! cross-test interference (leaked sessions autoconnecting to each other
 //! via scouting/multicast, which is enabled by default).
 
-use std::sync::mpsc;
-use std::time::Duration;
-use zenoh::config::{Config, WhatAmI};
-use zenoh::Wait;
+use std::{sync::mpsc, time::Duration};
+
+use zenoh::{
+    config::{Config, WhatAmI},
+    Wait,
+};
 
 const N_TOKENS: usize = 300;
 const TIMEOUT: Duration = Duration::from_secs(15);
@@ -123,7 +125,10 @@ fn candidate_a_router_mode() {
     std::mem::forget(handle);
 
     eprintln!("=== candidate_a_router_mode result: {outcome}, {reply_count} replies ===");
-    assert_eq!(outcome, "no-hang", "router mode self-declare/self-query hung");
+    assert_eq!(
+        outcome, "no-hang",
+        "router mode self-declare/self-query hung"
+    );
     assert_eq!(
         reply_count, N_TOKENS,
         "expected all {N_TOKENS} tokens to be replayed as replies, got {reply_count}"

From f2b77da1d7ca825a0581e73dad89fbef804cd83e Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 03:54:50 +0800
Subject: [PATCH 10/24] test(liveliness): remove abandoned candidate-b repro
 file

liveliness_deadlock_repro.rs was leftover exploratory code from an
early, never-confirmed investigation branch (candidate B, direct-link
two-session repro) -- superseded by the candidate-a-based regression
tests that actually ship. It was also never rustfmt-clean, which was
failing CI. Drop it; the confirmed regression coverage lives in
liveliness_self_deadlock_{peer,router}.rs.
---
 zenoh/tests/liveliness_deadlock_repro.rs | 201 -----------------------
 1 file changed, 201 deletions(-)
 delete mode 100644 zenoh/tests/liveliness_deadlock_repro.rs

diff --git a/zenoh/tests/liveliness_deadlock_repro.rs b/zenoh/tests/liveliness_deadlock_repro.rs
deleted file mode 100644
index 10f829f197..0000000000
--- a/zenoh/tests/liveliness_deadlock_repro.rs
+++ /dev/null
@@ -1,201 +0,0 @@
-//
-// Candidate B repro: two sessions in the same process, connected via an
-// EXPLICIT direct TCP link (connect.endpoints), gossip and multicast
-// scouting fully disabled on both sides.
-//
-// Session A (Peer, listener) declares N liveliness tokens on itself.
-// Session B (Peer, connector, direct link only, no discovery) issues
-// `liveliness().get(pattern).wait()` with the DEFAULT handler (256-slot
-// bounded FIFO) against A's tokens, which must propagate to B purely via
-// the direct TCP link's interest/declare mechanism (no gossip).
-//
-// Two sub-variants:
-//   - `immediate`: B queries right after connecting, racing propagation.
-//   - `settled`: B first declares its own liveliness subscriber and spins
-//     until it has independently observed all N tokens locally, THEN
-//     issues the query -- controlling for the timing variable.
-//
-// Everything runs on a background OS thread with a bounded wall-clock
-// timeout so a genuine deadlock fails the test cleanly instead of wedging
-// the whole `cargo test` process.
-
-#![cfg(feature = "unstable")]
-
-use std::{
-    sync::mpsc,
-    thread,
-    time::Duration,
-};
-
-use zenoh::{config::WhatAmI, Config, Wait};
-
-const N_TOKENS: usize = 300;
-const LIVELINESS_PREFIX: &str = "test/liveliness/deadlock/repro";
-const TEST_TIMEOUT: Duration = Duration::from_secs(15);
-
-fn make_a_config(port: u16) -> Config {
-    let mut config = Config::default();
-    config.set_mode(Some(WhatAmI::Peer)).unwrap();
-    config
-        .listen
-        .endpoints
-        .set(vec![format!("tcp/127.0.0.1:{port}").parse().unwrap()])
-        .unwrap();
-    config.scouting.multicast.set_enabled(Some(false)).unwrap();
-    config.scouting.gossip.set_enabled(Some(false)).unwrap();
-    config
-}
-
-fn make_b_config(port: u16) -> Config {
-    let mut config = Config::default();
-    config.set_mode(Some(WhatAmI::Peer)).unwrap();
-    // Explicit direct link ONLY -- no listen endpoints, no discovery.
-    config
-        .connect
-        .endpoints
-        .set(vec![format!("tcp/127.0.0.1:{port}")
-            .parse::<zenoh_protocol::core::EndPoint>()
-            .unwrap()
-            .into()])
-        .unwrap();
-    config.scouting.multicast.set_enabled(Some(false)).unwrap();
-    config.scouting.gossip.set_enabled(Some(false)).unwrap();
-    config
-}
-
-/// Runs the core repro body on the calling thread. `settle_before_query`
-/// selects between the "immediate" and "settled" sub-variants.
-fn run_repro(settle_before_query: bool, handler_capacity: Option<usize>) {
-    let port = zenoh_test::get_free_tcp_port();
-
-    let session_a = zenoh::open(make_a_config(port)).wait().unwrap();
-    println!("[A] zid = {}", session_a.zid());
-
-    // Declare N liveliness tokens on session A, on itself.
-    let mut tokens = Vec::with_capacity(N_TOKENS);
-    for i in 0..N_TOKENS {
-        let ke = format!("{LIVELINESS_PREFIX}/{i}");
-        let token = session_a.liveliness().declare_token(ke).wait().unwrap();
-        tokens.push(token);
-    }
-    println!("[A] declared {N_TOKENS} liveliness tokens");
-
-    let session_b = zenoh::open(make_b_config(port)).wait().unwrap();
-    println!("[B] zid = {}", session_b.zid());
-
-    if settle_before_query {
-        // Independently confirm B has already received all N tokens
-        // locally (e.g. via its own subscriber) BEFORE issuing the query,
-        // to control for the timing/settle variable.
-        let sub = session_b
-            .liveliness()
-            .declare_subscriber(format!("{LIVELINESS_PREFIX}/**"))
-            .wait()
-            .unwrap();
-
-        let deadline = std::time::Instant::now() + Duration::from_secs(10);
-        let mut seen = 0usize;
-        while seen < N_TOKENS && std::time::Instant::now() < deadline {
-            if let Ok(Some(_sample)) = sub.recv_timeout(Duration::from_millis(200)) {
-                seen += 1;
-            }
-        }
-        println!("[B] settled: observed {seen}/{N_TOKENS} tokens via subscriber before query");
-        sub.undeclare().wait().unwrap();
-        assert_eq!(seen, N_TOKENS, "did not observe all tokens before querying");
-    }
-
-    println!(
-        "[B] issuing liveliness().get() with {} handler, settle_before_query={settle_before_query}",
-        match handler_capacity {
-            Some(c) => format!("bounded({c})"),
-            None => "DEFAULT (256)".to_string(),
-        }
-    );
-
-    let query = session_b.liveliness().get(format!("{LIVELINESS_PREFIX}/**"));
-    let replies = if let Some(cap) = handler_capacity {
-        query
-            .with(zenoh::handlers::FifoChannel::new(cap))
-            .wait()
-            .unwrap()
-    } else {
-        // Default handler = bounded FIFO, capacity 256.
-        query.wait().unwrap()
-    };
-
-    let mut count = 0;
-    while let Ok(reply) = replies.recv() {
-        if reply.result().is_ok() {
-            count += 1;
-        }
-    }
-    println!("[B] drained {count} replies");
-    assert_eq!(count, N_TOKENS, "did not receive all tokens in reply");
-
-    for token in tokens {
-        token.undeclare().wait().unwrap();
-    }
-    session_b.close().wait().unwrap();
-    session_a.close().wait().unwrap();
-}
-
-/// Spawns `run_repro` on a background thread and enforces a wall-clock
-/// timeout, reporting hang vs completion vs panic.
-fn run_bounded(name: &str, settle_before_query: bool, handler_capacity: Option<usize>) {
-    let (tx, rx) = mpsc::channel();
-    let handle = thread::spawn(move || {
-        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-            run_repro(settle_before_query, handler_capacity)
-        }));
-        let _ = tx.send(result);
-    });
-
-    match rx.recv_timeout(TEST_TIMEOUT) {
-        Ok(Ok(())) => {
-            println!("[{name}] completed within timeout -- NO HANG");
-        }
-        Ok(Err(e)) => {
-            // Don't join; thread panicked, that's fine, propagate as test failure.
-            std::panic::resume_unwind(e);
-        }
-        Err(mpsc::RecvTimeoutError::Timeout) => {
-            panic!(
-                "[{name}] TIMED OUT after {:?} -- deadlock reproduced (thread still alive: {})",
-                TEST_TIMEOUT,
-                !handle.is_finished()
-            );
-        }
-        Err(mpsc::RecvTimeoutError::Disconnected) => {
-            panic!("[{name}] worker thread died without sending a result");
-        }
-    }
-}
-
-/// Candidate B, immediate variant, default (256) bounded handler.
-/// Expected (per prior hiroz findings on direct-link / non-gossip paths):
-/// TBD -- this is the very case under test.
-#[test]
-fn candidate_b_immediate_default_handler() {
-    zenoh_util::init_log_from_env_or("error");
-    run_bounded("candidate_b_immediate_default_handler", false, None);
-}
-
-/// Candidate B, settled variant (subscriber-confirmed all tokens observed
-/// locally before query), default (256) bounded handler.
-#[test]
-fn candidate_b_settled_default_handler() {
-    zenoh_util::init_log_from_env_or("error");
-    run_bounded("candidate_b_settled_default_handler", true, None);
-}
-
-/// Control: oversized handler capacity should never hang, immediate variant.
-#[test]
-fn candidate_b_immediate_oversized_handler_control() {
-    zenoh_util::init_log_from_env_or("error");
-    run_bounded(
-        "candidate_b_immediate_oversized_handler_control",
-        false,
-        Some(N_TOKENS + 100),
-    );
-}

From 0da9fb5c0878ff68680e062fdbec3b0f09ebc31d Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 03:57:38 +0800
Subject: [PATCH 11/24] fix(liveliness): satisfy clippy while_let_loop lint

Rewrite the manual loop{match{...break}} reply-drain as a while loop,
per clippy::while_let_loop (-D warnings in CI).
---
 zenoh/tests/liveliness_self_deadlock_peer.rs   | 7 ++-----
 zenoh/tests/liveliness_self_deadlock_router.rs | 7 ++-----
 2 files changed, 4 insertions(+), 10 deletions(-)

diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index 2a0976a5b6..198dbde258 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -63,11 +63,8 @@ fn candidate_a_peer_mode() {
                     // Drain with a short per-recv timeout too, in case get()
                     // itself returned but the channel is wedged some other
                     // way.
-                    loop {
-                        match replies.recv_timeout(Duration::from_secs(5)) {
-                            Ok(_reply) => count += 1,
-                            Err(_) => break,
-                        }
+                    while replies.recv_timeout(Duration::from_secs(5)).is_ok() {
+                        count += 1;
                     }
                     eprintln!("[candidate-a] drained {count} replies");
                     count
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index 3afb2564ee..6ffe19b4b2 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -63,11 +63,8 @@ fn candidate_a_router_mode() {
                     // Drain with a short per-recv timeout too, in case get()
                     // itself returned but the channel is wedged some other
                     // way.
-                    loop {
-                        match replies.recv_timeout(Duration::from_secs(5)) {
-                            Ok(_reply) => count += 1,
-                            Err(_) => break,
-                        }
+                    while replies.recv_timeout(Duration::from_secs(5)).is_ok() {
+                        count += 1;
                     }
                     eprintln!("[candidate-a] drained {count} replies");
                     count

From 8d7c9e47084abe7722244bb8c3042645f5cf9d0f Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 04:03:07 +0800
Subject: [PATCH 12/24] fix(net): also settle async replay in region test
 inject() helper

Missed one send_interest call site in the earlier fix -- the
forwarding-test inject() helper injects Interest messages directly
and its callers assert on recorded message counts immediately after.
---
 zenoh/src/net/tests/regions/mod.rs | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs
index be8862bdcd..c549fde0c4 100644
--- a/zenoh/src/net/tests/regions/mod.rs
+++ b/zenoh/src/net/tests/regions/mod.rs
@@ -1107,7 +1107,13 @@ impl EstablishedConnection {
             Message::Request(r) => target.face.send_request(&mut r.clone()),
             Message::Response(r) => target.face.send_response(&mut r.clone()),
             Message::ResponseFinal(r) => target.face.send_response_final(&mut r.clone()),
-            Message::Interest(i) => target.face.send_interest(&mut i.clone()),
+            Message::Interest(i) => {
+                target.face.send_interest(&mut i.clone());
+                // See the comment on `MockFace::interest` -- send_interest's
+                // local-face replay now runs on a spawned task, not
+                // synchronously; give it a moment to land before returning.
+                thread::sleep(Duration::from_millis(50));
+            }
             Message::Oam(o) => {
                 if let Some(demux) = &target.demux {
                     let _ = demux.handle_message(NetworkMessageMut {

From c1f29238871ef1c6aa940218fa2d350eef1a8285 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 14:44:50 +0800
Subject: [PATCH 13/24] test(routing): give three_node_combination more timing
 headroom

three_node_combination consistently timed out on CI (0/3 nextest
retries passing on both macos and ubuntu-shared-memory), stalled on
LivelinessSub's initial declare_subscriber -- the one-time Current
interest replay that now runs on a spawned task instead of
synchronously. The recipe's 10s budget (shared TIMEOUT const, also
used for MSG_COUNT=50 message delivery) was already tight with
randomized per-node startup delays; the async replay's scheduling hop
pushed it over. Double the budget to 20s.
---
 zenoh/tests/routing.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/zenoh/tests/routing.rs b/zenoh/tests/routing.rs
index 69ef5ba035..aefc498902 100644
--- a/zenoh/tests/routing.rs
+++ b/zenoh/tests/routing.rs
@@ -30,7 +30,7 @@ use zenoh_link::EndPoint;
 use zenoh_result::bail;
 use zenoh_test::{get_free_tcp_port, get_tcp_locator};
 
-const TIMEOUT: Duration = Duration::from_secs(10);
+const TIMEOUT: Duration = Duration::from_secs(20);
 const MSG_COUNT: usize = 50;
 const LIVELINESSGET_DELAY: Duration = Duration::from_millis(10);
 

From 1ed4dd198c5aef8fddeb4f69105c7de31bb6009d Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 15:05:13 +0800
Subject: [PATCH 14/24] test(routing): bump three_node_combination timeout
 further

20s still wasn't enough under the heavier unstable-feature test
binary (836 tests vs 452 in the plain build, more concurrent load in
the same process) -- all 3 nextest retries failed consistently, not
marginally. Go to 45s for real headroom instead of inching up again.
---
 zenoh/tests/routing.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/zenoh/tests/routing.rs b/zenoh/tests/routing.rs
index aefc498902..869438eb73 100644
--- a/zenoh/tests/routing.rs
+++ b/zenoh/tests/routing.rs
@@ -30,7 +30,7 @@ use zenoh_link::EndPoint;
 use zenoh_result::bail;
 use zenoh_test::{get_free_tcp_port, get_tcp_locator};
 
-const TIMEOUT: Duration = Duration::from_secs(20);
+const TIMEOUT: Duration = Duration::from_secs(45);
 const MSG_COUNT: usize = 50;
 const LIVELINESSGET_DELAY: Duration = Duration::from_millis(10);
 

From 64c8cd1c4eb7033d81691ab70fe0a7bc6ec07c7a Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 16:25:46 +0800
Subject: [PATCH 15/24] test(routing): raise TIMEOUT to 90s for
 three_node_combination flakiness

CI observed failures at ~2x the prior 45s budget on macOS, ubuntu
(shared memory), and windows (unstable), consistent with chained
ztimeout! calls plus the recipe's own outer watchdog racing on the
same shared constant under load.
---
 zenoh/tests/routing.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/zenoh/tests/routing.rs b/zenoh/tests/routing.rs
index 869438eb73..908a0ffea7 100644
--- a/zenoh/tests/routing.rs
+++ b/zenoh/tests/routing.rs
@@ -30,7 +30,7 @@ use zenoh_link::EndPoint;
 use zenoh_result::bail;
 use zenoh_test::{get_free_tcp_port, get_tcp_locator};
 
-const TIMEOUT: Duration = Duration::from_secs(45);
+const TIMEOUT: Duration = Duration::from_secs(90);
 const MSG_COUNT: usize = 50;
 const LIVELINESSGET_DELAY: Duration = Duration::from_millis(10);
 

From 830e429f19d357d642f057a5ec6e972f0e1076f2 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 16:35:30 +0800
Subject: [PATCH 16/24] fix(face): use spawn_with_rt instead of
 spawn_abortable_with_rt for interest replay

Aborting the replay task on teardown drops its JoinHandle without
stopping the nested spawn_blocking closure, which runs to completion
regardless -- leaving Arc clones of session/face state alive past
Session::close(). This raced dead_after_drop_many_times, which asserts
all state is dropped immediately after close(). spawn_with_rt makes
close()'s task_controller wait for genuine completion instead; the
core deadlock fix's correctness does not depend on abort semantics.
---
 zenoh/src/net/routing/dispatcher/face.rs | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index c51c06d137..78aedf0c86 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -548,7 +548,15 @@ impl Primitives for Face {
             let mut declares = vec![];
             self.interest(msg, &mut |p, m| declares.push((p.clone(), m)));
             drop(ctrl_lock);
-            self.state.task_controller.spawn_abortable_with_rt(
+            // Not spawn_abortable_with_rt: aborting this task on teardown would
+            // drop the JoinHandle below without stopping the spawn_blocking
+            // closure it awaits (spawn_blocking closures run to completion
+            // regardless of handle drop), leaving Arc clones captured by
+            // `declares` alive past Session::close() -- racing tests/callers
+            // that expect all state dropped immediately after close(). Plain
+            // spawn_with_rt makes Session::close()'s task_controller wait for
+            // genuine completion instead, bounded by its own timeout either way.
+            self.state.task_controller.spawn_with_rt(
                 zenoh_runtime::ZRuntime::Net,
                 async move {
                     // `send_declare` may invoke a blocking user callback, which

From a1585ce21be53e2d367327014a7435acd1559f98 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Thu, 16 Jul 2026 16:38:13 +0800
Subject: [PATCH 17/24] style(face): rustfmt fix for spawn_with_rt call

---
 zenoh/src/net/routing/dispatcher/face.rs | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index 78aedf0c86..e414b25da2 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -556,9 +556,9 @@ impl Primitives for Face {
             // that expect all state dropped immediately after close(). Plain
             // spawn_with_rt makes Session::close()'s task_controller wait for
             // genuine completion instead, bounded by its own timeout either way.
-            self.state.task_controller.spawn_with_rt(
-                zenoh_runtime::ZRuntime::Net,
-                async move {
+            self.state
+                .task_controller
+                .spawn_with_rt(zenoh_runtime::ZRuntime::Net, async move {
                     // `send_declare` may invoke a blocking user callback, which
                     // would otherwise starve ZRuntime::Net's small async-worker
                     // pool (shared with Session::close_inner's teardown). Run it
@@ -573,8 +573,7 @@ impl Primitives for Face {
                     {
                         tracing::error!(?e, "panic while replaying declares for send_interest");
                     }
-                },
-            );
+                });
         } else {
             self.interest_final(msg);
         }

From 064f667b2037f4959d23014fc2d3fdc231158f92 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 02:42:04 +0800
Subject: [PATCH 18/24] test(routing): scope timeout bump to
 three_node_combination only

The prior fix widened the shared TIMEOUT const to 90s for every test
in routing.rs, but only three_node_combination actually needs the
extra headroom: its recipes exercise Face::send_interest's now-async
declare replay, which queues on ZRuntime::Net's single worker thread
and can occasionally exceed the old 10s budget under CI contention.
Restore TIMEOUT to 10s and give three_node_combination its own
THREE_NODE_COMBINATION_TIMEOUT via a new Recipe::run_with_timeout,
leaving the deadline unchanged for gossip*, static_failover_brokering,
two_node_combination, three_node_combination_multicast, and
router_linkstate.
---
 zenoh/tests/routing.rs | 30 ++++++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/zenoh/tests/routing.rs b/zenoh/tests/routing.rs
index 908a0ffea7..34b12b5e98 100644
--- a/zenoh/tests/routing.rs
+++ b/zenoh/tests/routing.rs
@@ -30,9 +30,17 @@ use zenoh_link::EndPoint;
 use zenoh_result::bail;
 use zenoh_test::{get_free_tcp_port, get_tcp_locator};
 
-const TIMEOUT: Duration = Duration::from_secs(90);
+const TIMEOUT: Duration = Duration::from_secs(10);
 const MSG_COUNT: usize = 50;
 const LIVELINESSGET_DELAY: Duration = Duration::from_millis(10);
+// `three_node_combination` runs its recipes' liveliness-token declare/notify
+// through `Face::send_interest`'s replay, which is now dispatched onto a
+// shared async runtime instead of running inline on the calling thread (see
+// `Face::send_interest`). Under CI's contended runners that adds scheduling
+// latency, occasionally past the default `TIMEOUT`; give this test alone a
+// wider budget rather than loosening the deadline for every other test in
+// this file.
+const THREE_NODE_COMBINATION_TIMEOUT: Duration = Duration::from_secs(90);
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 enum Task {
@@ -344,6 +352,10 @@ impl Recipe {
     }
 
     async fn run(&self) -> Result<()> {
+        self.run_with_timeout(TIMEOUT).await
+    }
+
+    async fn run_with_timeout(&self, timeout: Duration) -> Result<()> {
         let num_checkpoints = self.num_checkpoints();
         let remaining_checkpoints = Arc::new(AtomicUsize::new(num_checkpoints));
         println!(
@@ -439,7 +451,7 @@ impl Recipe {
         // All tasks of the recipe run together
         loop {
             tokio::select! {
-                _ = tokio::time::sleep(TIMEOUT) => {
+                _ = tokio::time::sleep(timeout) => {
                     println!("Recipe {self} Timeout.");
 
                     // Termination
@@ -1030,10 +1042,16 @@ async fn three_node_combination() -> Result<()> {
         let mut join_set = tokio::task::JoinSet::new();
         for (pubsub, getqueryable, getliveliness, subliveliness) in chunks {
             join_set.spawn(async move {
-                pubsub.run().await?;
-                getqueryable.run().await?;
-                getliveliness.run().await?;
-                subliveliness.run().await?;
+                pubsub.run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT).await?;
+                getqueryable
+                    .run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT)
+                    .await?;
+                getliveliness
+                    .run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT)
+                    .await?;
+                subliveliness
+                    .run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT)
+                    .await?;
                 Result::Ok(())
             });
         }

From ec50ea95146b233ea57e3f07c8230f65e55f27f0 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 02:44:15 +0800
Subject: [PATCH 19/24] style(routing): rustfmt wrap for run_with_timeout call

---
 zenoh/tests/routing.rs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/zenoh/tests/routing.rs b/zenoh/tests/routing.rs
index 34b12b5e98..9f98c0cc50 100644
--- a/zenoh/tests/routing.rs
+++ b/zenoh/tests/routing.rs
@@ -1042,7 +1042,9 @@ async fn three_node_combination() -> Result<()> {
         let mut join_set = tokio::task::JoinSet::new();
         for (pubsub, getqueryable, getliveliness, subliveliness) in chunks {
             join_set.spawn(async move {
-                pubsub.run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT).await?;
+                pubsub
+                    .run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT)
+                    .await?;
                 getqueryable
                     .run_with_timeout(THREE_NODE_COMBINATION_TIMEOUT)
                     .await?;

From c288dd3c633878828c2702ea065f2a0bb5392884 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 15:57:53 +0800
Subject: [PATCH 20/24] docs(tests): trim overlong comments in deadlock
 regression tests

---
 zenoh/tests/liveliness_self_deadlock_peer.rs  | 54 ++++++-------------
 .../tests/liveliness_self_deadlock_router.rs  | 54 ++++++-------------
 2 files changed, 34 insertions(+), 74 deletions(-)

diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index 198dbde258..542f05f8f2 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -1,16 +1,11 @@
-//! Candidate A, vulnerable path (regression test post-fix): single `Session`
-//! in peer mode declares 300 liveliness tokens on itself, then queries its
-//! own local table with the default (256-slot) bounded FIFO handler.
-//! Pre-fix, this self-deadlocked (see KNOWLEDGE.md); post-fix (async
-//! replay in `Face::send_interest`), it must complete promptly and deliver
-//! every reply.
+//! Regression test: a peer-mode `Session` declares 300 liveliness tokens on
+//! itself, then queries its own local table with the default 256-slot
+//! handler -- the exact path that self-deadlocked pre-fix (see
+//! KNOWLEDGE.md). Must complete promptly and deliver every reply.
 //!
-//! Deliberately its own single-test file (its own cargo test binary / OS
-//! process, not merged with `liveliness_self_deadlock_router.rs`) so this
-//! test's deliberately-leaked session/tokens/thread can never contaminate
-//! another test -- running both in the same process previously caused real
-//! cross-test interference (leaked sessions autoconnecting to each other
-//! via scouting/multicast, which is enabled by default).
+//! Own test binary (own OS process) so its deliberately-leaked
+//! session/tokens/thread can't autoconnect to and contaminate
+//! `liveliness_self_deadlock_router.rs` via default scouting/multicast.
 
 use std::{sync::mpsc, time::Duration};
 
@@ -46,11 +41,9 @@ fn candidate_a_peer_mode() {
             }
             eprintln!("[candidate-a] declared {} tokens", tokens.len());
 
-            // This is the synchronous, blocking call under test. The
-            // hypothesis is that it self-deadlocks regardless of executor
-            // because it's the *same* calling thread doing both the
-            // declare-replay writes and (would-be) the channel drain, and
-            // replay happens before the function returns.
+            // The blocking call under test: replay and the reply-channel
+            // drain would run on this same thread, and replay happens
+            // before the call returns.
             eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
             let replies = session
                 .liveliness()
@@ -75,18 +68,10 @@ fn candidate_a_peer_mode() {
                 }
             };
 
-            // Leak `tokens` and `session` deliberately. Dropping 300
-            // LivelinessTokens (and the Session itself) triggers a second
-            // round of synchronous undeclare/cleanup work that can itself
-            // take a long time (or hang) -- and since Drop runs as part of
-            // this async block's scope exit, it would run *before*
-            // `rt.block_on` returns to the outer thread, confounding
-            // "did the query hang" with "did cleanup hang". Leaking here
-            // decouples the two: the moment the query genuinely completes
-            // (or doesn't), that's what determines the test's timing, not
-            // teardown cost. This test is its own cargo test binary (own
-            // OS process), so leaking here cannot contaminate any other
-            // test.
+            // Leak deliberately: dropping 300 tokens triggers its own
+            // synchronous cleanup, which would run before `block_on`
+            // returns and confound "did the query hang" with "did
+            // teardown hang". Safe to leak -- own process, exits after.
             std::mem::forget(tokens);
             std::mem::forget(session);
 
@@ -111,14 +96,9 @@ fn candidate_a_peer_mode() {
         }
     };
 
-    // Deliberately never join() the background thread, in either branch.
-    // The query outcome is already known from the channel recv above; the
-    // background thread's tokio Runtime, when it eventually drops (in the
-    // no-hang case) or never does (in the HUNG case), blocks until all of
-    // the session's own spawned background tasks terminate -- which is
-    // teardown cost unrelated to the query itself and must not be allowed
-    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
-    // process exit reaps it.
+    // Never join(): the Runtime's own drop would block on teardown,
+    // which is unrelated to (and could wedge) the query outcome we
+    // already have. Leak the thread; process exit reaps it.
     std::mem::forget(handle);
 
     eprintln!("=== candidate_a_peer_mode result: {outcome}, {reply_count} replies ===");
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index 6ffe19b4b2..e28009576f 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -1,16 +1,11 @@
-//! Candidate A, vulnerable path (regression test post-fix): single `Session`
-//! in router mode declares 300 liveliness tokens on itself, then queries
-//! its own local table with the default (256-slot) bounded FIFO handler.
-//! Pre-fix, this self-deadlocked (see KNOWLEDGE.md); post-fix (async
-//! replay in `Face::send_interest`), it must complete promptly and deliver
-//! every reply.
+//! Regression test: a router-mode `Session` declares 300 liveliness tokens
+//! on itself, then queries its own local table with the default 256-slot
+//! handler -- the exact path that self-deadlocked pre-fix (see
+//! KNOWLEDGE.md). Must complete promptly and deliver every reply.
 //!
-//! Deliberately its own single-test file (its own cargo test binary / OS
-//! process, not merged with `liveliness_self_deadlock_peer.rs`) so this
-//! test's deliberately-leaked session/tokens/thread can never contaminate
-//! another test -- running both in the same process previously caused real
-//! cross-test interference (leaked sessions autoconnecting to each other
-//! via scouting/multicast, which is enabled by default).
+//! Own test binary (own OS process) so its deliberately-leaked
+//! session/tokens/thread can't autoconnect to and contaminate
+//! `liveliness_self_deadlock_peer.rs` via default scouting/multicast.
 
 use std::{sync::mpsc, time::Duration};
 
@@ -46,11 +41,9 @@ fn candidate_a_router_mode() {
             }
             eprintln!("[candidate-a] declared {} tokens", tokens.len());
 
-            // This is the synchronous, blocking call under test. The
-            // hypothesis is that it self-deadlocks regardless of executor
-            // because it's the *same* calling thread doing both the
-            // declare-replay writes and (would-be) the channel drain, and
-            // replay happens before the function returns.
+            // The blocking call under test: replay and the reply-channel
+            // drain would run on this same thread, and replay happens
+            // before the call returns.
             eprintln!("[candidate-a] issuing liveliness_query (blocking wait)...");
             let replies = session
                 .liveliness()
@@ -75,18 +68,10 @@ fn candidate_a_router_mode() {
                 }
             };
 
-            // Leak `tokens` and `session` deliberately. Dropping 300
-            // LivelinessTokens (and the Session itself) triggers a second
-            // round of synchronous undeclare/cleanup work that can itself
-            // take a long time (or hang) -- and since Drop runs as part of
-            // this async block's scope exit, it would run *before*
-            // `rt.block_on` returns to the outer thread, confounding
-            // "did the query hang" with "did cleanup hang". Leaking here
-            // decouples the two: the moment the query genuinely completes
-            // (or doesn't), that's what determines the test's timing, not
-            // teardown cost. This test is its own cargo test binary (own
-            // OS process), so leaking here cannot contaminate any other
-            // test.
+            // Leak deliberately: dropping 300 tokens triggers its own
+            // synchronous cleanup, which would run before `block_on`
+            // returns and confound "did the query hang" with "did
+            // teardown hang". Safe to leak -- own process, exits after.
             std::mem::forget(tokens);
             std::mem::forget(session);
 
@@ -111,14 +96,9 @@ fn candidate_a_router_mode() {
         }
     };
 
-    // Deliberately never join() the background thread, in either branch.
-    // The query outcome is already known from the channel recv above; the
-    // background thread's tokio Runtime, when it eventually drops (in the
-    // no-hang case) or never does (in the HUNG case), blocks until all of
-    // the session's own spawned background tasks terminate -- which is
-    // teardown cost unrelated to the query itself and must not be allowed
-    // to confound (or wedge) this test's pass/fail signal. Leak the thread;
-    // process exit reaps it.
+    // Never join(): the Runtime's own drop would block on teardown,
+    // which is unrelated to (and could wedge) the query outcome we
+    // already have. Leak the thread; process exit reaps it.
     std::mem::forget(handle);
 
     eprintln!("=== candidate_a_router_mode result: {outcome}, {reply_count} replies ===");

From 770eedd4bc784ea22ca241d67d155ff071d0ad52 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 17:38:44 +0800
Subject: [PATCH 21/24] fix(liveliness): address Copilot review feedback

- Replace fixed 50ms sleeps in the mock routing-test harness with a
  bounded poll on the DeclareFinal completion marker, so the wait is
  correctness-driven instead of timing-dependent.
- Disable scouting (multicast + gossip) in the self-deadlock regression
  tests so they can't autoconnect to unrelated sessions from other test
  binaries running concurrently.
- Drop the dead KNOWLEDGE.md reference (not part of this repo); restate
  the root cause inline instead.
- Bind send_interest's spawned JoinHandle instead of discarding it
  silently, and distinguish a genuine panic from cancellation when
  logging a spawn_blocking failure.
---
 zenoh/src/net/routing/dispatcher/face.rs      | 45 ++++++++------
 zenoh/src/net/tests/regions/mod.rs            | 58 +++++++++++++++----
 zenoh/tests/liveliness_self_deadlock_peer.rs  | 12 +++-
 .../tests/liveliness_self_deadlock_router.rs  | 12 +++-
 4 files changed, 94 insertions(+), 33 deletions(-)

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index e414b25da2..c7155104b0 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -556,24 +556,35 @@ impl Primitives for Face {
             // that expect all state dropped immediately after close(). Plain
             // spawn_with_rt makes Session::close()'s task_controller wait for
             // genuine completion instead, bounded by its own timeout either way.
-            self.state
-                .task_controller
-                .spawn_with_rt(zenoh_runtime::ZRuntime::Net, async move {
-                    // `send_declare` may invoke a blocking user callback, which
-                    // would otherwise starve ZRuntime::Net's small async-worker
-                    // pool (shared with Session::close_inner's teardown). Run it
-                    // on Net's separate blocking-thread pool instead.
-                    if let Err(e) = zenoh_runtime::ZRuntime::Net
-                        .spawn_blocking(move || {
-                            for (p, m) in declares {
-                                m.with_mut(|m| p.send_declare(m));
+            let _handle =
+                self.state
+                    .task_controller
+                    .spawn_with_rt(zenoh_runtime::ZRuntime::Net, async move {
+                        // `send_declare` may invoke a blocking user callback, which
+                        // would otherwise starve ZRuntime::Net's small async-worker
+                        // pool (shared with Session::close_inner's teardown). Run it
+                        // on Net's separate blocking-thread pool instead.
+                        if let Err(e) = zenoh_runtime::ZRuntime::Net
+                            .spawn_blocking(move || {
+                                for (p, m) in declares {
+                                    m.with_mut(|m| p.send_declare(m));
+                                }
+                            })
+                            .await
+                        {
+                            if e.is_panic() {
+                                tracing::error!(
+                                    ?e,
+                                    "panic while replaying declares for send_interest"
+                                );
+                            } else {
+                                tracing::debug!(
+                                    ?e,
+                                    "declare replay for send_interest was cancelled"
+                                );
                             }
-                        })
-                        .await
-                    {
-                        tracing::error!(?e, "panic while replaying declares for send_interest");
-                    }
-                });
+                        }
+                    });
         } else {
             self.interest_final(msg);
         }
diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs
index c549fde0c4..57d937f60f 100644
--- a/zenoh/src/net/tests/regions/mod.rs
+++ b/zenoh/src/net/tests/regions/mod.rs
@@ -75,6 +75,9 @@ use crate::net::{
     runtime::{Runtime, RuntimeBuilder},
 };
 
+/// Bound on how long [`MockFace::wait_for_declare_final`] polls before panicking.
+const DECLARE_FINAL_TIMEOUT: Duration = Duration::from_secs(5);
+
 pub(crate) fn try_init_tracing_subscriber() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -384,6 +387,20 @@ impl RecordingPrimitives {
     pub(crate) fn clear(&self) {
         self.messages.lock().unwrap().clear();
     }
+
+    /// Whether a `DeclareFinal` for `interest_id` has been recorded.
+    pub(crate) fn has_declare_final(&self, interest_id: InterestId) -> bool {
+        self.messages.lock().unwrap().iter().any(|m| {
+            matches!(
+                m,
+                Message::Declare(Declare {
+                    interest_id: Some(id),
+                    body: DeclareBody::DeclareFinal(_),
+                    ..
+                }) if *id == interest_id
+            )
+        })
+    }
 }
 
 impl Primitives for RecordingPrimitives {
@@ -682,11 +699,11 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
-        // `send_interest`'s local-face replay (declares + DeclareFinal) runs on a
-        // spawned task now, not synchronously on this thread -- give it a moment
-        // to land before returning, since callers immediately inspect the
-        // recorder synchronously.
-        thread::sleep(Duration::from_millis(50));
+        // A `Final`-mode interest tears down a prior registration and never
+        // produces its own `DeclareFinal`.
+        if mode != InterestMode::Final {
+            self.wait_for_declare_final(id);
+        }
     }
 
     /// Declare an interest without a key expression.
@@ -705,7 +722,27 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
-        thread::sleep(Duration::from_millis(50));
+        // A `Final`-mode interest tears down a prior registration and never
+        // produces its own `DeclareFinal`.
+        if mode != InterestMode::Final {
+            self.wait_for_declare_final(id);
+        }
+    }
+
+    /// Block until a `DeclareFinal` for `interest_id` is recorded, or panic after
+    /// [`DECLARE_FINAL_TIMEOUT`]. `send_interest`'s local-face replay (declares +
+    /// `DeclareFinal`) runs on a spawned task, not synchronously, so callers that
+    /// immediately inspect the recorder must wait for this observable completion
+    /// marker rather than a fixed sleep.
+    pub(crate) fn wait_for_declare_final(&self, interest_id: InterestId) {
+        let deadline = std::time::Instant::now() + DECLARE_FINAL_TIMEOUT;
+        while !self.recorder.has_declare_final(interest_id) {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "timed out after {DECLARE_FINAL_TIMEOUT:?} waiting for DeclareFinal(interest_id={interest_id})"
+            );
+            thread::sleep(Duration::from_millis(1));
+        }
     }
 
     /// Declare a liveliness token for `key_expr` from this face's perspective.
@@ -1109,10 +1146,11 @@ impl EstablishedConnection {
             Message::ResponseFinal(r) => target.face.send_response_final(&mut r.clone()),
             Message::Interest(i) => {
                 target.face.send_interest(&mut i.clone());
-                // See the comment on `MockFace::interest` -- send_interest's
-                // local-face replay now runs on a spawned task, not
-                // synchronously; give it a moment to land before returning.
-                thread::sleep(Duration::from_millis(50));
+                // A `Final`-mode interest tears down a prior registration and
+                // never produces its own `DeclareFinal`.
+                if i.mode != InterestMode::Final {
+                    target.wait_for_declare_final(i.id);
+                }
             }
             Message::Oam(o) => {
                 if let Some(demux) = &target.demux {
diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs
index 542f05f8f2..ba4bcd43ed 100644
--- a/zenoh/tests/liveliness_self_deadlock_peer.rs
+++ b/zenoh/tests/liveliness_self_deadlock_peer.rs
@@ -1,7 +1,9 @@
 //! Regression test: a peer-mode `Session` declares 300 liveliness tokens on
 //! itself, then queries its own local table with the default 256-slot
-//! handler -- the exact path that self-deadlocked pre-fix (see
-//! KNOWLEDGE.md). Must complete promptly and deliver every reply.
+//! handler -- the exact path that self-deadlocked pre-fix, because the
+//! synchronous token replay in `Face::send_interest` ran on the same
+//! thread that was about to drain the reply channel. Must complete
+//! promptly and deliver every reply.
 //!
 //! Own test binary (own OS process) so its deliberately-leaked
 //! session/tokens/thread can't autoconnect to and contaminate
@@ -30,7 +32,11 @@ fn candidate_a_peer_mode() {
         let result = rt.block_on(async {
             let mut config = Config::default();
             config.set_mode(Some(WhatAmI::Peer)).unwrap();
-            // No network: no listen/connect endpoints configured.
+            // No network: no listen/connect endpoints, and scouting disabled
+            // so this session can't discover or autoconnect to unrelated
+            // sessions from other test binaries running concurrently.
+            config.scouting.multicast.set_enabled(Some(false)).unwrap();
+            config.scouting.gossip.set_enabled(Some(false)).unwrap();
             let session = zenoh::open(config).await.unwrap();
 
             let mut tokens = Vec::with_capacity(N_TOKENS);
diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs
index e28009576f..f6f2fba767 100644
--- a/zenoh/tests/liveliness_self_deadlock_router.rs
+++ b/zenoh/tests/liveliness_self_deadlock_router.rs
@@ -1,7 +1,9 @@
 //! Regression test: a router-mode `Session` declares 300 liveliness tokens
 //! on itself, then queries its own local table with the default 256-slot
-//! handler -- the exact path that self-deadlocked pre-fix (see
-//! KNOWLEDGE.md). Must complete promptly and deliver every reply.
+//! handler -- the exact path that self-deadlocked pre-fix, because the
+//! synchronous token replay in `Face::send_interest` ran on the same
+//! thread that was about to drain the reply channel. Must complete
+//! promptly and deliver every reply.
 //!
 //! Own test binary (own OS process) so its deliberately-leaked
 //! session/tokens/thread can't autoconnect to and contaminate
@@ -30,7 +32,11 @@ fn candidate_a_router_mode() {
         let result = rt.block_on(async {
             let mut config = Config::default();
             config.set_mode(Some(WhatAmI::Router)).unwrap();
-            // No network: no listen/connect endpoints configured.
+            // No network: no listen/connect endpoints, and scouting disabled
+            // so this session can't discover or autoconnect to unrelated
+            // sessions from other test binaries running concurrently.
+            config.scouting.multicast.set_enabled(Some(false)).unwrap();
+            config.scouting.gossip.set_enabled(Some(false)).unwrap();
             let session = zenoh::open(config).await.unwrap();
 
             let mut tokens = Vec::with_capacity(N_TOKENS);

From 53bbda423d123510f6815717a868483bd24f27f4 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 17:47:39 +0800
Subject: [PATCH 22/24] fix(ci): fix rustfmt violation and widen regions-test
 timeout

rustfmt: reflow the spawn_with_rt call in face.rs to the formatter's
preferred layout.

nextest: net::tests::regions tests using MockFace::interest now wait
on send_interest's replay, dispatched onto the shared, single-worker
ZRuntime::Net instead of running synchronously. Under CI's parallel
test execution that adds real scheduling latency that blew past the
previous 2s/no-retry budget (several *_with_interest tests timed out
consistently). Widen to 10s with one retry.
---
 .config/nextest.toml                     | 10 ++++-
 zenoh/src/net/routing/dispatcher/face.rs | 48 +++++++++++-------------
 2 files changed, 29 insertions(+), 29 deletions(-)

diff --git a/.config/nextest.toml b/.config/nextest.toml
index 74e7f2e85f..55302223e0 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -38,5 +38,11 @@ test-group = 'serial'
 filter = """
 package(zenoh) and test(net::tests::regions)
 """
-retries = 0
-slow-timeout = { period = "2s", terminate-after = 1 }
+# Tests using `MockFace::interest` now wait on `send_interest`'s replay,
+# which is dispatched onto the shared, single-worker `ZRuntime::Net` (see
+# `Face::send_interest`) instead of running synchronously. Under CI's
+# parallel test execution that adds real scheduling latency, so the
+# previous 2s/no-retry budget (calibrated for fully-synchronous behavior)
+# is too tight.
+retries = 1
+slow-timeout = { period = "10s", terminate-after = 1 }
diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index c7155104b0..9feb1ec128 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -556,35 +556,29 @@ impl Primitives for Face {
             // that expect all state dropped immediately after close(). Plain
             // spawn_with_rt makes Session::close()'s task_controller wait for
             // genuine completion instead, bounded by its own timeout either way.
-            let _handle =
-                self.state
-                    .task_controller
-                    .spawn_with_rt(zenoh_runtime::ZRuntime::Net, async move {
-                        // `send_declare` may invoke a blocking user callback, which
-                        // would otherwise starve ZRuntime::Net's small async-worker
-                        // pool (shared with Session::close_inner's teardown). Run it
-                        // on Net's separate blocking-thread pool instead.
-                        if let Err(e) = zenoh_runtime::ZRuntime::Net
-                            .spawn_blocking(move || {
-                                for (p, m) in declares {
-                                    m.with_mut(|m| p.send_declare(m));
-                                }
-                            })
-                            .await
-                        {
-                            if e.is_panic() {
-                                tracing::error!(
-                                    ?e,
-                                    "panic while replaying declares for send_interest"
-                                );
-                            } else {
-                                tracing::debug!(
-                                    ?e,
-                                    "declare replay for send_interest was cancelled"
-                                );
+            let _handle = self.state.task_controller.spawn_with_rt(
+                zenoh_runtime::ZRuntime::Net,
+                async move {
+                    // `send_declare` may invoke a blocking user callback, which
+                    // would otherwise starve ZRuntime::Net's small async-worker
+                    // pool (shared with Session::close_inner's teardown). Run it
+                    // on Net's separate blocking-thread pool instead.
+                    if let Err(e) = zenoh_runtime::ZRuntime::Net
+                        .spawn_blocking(move || {
+                            for (p, m) in declares {
+                                m.with_mut(|m| p.send_declare(m));
                             }
+                        })
+                        .await
+                    {
+                        if e.is_panic() {
+                            tracing::error!(?e, "panic while replaying declares for send_interest");
+                        } else {
+                            tracing::debug!(?e, "declare replay for send_interest was cancelled");
                         }
-                    });
+                    }
+                },
+            );
         } else {
             self.interest_final(msg);
         }

From 031d4b44e435006ff8073388f2a5f033d696f860 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 17:54:05 +0800
Subject: [PATCH 23/24] fix(tests): make post-interest settle best-effort, not
 a hard assert

DeclareFinal is only sent when RouteInterestResult::ResolvedCurrentInterest
(see dispatcher/interests.rs) -- interests requiring cross-region
forwarding stay unresolved until the caller's own bi_fwd_all() pumps
messages across the mock connections. The previous wait_for_declare_final
asserted on a timeout, which deadlocked (by construction) any test whose
interest needs multi-hop resolution: the code that would make forwarding
progress can't run until the assert-wait returns, and the wait can't
succeed until that code runs.

Replace with settle_after_send_interest, a short (500ms) best-effort poll
that returns silently if DeclareFinal never lands locally, instead of
hard-blocking or panicking.
---
 zenoh/src/net/tests/regions/mod.rs | 40 ++++++++++++++++++------------
 1 file changed, 24 insertions(+), 16 deletions(-)

diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs
index 57d937f60f..166e061aa5 100644
--- a/zenoh/src/net/tests/regions/mod.rs
+++ b/zenoh/src/net/tests/regions/mod.rs
@@ -75,8 +75,15 @@ use crate::net::{
     runtime::{Runtime, RuntimeBuilder},
 };
 
-/// Bound on how long [`MockFace::wait_for_declare_final`] polls before panicking.
-const DECLARE_FINAL_TIMEOUT: Duration = Duration::from_secs(5);
+/// Bound on how long [`MockFace::settle_after_send_interest`] polls before
+/// giving up. `DeclareFinal` is only sent when the interest resolves
+/// entirely locally (`RouteInterestResult::ResolvedCurrentInterest`, see
+/// `dispatcher/interests.rs`); interests requiring cross-region forwarding
+/// stay unresolved until the caller's own `bi_fwd_all()` pumps messages
+/// across the mock connections, so this settle window must not block
+/// indefinitely (or assert) waiting for something that may legitimately
+/// not have happened yet.
+const DECLARE_FINAL_SETTLE: Duration = Duration::from_millis(500);
 
 pub(crate) fn try_init_tracing_subscriber() {
     let _ = tracing_subscriber::fmt()
@@ -702,7 +709,7 @@ impl MockFace {
         // A `Final`-mode interest tears down a prior registration and never
         // produces its own `DeclareFinal`.
         if mode != InterestMode::Final {
-            self.wait_for_declare_final(id);
+            self.settle_after_send_interest(id);
         }
     }
 
@@ -725,22 +732,23 @@ impl MockFace {
         // A `Final`-mode interest tears down a prior registration and never
         // produces its own `DeclareFinal`.
         if mode != InterestMode::Final {
-            self.wait_for_declare_final(id);
+            self.settle_after_send_interest(id);
         }
     }
 
-    /// Block until a `DeclareFinal` for `interest_id` is recorded, or panic after
-    /// [`DECLARE_FINAL_TIMEOUT`]. `send_interest`'s local-face replay (declares +
-    /// `DeclareFinal`) runs on a spawned task, not synchronously, so callers that
-    /// immediately inspect the recorder must wait for this observable completion
-    /// marker rather than a fixed sleep.
-    pub(crate) fn wait_for_declare_final(&self, interest_id: InterestId) {
-        let deadline = std::time::Instant::now() + DECLARE_FINAL_TIMEOUT;
+    /// Give `send_interest`'s spawned replay a chance to land before
+    /// returning, by polling for `interest_id`'s `DeclareFinal` up to
+    /// [`DECLARE_FINAL_SETTLE`]. Best-effort: returns as soon as the
+    /// completion marker is observed, but does *not* fail if it never
+    /// shows up locally -- an interest requiring cross-region forwarding
+    /// only resolves once the caller pumps messages (e.g. `bi_fwd_all()`),
+    /// so this must never hard-block or assert.
+    pub(crate) fn settle_after_send_interest(&self, interest_id: InterestId) {
+        let deadline = std::time::Instant::now() + DECLARE_FINAL_SETTLE;
         while !self.recorder.has_declare_final(interest_id) {
-            assert!(
-                std::time::Instant::now() < deadline,
-                "timed out after {DECLARE_FINAL_TIMEOUT:?} waiting for DeclareFinal(interest_id={interest_id})"
-            );
+            if std::time::Instant::now() >= deadline {
+                return;
+            }
             thread::sleep(Duration::from_millis(1));
         }
     }
@@ -1149,7 +1157,7 @@ impl EstablishedConnection {
                 // A `Final`-mode interest tears down a prior registration and
                 // never produces its own `DeclareFinal`.
                 if i.mode != InterestMode::Final {
-                    target.wait_for_declare_final(i.id);
+                    target.settle_after_send_interest(i.id);
                 }
             }
             Message::Oam(o) => {

From 2a47a4ed6056594c7162fc52613793601fcdc926 Mon Sep 17 00:00:00 2001
From: yuanyuyuan <az6980522@gmail.com>
Date: Tue, 21 Jul 2026 20:10:19 +0800
Subject: [PATCH 24/24] fix(routing): dispatch send_interest replay on
 Application runtime, not Net

ZRuntime::Net has a single worker thread and is also where transport
teardown runs (handle_close in the transport RX loop), which calls
TaskController::terminate_all -- a blocking call via block_in_place
that stalls Net's one worker for up to its own timeout. Queuing the
send_interest replay task on Net too serialized it behind every
concurrent face teardown, causing three_node_combination to hang under
CI load with many sessions closing at once. Application's multi-worker
pool avoids that bottleneck.
---
 zenoh/src/net/routing/dispatcher/face.rs | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/zenoh/src/net/routing/dispatcher/face.rs b/zenoh/src/net/routing/dispatcher/face.rs
index 9feb1ec128..989dccb445 100644
--- a/zenoh/src/net/routing/dispatcher/face.rs
+++ b/zenoh/src/net/routing/dispatcher/face.rs
@@ -556,14 +556,21 @@ impl Primitives for Face {
             // that expect all state dropped immediately after close(). Plain
             // spawn_with_rt makes Session::close()'s task_controller wait for
             // genuine completion instead, bounded by its own timeout either way.
+            //
+            // Application, not Net: ZRuntime::Net is configured with a single
+            // worker thread and is also where transport teardown runs (see
+            // handle_close in io/zenoh-transport/.../rx.rs), which calls
+            // TaskController::terminate_all -- a blocking call that stalls
+            // Net's one worker for up to its own timeout. Queuing this replay
+            // on Net too serializes it behind every concurrent face teardown;
+            // Application's multi-worker pool avoids that bottleneck.
             let _handle = self.state.task_controller.spawn_with_rt(
-                zenoh_runtime::ZRuntime::Net,
+                zenoh_runtime::ZRuntime::Application,
                 async move {
                     // `send_declare` may invoke a blocking user callback, which
-                    // would otherwise starve ZRuntime::Net's small async-worker
-                    // pool (shared with Session::close_inner's teardown). Run it
-                    // on Net's separate blocking-thread pool instead.
-                    if let Err(e) = zenoh_runtime::ZRuntime::Net
+                    // would otherwise starve the async worker pool. Run it on
+                    // a separate blocking-thread pool instead.
+                    if let Err(e) = zenoh_runtime::ZRuntime::Application
                         .spawn_blocking(move || {
                             for (p, m) in declares {
                                 m.with_mut(|m| p.send_declare(m));