Leave a destroyed awaiter alone when its cancellation runs late

[yukiendo-pfr]

register_cancel queues the cancellation on the executor and, when it runs, calls is_done() and action() before it checks that the awaiter is still alive. TopicStream and GoalStream pass closures that capture the awaiter (this), so a task destroyed after cancel() but before that queued callback runs makes it read the destroyed frame. ~Task() itself requests a stop before destroying the frame, so dropping a suspended task is enough to trigger it.

The same streams also keep the destroyed frame as their waiter_, since only the deferred cancellation would clear it. A stream that outlives the coroutine (held outside it) then resumes the destroyed frame on its next message.

Check that the awaiter is alive before calling is_done() / action(), and clear the stream's waiter_ from the NextAwaiter destructor when it is still the one registered. TimerStream and Channel already clear their waiter synchronously in the stop callback and check the awaiter before resuming, so they are unchanged.

Add two TopicStream tests that destroy a waiting task (after and without cancel()) and then keep using the stream. Built with -fsanitize=address, the first reports a heap-use-after-free in the register_cancel closure without this change and passes with it. (LeakSanitizer was off for that run: create_timer's stream and timer keep each other alive, which reports leaks independently of this change.)

GoalStream's deferred cancellation also cancels the goal (set_auto_cancel_on_stop). Destroying a task that waits for feedback relied on that callback running against the destroyed awaiter. With the awaiter check, it no longer runs, so NextAwaiter's destructor now cancels the goal itself when it is destroyed while still waiting. test_task_destroy gains DestroyingATaskAwaitingFeedbackCancelsTheGoal, which fails without that and passes with it. (Without it, the test server's feedback thread outlived the test and crashed the lyrical Release job at exit.)

[otamachan]

Thanks for tracking this down — the GoalStream fix works: on main, DestroyGoalTaskWhileAwaitingFeedback and DestroyingATaskAwaitingFeedbackCancelsTheGoal hit a heap-use-after-free under ASan, and with this PR they're clean.

However, the early return in register_cancel introduces a regression for the other awaiters. ~Task() requests stop and destroys the frame right away, so by the time the posted cancellation runs, weak.lock() always fails and action() is skipped. Before this PR, action() still ran and only the resume was skipped. That action() is what unregisters the waiter:

• Event::wait / Mutex::lock: *active = false
• send_request / send_goal: state->done = true (the comment in SendRequestAwaiter::await_suspend relies on this)
• TfBuffer::lookup_transform: remove_pending and *active = false

Without it, a later set() / unlock() / response resumes the destroyed frame.

These tests pass on main and crash with this PR (segfault 5/5 runs in a plain Debug build, heap-use-after-free under ASan):

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, 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());
}

(spin_for just calls executor_.spin_some() in a loop for the given duration. The same pattern with a deferred service response reproduces it for send_request.)

Since the existing tests only destroy TopicStream / GoalStream waiters, CI doesn't catch this.

Suggestion: do what you already did for GoalStream::NextAwaiter in every awaiter — unregister the waiter synchronously in its destructor (*active = false for Event/Mutex, state->done = true for send_request/send_goal, remove_pending + *active = false for TfBuffer). That also fixes a race that already exists on main: if set() is called after the task is destroyed but before the posted cancellation has run, main resumes the destroyed frame too:

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);
}

Could you add these tests along with the fix?

[yukiendo-pfr]

Thanks for the careful review — you're right on all counts. Skipping action() for a destroyed awaiter left Event/Mutex/send_request/send_goal/TfBuffer waiters registered, so a later set() / unlock() / response / transform resumed the destroyed frame.

I've followed your suggestion: each of those awaiters now withdraws its waiter in its destructor (*active = false for Event/Mutex, state_->done = true for send_request/send_goal, remove_pending + *active = false for TfBuffer). For send_request/send_goal the destructor only acts while cancel_cb_ is set, i.e. while actually suspended, so a copy made before the co_await (e.g. by wait_for) doesn't mark the shared state done. The early return in register_cancel stays, since TopicStream/GoalStream's is_done reads this.

I added your three tests, plus ResponseAfterRequesterDestroyed (deferred service response) and TfBufferTest.TransformAfterWaiterDestroyed. Without the fix all five segfault in a Debug build and report heap-use-after-free under ASan (3/3 runs each); with it they pass, and the full suite passes in Debug.

Two related things I noticed but left out of this PR:

• Channel::send() / close() and TfBuffer::check_pending() take the waiter and post its resume without a liveness check, so a task destroyed between that and the posted resume is still resumed. The destructor can't help there since the waiter has already been taken; this needs a separate fix.
• Under ASan, TimerStreamTest.BasicNext reports a stack-use-after-scope at test_timer_stream.cpp:76 on main as well, in the test itself.

Happy to send follow-ups for either.

[otamachan]

Thanks a lot for the quick and thorough fix, and for the extra tests!