fix: the drain loops must terminate, and must drain the right channels

shutdown()'s drain step was an unbounded

    while (anything, anywhere, is non-empty) poll every channel

Three defects in one loop.

It drained the wrong thing. Stopping a node should wait for that node's own
outputs before moving to the next layer; this waited for the entire graph to
fall idle each time. The dynamic Network's version made it explicit — it took
a node name and ignored it. Both now track which node feeds each probe and
wait only on those.

It had no deadline, so anything wedged downstream turned a graceful shutdown
into the hang it exists to avoid. Now bounded two ways, because a stalled
consumer and a slow one fail differently: a deadline for fill that never
changes, and a no-progress counter that keeps waiting as long as the queue is
shrinking, so a slow drain is not cut short merely for taking a while.

And it could fail to terminate with nothing wedged at all. current_fill came
from a snapshot that loaded tail_ before head_. A concurrent pop between the
two reads yields a head_ past the sampled tail_, and the unsigned difference
wraps to ~2^64 — so a poll for "is it empty yet" runs forever on a channel
that is in fact empty. Both indices only ever increase, so loading head_
first can at worst under-report a concurrent push, which this loop tolerates
and a wrap does not. size() and snapshot() are both corrected; size() feeds
approx_size(), which is what node readiness checks call.

Giving up is now reported rather than silent, because undrained data at that
point is about to be discarded by the stop that follows, and a graceful
shutdown quietly dropping values is the thing worth knowing about.

Verified in both directions: with the old loop the new case is killed at a
30 s timeout; with this it returns in under a second, having reported four
items its wedged consumer never took. Full suite 138/138.

Note the drain timeout is per node and defaults to 5 s, so a graph of N
stalled nodes can still take N x 5 s to shut down. That is a deliberate
trade against cutting off legitimate slow drains, and set_drain_timeout()
exists for callers who want it tighter.

include/kpn/channel.hpp

@@ -382,9 +382,15 @@ public:
 
     // Ring occupancy, derived lazily from indices — no separate counter on the
     // hot path. Excludes any out-of-band sentinel (that lives outside the ring).
+    // head_ is loaded first, deliberately. Both indices only ever increase, so
+    // reading head_ before tail_ can at worst under-report a concurrent push;
+    // the other order can read a head_ that has advanced past the tail_ already
+    // sampled, and the unsigned difference then wraps to ~2^64. A caller
+    // polling "is this channel empty yet" against that value never terminates.
     std::size_t size() const {
-        return tail_.load(std::memory_order_relaxed)
+        const std::size_t h = head_.load(std::memory_order_relaxed);
-             - head_.load(std::memory_order_relaxed);
+        const std::size_t t = tail_.load(std::memory_order_acquire);
+        return t - h;
     }
 
     // A pending out-of-band sentinel (EOF) counts as consumable work here even
@@ -401,8 +407,9 @@ public:
     const ChannelStats& stats() const { return stats_; }
 
     ChannelSnapshot snapshot(const std::string& name) const {
-        const std::size_t t = tail_.load(std::memory_order_relaxed);
+        // head_ before tail_, for the reason given on size().
         const std::size_t h = head_.load(std::memory_order_relaxed);
+        const std::size_t t = tail_.load(std::memory_order_acquire);
         return {
             name,
             capacity_,

include/kpn/network.hpp

@@ -91,6 +91,7 @@ public:
                             + " → " + dst_name + ":" + std::to_string(DstIdx);
         channel_probes_.push_back(
             std::make_unique<ChannelProbe<out_t>>(in_ch, ch_name));
+        channel_src_names_.push_back(src_name);
 
         adj_[src_name].push_back(dst_name);
         return *this;
@@ -213,6 +214,10 @@ public:
         watchdog_interval_ = interval;
     }
 
+    /// How long shutdown() waits for one node's outputs to drain before giving
+    /// up on them and stopping the next layer anyway.
+    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
+
     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); }
