Leave a destroyed awaiter alone when its cancellation runs late

rclcpp_async/include/rclcpp_async/co_context.hpp

@@ -445,6 +445,7 @@ inline void SleepAwaiter::await_suspend(std::coroutine_handle<> h)
 template <typename MsgT>
 void TopicStream<MsgT>::NextAwaiter::await_suspend(std::coroutine_handle<> h)
 {
+  waiting_ = h;
   stream.waiter_ = h;
   register_cancel(
     cancel_cb_, token, stream.ctx_, h, [this, h]() { return stream.waiter_ != h; },
@@ -474,17 +475,14 @@ void GoalStream<ActionT>::resume_waiter(std::coroutine_handle<> h)
 template <typename ActionT>
 void GoalStream<ActionT>::NextAwaiter::await_suspend(std::coroutine_handle<> h)
 {
+  waiting_ = h;
   stream.waiter_ = h;
   register_cancel(
     cancel_cb_, token, stream.ctx_, h, [this, h]() { return stream.waiter_ != h; },
     [this]() {
       stream.waiter_ = nullptr;
       cancelled = true;
-      if (
-        stream.auto_cancel_on_stop_ && !stream.completed_ && stream.goal_handle_ &&
-        stream.client_) {
-        stream.client_->async_cancel_goal(stream.goal_handle_);
-      }
+      stream.cancel_goal_on_stop();
     });
 }
 

rclcpp_async/include/rclcpp_async/event.hpp

@@ -56,6 +56,16 @@ class Event
     std::shared_ptr<StopCb> cancel_cb_;
     std::shared_ptr<bool> active;
 
+    // A frame destroyed while suspended here must not stay registered, or a
+    // later set() would resume it. The deferred cancellation no longer runs
+    // for a destroyed awaiter, so the waiter is withdrawn here.
+    ~WaitAwaiter()
+    {
+      if (active) {
+        *active = false;
+      }
+    }
+
     void set_token(std::stop_token t) { token = std::move(t); }
 
     bool await_ready()

rclcpp_async/include/rclcpp_async/executor.hpp

@@ -46,13 +46,17 @@ inline void register_cancel(
 {
   out = std::make_shared<StopCb>(token, [&ctx, h, is_done, action, &out]() {
     ctx.post([&ctx, h, is_done, action, weak = std::weak_ptr(out)]() {
+      // The awaiter owns `out`, and is_done / action may refer to it. Once it
+      // is gone (resumed, or its frame destroyed after the stop was
+      // requested) neither may be called.
+      if (!weak.lock()) {
+        return;
+      }
       if (is_done()) {
         return;
       }
       action();
-      if (weak.lock()) {
-        ctx.resume(h);
-      }
+      ctx.resume(h);
     });
   });
 }

rclcpp_async/include/rclcpp_async/goal_stream.hpp

@@ -92,6 +92,14 @@ class GoalStream
 
   void resume_waiter(std::coroutine_handle<> h);
 
+  // The waiting coroutine was stopped: cancel the goal it was following.
+  void cancel_goal_on_stop()
+  {
+    if (auto_cancel_on_stop_ && !completed_ && goal_handle_ && client_) {
+      client_->async_cancel_goal(goal_handle_);
+    }
+  }
+
 public:
   explicit GoalStream(CoContext & ctx, size_t max_depth = kDefaultStreamDepth)
   : ctx_(ctx), max_depth_(max_depth)
@@ -106,6 +114,19 @@ class GoalStream
     std::stop_token token;
     std::shared_ptr<StopCb> cancel_cb_;
     bool cancelled = false;
+    std::coroutine_handle<> waiting_{};
+
+    // A frame destroyed while suspended here must not stay registered as the
+    // stream's waiter, or the next message would resume it. Destroying the
+    // task stops it first, and the deferred cancellation no longer runs for a
+    // destroyed awaiter, so the goal is cancelled here instead.
+    ~NextAwaiter()
+    {
+      if (waiting_ && stream.waiter_ == waiting_) {
+        stream.waiter_ = nullptr;
+        stream.cancel_goal_on_stop();
+      }
+    }
 
     void set_token(std::stop_token t) { token = std::move(t); }
 
@@ -183,6 +204,15 @@ struct SendGoalAwaiter
   std::shared_ptr<StopCb> cancel_cb_;
   std::shared_ptr<State> state_;
 
+  // Same as SendRequestAwaiter: the goal response must not resume a frame
+  // destroyed while suspended here.
+  ~SendGoalAwaiter()
+  {
+    if (cancel_cb_) {
+      state_->done = true;
+    }
+  }
+
   void set_token(std::stop_token t) { token = std::move(t); }
 
   bool await_ready()

rclcpp_async/include/rclcpp_async/mutex.hpp

@@ -52,6 +52,16 @@ class Mutex
     std::shared_ptr<StopCb> cancel_cb_;
     std::shared_ptr<bool> active;
 
+    // A frame destroyed while suspended here must not stay registered, or a
+    // later unlock() would resume it. The deferred cancellation no longer runs
+    // for a destroyed awaiter, so the waiter is withdrawn here.
+    ~LockAwaiter()
+    {
+      if (active) {
+        *active = false;
+      }
+    }
+
     void set_token(std::stop_token t) { token = std::move(t); }
 
     bool await_ready()

rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp

@@ -53,6 +53,17 @@ struct SendRequestAwaiter
   std::shared_ptr<StopCb> cancel_cb_;
   std::shared_ptr<State> state_;
 
+  // A frame destroyed while suspended here must not be resumed by the
+  // response. The deferred cancellation no longer runs for a destroyed
+  // awaiter, so the request is marked done here. cancel_cb_ is set only while
+  // suspended, which leaves copies made before the co_await alone.
+  ~SendRequestAwaiter()
+  {
+    if (cancel_cb_) {
+      state_->done = true;
+    }
+  }
+
   void set_token(std::stop_token t) { token = std::move(t); }
 
   bool await_ready()

rclcpp_async/include/rclcpp_async/tf_buffer.hpp

@@ -111,6 +111,17 @@ class TfBuffer
     bool cancelled = false;
     std::shared_ptr<PendingRequest> request;
 
+    // A frame destroyed while suspended here must not stay pending, or the
+    // transform's arrival would resume it. The deferred cancellation no longer
+    // runs for a destroyed awaiter, so the request is withdrawn here.
+    ~LookupTransformAwaiter()
+    {
+      if (request && request->active && *request->active) {
+        tf.remove_pending(request);
+        *request->active = false;
+      }
+    }
+
     void set_token(std::stop_token t) { token = std::move(t); }
 
     bool await_ready()

rclcpp_async/include/rclcpp_async/topic_stream.hpp

@@ -59,6 +59,16 @@ class TopicStream
     std::stop_token token;
     std::shared_ptr<StopCb> cancel_cb_;
     bool cancelled = false;
+    std::coroutine_handle<> waiting_{};
+
+    // A frame destroyed while suspended here must not stay registered as the
+    // stream's waiter, or the next message would resume it.
+    ~NextAwaiter()
+    {
+      if (waiting_ && stream.waiter_ == waiting_) {
+        stream.waiter_ = nullptr;
+      }
+    }
 
     void set_token(std::stop_token t) { token = std::move(t); }
 

rclcpp_async/test/test_task_destroy.cpp

@@ -14,8 +14,10 @@
 
 #include <gtest/gtest.h>
 
+#include <atomic>
 #include <chrono>
 #include <example_interfaces/action/fibonacci.hpp>
+#include <example_interfaces/srv/add_two_ints.hpp>
 #include <memory>
 #include <rclcpp/rclcpp.hpp>
 #include <rclcpp_action/rclcpp_action.hpp>
@@ -26,6 +28,7 @@ using namespace rclcpp_async;          // NOLINT(build/namespaces)
 using namespace std::chrono_literals;  // NOLINT(build/namespaces)
 using Fibonacci = example_interfaces::action::Fibonacci;
 using GoalHandle = rclcpp_action::ServerGoalHandle<Fibonacci>;
+using AddTwoInts = example_interfaces::srv::AddTwoInts;
 
 class TaskDestroyTest : public ::testing::Test
 {
@@ -58,14 +61,26 @@ class TaskDestroyTest : public ::testing::Test
     }
   }
 
+  void spin_for(std::chrono::milliseconds duration)
+  {
+    auto deadline = std::chrono::steady_clock::now() + duration;
+    while (std::chrono::steady_clock::now() < deadline) {
+      executor_.spin_some();
+      std::this_thread::sleep_for(1ms);
+    }
+  }
+
   void create_feedback_server(int count = 30, int interval_ms = 100)
   {
     action_server_ = rclcpp_action::create_server<Fibonacci>(
       node_, "test_destroy_action",
       [](const rclcpp_action::GoalUUID &, std::shared_ptr<const Fibonacci::Goal>) {
         return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE;
       },
-      [](const std::shared_ptr<GoalHandle>) { return rclcpp_action::CancelResponse::ACCEPT; },
+      [this](const std::shared_ptr<GoalHandle>) {
+        cancel_requested_ = true;
+        return rclcpp_action::CancelResponse::ACCEPT;
+      },
       [count, interval_ms](const std::shared_ptr<GoalHandle> goal_handle) {
         std::thread([goal_handle, count, interval_ms]() {
           std::this_thread::sleep_for(200ms);
@@ -107,6 +122,7 @@ class TaskDestroyTest : public ::testing::Test
   rclcpp::executors::SingleThreadedExecutor executor_;
   rclcpp_action::Client<Fibonacci>::SharedPtr action_client_;
   rclcpp_action::Server<Fibonacci>::SharedPtr action_server_;
+  std::atomic<bool> cancel_requested_{false};
 };
 
 TEST_F(TaskDestroyTest, DestroyWhileSuspendedOnTimer)
@@ -191,3 +207,110 @@ TEST_F(TaskDestroyTest, DestroyGoalTaskWhileAwaitingFeedback)
   spin_until_done(task, 15s);
   EXPECT_TRUE(outer_done);
 }
+
+TEST_F(TaskDestroyTest, DestroyingATaskAwaitingFeedbackCancelsTheGoal)
+{
+  create_feedback_server(30, 100);
+  wait_for_server();
+
+  auto follow = [this]() -> Task<void> {
+    Fibonacci::Goal goal;
+    goal.order = 30;
+    auto goal_result = co_await ctx_->send_goal<Fibonacci>(action_client_, goal);
+    if (!goal_result.ok()) {
+      co_return;
+    }
+    auto stream = *goal_result.value;
+    while (co_await stream->next()) {
+    }
+  };
+  {
+    auto task = ctx_->create_task(follow());
+    auto deadline = std::chrono::steady_clock::now() + 2s;
+    while (std::chrono::steady_clock::now() < deadline) {
+      executor_.spin_some();
+      std::this_thread::sleep_for(1ms);
+    }
+  }
+
+  auto deadline = std::chrono::steady_clock::now() + 2s;
+  while (!cancel_requested_ && std::chrono::steady_clock::now() < deadline) {
+    executor_.spin_some();
+    std::this_thread::sleep_for(1ms);
+  }
+  EXPECT_TRUE(cancel_requested_);
+}
+
+static Task<void> wait_event(Event & event) { co_await event.wait(); }
+static Task<void> lock_mutex(Mutex & mutex) { co_await mutex.lock(); }
+
+TEST_F(TaskDestroyTest, EventSetAfterWaiterDestroyed)
+{
+  Event event(*ctx_);
+  {
+    auto task = ctx_->create_task(wait_event(event));
+    spin_for(50ms);
+  }
+  spin_for(50ms);  // let the deferred cancellation run
+  event.set();
+  spin_for(50ms);
+}
+
+TEST_F(TaskDestroyTest, EventSetBeforeDeferredCancellationRuns)
+{
+  Event event(*ctx_);
+  {
+    auto task = ctx_->create_task(wait_event(event));
+    spin_for(50ms);
+  }
+  event.set();  // the deferred cancellation has not run yet
+  spin_for(50ms);
+}
+
+TEST_F(TaskDestroyTest, MutexUnlockAfterWaiterDestroyed)
+{
+  Mutex mutex(*ctx_);
+  auto holder = ctx_->create_task(lock_mutex(mutex));
+  spin_for(50ms);
+  ASSERT_TRUE(mutex.is_locked());
+  {
+    auto task = ctx_->create_task(lock_mutex(mutex));
+    spin_for(50ms);
+  }
+  spin_for(50ms);
+  mutex.unlock();
+  spin_for(50ms);
+  EXPECT_FALSE(mutex.is_locked());
+}
+
+TEST_F(TaskDestroyTest, ResponseAfterRequesterDestroyed)
+{
+  std::shared_ptr<rmw_request_id_t> pending_header;
+  auto service = node_->create_service<AddTwoInts>(
+    "test_destroy_service",
+    [&pending_header](
+      std::shared_ptr<rclcpp::Service<AddTwoInts>>, std::shared_ptr<rmw_request_id_t> header,
+      std::shared_ptr<AddTwoInts::Request>) { pending_header = header; });
+  auto client = node_->create_client<AddTwoInts>("test_destroy_service");
+  auto deadline = std::chrono::steady_clock::now() + 5s;
+  while (!client->service_is_ready() && std::chrono::steady_clock::now() < deadline) {
+    spin_for(10ms);
+  }
+  ASSERT_TRUE(client->service_is_ready());
+
+  auto request = [this, &client]() -> Task<void> {
+    co_await ctx_->send_request<AddTwoInts>(client, std::make_shared<AddTwoInts::Request>());
+  };
+  {
+    auto task = ctx_->create_task(request());
+    deadline = std::chrono::steady_clock::now() + 5s;
+    while (!pending_header && std::chrono::steady_clock::now() < deadline) {
+      spin_for(10ms);
+    }
+    ASSERT_TRUE(pending_header);
+  }
+  spin_for(50ms);  // let the deferred cancellation run
+  AddTwoInts::Response response;
+  service->send_response(*pending_header, response);
+  spin_for(200ms);
+}

rclcpp_async/test/test_tf_buffer.cpp

@@ -219,6 +219,31 @@ TEST_F(TfBufferTest, CancellationRemovesPendingRequest)
   EXPECT_TRUE(was_cancelled);
 }
 
+TEST_F(TfBufferTest, TransformAfterWaiterDestroyed)
+{
+  auto coro = [&]() -> Task<void> {
+    co_await tf_buffer_->lookup_transform("p", "q", rclcpp::Time(0));
+  };
+  {
+    auto task = ctx_->create_task(coro());
+    for (int i = 0; i < 5; i++) {
+      executor_.spin_some();
+      std::this_thread::sleep_for(10ms);
+    }
+  }
+  // Let the deferred cancellation run, then deliver the awaited transform.
+  for (int i = 0; i < 5; i++) {
+    executor_.spin_some();
+    std::this_thread::sleep_for(10ms);
+  }
+  auto broadcaster = make_static_broadcaster(node_);
+  broadcaster.sendTransform(make_transform("p", "q"));
+  for (int i = 0; i < 30; i++) {
+    executor_.spin_some();
+    std::this_thread::sleep_for(10ms);
+  }
+}
+
 TEST_F(TfBufferTest, MultipleConcurrentLookups)
 {
   geometry_msgs::msg::TransformStamped r1, r2;