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 206e658fad..989dccb445 100644 --- a/zenoh/src/net/routing/dispatcher/face.rs +++ b/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); } diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs index 0acceb456a..166e061aa5 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; @@ -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 { diff --git a/zenoh/tests/liveliness_self_deadlock_peer.rs b/zenoh/tests/liveliness_self_deadlock_peer.rs new file mode 100644 index 0000000000..ba4bcd43ed --- /dev/null +++ b/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}" + ); +} diff --git a/zenoh/tests/liveliness_self_deadlock_router.rs b/zenoh/tests/liveliness_self_deadlock_router.rs new file mode 100644 index 0000000000..f6f2fba767 --- /dev/null +++ b/zenoh/tests/liveliness_self_deadlock_router.rs @@ -0,0 +1,119 @@ +//! 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, 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_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, 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_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/routing.rs b/zenoh/tests/routing.rs index 69ef5ba035..9f98c0cc50 100644 --- a/zenoh/tests/routing.rs +++ b/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(()) }); }