@@ -370,19 +375,46 @@ private:
         return true;
     }
 
-    void drain_output_channels(const std::string& /*name*/) const {
+    /// Wait for the channels fed by `name` to empty, or give up.
-        // Poll all channel probes until none report non-zero fill.
+    ///
-        // A short sleep prevents busy-spin; 1 ms is fine for drain purposes.
+    /// This took a node name and ignored it, polling *every* channel in the
-        bool any_full = true;
+    /// graph instead — so shutdown() waited for the whole network to be idle
-        while (any_full) {
+    /// before stopping each successive layer. With no deadline either, anything
-            any_full = false;
+    /// wedged downstream turned a graceful shutdown into the hang it exists to
-            for (auto& probe : channel_probes_) {
+    /// avoid.
-                auto snap = probe->snapshot();
+    ///
-                if (snap.current_fill > 0) { any_full = true; break; }
+    /// Two bounds, because they fail differently. The deadline covers a
-            }
+    /// consumer that has stopped consuming, where fill never changes and
-            if (any_full)
+    /// waiting cannot help. The no-progress counter covers one that is merely
-                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    /// slow: it keeps waiting while the queue is shrinking, so a slow drain is
+    /// not cut short just for taking a while.
+    void drain_output_channels(const std::string& name) const {
+        const auto deadline = clock_t::now() + drain_timeout_;
+        std::size_t last_fill = static_cast<std::size_t>(-1);
+        int         stalls    = 0;
+
+        auto fill_of = [&] {
+            std::size_t fill = 0;
+            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+                if (channel_src_names_[i] == name)
+                    fill += channel_probes_[i]->snapshot().current_fill;
+            return fill;
+        };
+
+        for (;;) {
+            const std::size_t fill = fill_of();
+            if (fill == 0) return;
+            if (fill >= last_fill) { if (++stalls > 100) break; }
+            else                   { stalls = 0; }
+            last_fill = fill;
+            if (clock_t::now() >= deadline) break;
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
+
+        if (const std::size_t left = fill_of())
+            std::cerr << "[kpn] shutdown: '" << name << "' still has " << left
+                      << " queued item(s) its consumer did not take; "
+                         "they are discarded\n";
     }
 
     // ── Cycle detection / topological sort ───────────────────────────────────
@@ -447,6 +479,10 @@ private:
     std::map<std::string, std::string>                     exposed_outputs_;
     std::set<std::pair<std::string, std::size_t>>          connected_outputs_;
     std::vector<std::unique_ptr<IChannelProbe>>            channel_probes_;
+    /// Name of the node feeding each probe, parallel to channel_probes_.
+    /// shutdown() drains a node's own outputs, so it has to know which they are.
+    std::vector<std::string>                               channel_src_names_;
+    std::chrono::milliseconds                              drain_timeout_{5000};
     std::vector<std::pair<std::string, IPoolProbe*>>       pool_probes_;
     ErrorHandler                                           error_handler_;
     DiagnosticsHandler                                     diag_handler_;

include/kpn/static_network.hpp

@@ -98,13 +98,15 @@ public:
                   std::vector<INode*>        fanout_ptrs,
                   std::vector<std::string>   user_node_names,
                   std::vector<std::string>   fanout_node_names,
-                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes)
+                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes,
+                  std::vector<std::string>   channel_src_names)
         : fanouts_(std::move(fanouts))
         , user_nodes_topo_(std::move(user_nodes_topo))
         , fanout_nodes_ptr_(std::move(fanout_ptrs))
         , user_node_names_(std::move(user_node_names))
         , fanout_node_names_(std::move(fanout_node_names))
         , channel_probes_(std::move(channel_probes))
+        , channel_src_names_(std::move(channel_src_names))
     {}
 
     ~StaticNetwork() override { stop(); }
@@ -173,9 +175,9 @@ public:
 #endif
         // user_nodes_topo_ is already in sources-first order.
         // Stop each node and drain its output channels before moving on.
-        for (auto* n : user_nodes_topo_) {
+        for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
-            n->stop();
+            user_nodes_topo_[i]->stop();
-            drain_all_channels();
+            drain_outputs_of(user_node_names_[i]);
         }
         for (auto* n : fanout_nodes_ptr_) n->stop();
     }
