exec::merge_each never completes when the sender of a nested sequence completes with set_error or set_stopped #2445

// exec::merge_each never completes when the sender of a nested sequence
// completes with set_error or set_stopped instead of producing a sequence.
//
// Build, from the stdexec checkout:
//   c++ -std=c++20 -pthread -I include repro.cpp -o repro
// Exit code: 0 if every operation completed, 1 otherwise.
#include <stdexec/execution.hpp>

#include <exec/sequence/ignore_all_values.hpp>
#include <exec/sequence/iterate.hpp>
#include <exec/sequence/merge_each.hpp>
#include <exec/sequence/transform_each.hpp>
#include <exec/variant_sender.hpp>

#include <cstdio>
#include <ranges>

namespace ex = stdexec;

struct receiver
{
  using receiver_concept = ex::receiver_tag;

  void set_value() noexcept
  {
    *result = "set_value()";
  }

  template <class Error>
  void set_error(Error&&) noexcept
  {
    *result = "set_error(...)";
  }

  void set_stopped() noexcept
  {
    *result = "set_stopped()";
  }

  auto get_env() const noexcept
  {
    return ex::prop{ex::get_stop_token, token};
  }

  char const **          result;
  ex::inplace_stop_token token{};
};

template <class Sender>
bool run(char const * what, Sender&& sndr, ex::inplace_stop_token token = {})
{
  char const * result = nullptr;
  {
    auto op = ex::connect(static_cast<Sender&&>(sndr), receiver{&result, token});
    ex::start(op);
  }
  // Everything above runs inline, so the receiver has been called by now, or
  // it never will be.
  std::printf("%-58s -> %s\n", what, result ? result : "NEVER COMPLETED");
  return result != nullptr;
}

int main()
{
  bool ok = true;

  // 1. The sender of the nested sequence completes with set_error.
  ok &= run("sender of nested sequence throws",
            exec::merge_each(ex::just(1)
                             | ex::then(
                               [](int i)
                               {
                                 if (i == 1)
                                   throw 42;
                                 return ex::just(i);
                               }))
              | exec::ignore_all_values());

  // 2. The sender of the nested sequence completes with set_stopped.
  using maybe_sequence_t =
    exec::variant_sender<decltype(ex::just(ex::just(0))), decltype(ex::just_stopped())>;
  ok &= run("sender of nested sequence is stopped",
            exec::merge_each(
              ex::just(1)
              | ex::let_value([](int) noexcept -> maybe_sequence_t { return ex::just_stopped(); }))
              | exec::ignore_all_values());

  // 3. Stop is requested while the second of three nested sequences runs. The
  //    sender of the third nested sequence then completes with set_stopped.
  ex::inplace_stop_source stop_source;
  int                     count = 0;
  ok &= run("stop requested between nested sequences",
            exec::merge_each(
              exec::iterate(std::views::iota(0, 3))
              | exec::transform_each(
                ex::then([](int) noexcept { return exec::iterate(std::views::iota(0, 3)); })))
              | exec::transform_each(ex::then(
                [&](int) noexcept
                {
                  if (++count == 4)
                    stop_source.request_stop();
                }))
              | exec::ignore_all_values(),
            stop_source.get_token());

  return ok ? 0 : 1;
}