From 188225ed90914ec1be962a83438eee2c29d7b4dc Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Mon, 10 Aug 2026 11:24:49 +0200 Subject: [PATCH 1/3] Fix merge_each error propagation --- include/exec/sequence/merge_each.hpp | 17 ++-- test/exec/sequence/test_merge_each.cpp | 136 +++++++++++++++++++++---- 2 files changed, 123 insertions(+), 30 deletions(-) diff --git a/include/exec/sequence/merge_each.hpp b/include/exec/sequence/merge_each.hpp index 923b28ddd..397e312aa 100644 --- a/include/exec/sequence/merge_each.hpp +++ b/include/exec/sequence/merge_each.hpp @@ -383,16 +383,12 @@ namespace experimental::execution } void set_break() noexcept { - switch (__completion_.exchange(__completion_t::__stopped)) + auto __expected = __completion_t::__started; + if (__completion_.compare_exchange_strong(__expected, __completion_t::__stopped)) { - case __completion_t::__started: - // We must request stop. When the previous state is __error or __stopped, then stop has - // already been requested. + // We transitioned from started to stopped, so we must request stop. When the state is + // already error or stopped, stop has already been requested. __nested_stop_.request_stop(); - break; - case __completion_t::__stopped: - [[fallthrough]]; // We're already in the "stopped" state. Ignore the break. - case __completion_t::__error:; // We're already in the "error" state. Ignore the break. } } @@ -455,8 +451,9 @@ namespace experimental::execution } void error_complete() noexcept override { - // do not double report error - STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); + // The error has been delivered as an item. Complete the sequence so the consumer can + // publish it, unless the operation was independently stopped. + exec::__set_value_unless_stopped(static_cast<_Receiver&&>(__rcvr_)); } void complete_if_none_active() noexcept diff --git a/test/exec/sequence/test_merge_each.cpp b/test/exec/sequence/test_merge_each.cpp index 2160a49c4..090c1262d 100644 --- a/test/exec/sequence/test_merge_each.cpp +++ b/test/exec/sequence/test_merge_each.cpp @@ -33,6 +33,7 @@ #include #include +#include using namespace std::chrono_literals; using namespace exec; @@ -227,6 +228,86 @@ namespace auto op = subscribe(std::move(merged), null_receiver{}); } + template + struct error_sequence + { + using sender_concept = sequence_sender_tag; + using item_types = exec::item_types<>; + using completion_signatures = ex::completion_signatures; + + template + struct operation + { + void start() & noexcept + { + ex::set_error(static_cast(receiver_), static_cast(error_)); + } + + Receiver receiver_; + Error error_; + }; + + template + auto subscribe(Receiver receiver) && noexcept -> operation + { + return {static_cast(receiver), static_cast(error_)}; + } + + Error error_; + }; + + struct error_state + { + std::optional error_{}; + bool completed_ = false; + bool stopped_ = false; + }; + + struct record_error_receiver + { + using receiver_concept = ex::receiver_tag; + + template + auto set_next(Item&& item) & + { + auto record_error = [state = state_](Error&& error) noexcept + { + if constexpr (std::same_as, int>) + { + state->error_ = static_cast(error); + } + }; + return ex::upon_stopped(ex::upon_error(static_cast(item), record_error), + []() noexcept {}); + } + + void set_value() noexcept + { + state_->completed_ = true; + } + + template + void set_error(Error&& error) noexcept + { + if constexpr (std::same_as, int>) + { + state_->error_ = static_cast(error); + } + } + + void set_stopped() noexcept + { + state_->stopped_ = true; + } + + auto get_env() const noexcept -> ex::env<> + { + return {}; + } + + error_state* state_; + }; + TEST_CASE("merge_each - merge two sequence senders of no elements", "[sequence_senders][merge_each][empty_sequence]") { @@ -368,31 +449,46 @@ namespace CHECK(v.has_value() == true); } -// TODO - fix problem with stopping -#if 0 - TEST_CASE( - "merge_each - merge_each sender stops when a nested sequence fails", - "[sequence_senders][static_thread_pool][merge_each][merge][iterate]") { + TEST_CASE("merge_each - preserves errors from nested value senders", + "[sequence_senders][merge_each]") + { + error_state state{}; + auto merged = merge_each(ex::just(ex::just_error(42))); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); - auto sequences = merge( - log_start(range(100, 120), "range 100-120"), - ex::just(emits_error(std::runtime_error{"failed sequence "})), - log_start(range(200, 220), "range 200-220") - ); + ex::start(op); - [[maybe_unused]] auto merged = merge_each(std::move(sequences)); + CHECK(state.error_ == 42); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } - int count = 0; + TEST_CASE("merge_each - preserves errors from nested sequences", "[sequence_senders][merge_each]") + { + error_state state{}; + auto nested_sequence = error_sequence{42}; + auto merged = merge_each(ex::just(std::move(nested_sequence))); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); + + ex::start(op); + + CHECK(state.error_ == 42); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } + + TEST_CASE("merge_each - preserves errors from the outer sequence", + "[sequence_senders][merge_each]") + { + error_state state{}; + auto merged = merge_each(error_sequence{42}); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); - auto v = ex::sync_wait(ignore_all_values(merged | then_each([&count](int x){ - ++count; - UNSCOPED_INFO("item: " << x - << ", on thread id: " << std::this_thread::get_id()); - }))); + ex::start(op); - CHECK(count == 20); - CHECK(v.has_value() == false); + CHECK(state.error_ == 42); + CHECK_FALSE(state.completed_); + CHECK_FALSE(state.stopped_); } -#endif // 0 } // namespace From 9ee0c9970b23e5c42d7961d0094914766d149dd8 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 15 Aug 2026 21:08:41 +0200 Subject: [PATCH 2/3] Guard merge_each outer error test without exceptions --- test/exec/sequence/test_merge_each.cpp | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/test/exec/sequence/test_merge_each.cpp b/test/exec/sequence/test_merge_each.cpp index 090c1262d..03802fa16 100644 --- a/test/exec/sequence/test_merge_each.cpp +++ b/test/exec/sequence/test_merge_each.cpp @@ -259,8 +259,9 @@ namespace struct error_state { std::optional error_{}; - bool completed_ = false; - bool stopped_ = false; + bool outer_error_ = false; + bool completed_ = false; + bool stopped_ = false; }; struct record_error_receiver @@ -293,6 +294,10 @@ namespace { state_->error_ = static_cast(error); } + else if constexpr (std::same_as, std::exception_ptr>) + { + state_->outer_error_ = true; + } } void set_stopped() noexcept @@ -477,6 +482,7 @@ namespace CHECK_FALSE(state.stopped_); } +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() TEST_CASE("merge_each - preserves errors from the outer sequence", "[sequence_senders][merge_each]") { @@ -486,9 +492,10 @@ namespace ex::start(op); - CHECK(state.error_ == 42); + CHECK(state.outer_error_); CHECK_FALSE(state.completed_); CHECK_FALSE(state.stopped_); } +#endif } // namespace From 7c08e0c7f776d261786b9b5cac122f8d2e0120c9 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 15 Aug 2026 21:13:02 +0200 Subject: [PATCH 3/3] Record merge_each outer error in regression test --- test/exec/sequence/test_merge_each.cpp | 32 ++++++++++++++++++-------- 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/test/exec/sequence/test_merge_each.cpp b/test/exec/sequence/test_merge_each.cpp index 03802fa16..4d53398c7 100644 --- a/test/exec/sequence/test_merge_each.cpp +++ b/test/exec/sequence/test_merge_each.cpp @@ -33,6 +33,7 @@ #include #include +#include #include using namespace std::chrono_literals; @@ -258,10 +259,10 @@ namespace struct error_state { - std::optional error_{}; - bool outer_error_ = false; - bool completed_ = false; - bool stopped_ = false; + std::optional nested_error_{}; + std::exception_ptr outer_error_{}; + bool completed_ = false; + bool stopped_ = false; }; struct record_error_receiver @@ -275,7 +276,7 @@ namespace { if constexpr (std::same_as, int>) { - state->error_ = static_cast(error); + state->nested_error_ = static_cast(error); } }; return ex::upon_stopped(ex::upon_error(static_cast(item), record_error), @@ -292,11 +293,11 @@ namespace { if constexpr (std::same_as, int>) { - state_->error_ = static_cast(error); + state_->nested_error_ = static_cast(error); } else if constexpr (std::same_as, std::exception_ptr>) { - state_->outer_error_ = true; + state_->outer_error_ = static_cast(error); } } @@ -463,7 +464,7 @@ namespace ex::start(op); - CHECK(state.error_ == 42); + CHECK(state.nested_error_ == 42); CHECK(state.completed_); CHECK_FALSE(state.stopped_); } @@ -477,7 +478,7 @@ namespace ex::start(op); - CHECK(state.error_ == 42); + CHECK(state.nested_error_ == 42); CHECK(state.completed_); CHECK_FALSE(state.stopped_); } @@ -492,7 +493,18 @@ namespace ex::start(op); - CHECK(state.outer_error_); + CHECK(state.outer_error_ != nullptr); + bool caught = false; + try + { + std::rethrow_exception(state.outer_error_); + } + catch (int error) + { + caught = true; + CHECK(error == 42); + } + CHECK(caught); CHECK_FALSE(state.completed_); CHECK_FALSE(state.stopped_); }