fix(core): prevent duplicate registration after registrar moves

phlex/core/registrar.hpp

@@ -80,8 +80,25 @@ namespace phlex::detail {
     registrar(registrar const&) = delete;
     registrar& operator=(registrar const&) = delete;
 
-    registrar(registrar&&) = default;
-    registrar& operator=(registrar&&) = default;
+    // Moving must clear the source creator so its destructor cannot register the node again.
+    registrar(registrar&& other) noexcept :
+      nodes_{other.nodes_},
+      errors_{other.errors_},
+      creator_{std::exchange(other.creator_, node_creator{})},
+      predicates_{std::exchange(other.predicates_, std::nullopt)},
+      output_product_suffixes_{std::move(other.output_product_suffixes_)}
+    {
+    }
+
+    registrar& operator=(registrar&& other) noexcept
+    {
+      nodes_ = other.nodes_;
+      errors_ = other.errors_;
+      creator_ = std::exchange(other.creator_, node_creator{});
+      predicates_ = std::exchange(other.predicates_, std::nullopt);
+      output_product_suffixes_ = std::move(other.output_product_suffixes_);
+      return *this;
+    }
 
     bool has_predicates() const { return predicates_.has_value(); }
 

test/core_misc_test.cpp

@@ -1,4 +1,5 @@
 #include "phlex/configuration.hpp"
+#include "phlex/core/declared_output.hpp"
 #include "phlex/core/glue.hpp"
 #include "phlex/core/registrar.hpp"
 #include "phlex/model/algorithm_name.hpp"
@@ -7,9 +8,13 @@
 
 #include <boost/json/object.hpp>
 #include <catch2/catch_test_macros.hpp>
+#include <catch2/matchers/catch_matchers.hpp>
+#include <oneapi/tbb/flow_graph.h>
 
+#include <memory>
 #include <stdexcept>
 #include <string>
+#include <utility>
 #include <vector>
 
 TEST_CASE("algorithm_name tests", "[model]")
@@ -95,3 +100,65 @@ TEST_CASE("add_to_error_messages tests", "[core]")
   REQUIRE(errors.size() == 1);
   CHECK(errors[0].contains("duplicate_node"));
 }
+
+TEST_CASE("A throwing registrar creator is not retried by the destructor", "[core]")
+{
+  using namespace phlex::detail;
+
+  declared_outputs outputs;
+  std::vector<std::string> errors;
+  int creations = 0;
+  {
+    registrar<declared_output_ptr> reg{outputs, errors};
+    reg.set_creator([&](auto const&, auto const&) -> declared_output_ptr {
+      ++creations;
+      throw std::runtime_error{"creator failed"};
+    });
+    CHECK_THROWS_WITH(reg.set_output_product_suffixes({}), "creator failed");
+  }
+  CHECK(creations == 1);
+  CHECK(errors.empty());
+}
+
+TEST_CASE("Moving a registrar transfers node creation", "[core]")
+{
+  using namespace phlex::detail;
+
+  tbb::flow::graph graph;
+  declared_outputs outputs;
+  std::vector<std::string> errors;
+  int creations = 0;
+  auto creator = [&](auto const&, auto const&) -> declared_output_ptr {
+    ++creations;
+    return std::make_unique<declared_output>(phlex::experimental::algorithm_name::create("output"),
+                                             1,
+                                             std::vector<std::string>{},
+                                             graph,
+                                             [](auto const&) {});
+  };
+
+  SECTION("Move construction")
+  {
+    {
+      registrar<declared_output_ptr> original{outputs, errors};
+      original.set_creator(creator);
+      auto moved = std::move(original);
+    }
+    CHECK(creations == 1);
+    CHECK(outputs.get("output") != nullptr);
+  }
+
+  SECTION("Move assignment")
+  {
+    {
+      registrar<declared_output_ptr> original{outputs, errors};
+      original.set_creator(creator);
+      registrar<declared_output_ptr> moved{outputs, errors};
+      moved = std::move(original);
+    }
+    CHECK(creations == 1);
+    CHECK(outputs.get("output") != nullptr);
+  }
+
+  CHECK(errors.empty());
+}

