Deadlock in stdexec::task destructor when affine sender completes synchronously with set_stopped

[mika-fischer]

The test below deadlocks with current main.

The underlying issue seems to be that in __as_awaitable.hpp __async_receiver::set_value() and set_error() both call __done(), which CAS-sets __ready_ to true to signal that the inline (same-thread) completion has occurred. However, set_stopped() does not call __done(). It calls promise.unhandled_stopped() directly, leaving __ready_ as false.

The unhandled_stopped() chain eventually destroys the coroutine frame, which destroys the __sender_awaitable. The destructor calls __ready_.wait(false), blocking until __ready_ becomes non-false. Since nobody ever set it to true, this is a permanent deadlock on the same thread that called start().

I'm not sure what the proper fix is.

  struct inline_affine_stopped_sender
  {
    using sender_concept        = ex::sender_tag;
    using completion_signatures = ex::completion_signatures<ex::set_stopped_t()>;

    template <class Receiver>
    struct operation
    {
      Receiver rcvr_;

      void start() & noexcept
      {
        ex::set_stopped(std::move(rcvr_));
      }
    };

    template <class Receiver>
    auto connect(Receiver rcvr) && -> operation<Receiver>
    {
      return {std::move(rcvr)};
    }

    struct attrs
    {
      [[nodiscard]]
      static constexpr auto query(ex::__get_completion_behavior_t<ex::set_stopped_t>) noexcept
      {
        return ex::__completion_behavior::__inline_completion
             | ex::__completion_behavior::__asynchronous_affine;
      }
    };

    [[nodiscard]]
    auto get_env() const noexcept -> attrs
    {
      return {};
    }
  };

  TEST_CASE("task co_awaiting inline|async_affine stopped sender does not deadlock",
            "[types][task]")
  {
    auto res = ex::sync_wait([]() -> ex::task<int> {
        co_await inline_affine_stopped_sender{};
        FAIL("Expected co_awaiting inline_affine_stopped_sender to stop the task");
        co_return 42;
    }());
    CHECK(!res.has_value());
  }