@@ -194,6 +196,10 @@ public:
 
     void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
 
+    /// How long shutdown() waits for one node's outputs to drain before giving
+    /// up on them and stopping the next layer anyway.
+    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
+
     /// 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,
@@ -274,16 +280,50 @@ private:
         return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s};
     }
 
-    void drain_all_channels() const {
+    /// Wait for the channels fed by `src` to empty, or give up.
-        bool any_full = true;
+    ///
-        while (any_full) {
+    /// This was an unbounded `while (anything anywhere is non-empty)` poll over
-            any_full = false;
+    /// *every* channel in the graph, which made shutdown() wait for the whole
-            for (auto& probe : channel_probes_) {
+    /// network to be idle before stopping each successive layer, and wait
-                if (probe->snapshot().current_fill > 0) { any_full = true; break; }
+    /// forever if anything downstream was wedged — turning a graceful shutdown
-            }
+    /// into the hang it exists to avoid.
-            if (any_full)
+    ///
-                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    /// Two bounds, because they fail differently. The deadline covers a
+    /// consumer that has stopped consuming: fill never changes and no amount of
+    /// waiting helps. The no-progress counter covers a consumer that is merely
+    /// slow — it keeps waiting as long as the queue is shrinking, so a slow
+    /// drain is not cut short just for exceeding a fixed time.
+    ///
+    /// Giving up is reported rather than silent: undrained data at this point
+    /// means values are about to be discarded by the stop that follows.
+    void drain_outputs_of(const std::string& src) const {
+        const auto deadline = clock_t::now() + drain_timeout_;
+        std::size_t last_fill = static_cast<std::size_t>(-1);
+        int         stalls    = 0;
+
+        for (;;) {
+            std::size_t fill = 0;
+            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+                if (channel_src_names_[i] == src)
+                    fill += channel_probes_[i]->snapshot().current_fill;
+
+            if (fill == 0) return;
+            if (fill >= last_fill) { if (++stalls > 100) break; }
+            else                   { stalls = 0; }
+            last_fill = fill;
+
+            if (clock_t::now() >= deadline) break;
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
+
+        std::size_t left = 0;
+        for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+            if (channel_src_names_[i] == src)
+                left += channel_probes_[i]->snapshot().current_fill;
+        if (left)
+            std::cerr << "[kpn] shutdown: '" << src << "' still has " << left
+                      << " queued item(s) its consumer did not take; "
+                         "they are discarded\n";
     }
 
     std::string                                         name_;
@@ -294,6 +334,10 @@ private:
     std::vector<std::string>                            user_node_names_;
     std::vector<std::string>                            fanout_node_names_;
     std::vector<std::unique_ptr<IChannelProbe>>         channel_probes_;
+    /// Display name of the node feeding each probe, parallel to channel_probes_.
+    /// shutdown() drains a node's own outputs, so it has to know which they are.
+    std::vector<std::string>                            channel_src_names_;
+    std::chrono::milliseconds                           drain_timeout_{5000};
     std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
     std::vector<std::pair<std::string, IPoolProbe*>>     pool_probes_;
     EventHandler                                        event_handler_;
@@ -401,6 +445,7 @@ auto make_network(Edges&&... edges) {
     };
 
     std::vector<std::unique_ptr<IChannelProbe>> channel_probes;
+    std::vector<std::string>                    channel_src_names;
 
     auto wire_one = [&]<typename SE>(SE) {
         using SrcNode = typename SE::src_node_t;
@@ -418,6 +463,7 @@ auto make_network(Edges&&... edges) {
                                 + " \xe2\x86\x92 "  // UTF-8 →
                                 + node_name.template operator()<DstNode>() + ":" + std::to_string(DstIdx);
             channel_probes.push_back(std::make_unique<ChannelProbe<out_t>>(ch, ch_name));
+            channel_src_names.push_back(node_name.template operator()<SrcNode>());
         }
     };
 
@@ -445,7 +491,8 @@ auto make_network(Edges&&... edges) {
                std::move(fanout_ptrs),
                std::move(user_node_names),
                std::move(fanout_node_names),
-               std::move(channel_probes));
+               std::move(channel_probes),
+               std::move(channel_src_names));
 }
 
 } // namespace kpn