test/multilayer_join_test.cpp

@@ -1,5 +1,7 @@
+#include "phlex/core/fold_join_node.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/product_selector.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/fwd.hpp"
 #include "phlex/model/identifier.hpp"
@@ -10,7 +12,13 @@
 #include <gsl/pointers>
 #include <oneapi/tbb/flow_graph.h>
 
+#include <array>
+#include <cstddef>
+#include <limits>
+#include <memory>
+#include <stdexcept>
 #include <string>
+#include <tuple>
 #include <utility>
 #include <vector>
 
@@ -73,6 +81,8 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]")
   auto& left_port = receiver_for<0ull, 2>(join, 0u);
   auto& right_port = receiver_for<0ull, 2>(join, 1u);
 
+  CHECK_THROWS_AS(receiver_for<0ull>(join, 2u), std::runtime_error);
+
   REQUIRE(left_port.try_put({.store = left_store, .id = message_id}));
   graph.wait_for_all();
 
@@ -85,6 +95,10 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]")
 
   auto index_ports = join.index_ports();
   REQUIRE(index_ports.size() == 2u);
+  CHECK(index_ports[0].layer == "left_layer"_id);
+  CHECK(index_ports[1].layer == "right_layer"_id);
+  CHECK_FALSE(index_ports[0].counting_layer.has_value());
+  CHECK_FALSE(index_ports[1].counting_layer.has_value());
   REQUIRE(index_ports[0].index_port->try_put(
     {.index = left_store->index(), .msg_id = message_id, .cache = true}));
   graph.wait_for_all();
@@ -110,3 +124,75 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]")
   REQUIRE(index_ports[1].token_port->try_put({.index = right_store->index(), .count = 1}));
   graph.wait_for_all();
 }
+
+TEST_CASE("Same-layer joins wait for all inputs and preserve selector order", "[join]")
+{
+  tbb::flow::graph graph;
+  multilayer_join_node<3> join{graph, "same_layer_join", {"event"_id, "event"_id, "event"_id}};
+  CHECK(join.index_ports().empty());
+
+  auto const left_creator = algorithm_name::create("left");
+  auto const middle_creator = algorithm_name::create("middle");
+  auto const right_creator = algorithm_name::create("right");
+
+  product_selectors const selectors{{.creator = left_creator.to_string()},
+                                    {.creator = middle_creator.to_string()},
+                                    {.creator = right_creator.to_string()}};
+  auto const stage = "test"_id;
+  std::array const stores{product_store::base(&left_creator, &stage),
+                          product_store::base(&middle_creator, &stage),
+                          product_store::base(&right_creator, &stage)};
+  tbb::flow::queue_node<message_tuple<3>> sink{graph};
+
+  auto try_put_for_receiver = [&join, &selectors, &stores](std::size_t const index) {
+    return internal::receiver_for<3>(join, selectors, selectors[index])
+      .try_put({.store = stores.at(index), .id = message_id});
+  };
+
+  using internal::receiver_for;
+  make_edge(output_port<0>(join), sink);
+  REQUIRE(try_put_for_receiver(2));
+  REQUIRE(try_put_for_receiver(0));
+  graph.wait_for_all();
+  message_tuple<3> output;
+  CHECK_FALSE(sink.try_get(output));
+  REQUIRE(try_put_for_receiver(1));
+  graph.wait_for_all();
+  REQUIRE(sink.try_get(output));
+  CHECK(std::get<0>(output).id == message_id);
+  CHECK(std::get<1>(output).id == message_id);
+  CHECK(std::get<2>(output).id == message_id);
+  CHECK(std::get<0>(output).store == stores[0]);
+  CHECK(std::get<1>(output).store == stores[1]);
+  CHECK(std::get<2>(output).store == stores[2]);
+  CHECK_FALSE(sink.try_get(output));
+  CHECK_THROWS_AS(receiver_for<3>(join, selectors, {.creator = "missing"_id}), std::runtime_error);
+}
+
+TEST_CASE("Fold join data receivers exclude the partition port", "[join]")
+{
+  tbb::flow::graph graph;
+  auto const name = algorithm_name::create("fold_join_test");
+  fold_join_node<int, 2> join{graph,
+                              name,
+                              "test_stage"_id,
+                              "job"_id,
+                              {"event"_id, "run"_id},
+                              product_specifications(1),
+                              [](data_cell_index const&) { return std::make_unique<int>(0); }};
+  CHECK_THROWS_AS(receiver_for<1ull>(join, 2u), std::runtime_error);
+  CHECK_THROWS_AS(receiver_for<1ull>(join, std::numeric_limits<std::size_t>::max()),
+                  std::runtime_error);
+  product_selectors const selectors{{.creator = "left"_id}, {.creator = "right"_id}};
+  CHECK(&receiver_for(join, selectors, selectors[0]) == &input_port<1>(join));
+  CHECK(&receiver_for(join, selectors, selectors[1]) == &input_port<2>(join));
+
+  auto const ports = join.index_ports();
+  REQUIRE(ports.size() == 3);
+  CHECK(ports[0].layer == "job"_id);
+  CHECK(ports[0].counting_layer == "event"_id);
+  CHECK(ports[1].layer == "event"_id);
+  CHECK(ports[1].counting_layer == "event"_id);
+  CHECK(ports[2].layer == "run"_id);
+  CHECK(ports[2].counting_layer == "run"_id);
+}

