fix(liveliness): decouple send_interest replay from calling thread

.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 }

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

@@ -548,9 +548,44 @@ 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.
+            //
+            // 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::Application,
+                async move {
+                    // `send_declare` may invoke a blocking user callback, which
+                    // 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));
+                            }
+                        })
+                        .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);
         }

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;
@@ -73,6 +75,16 @@ use crate::net::{
     runtime::{Runtime, RuntimeBuilder},
 };
 
+/// 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()
         .with_env_filter(EnvFilter::from_default_env())
@@ -382,6 +394,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 {
@@ -680,6 +706,11 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
+        // A `Final`-mode interest tears down a prior registration and never
+        // produces its own `DeclareFinal`.
+        if mode != InterestMode::Final {
+            self.settle_after_send_interest(id);
+        }
     }
 
     /// Declare an interest without a key expression.
@@ -698,6 +729,28 @@ impl MockFace {
             ext_tstamp: None,
             ext_nodeid: NodeIdType::DEFAULT,
         });
+        // A `Final`-mode interest tears down a prior registration and never
+        // produces its own `DeclareFinal`.
+        if mode != InterestMode::Final {
+            self.settle_after_send_interest(id);
+        }
+    }
+
+    /// 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) {
+            if std::time::Instant::now() >= deadline {
+                return;
+            }
+            thread::sleep(Duration::from_millis(1));
+        }
     }
 
     /// Declare a liveliness token for `key_expr` from this face's perspective.
@@ -1099,7 +1152,14 @@ 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());
+                // A `Final`-mode interest tears down a prior registration and
+                // never produces its own `DeclareFinal`.
+                if i.mode != InterestMode::Final {
+                    target.settle_after_send_interest(i.id);
+                }
+            }
             Message::Oam(o) => {
                 if let Some(demux) = &target.demux {
                     let _ = demux.handle_message(NetworkMessageMut {

zenoh/tests/liveliness_self_deadlock_peer.rs

@@ -0,0 +1,116 @@
+//! 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, 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
+//! `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, 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);
+            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}"
+    );
+}