fix(liveliness): decouple send_interest replay from calling thread

zenoh/src/net/routing/dispatcher/face.rs

@@ -548,9 +548,32 @@ 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));
-            }
+            // 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
+                    // 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
+                    {
+                        tracing::error!(?e, "panic while replaying declares for send_interest");
+                    }
+                });
         } else {
             self.interest_final(msg);
         }

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.
@@ -1099,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 {

zenoh/tests/liveliness_self_deadlock_peer.rs

@@ -0,0 +1,110 @@
+//! 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.
+//!
+//! 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};
+
+use zenoh::{
+    config::{Config, WhatAmI},
+    Wait,
+};
+
+const N_TOKENS: usize = 300;
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+#[test]
+fn candidate_a_peer_mode() {
+    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());
+
+            // 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()
+                .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.
+                    while replies.recv_timeout(Duration::from_secs(5)).is_ok() {
+                        count += 1;
+                    }
+                    eprintln!("[candidate-a] drained {count} replies");
+                    count
+                }
+                Err(e) => {
+                    eprintln!("[candidate-a] get() errored: {e:?}");
+                    0
+                }
+            };
+
+            // 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);
+
+            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)
+        }
+    };
+
+    // 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 ===");
+    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}"
+    );
+}

zenoh/tests/liveliness_self_deadlock_router.rs

@@ -0,0 +1,113 @@
+//! 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.
+//!
+//! 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};
+
+use zenoh::{
+    config::{Config, WhatAmI},
+    Wait,
+};
+
+const N_TOKENS: usize = 300;
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+#[test]
+fn candidate_a_router_mode() {
+    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());
+
+            // 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()
+                .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.
+                    while replies.recv_timeout(Duration::from_secs(5)).is_ok() {
+                        count += 1;
+                    }
+                    eprintln!("[candidate-a] drained {count} replies");
+                    count
+                }
+                Err(e) => {
+                    eprintln!("[candidate-a] get() errored: {e:?}");
+                    0
+                }
+            };
+
+            // 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);
+
+            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)
+        }
+    };
+
+    // 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 ===");
+    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}"
+    );
+}

zenoh/tests/routing.rs

@@ -33,6 +33,14 @@ use zenoh_test::{get_free_tcp_port, get_tcp_locator};
 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,18 @@ 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(())
             });
         }