diff --git a/crates/hiroz-tests/tests/reentrant_lifecycle.rs b/crates/hiroz-tests/tests/reentrant_lifecycle.rs new file mode 100644 index 000000000..e66a56fed --- /dev/null +++ b/crates/hiroz-tests/tests/reentrant_lifecycle.rs @@ -0,0 +1,376 @@ +//! Re-entrancy audit for lifecycle transition callbacks. +//! +//! `ZLifecycleNode::trigger_transition` used to invoke the user's transition +//! callback from inside `self.state_machine.lock().unwrap().trigger(..)`, i.e. +//! with the state-machine mutex held for the whole duration of the callback. +//! +//! That mutex is not private to the transition: it is shared with the node's +//! own `~/get_state`, `~/change_state` and `~/get_available_transitions` service +//! handlers, and with `get_current_state()`. So while a transition callback ran, +//! the node could not answer any question about its own state. A callback that +//! asks — directly, or by waiting on anything that asks — waits forever. +//! +//! The fix splits `StateMachine::trigger` into `begin` (enter the intermediate +//! "busy" state) and `finish` (apply the callback's verdict), so the node drops +//! the guard while the callback runs. +//! +//! The intermediate state therefore becomes *observable*, which it was not +//! before — and that is a behaviour change, not a preservation. Previously a +//! `~/get_state` issued during a transition blocked for the callback's whole +//! duration (the deadlock above), and on the rare path where a reply did come +//! back, `ZLifecycleClient` mapped every transition-state id onto +//! `Unconfigured`. Callers now see `configuring`, `activating`, … while the +//! transition runs, which is what rclcpp reports and what a lifecycle manager +//! polling the node expects. +//! +//! Every scenario runs on a dedicated thread behind a hard deadline, so a +//! re-entrancy deadlock fails the test instead of wedging the suite — the same +//! shape as `reentrant_publish.rs` and `reentrant_service.rs`. + +mod common; + +use std::{ + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + mpsc, + }, + thread, + time::Duration, +}; + +use common::{TestRouter, create_hiroz_context_with_endpoint}; +use hiroz::{ + Builder, + lifecycle::{CallbackReturn, LifecycleState, ZLifecycleClient}, +}; +use serial_test::serial; + +/// Budget for one scenario. Generous relative to the work done — anything slower +/// than this is a hang, not slowness. +const SCENARIO_TIMEOUT: Duration = Duration::from_secs(45); + +/// How long the in-callback service call is allowed to take. Shorter than the +/// scenario budget so a blocked handler surfaces as a failed assertion rather +/// than as an unhelpful whole-scenario timeout. +const CALL_TIMEOUT: Duration = Duration::from_secs(8); + +fn with_deadline(name: &'static str, scenario: impl FnOnce() + Send + 'static) { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + scenario(); + let _ = tx.send(()); + }); + match rx.recv_timeout(SCENARIO_TIMEOUT) { + Ok(()) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => { + panic!("{name}: scenario panicked — see the worker thread's panic above") + } + Err(mpsc::RecvTimeoutError::Timeout) => { + panic!("{name}: scenario did not finish within {SCENARIO_TIMEOUT:?} — deadlock") + } + } +} + +/// Query `~/get_state` from a thread with its own runtime and return the answer. +/// +/// The transition callback runs on the caller's thread, which may already be +/// inside a runtime, so the query gets a plain std thread of its own — the same +/// pattern `reentrant_service.rs` uses for a nested service call. +fn get_state_blocking(client: Arc) -> zenoh::Result { + thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async { client.get_state(CALL_TIMEOUT).await }) + }) + .join() + .expect("get_state worker panicked") +} + +/// A transition callback that asks its own node what state it is in. +/// +/// The `~/get_state` handler runs on a zenoh RX thread and takes the same +/// state-machine mutex the transition holds. Under the old code it blocks until +/// the transition completes — but the transition cannot complete, because its +/// callback is waiting for that very reply. Deadlock, broken only by the client +/// timeout, which then fails the assertion. +/// +/// Introspecting your own lifecycle state during a transition is ordinary +/// (rclcpp exposes `get_current_state()` for exactly this), and a lifecycle +/// manager polling `~/get_state` while a node configures is even more ordinary. +#[test] +#[serial] +fn transition_callback_querying_own_state_does_not_deadlock() { + with_deadline("lifecycle_transition_get_state", || { + const NODE_NAME: &str = "lc_reentrant_get_state"; + + let router = TestRouter::new(); + + let ctx_node = create_hiroz_context_with_endpoint(router.endpoint()).expect("node ctx"); + let mut lc_node = ctx_node + .create_lifecycle_node(NODE_NAME) + .build() + .expect("lifecycle node"); + + // The querying side lives in its own context, as a real lifecycle + // manager would. + let ctx_client = create_hiroz_context_with_endpoint(router.endpoint()).expect("client ctx"); + let mgr_node = ctx_client + .create_node("lc_manager") + .build() + .expect("mgr node"); + + thread::sleep(Duration::from_millis(1000)); + let client = + Arc::new(ZLifecycleClient::new(&mgr_node, NODE_NAME).expect("lifecycle client")); + thread::sleep(Duration::from_millis(1000)); + + let ran = Arc::new(AtomicBool::new(false)); + let seen: Arc>>> = Arc::new(Mutex::new(None)); + + let ran_c = ran.clone(); + let seen_c = seen.clone(); + let client_c = client.clone(); + lc_node.on_configure = Box::new(move |_prev| { + ran_c.store(true, Ordering::SeqCst); + // Re-entrant query of this node's own state, from inside its own + // transition callback. + *seen_c.lock().unwrap() = Some(get_state_blocking(client_c.clone())); + CallbackReturn::Success + }); + + let final_state = lc_node.configure().expect("configure"); + + assert!(ran.load(Ordering::SeqCst), "on_configure never ran"); + + let seen = seen.lock().unwrap().take().expect("no state recorded"); + let seen = seen.expect( + "~/get_state did not answer while the transition callback was running — \ + the state-machine mutex was held across the callback", + ); + assert_eq!( + seen, + LifecycleState::Configuring, + "the callback must observe the intermediate transition state" + ); + assert_eq!(final_state, LifecycleState::Inactive); + }); +} + +/// The same hazard on the failure path: a callback that returns `Failure` must +/// still have been able to introspect, and the node must still land on the +/// pre-transition state. +/// +/// This pins the ordering the split introduced: `begin` publishes the +/// intermediate state, `finish` applies the verdict *after* the callback +/// returns. If the two halves were reordered, the observed state here would be +/// `Unconfigured` rather than `Activating`. +#[test] +#[serial] +fn failing_transition_callback_still_observes_intermediate_state() { + with_deadline("lifecycle_failing_transition_get_state", || { + const NODE_NAME: &str = "lc_reentrant_fail"; + + let router = TestRouter::new(); + + let ctx_node = create_hiroz_context_with_endpoint(router.endpoint()).expect("node ctx"); + let mut lc_node = ctx_node + .create_lifecycle_node(NODE_NAME) + .build() + .expect("lifecycle node"); + + let ctx_client = create_hiroz_context_with_endpoint(router.endpoint()).expect("client ctx"); + let mgr_node = ctx_client + .create_node("lc_manager_fail") + .build() + .expect("mgr node"); + + thread::sleep(Duration::from_millis(1000)); + let client = + Arc::new(ZLifecycleClient::new(&mgr_node, NODE_NAME).expect("lifecycle client")); + thread::sleep(Duration::from_millis(1000)); + + // Get to Inactive first, so `activate` is a legal transition. + assert_eq!( + lc_node.configure().expect("configure"), + LifecycleState::Inactive + ); + + let seen: Arc>>> = Arc::new(Mutex::new(None)); + let seen_c = seen.clone(); + let client_c = client.clone(); + lc_node.on_activate = Box::new(move |_prev| { + *seen_c.lock().unwrap() = Some(get_state_blocking(client_c.clone())); + CallbackReturn::Failure + }); + + let final_state = lc_node.activate().expect("activate"); + + let seen = seen.lock().unwrap().take().expect("no state recorded"); + let seen = seen.expect( + "~/get_state did not answer while the transition callback was running — \ + the state-machine mutex was held across the callback", + ); + assert_eq!( + seen, + LifecycleState::Activating, + "the callback must observe the intermediate transition state" + ); + // Failure reverts to the start state. + assert_eq!(final_state, LifecycleState::Inactive); + assert_eq!(lc_node.get_current_state(), LifecycleState::Inactive); + }); +} + +/// A transition callback that panics must not leave the node wedged. +/// +/// This is the hazard the `begin`/`finish` split introduced. `begin` moves the +/// state machine into the intermediate ("busy") state and releases the guard; +/// the callback then runs unlocked. If it panics, the unwind used to escape +/// `trigger_transition` before `finish` ran — so `current` stayed at +/// `Configuring` forever. Worse, because the guard had already been dropped, +/// the mutex was *not* poisoned, so nothing ever reported it: every later +/// `begin` returned `None` and `~/get_state` answered `configuring` for the +/// life of the process. +/// +/// That is strictly worse than the behaviour it replaced. While the callback +/// ran under the guard, a panic unwound through a live `MutexGuard` and +/// poisoned the mutex, so the very next `lock().unwrap()` panicked loudly. +/// Fail-fast became a silent permanent wedge. +/// +/// The fix catches the unwind, settles the state machine on `Failure` (revert +/// to the start state — "the transition did not happen"), and re-raises the +/// payload, so the panic is still as loud as before while the node stays +/// usable. +/// +/// Reverting the `catch_unwind` in `ZLifecycleNode::trigger_transition` makes +/// both assertions below fail: the state reads `Configuring`, and the recovery +/// transition returns `Configuring` because `begin` refuses to start. +#[test] +#[serial] +fn panicking_transition_callback_does_not_wedge_the_node() { + with_deadline("lifecycle_transition_panic", || { + const NODE_NAME: &str = "lc_panicking_callback"; + + let router = TestRouter::new(); + let ctx_node = create_hiroz_context_with_endpoint(router.endpoint()).expect("node ctx"); + let mut lc_node = ctx_node + .create_lifecycle_node(NODE_NAME) + .build() + .expect("lifecycle node"); + + lc_node.on_configure = Box::new(|_prev| panic!("deliberate panic from on_configure")); + + // The panic must still propagate — silently swallowing it would be its + // own defect. Keep the default hook quiet for this one call so the + // expected backtrace does not look like a test failure. + let prev_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let outcome = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| lc_node.configure())); + std::panic::set_hook(prev_hook); + + assert!( + outcome.is_err(), + "the panic was swallowed; it must still reach the caller" + ); + + // The node must be back at its pre-transition primary state, not + // stranded in the intermediate one. + assert_eq!( + lc_node.get_current_state(), + LifecycleState::Unconfigured, + "node wedged in an intermediate state after a panicking callback" + ); + + // And it must still be usable: a subsequent transition has to work. + lc_node.on_configure = Box::new(|_prev| CallbackReturn::Success); + let recovered = lc_node.configure().expect("configure after panic"); + assert_eq!( + recovered, + LifecycleState::Inactive, + "node could not transition after a panicking callback" + ); + }); +} + +/// The `on_error` callback must also run with the state-machine lock released. +/// +/// `trigger_transition` splits the *error* path the same way it splits the +/// normal one: `finish` records `CallbackReturn::Error` and drops the guard, +/// `on_error` runs unlocked, `finish_error_processing` applies its verdict. +/// Neither existing scenario reaches those lines — one callback returns +/// `Success` and the other `Failure` — so a regression that re-ran `on_error` +/// under the mutex would have passed the whole suite. +/// +/// The detector is the same shape as the `on_configure` one: `on_error` asks +/// the node, through a real service client in another context, what state it is +/// in. Under a regression that query blocks on the guard `on_error` is running +/// beneath, the client times out, and the assertion fails on the error rather +/// than hanging the suite. +/// +/// It also pins the state that should be observable there — `ErrorProcessing`, +/// not the pre-transition state and not the goal state. +#[test] +#[serial] +fn on_error_callback_querying_own_state_does_not_deadlock() { + with_deadline("lifecycle_on_error_get_state", || { + const NODE_NAME: &str = "lc_reentrant_on_error"; + + let router = TestRouter::new(); + + let ctx_node = create_hiroz_context_with_endpoint(router.endpoint()).expect("node ctx"); + let mut lc_node = ctx_node + .create_lifecycle_node(NODE_NAME) + .build() + .expect("lifecycle node"); + + let ctx_client = create_hiroz_context_with_endpoint(router.endpoint()).expect("client ctx"); + let mgr_node = ctx_client + .create_node("lc_error_manager") + .build() + .expect("mgr node"); + + thread::sleep(Duration::from_millis(1000)); + let client = + Arc::new(ZLifecycleClient::new(&mgr_node, NODE_NAME).expect("lifecycle client")); + thread::sleep(Duration::from_millis(1000)); + + // Drive the transition into the error path. + lc_node.on_configure = Box::new(|_prev| CallbackReturn::Error); + + let ran = Arc::new(AtomicBool::new(false)); + let seen: Arc>>> = Arc::new(Mutex::new(None)); + + let ran_c = ran.clone(); + let seen_c = seen.clone(); + let client_c = client.clone(); + lc_node.on_error = Box::new(move |_prev| { + ran_c.store(true, Ordering::SeqCst); + *seen_c.lock().unwrap() = Some(get_state_blocking(client_c.clone())); + // Success from on_error means "recovered" -> Unconfigured. + CallbackReturn::Success + }); + + let final_state = lc_node.configure().expect("configure"); + + assert!(ran.load(Ordering::SeqCst), "on_error never ran"); + + let seen = seen.lock().unwrap().take().expect("no state recorded"); + let seen = seen.expect( + "the ~/get_state query from inside on_error did not complete; the error path is \ + running the callback under the state-machine guard again", + ); + assert_eq!( + seen, + LifecycleState::ErrorProcessing, + "on_error should observe the node in ErrorProcessing" + ); + + // Success from on_error recovers the node to Unconfigured. + assert_eq!(final_state, LifecycleState::Unconfigured); + assert_eq!(lc_node.get_current_state(), LifecycleState::Unconfigured); + }); +} diff --git a/crates/hiroz/src/lifecycle/client.rs b/crates/hiroz/src/lifecycle/client.rs index f96d17af7..d9d292ed1 100644 --- a/crates/hiroz/src/lifecycle/client.rs +++ b/crates/hiroz/src/lifecycle/client.rs @@ -173,6 +173,16 @@ fn state_from_lc(s: &LcState) -> LifecycleState { 2 => LifecycleState::Inactive, 3 => LifecycleState::Active, 4 => LifecycleState::Finalized, + // Transition ("busy") states. A node genuinely reports these while a + // transition callback is running; mapping them onto `Unconfigured` told + // a lifecycle manager the node had reset itself, which is both wrong and + // indistinguishable from the real thing. + 10 => LifecycleState::Configuring, + 11 => LifecycleState::CleaningUp, + 12 => LifecycleState::ShuttingDown, + 13 => LifecycleState::Activating, + 14 => LifecycleState::Deactivating, + 15 => LifecycleState::ErrorProcessing, _ => LifecycleState::Unconfigured, } } diff --git a/crates/hiroz/src/lifecycle/node.rs b/crates/hiroz/src/lifecycle/node.rs index 0169c9ebe..79f8de2ef 100644 --- a/crates/hiroz/src/lifecycle/node.rs +++ b/crates/hiroz/src/lifecycle/node.rs @@ -1,4 +1,7 @@ -use std::sync::{Arc, Mutex}; +use std::{ + panic::{AssertUnwindSafe, catch_unwind, resume_unwind}, + sync::{Arc, Mutex}, +}; use tracing::{debug, info, warn}; use zenoh::{Result, Wait, query::Query}; @@ -129,44 +132,123 @@ impl ZLifecycleNode { let start = self.get_current_state(); debug!(node=%self.inner.entity.name, ?transition, ?start, "triggering lifecycle transition"); - let cb_result = { - let callback: &dyn Fn(State) -> CallbackReturn = match transition { - TransitionId::Configure => self.on_configure.as_ref(), - TransitionId::Activate => self.on_activate.as_ref(), - TransitionId::Deactivate => self.on_deactivate.as_ref(), - TransitionId::Cleanup => self.on_cleanup.as_ref(), - TransitionId::UnconfiguredShutdown - | TransitionId::InactiveShutdown - | TransitionId::ActiveShutdown => self.on_shutdown.as_ref(), - }; - self.state_machine - .lock() - .unwrap() - .trigger(transition, callback) + let callback: &dyn Fn(State) -> CallbackReturn = match transition { + TransitionId::Configure => self.on_configure.as_ref(), + TransitionId::Activate => self.on_activate.as_ref(), + TransitionId::Deactivate => self.on_deactivate.as_ref(), + TransitionId::Cleanup => self.on_cleanup.as_ref(), + TransitionId::UnconfiguredShutdown + | TransitionId::InactiveShutdown + | TransitionId::ActiveShutdown => self.on_shutdown.as_ref(), + }; + + // The transition is driven in two locked steps with the user callback in + // between, *outside* the lock. `state_machine` is shared with the + // `~/get_state` / `~/change_state` service handlers and with + // `get_current_state`, so running the callback under the guard would + // deadlock any callback that inspects its own node. + // + // NOTE: the `begin` call is bound to its own `let` on purpose. Writing + // `match self.state_machine.lock().unwrap().begin(..) { .. }` keeps the + // temporary guard alive for the whole `match` body — which silently + // reintroduces exactly the deadlock this split exists to remove. + let begun = self.state_machine.lock().unwrap().begin(transition); + let cb_result = match begun { + Some(start_state) => { + // Lock released; observers see the intermediate ("busy") state. + // + // A panic here must not escape without settling the state + // machine. `begin` has already moved `current` to the + // intermediate state, and the guard it took is gone — so an + // unwind straight out of this function would leave `current` + // at `Configuring`/`Activating`/... permanently, *and* leave + // the mutex unpoisoned, so nothing would ever report it. Every + // later `begin` returns `None` and `~/get_state` answers + // `configuring` for the life of the process: a silent wedge. + // Before the callback was moved out of the lock, the unwind + // passed through a live guard and poisoned the mutex, so the + // next access failed loudly. + // + // Settle on `Failure`, which reverts to `start_state` — the + // same "nothing changed" outcome as the invalid-transition arm + // below, and the only result that is both well-defined and + // leaves the node usable. `on_error` is deliberately not run: + // we are already unwinding, and running more user code on the + // way out would risk a second panic. The payload is then + // re-raised, so a panicking callback still fails exactly as + // loudly as it did before the split. + let ret = match catch_unwind(AssertUnwindSafe(|| callback(start_state))) { + Ok(ret) => ret, + Err(payload) => { + self.state_machine.lock().unwrap().finish( + transition, + start_state, + CallbackReturn::Failure, + ); + resume_unwind(payload); + } + }; + self.state_machine + .lock() + .unwrap() + .finish(transition, start_state, ret) + } + // Invalid transition from the current state: nothing changed. + None => self.get_current_state(), }; let final_state = if cb_result == State::ErrorProcessing { + // Same split, and the same unwind guard, for the error path: a + // panic in `on_error` would otherwise strand the node in + // `ErrorProcessing` with no way out. + let cb = AssertUnwindSafe(|| (self.on_error)(State::ErrorProcessing)); + let ret = match catch_unwind(cb) { + Ok(ret) => ret, + Err(payload) => { + self.state_machine + .lock() + .unwrap() + .finish_error_processing(CallbackReturn::Error); + resume_unwind(payload); + } + }; self.state_machine .lock() .unwrap() - .trigger_error_processing(|prev| (self.on_error)(prev)) + .finish_error_processing(ret) } else { cb_result }; - // Bulk-activate / bulk-deactivate managed entities - match (start, final_state) { - (_, State::Active) if start != State::Active => { - for e in self.managed_entities.lock().unwrap().iter() { + // Bulk-activate / bulk-deactivate managed entities. + // + // Snapshot the list under the guard and notify after dropping it, the + // same collect-then-invoke shape used for event callbacks. Today every + // registered entity is a `ZLifecyclePublisher` created by + // `create_publisher`, whose `on_activate`/`on_deactivate` only flip an + // atomic and cannot re-enter — so this is preventive, not a live fix. + // It stops being preventive the moment `ManagedEntity` becomes + // registerable from outside this module, because arbitrary + // implementations would then run under a guard that `create_publisher` + // also takes. + let activate = match (start, final_state) { + (_, State::Active) if start != State::Active => Some(true), + (State::Active, _) if final_state != State::Active => Some(false), + _ => None, + }; + if let Some(activate) = activate { + // The snapshot is its own `let` statement so the guard is dropped + // before the loop — writing the lock inline in the `for` header + // would hold it across every notification instead. + let entities: Vec> = + self.managed_entities.lock().unwrap().clone(); + for e in entities { + if activate { e.on_activate(); - } - } - (State::Active, _) if final_state != State::Active => { - for e in self.managed_entities.lock().unwrap().iter() { + } else { e.on_deactivate(); } } - _ => {} } // Publish transition event diff --git a/crates/hiroz/src/lifecycle/state_machine.rs b/crates/hiroz/src/lifecycle/state_machine.rs index 5ae880e9e..0b1ce0e19 100644 --- a/crates/hiroz/src/lifecycle/state_machine.rs +++ b/crates/hiroz/src/lifecycle/state_machine.rs @@ -162,6 +162,29 @@ impl StateMachine { where F: FnOnce(State) -> CallbackReturn, { + match self.begin(transition) { + Some(start_state) => { + let cb_result = callback(start_state); + self.finish(transition, start_state, cb_result) + } + // Invalid transition: state unchanged + None => self.current, + } + } + + /// First half of [`Self::trigger`]: validate the transition and enter the + /// intermediate ("busy") state. + /// + /// Returns the *start* state to hand to the user callback, or `None` if the + /// transition is not legal from the current state (in which case nothing + /// changed). + /// + /// Split out from `trigger` so a caller holding a lock around the state + /// machine can release it before running the user callback and re-acquire it + /// for [`Self::finish`]. The user callback is free to call back into the + /// node (e.g. `get_current_state`, or the `~/get_state` service), which would + /// self-deadlock on a non-reentrant mutex if it ran inside `trigger`. + pub fn begin(&mut self, transition: TransitionId) -> Option { let start_state = self.current; // Validate transition is legal from the current primary state @@ -177,17 +200,25 @@ impl StateMachine { ); if !is_valid { - // Invalid transition: state unchanged - return self.current; + return None; } - // Enter the intermediate transition state - let transition_state = Self::intermediate_state(transition); - self.current = transition_state; + // Enter the intermediate transition state. This is what an observer + // (including the callback itself) sees while the callback runs. + self.current = Self::intermediate_state(transition); - // Invoke user callback - let cb_result = callback(start_state); + Some(start_state) + } + /// Second half of [`Self::trigger`]: apply the user callback's verdict. + /// + /// `start_state` must be the value returned by the matching [`Self::begin`]. + pub fn finish( + &mut self, + transition: TransitionId, + start_state: State, + cb_result: CallbackReturn, + ) -> State { match cb_result { CallbackReturn::Success => { // Happy path: move to goal state @@ -202,7 +233,7 @@ impl StateMachine { self.current = State::ErrorProcessing; // Error processing returns to ErrorProcessing state — the // on_error callback is invoked by the node, not here. - // The node calls trigger_error_processing() after this. + // The node calls finish_error_processing() after this. } } @@ -217,6 +248,15 @@ impl StateMachine { { debug_assert_eq!(self.current, State::ErrorProcessing); let cb = on_error(State::ErrorProcessing); + self.finish_error_processing(cb) + } + + /// Apply the `on_error` callback's verdict, without invoking it. + /// + /// The counterpart of [`Self::begin`]/[`Self::finish`] for the error path: + /// the caller runs `on_error` with no lock held and passes its result here. + pub fn finish_error_processing(&mut self, cb: CallbackReturn) -> State { + debug_assert_eq!(self.current, State::ErrorProcessing); self.current = match cb { CallbackReturn::Success => State::Unconfigured, // FAILURE or ERROR both lead to Finalized