Stop libdispatch bulk after value capture failure

include/exec/libdispatch_queue.hpp

@@ -305,7 +305,8 @@ namespace experimental::execution
                 _WITH_PRETTY_SENDER_<__copy_cvref_t<Self, Sender>>,
                 _WITH_ENVIRONMENT_(Env...)>();
             }
-            else if constexpr (__nothrow_applicable<Fun &, arg_pack_t>)
+            else if constexpr (__nothrow_applicable<Fun &, arg_pack_t>
+                               && __nothrow_decay_copyable<Args...>)
             {
               return completion_signatures<value_sig_t>();
             }
@@ -476,16 +477,17 @@ namespace experimental::execution
         {
           STDEXEC_TRY
           {
-            shared_state_.data_.template emplace<tuple_t>(std::move(as)...);
+            shared_state_.data_.template emplace<tuple_t>(static_cast<As &&>(as)...);
           }
           STDEXEC_CATCH_ALL
           {
             STDEXEC::set_error(std::move(shared_state_.rcvr_), std::current_exception());
+            return;
           }
         }
         else
         {
-          shared_state_.data_.template emplace<tuple_t>(std::move(as)...);
+          shared_state_.data_.template emplace<tuple_t>(static_cast<As &&>(as)...);
         }
 
         if (shared_state_.shape_)

test/exec/test_libdispatch.cpp

@@ -17,6 +17,7 @@
 #include "exec/libdispatch_queue.hpp"
 #include "stdexec/execution.hpp"
 #include "test_common/catch2.hpp"
+#include "test_common/type_helpers.hpp"
 
 #include <numeric>
 #include <utility>
@@ -30,7 +31,7 @@ namespace
     auto                    sch = queue.get_scheduler();
 
     std::vector<int> data{1, 2, 3, 4, 5};
-    auto             add = [](auto const & data)
+    auto             add = [](auto const &data)
     {
       return std::accumulate(std::begin(data), std::end(data), 0);
     };
@@ -52,11 +53,11 @@ namespace
 
     std::vector<int> data{1, 2, 3, 4, 5};
     auto             size                  = data.size();
-    auto             expensive_computation = [](auto i, auto& data)
+    auto             expensive_computation = [](auto i, auto &data)
     {
       data[i] = 2 * data[i];
     };
-    auto add = [](auto const & data)
+    auto add = [](auto const &data)
     {
       return std::accumulate(std::begin(data), std::end(data), 0);
     };
@@ -85,7 +86,7 @@ namespace
         throw 999;
       return 2 * data[i];
     };
-    auto add = [](auto const & data)
+    auto add = [](auto const &data)
     {
       return std::accumulate(std::begin(data), std::end(data), 0);
     };
@@ -106,5 +107,98 @@ namespace
       FAIL("invalid exception caught");
     }
   }
+
+  TEST_CASE("libdispatch bulk stops after value capture fails")
+  {
+    struct value_capture_error
+    {};
+
+    struct throwing_value
+    {
+      throwing_value() = default;
+
+      throwing_value(throwing_value const &)
+      {
+        throw value_capture_error{};
+      }
+
+      throwing_value(throwing_value &&)
+      {
+        throw value_capture_error{};
+      }
+    };
+
+    exec::libdispatch_queue queue;
+    auto                    sch = queue.get_scheduler();
+
+    auto sender = STDEXEC::schedule(sch) | STDEXEC::then([]() noexcept { return throwing_value{}; })
+                | STDEXEC::bulk(STDEXEC::par, 0, [](int, throwing_value &) noexcept {});
+
+    STATIC_REQUIRE(
+      set_equivalent<STDEXEC::completion_signatures_of_t<decltype(sender), STDEXEC::env<>>,
+                     STDEXEC::completion_signatures<STDEXEC::set_value_t(throwing_value),
+                                                    STDEXEC::set_error_t(std::exception_ptr),
+                                                    STDEXEC::set_stopped_t()>>);
+
+    STDEXEC_TRY
+    {
+      STDEXEC::sync_wait(std::move(sender));
+      CHECK(false);
+    }
+    STDEXEC_CATCH(value_capture_error const &)
+    {
+    }
+    STDEXEC_CATCH_ALL
+    {
+      FAIL("invalid exception caught");
+    }
+  }
+
 #endif
+
+  TEST_CASE("libdispatch bulk preserves lvalue-reference value categories")
+  {
+    struct lvalue_value
+    {
+      lvalue_value()
+        : value(0)
+      {}
+
+      explicit lvalue_value(int value)
+        : value(value)
+      {}
+
+      lvalue_value(lvalue_value const &) noexcept = default;
+
+      lvalue_value(lvalue_value &&other) noexcept(false)
+        : value(other.value)
+      {
+        other.moved_from = true;
+      }
+
+      int  value;
+      bool moved_from = false;
+    } value{42};
+
+    exec::libdispatch_queue queue;
+    auto                    sch  = queue.get_scheduler();
+    int                     seen = 0;
+
+    auto sender = STDEXEC::schedule(sch)
+                | STDEXEC::then([&]() noexcept -> lvalue_value & { return value; })
+                | STDEXEC::bulk(STDEXEC::par,
+                                1,
+                                [&](int, lvalue_value &item) noexcept { seen = item.value; });
+
+    STATIC_REQUIRE(
+      set_equivalent<STDEXEC::completion_signatures_of_t<decltype(sender), STDEXEC::env<>>,
+                     STDEXEC::completion_signatures<STDEXEC::set_value_t(lvalue_value),
+                                                    STDEXEC::set_stopped_t()>>);
+
+    auto result = STDEXEC::sync_wait(std::move(sender));
+
+    REQUIRE(result.has_value());
+    CHECK(seen == 42);
+    CHECK_FALSE(value.moved_from);
+  }
 }  // namespace