SIGSEGV with `co_await (stdexec::get_scheduler() | ...);`

Run the Godbolt reproducer and trace the await path through include/stdexec/__detail/__as_awaitable.hpp, especially __sender_awaiter::await_suspend and __get_continuation. Compare the reported regression with the workaround and the intended ordering example. Done means sync_wait does not return prematurely, the task coroutine is not destroyed before resumption, and the reproducer no longer reaches SIGSEGV.

Summary

Using co_await (ex::get_scheduler() | ex::let_value([](auto sch) { return ex::schedule(sch); })); in a task leads to a premature exit of sync_wait and a subsequent use-after-free of the coroutine.
This is a regression introduced with 482c2605918d3aadbd472960e207126a21809140 #2078

Tested with Clang and GCC on x86_64

Workaround

            auto sch = co_await ex::get_scheduler();
            co_await (ex::just(sch) | ex::let_value([](auto sch) { return ex::schedule(sch); }));

Reproducer

https://godbolt.org/z/WE9fs5WPv

#include <coroutine>
#include <stdexec/execution.hpp>

namespace ex = STDEXEC;

void foo()
{
    int order[2] = {0, 0};
    size_t idx = 0;
    ex::sync_wait(
        [&]() -> ex::task<void> {
            order[idx++] = 1;
            co_await (ex::get_scheduler() | ex::let_value([](auto sch) { return ex::schedule(sch); }));
            order[idx++] = 2;
        }());
}

void foo_workaround()
{
    int order[2] = {0, 0};
    size_t idx = 0;
    ex::sync_wait(
        [&]() -> ex::task<void> {
            order[idx++] = 1;
            auto sch = co_await ex::get_scheduler();
            co_await (ex::just(sch) | ex::let_value([](auto sch) { return ex::schedule(sch); }));
            order[idx++] = 2;
        }());
}

int main()
{
    foo(); // crashes with SIGSEGV probaly use after free
    foo_workaround();
}

/* my usage:
namespace Tools
{
    // Give control back to the scheduler.
    struct Yield : ex::sender_adaptor_closure<Yield>
    {
        auto operator()() const
        {
            return ex::get_scheduler() | ex::let_value([](auto sch) { return ex::schedule(sch); });
        }

        template<class Sndr>
        auto operator()(Sndr sndr) const;
    };

    inline constexpr Yield yield{};

    template<class Sndr>
    auto Yield::operator()(Sndr sndr) const
    {
        return std::move(sndr) | ex::let_value([](auto... value) {
                   return ex::when_all(ex::just(std::move(value)...), Tools::yield());
               });
    }
} 

void foo_intended_use()
{
    int order[4] = {0, 0, 0, 0};
    size_t idx = 0;
    ex::sync_wait(
        ex::when_all(
            [&]() -> ex::task<void> {
                order[idx++] = 1;
                auto sch = co_await ex::get_scheduler();
                co_await Tools::yield();
                order[idx++] = 2;
            }(),
            [&]() -> ex::task<void> {
                order[idx++] = 3;
                auto sch = co_await ex::get_scheduler();
                co_await Tools::yield();
                order[idx++] = 4;
            }()));
    assert(order[0] == 1);
    assert(order[1] == 3);
    assert(order[2] == 2);
    assert(order[3] == 4);
}
*/

Analysis

We analyzed this on 307b83c5689ea7c2e5b31561cdc428697705333e and fee4d651494014610a277540f209cae56011e47f with our intended code, but without the when_all.
We debugged it and it seems, that after the co_await Tools::yield() the sync_wait returns, because __coroutine_unhandled_stopped is called. This leads then to the destruction of the coroutine handle and then a use-after-free once the run_loop wants to execute the rest of the task:

include/stdexec/__detail/__as_awaitable.hpp

    // When the sender is known to complete inline, we can connect and start the operation
    // in await_suspend.
    template <class _Promise, sender_in<env_of_t<_Promise&>> _Sender>
      requires __completes_inline<_Sender, env_of_t<_Promise&>>
    struct __sender_awaiter<_Promise, _Sender>
      : __sender_awaiter_base<__value_t<_Sender, _Promise>, true>
    {
...
      auto await_suspend([[maybe_unused]] __std::coroutine_handle<> __continuation)
        -> __std::coroutine_handle<>
      {
        STDEXEC_ASSERT(this->__continuation_.handle() == __continuation);
        {
          auto __opstate = STDEXEC::connect(static_cast<_Sender&&>(__sndr_), __receiver_t(*this));
          // The following call to start will complete synchronously, writing its result
          // into the __result_ variant.
          STDEXEC::start(__opstate);         // yields continuation was added here
        }

        return this->__get_continuation();   // sync_wait will return after this.
      }
...
    };
...
      [[nodiscard]]
      constexpr auto __get_continuation() const noexcept -> __std::coroutine_handle<>
      {
        // If the operation was stopped (__result_ is valueless), we should use the
        // unhandled_stopped() continuation. Otherwise, should resume the __continuation_
        // as normal.
        if (__result_.__is_valueless()) // after Tools::yield() __result_ is considered valueless because set_value was not called
        {
          return STDEXEC::__coroutine_unhandled_stopped(__continuation_); // sync_wait returns because of this
        }
        else
        {
          return __continuation_.handle();
        }
      }