test/resources_test.cpp

@@ -44,6 +44,16 @@ namespace {
 
   struct default_unlimited_resource {};
 
+  struct counted_unlimited_resource {
+    explicit counted_unlimited_resource(int& constructions) { ++constructions; }
+  };
+
+  struct counted_serialized_resource {
+    using token_type = counted_serialized_resource const*;
+
+    explicit counted_serialized_resource(int& constructions) { ++constructions; }
+  };
+
   struct invalid_serialized_resource {
     using token_type = int*;
   };
@@ -276,18 +286,28 @@ TEST_CASE("resource catalog", "[resource]")
 
   SECTION("duplicate serialized registrations throw")
   {
-    catalog.add_serialized<catch2_resource>();
-    CHECK_THROWS_WITH(catalog.add_serialized<catch2_resource>(),
+    int constructions = 0;
+    catalog.add_serialized<counted_serialized_resource>(constructions);
+    auto& limiter = catalog.limiter_for<counted_serialized_resource>();
+    CHECK(constructions == 1);
+    CHECK_THROWS_WITH(catalog.add_serialized<counted_serialized_resource>(constructions),
                       Catch::Matchers::ContainsSubstring("Resource of type '") &&
                         Catch::Matchers::ContainsSubstring("' has already been registered"));
+    CHECK(constructions == 1);
+    CHECK(&catalog.limiter_for<counted_serialized_resource>() == &limiter);
   }
 
   SECTION("duplicate unlimited registrations throw")
   {
-    catalog.add_unlimited<unlimited_resource_type>(42);
-    CHECK_THROWS_WITH(catalog.add_unlimited<unlimited_resource_type>(43),
+    int constructions = 0;
+    catalog.add_unlimited<counted_unlimited_resource>(constructions);
+    auto const resource = catalog.access_for<counted_unlimited_resource>();
+    CHECK(constructions == 1);
+    CHECK_THROWS_WITH(catalog.add_unlimited<counted_unlimited_resource>(constructions),
                       Catch::Matchers::ContainsSubstring("Resource of type '") &&
                         Catch::Matchers::ContainsSubstring("' has already been registered"));
+    CHECK(constructions == 1);
+    CHECK(catalog.access_for<counted_unlimited_resource>() == resource);
   }
 
   SECTION("missing resources throw")