Fix WaitSetRunner cleanup before propagating callback panics

[coolcmyk]

Fixes #613.

A panic inside a callback was caught by its WaitSetRunner, but the basic
executor immediately resumed unwinding when it received the panic. This
allowed executor and ROS entities to be destroyed while other runner threads
were still active, causing RMW cleanup errors and potentially a segmentation
fault.

This change:

• Returns callback panics from WaitSetRunner as a distinct outcome.
• Saves the first panic payload and the runner that encountered it.
• Signals all remaining runners to stop and wakes runners blocked in the wait set.
• Waits for the runners to finish.
• Restores executor-owned state before resuming the original panic.

The panic still propagates to the caller, but worker shutdown and ROS resource
cleanup now happen first.

Validation

• Ran cargo check -p rclrs --lib.
• Reproduced the issue using example 613 with an intentional callback panic.
• Confirmed that the panic is propagated without the previous RMW cleanup
  errors or segmentation fault.
• Confirmed the panic can be caught by the caller with catch_unwind.

use rclrs::*;
use ros_env::std_msgs;
use std::time::Duration;

fn main() -> Result<(), RclrsError> {
  let mut executor = Context::default_from_env()?.create_basic_executor();
  let node = executor.create_node("worker_demo")?;

  let publisher = node.create_publisher("output_topic")?;

  // Keep a separate runner blocked in rcl_wait when the subscription callback
  // panics. The executor must stop and reclaim this runner before propagating
  // the panic.
  let idle_worker = node.create_worker(());
  let _idle_timer = idle_worker.create_timer_inert(Duration::from_secs(60))?;

  let worker = node.create_worker(String::new());
  let _subscription = worker.create_subscription(
      "input_topic",
      move |data: &mut String, msg: std_msgs::msg::String| {
          *data = msg.data;
          panic!();
      },
  )?;

  // Use this timer-based implementation when timers are available instead
  // of using std::thread::spawn.
  let _timer =
      worker.create_timer_repeating(Duration::from_secs(1), move |data: &mut String| {
          let msg = std_msgs::msg::String { data: data.clone() };

          publisher.publish(msg).ok();
      })?;

  println!(
      "Beginning repeater... \n >> \
      Publish a std_msg::msg::String to \"input_topic\" and we will periodically republish it to \"output_topic\".\n\n\
      To see this in action run the following commands in two different terminals:\n \
      $ ros2 topic echo output_topic\n \
      $ ros2 topic pub input_topic std_msgs/msg/String \"{{data: Hello}}\""
  );
  let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
      executor.spin(SpinOptions::default());
  }));

  assert!(result.is_err(), "callback panic was not propagated");

  Ok(())
}

[mxgrey]

Let's add UnwindSafe as a trait bound for all callbacks to help ensure that catching the unwind will be safe.

[coolcmyk]

Hi mxgrey, i've added UnwindSafe across the callbacks in this commit. Tests has passed with

cargo test -p rclrs --lib

test result: ok. 119 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 3.45s

Let's add UnwindSafe as a trait bound for all callbacks to help ensure that catching the unwind will be safe.