include/kpn/network.hpp

@@ -41,8 +41,15 @@ public:
 
 class Network : public INode {
 public:
-    using ErrorHandler =
-        std::function<void(std::string_view node_name, std::exception_ptr)>;
+    /// Application-level error listener. Receives the exception any node's
+    /// function throws, after that node's own handler (if any) declined it.
+    /// Return true to skip the failed invocation and keep the node running,
+    /// false to let it stop.
+    ///
+    /// Same type as StaticNetwork's, deliberately: this used to be a void
+    /// signature, which could not express the keep-running decision and, more
+    /// to the point, was never delivered anywhere.
+    using ErrorHandler = NodeErrorHandler;
     using DiagnosticsHandler =
         std::function<void(const std::vector<NodeSnapshot>&,
                            const std::vector<ChannelSnapshot>&)>;
@@ -137,6 +144,14 @@ public:
 
     void start() override {
         start_time_ = clock_t::now();
+        // Deliver the listener to the nodes. Without this the handler was
+        // stored and never read: a node's exception was discarded at the node
+        // boundary and the only surviving evidence was a Closed event, which
+        // says a node stopped but not why. StaticNetwork has always done this;
+        // Network accepted the handler and silently dropped it.
+        if (error_handler_)
+            for (auto& name : topo_)
+                nodes_.at(name)->set_network_error_callback(error_handler_);
         // Callbacks first, everywhere, before anything runs — see INode::prepare.
         for (auto& name : topo_)
             nodes_.at(name)->prepare();
@@ -220,6 +235,7 @@ public:
     /// up on them and stopping the next layer anyway.
     void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
 
+    /// Must be called before start(); the handler is delivered to nodes there.
     void set_error_handler(ErrorHandler h)             { error_handler_  = std::move(h); }
     void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_  = std::move(h); }
     void set_event_handler(EventHandler h)             { event_handler_  = std::move(h); }

tests/test_network.cpp

@@ -59,6 +59,63 @@ TEST_CASE("stop disables input channels — producer push is silently dropped",
     REQUIRE(in_ch.size() == 0);
 }
 
+// Regression: Network::set_error_handler must actually deliver the handler.
+//
+// The handler was stored in a member and never read. A node's exception was
+// discarded at the node boundary and the only surviving evidence was a Closed
+// event, which reports that a node stopped but not why — the difference between
+// a diagnosis and a guess. StaticNetwork has always wired this; Network
+// accepted the handler and silently dropped it, which is worse than not
+// offering the setter at all.
+//
+// The type changed with the fix. It was void(name, exception_ptr), which cannot
+// express the keep-running decision the node side needs, so it is now
+// NodeErrorHandler like StaticNetwork's.
+namespace {
+
+static int throwing_stage(int x) {
+    if (x == 42) throw std::runtime_error("boom");
+    return x;
+}
+
+}  // namespace
+
+TEST_CASE("network error handler receives the node's exception", "[network]") {
+    auto src  = kpn::make_node<throwing_stage>(kpn::in<"v">{}, kpn::out<"w">{}, 8);
+    kpn::Channel<int> out(8);
+    src.set_output_channel<0>(&out);
+
+    kpn::Network net;
+    net.add("stage", src).build();
+
+    std::atomic<int>  calls{0};
+    std::string       seen_name;
+    std::string       seen_what;
+    std::mutex        mx;
+
+    net.set_error_handler([&](std::string_view name, std::exception_ptr ep) {
+        std::lock_guard lk(mx);
+        seen_name = std::string(name);
+        try { if (ep) std::rethrow_exception(ep); }
+        catch (const std::exception& e) { seen_what = e.what(); }
+        calls.fetch_add(1, std::memory_order_relaxed);
+        return true;                       // handled: keep the node running
+    });
+
+    net.set_watchdog_interval(std::chrono::hours(1));   // keep the report quiet
+    net.start();
+    src.input_channel<0>().push(42);                    // throws
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    src.input_channel<0>().push(7);                     // must still be running
+    const int passed = out.pop();
+    net.stop();
+
+    CHECK(calls.load(std::memory_order_relaxed) == 1);
+    CHECK(seen_name == "stage");
+    CHECK(seen_what == "boom");
+    CHECK(passed == 7);
+}
+
 // Regression: stopping a network must not wait for the watchdog's next tick.
 //
 // The watchdog looped on std::this_thread::sleep_for(watchdog_interval_), and