tests/test_static_network.cpp

@@ -271,3 +271,81 @@ TEST_CASE("static_network: fanout with labelled same-function consumers", "[stat
     REQUIRE(outB.pop() == 7);
     net.stop();
 }
+
+// Regression: shutdown() must return even when a consumer stopped consuming.
+//
+// The drain step was an unbounded `while (anything anywhere is non-empty)` poll
+// over *every* channel in the graph. Two defects in one loop: it waited for the
+// whole network to be idle before stopping each successive layer rather than
+// just the node it had stopped — the dynamic Network's version even took a node
+// name and ignored it — and it had no deadline, so anything wedged downstream
+// turned a graceful shutdown into the hang it exists to avoid.
+//
+// It could also fail to terminate with nothing wedged at all. current_fill came
+// from a snapshot that loaded tail_ before head_; a concurrent pop between the
+// two reads yields a head_ past the sampled tail_, and the unsigned difference
+// wraps to ~2^64. Any poll for "is it empty yet" against that value runs
+// forever. Both indices only ever increase, so loading head_ first can at worst
+// under-report a push, which this loop tolerates and a wrap does not.
+//
+// Here the sink never takes anything, so its input cannot drain and the only
+// correct outcome is to give up and say so. The bound asserted is deliberately
+// loose: the point is that it terminates, not how fast.
+namespace {
+
+struct DrainSource {
+    static constexpr std::string_view label() { return "drain_source"; }
+    int n{0};
+    int operator()() {
+        std::this_thread::sleep_for(std::chrono::microseconds(100));
+        return n++;
+    }
+};
+
+struct NeverConsumes {
+    static constexpr std::string_view label() { return "never_consumes"; }
+    std::atomic<bool>* wedged;
+    void operator()(int) {
+        // Blocks for the duration of the test: the input channel behind it
+        // fills and stays full.
+        while (!wedged->load(std::memory_order_acquire))
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+};
+
+}  // namespace
+
+TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutdown]") {
+    std::atomic<bool> release{false};
+
+    DrainSource   src_fn;
+    NeverConsumes sink_fn{&release};
+
+    kpn::ObjectNode<DrainSource,   kpn::in<>,    kpn::out<"v">, "drain_source",   0> s(src_fn,  4);
+    kpn::ObjectNode<NeverConsumes, kpn::in<"v">, kpn::out<>,    "never_consumes", 0> k(sink_fn, 4);
+
+    auto net = kpn::make_network(kpn::edge(s.output<"v">(), k.input<"v">()));
+    net.set_drain_timeout(std::chrono::milliseconds(100));
+    net.start();
+
+    // Let the channel fill and the sink jam.
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    // Unjam the sink well after the drain timeout should have expired. Stopping
+    // a node joins its worker, so a sink blocked forever would hang the test in
+    // stop() rather than in the drain loop this case is about.
+    std::thread unjam([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(800));
+        release.store(true, std::memory_order_release);
+    });
+
+    const auto t0 = std::chrono::steady_clock::now();
+    net.shutdown();
+    const auto elapsed = std::chrono::steady_clock::now() - t0;
+
+    unjam.join();
+
+    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
+    INFO("shutdown took " << ms << " ms");
+    CHECK(ms < 3000);   // unbounded before; one 100 ms drain timeout after
+}