fix: never drop a wake — a node must not sleep with one outstanding

28e0667 stopped nodes blocking a worker on a full output, but replaced an
intermittent hang with a quieter one: the pipeline still wedged about 2
runs in 30, now with every worker idle in pthread_cond_wait rather than
asleep in a push. Nothing was blocked; nothing had been woken.

try_submit discarded any wake arriving while queued_ was up:

    if (queued_.compare_exchange_strong(expected, true, ...))
        scheduler_->submit(...);
    // else: silently gone

Wakes are edge-triggered — a channel fires its space callback once, on the
transition — so a dropped one never returns. A node could park a value,
release its worker, and sleep forever holding exactly the output its
consumer was waiting for, while its producer parked on an input channel
that would never drain.

try_submit now records the drop in wake_pending_, and release_and_recheck()
consumes it at every site that releases a node, giving one invariant: a
node never sleeps with a wake outstanding. This subsumes the two ad-hoc
re-checks added for the parked-retry and normal push paths, which only
moved the stall (1211 items to 6472) because each new early return was a
fresh chance to drop a wake. self_stop keeps a plain store — honouring a
pending wake there would resubmit a dead node.

Adds "a saturated chain never stalls", which reproduces this in about a
second where the pipeline needed ~30 runs. It asserts *progress does not
freeze* rather than a completion total: capacity-1 channels are slow, and
slow must never be reported as wedged. Verified in both directions — it
stalls after 1211 items on 28e0667 and passes here.

The existing chain test could not catch it: 40 items drain before any
strand occurs, and its producer emits forever, so fresh input keeps
re-triggering on_input_ready() and flushing the stranded value.

Tests: 122/122.

include/kpn/channel.hpp

@@ -136,6 +136,42 @@ public:
             push_callback_();
     }
 
+    /// Called when a pop frees a slot in a previously-full ring.
+    ///
+    /// The mirror of `set_push_callback`, and it exists for the same reason:
+    /// a producer must be able to *park* rather than spin. Without it the only
+    /// lossless option is `push_blocking`, which sleeps inside the caller's
+    /// thread — and when that thread is a scheduler worker, parking it starves
+    /// every node pinned to it (see the hold-and-wait note on push_sentinel).
+    void set_space_callback(std::function<void()> cb) { space_callback_ = std::move(cb); }
+
+    /// True when a push would currently succeed. Used to close the lost-wakeup
+    /// race: a producer that parks must re-check after clearing its queued flag,
+    /// because a space_callback fired in between would otherwise be swallowed.
+    bool has_space() const {
+        return tail_.load(std::memory_order_relaxed) -
+               head_.load(std::memory_order_acquire) < capacity_;
+    }
+
+    /// Non-blocking, lossless push. Returns false when the ring is full, having
+    /// changed nothing — the caller keeps the value and retries when woken.
+    bool try_push(T& value) {
+        if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); return true; }
+        const std::size_t t = tail_.load(std::memory_order_relaxed);
+        const std::size_t h = head_.load(std::memory_order_acquire);
+        if (t - h >= capacity_) return false;
+
+        const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
+        const bool was_empty = (t == h);
+        buf_[t & ring_mask_] = make_storage(std::move(value));
+        tail_.store(t + 1, std::memory_order_release);
+        stats_.record_push(t - h + 1, data_bytes);
+        wake_.fetch_add(1, std::memory_order_release);
+        wake_.notify_one();
+        if (was_empty && push_callback_) push_callback_();
+        return true;
+    }
+
     // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to
     // drain instead of dropping (the throwing push()) — the producer just runs slower.
     // Use when every value must be delivered (e.g. replaying a dump for scoring, where
@@ -248,6 +284,8 @@ public:
                 throw ChannelClosedError{};
             T value = extract(std::move(buf_[h & ring_mask_]));
             head_.store(h + 1, std::memory_order_release);
+            // A slot just freed: wake any producer parked on this channel.
+            if (t - h >= capacity_ && space_callback_) space_callback_();
             stats_.record_pop();
             return value;
         }
@@ -269,11 +307,15 @@ public:
     // so pool nodes — which pop only via this path — still receive the token.
     bool try_pop_now(T& out) {
         const std::size_t h = head_.load(std::memory_order_relaxed);
-        if (h == tail_.load(std::memory_order_acquire))
+        const std::size_t t = tail_.load(std::memory_order_acquire);
+        if (h == t)
             return take_sentinel(out);
         out = extract(std::move(buf_[h & ring_mask_]));
         head_.store(h + 1, std::memory_order_release);
         stats_.record_pop();
+        // Pool nodes pop only through here, so this is where a parked producer
+        // gets woken: the ring was full, and it no longer is.
+        if (t - h >= capacity_ && space_callback_) space_callback_();
         return true;
     }
 
@@ -363,6 +405,7 @@ private:
     std::size_t                     ring_mask_;
     std::unique_ptr<storage_type[]> buf_;
     std::function<void()>           push_callback_;
+    std::function<void()> space_callback_;
     ChannelStats                    stats_;
 
     // Out-of-band sentinel (EOF): stored outside the ring so its delivery never

include/kpn/inode.hpp

@@ -34,6 +34,14 @@ struct INode {
     virtual void set_network_overflow_callback(NodeEventCallback) {}
     virtual void set_network_closed_callback(NodeEventCallback)   {}
 
+    // Network-level error listener. Consulted when a node's function throws
+    // and no per-node handler resolved it. Without this the exception is
+    // discarded and the failure is only visible as a Closed event, which says
+    // a node stopped but not why — the difference between a diagnosis and a
+    // guess. Same contract as NodeErrorHandler: true to continue, false to
+    // stop the node.
+    virtual void set_network_error_callback(NodeErrorHandler) {}
+
     // halt(): alias for stop() — immediate, discards in-flight work.
     virtual void halt() { stop(); }
 

include/kpn/pool_node.hpp

@@ -15,6 +15,7 @@
 #include <iostream>
 #include <memory>
 #include <optional>
+#include <variant>
 #include <stdexcept>
 #include <thread>
 #include <tuple>
@@ -131,6 +132,7 @@ public:
 
     void set_name(std::string name) override { name_ = std::move(name); }
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
+    void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
@@ -234,6 +236,8 @@ private:
 
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
+        // A parked producer is re-submitted when its output drains.
+        register_space_callbacks(std::make_index_sequence<output_count>{});
         (std::get<Is>(input_channels_)->set_push_callback(
             [this] { on_input_ready(); }), ...);
     }
@@ -243,10 +247,26 @@ private:
         for (auto& cb : cbs) if (cb) cb(ts);
     }
 
+    template<std::size_t... Os>
+    bool outputs_have_space(std::index_sequence<Os...>) const {
+        return (... && (!std::get<Os>(output_channels_) ||
+                        std::get<Os>(output_channels_)->has_space()));
+    }
+
+    template<std::size_t... Os>
+    void register_space_callbacks(std::index_sequence<Os...>) {
+        ((std::get<Os>(output_channels_)
+            ? (void)std::get<Os>(output_channels_)->set_space_callback(
+                  [this] { try_submit(0.5f); })
+            : (void)0), ...);
+    }
+
     void self_stop() {
         disable_inputs(std::make_index_sequence<input_count>{});
         disable_outputs(std::make_index_sequence<output_count>{});
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
+        // Plain store, not release_and_recheck(): this node is stopping, and
+        // honouring a pending wake here would resubmit a dead node.
         queued_.store(false, std::memory_order_release);
         stop_flag_.store(true, std::memory_order_relaxed);
     }
@@ -280,11 +300,51 @@ private:
         return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
     }
 
+    /// Priority in [0,1], higher runs sooner.
+    ///
+    /// Input fill alone answers "how much work is waiting for me". That is
+    /// only half the question: a node whose *outputs* are already full cannot
+    /// deliver anything: running it produces a value with nowhere to go, so it
+    /// immediately parks and the slot is wasted. Meanwhile the node that would
+    /// have drained that full channel waits behind it.
+    ///
+    /// So occupancy of the outputs is deducted from occupancy of the inputs.
+    /// The scheduler then naturally favours whoever is furthest downstream of
+    /// a bottleneck — the node whose inputs are backed up but whose outputs
+    /// have room is exactly the one whose execution frees the most capacity —
+    /// and defers producers that would only deepen a queue that is already
+    /// full.
+    ///
+    /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
+    /// both terms are mean fills in [0,1], so the difference is in [-1,1], and
+    /// the affine map keeps the whole range distinguishable instead of
+    /// collapsing every output-saturated node onto the same value. 0.5 remains
+    /// the neutral point, matching the default used for source nodes.
     float compute_priority() {
         if constexpr (input_count == 0) return 0.5f;
-        float sum = 0.0f;
+        float in = 0.0f;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
+        sum_fill(in, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        in /= static_cast<float>(input_count);
+
+        if constexpr (output_count == 0) return in;
+
+        float out = 0.0f;
+        sum_output_fill(out, std::make_index_sequence<output_count>{});
+        out /= static_cast<float>(output_count);
+
+        const float p = 0.5f * (1.0f + in - out);
+        return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
+    }
+
+    /// Mean fill of the output channels, same normalisation as sum_fill.
+    /// An unconnected output holds nothing back, so it contributes 0.
+    template<std::size_t... Os>
+    void sum_output_fill(float& sum, std::index_sequence<Os...>) {
+        ((sum += (std::get<Os>(output_channels_) &&
+                  std::get<Os>(output_channels_)->capacity() > 0)
+            ? float(std::get<Os>(output_channels_)->approx_size())
+              / float(std::get<Os>(output_channels_)->capacity())
+            : 0.0f), ...);
     }
 
     template<std::size_t... Is>
@@ -295,10 +355,32 @@ private:
             : 0.5f), ...);
     }
 
+    /// Submit unless already queued. A wake that arrives while this node is
+    /// queued or running is *recorded*, never dropped.
+    ///
+    /// Wakes are edge-triggered: a channel fires its space callback on the
+    /// transition, once. If that lands while queued_ is up, the CAS below fails
+    /// and — before wake_pending_ — the wake was gone. A node could then park a
+    /// value, release its worker, and sleep forever holding output its consumer
+    /// was waiting for, with every worker idle in cond_wait and nothing left to
+    /// re-trigger it. Recording the drop turns the signal level-triggered: the
+    /// invariant is that a node never sleeps with a wake outstanding, enforced
+    /// by release_and_recheck() at every point that releases the node.
     void try_submit(float priority) {
         bool expected = false;
         if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
             scheduler_->submit([this] { fire_once(); }, priority);
+        else
+            wake_pending_.store(true, std::memory_order_release);
+    }
+
+    /// Clear queued_, then honour any wake that was dropped while it was up.
+    /// Every path that finishes or parks a firing must release the node through
+    /// here rather than storing queued_ directly.
+    void release_and_recheck(float priority = 0.5f) {
+        queued_.store(false, std::memory_order_release);
+        if (wake_pending_.exchange(false, std::memory_order_acq_rel))
+            try_submit(priority);
     }
 
     // ── Execution ─────────────────────────────────────────────────────────────
@@ -315,6 +397,77 @@ private:
             t0.time_since_epoch()).count();
         stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
 
+        // Parked from a previous firing: retry that value before touching the
+        // inputs. Returning here releases the worker — the channel's space
+        // callback re-submits this node when the consumer drains a slot.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
+                release_and_recheck();
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                return;
+            }
+        }
+
+        // Parked from a previous firing: retry that value before touching the
+        // inputs. Returning here releases the worker — the channel's space
+        // callback re-submits this node once the consumer drains a slot.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
+                release_and_recheck();
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                return;
+            }
+        }
+
+        // Woken by output space rather than by input arrival, with nothing
+        // parked left to flush: there is no work to do. Falling through would
+        // read an empty channel, and pop_one reports empty as
+        // ChannelClosedError — self-stopping a live node. Release the worker;
+        // on_input_ready() resubmits when data actually lands.
+        if constexpr (input_count > 0) {
+            if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
+                release_and_recheck();
+                on_input_ready();   // data may have arrived while we checked
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1 = clock_t::now();
@@ -340,7 +493,11 @@ private:
         } catch (const ChannelOverflowError&) {
             fire_callbacks(event_callbacks_);
         } catch (...) {
-            if (error_handler_ && error_handler_(name_, std::current_exception())) {
+            auto eptr = std::current_exception();
+            const bool handled =
+                (error_handler_     && error_handler_(name_, eptr)) ||
+                (net_error_handler_ && net_error_handler_(name_, eptr));
+            if (handled) {
                 // continue — fall through to resubmit check
             } else {
                 fire_callbacks(closed_callbacks_);
@@ -350,10 +507,27 @@ private:
         }
 
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        release_and_recheck();
 
         if (stop_flag_.load(std::memory_order_relaxed)) return;
 
+        // Parked by the push above. Same situation as the retry path at the top
+        // of fire_once — and the same lost-wakeup race, which that path closes
+        // and this one did not. A space callback that fired while queued_ was
+        // still up got swallowed by try_submit's CAS, and the resubmit below
+        // cannot cover it: this firing consumed its input, so inputs are empty
+        // and on_input_ready() will not resubmit. The node would then hold its
+        // value forever while its consumer waits for exactly that value and its
+        // producer parks on an input channel that never drains. Re-check now
+        // that the flag is down.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                    try_submit(0.5f);
+                return;                    // parked
+            }
+        }
+
         // Source nodes always resubmit; others resubmit only if inputs are ready.
         if constexpr (input_count == 0) {
             try_submit(0.5f);
@@ -385,27 +559,41 @@ private:
     }
 
     template<std::size_t... Is>
+    /// Pushes what it can and parks the rest. `done_` marks the elements that
+    /// were taken, so a retry never re-pushes one — a duplicate would be as
+    /// wrong as a drop, just harder to notice.
     void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
-        (push_one_out<Is>(std::get<Is>(std::move(result))), ...);
+        bool all = true;
+        ((pending_done_[Is] = pending_done_[Is] ||
+            push_one_out<Is>(std::get<Is>(std::move(result))),
+          all = all && pending_done_[Is]), ...);
+        if (all) { pending_.reset(); pending_done_.fill(false); }
+        else       pending_ = std::move(result);
     }
 
+    /// Returns false when the ring was full and the value was NOT taken; the
+    /// caller must keep it and retry after the channel signals space.
     template<std::size_t I>
-    void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
+    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
-        if (!ch) return;
+        if (!ch) return true;
         // Sentinels (EOF) must never be dropped: a lost token wedges every
         // downstream pop() forever. Deliver them out-of-band (push_sentinel),
         // which never overflows and never blocks this node's worker thread.
         if (is_sentinel_value(val)) {
             ch->push_sentinel(std::move(val));
-            return;
+            return true;
-        }
-        try {
-            ch->push(std::move(val));
-        } catch (const ChannelOverflowError&) {
-            throw ChannelOverflowError(ch->capacity(),
-                "pool node '" + name_ + "' " + output_port_label<I>());
         }
+        // Backpressure without parking the worker. A full channel means the
+        // consumer is behind; the value is kept and this node stops running
+        // until the channel signals space (set_space_callback re-submits it).
+        //
+        // Blocking here instead would sleep inside a scheduler worker, and
+        // nodes are pinned to workers — park enough of them and nothing is left
+        // to run the consumer that would drain the channel. That is the
+        // hold-and-wait deadlock channel.hpp warns about for sentinels; it
+        // applies to data pushes just as much.
+        return ch->try_push(val);
     }
 
     template<std::size_t I>
@@ -427,8 +615,20 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{false};
+    /// A wake that arrived while queued_ was up. See try_submit.
+    std::atomic<bool>                        wake_pending_{false};
+
+    /// The hidden one-slot output buffer (see push_outputs). Holding the value
+    /// here is what lets a node stop running without dropping it or occupying a
+    /// scheduler worker. One slot suffices because at most one fire_once() runs
+    /// per node at a time — with concurrent firing it would have to hold a whole
+    /// FIFO's worth.
+    std::conditional_t<std::is_void_v<return_raw>, std::monostate,
+                       std::optional<return_tuple>> pending_{};
+    std::array<bool, (output_count ? output_count : 1)> pending_done_{};
     NodeStats                                stats_;
     NodeErrorHandler                         error_handler_;
+    NodeErrorHandler                         net_error_handler_;
     std::chrono::milliseconds                max_exec_time_{0};
     std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
     std::array<NodeEventCallback, 2>         closed_callbacks_{};
@@ -496,6 +696,7 @@ public:
     bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); }
     void set_name(std::string name) override { name_ = std::move(name); }
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
+    void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
@@ -567,6 +768,8 @@ private:
     }
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
+         // A parked producer is re-submitted when its output drains.
+         register_space_callbacks(std::make_index_sequence<output_count>{});
         (std::get<Is>(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...);
     }
 
@@ -575,10 +778,26 @@ private:
         for (auto& cb : cbs) if (cb) cb(ts);
     }
 
+     template<std::size_t... Os>
+     bool outputs_have_space(std::index_sequence<Os...>) const {
+         return (... && (!std::get<Os>(output_channels_) ||
+                         std::get<Os>(output_channels_)->has_space()));
+     }
+
+     template<std::size_t... Os>
+     void register_space_callbacks(std::index_sequence<Os...>) {
+         ((std::get<Os>(output_channels_)
+             ? (void)std::get<Os>(output_channels_)->set_space_callback(
+                   [this] { try_submit(0.5f); })
+             : (void)0), ...);
+     }
+
     void self_stop() {
         disable_inputs(std::make_index_sequence<input_count>{});
         disable_outputs(std::make_index_sequence<output_count>{});
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
+        // Plain store, not release_and_recheck(): this node is stopping, and
+        // honouring a pending wake here would resubmit a dead node.
         queued_.store(false, std::memory_order_release);
         stop_flag_.store(true, std::memory_order_relaxed);
     }
@@ -606,11 +825,51 @@ private:
         return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
     }
 
+    /// Priority in [0,1], higher runs sooner.
+    ///
+    /// Input fill alone answers "how much work is waiting for me". That is
+    /// only half the question: a node whose *outputs* are already full cannot
+    /// deliver anything: running it produces a value with nowhere to go, so it
+    /// immediately parks and the slot is wasted. Meanwhile the node that would
+    /// have drained that full channel waits behind it.
+    ///
+    /// So occupancy of the outputs is deducted from occupancy of the inputs.
+    /// The scheduler then naturally favours whoever is furthest downstream of
+    /// a bottleneck — the node whose inputs are backed up but whose outputs
+    /// have room is exactly the one whose execution frees the most capacity —
+    /// and defers producers that would only deepen a queue that is already
+    /// full.
+    ///
+    /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
+    /// both terms are mean fills in [0,1], so the difference is in [-1,1], and
+    /// the affine map keeps the whole range distinguishable instead of
+    /// collapsing every output-saturated node onto the same value. 0.5 remains
+    /// the neutral point, matching the default used for source nodes.
     float compute_priority() {
         if constexpr (input_count == 0) return 0.5f;
-        float sum = 0.0f;
+        float in = 0.0f;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
+        sum_fill(in, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        in /= static_cast<float>(input_count);
+
+        if constexpr (output_count == 0) return in;
+
+        float out = 0.0f;
+        sum_output_fill(out, std::make_index_sequence<output_count>{});
+        out /= static_cast<float>(output_count);
+
+        const float p = 0.5f * (1.0f + in - out);
+        return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
+    }
+
+    /// Mean fill of the output channels, same normalisation as sum_fill.
+    /// An unconnected output holds nothing back, so it contributes 0.
+    template<std::size_t... Os>
+    void sum_output_fill(float& sum, std::index_sequence<Os...>) {
+        ((sum += (std::get<Os>(output_channels_) &&
+                  std::get<Os>(output_channels_)->capacity() > 0)
+            ? float(std::get<Os>(output_channels_)->approx_size())
+              / float(std::get<Os>(output_channels_)->capacity())
+            : 0.0f), ...);
     }
     template<std::size_t... Is>
     void sum_fill(float& sum, std::index_sequence<Is...>) {
@@ -620,10 +879,32 @@ private:
             : 0.5f), ...);
     }
 
+    /// Submit unless already queued. A wake that arrives while this node is
+    /// queued or running is *recorded*, never dropped.
+    ///
+    /// Wakes are edge-triggered: a channel fires its space callback on the
+    /// transition, once. If that lands while queued_ is up, the CAS below fails
+    /// and — before wake_pending_ — the wake was gone. A node could then park a
+    /// value, release its worker, and sleep forever holding output its consumer
+    /// was waiting for, with every worker idle in cond_wait and nothing left to
+    /// re-trigger it. Recording the drop turns the signal level-triggered: the
+    /// invariant is that a node never sleeps with a wake outstanding, enforced
+    /// by release_and_recheck() at every point that releases the node.
     void try_submit(float priority) {
         bool expected = false;
         if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
             scheduler_->submit([this] { fire_once(); }, priority);
+        else
+            wake_pending_.store(true, std::memory_order_release);
+    }
+
+    /// Clear queued_, then honour any wake that was dropped while it was up.
+    /// Every path that finishes or parks a firing must release the node through
+    /// here rather than storing queued_ directly.
+    void release_and_recheck(float priority = 0.5f) {
+        queued_.store(false, std::memory_order_release);
+        if (wake_pending_.exchange(false, std::memory_order_acq_rel))
+            try_submit(priority);
     }
 
     void fire_once() {
@@ -636,6 +917,46 @@ private:
             t0.time_since_epoch()).count();
         stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
 
+        // Parked from a previous firing: retry that value before touching the
+        // inputs. Returning here releases the worker — the channel's space
+        // callback re-submits this node once the consumer drains a slot.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
+                release_and_recheck();
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                return;
+            }
+        }
+
+        // See the equivalent guard in PoolNode::fire_once: a space-callback
+        // wake with nothing parked must release the worker, not fall through
+        // into pop_inputs on an empty channel.
+        if constexpr (input_count > 0) {
+            if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
+                release_and_recheck();
+                on_input_ready();
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1   = clock_t::now();
@@ -659,7 +980,11 @@ private:
         } catch (const ChannelOverflowError&) {
             fire_callbacks(event_callbacks_);
         } catch (...) {
-            if (error_handler_ && error_handler_(name_, std::current_exception())) {
+            auto eptr = std::current_exception();
+            const bool handled =
+                (error_handler_     && error_handler_(name_, eptr)) ||
+                (net_error_handler_ && net_error_handler_(name_, eptr));
+            if (handled) {
             } else {
                 fire_callbacks(closed_callbacks_);
                 self_stop();
@@ -668,8 +993,26 @@ private:
         }
 
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        release_and_recheck();
         if (stop_flag_.load(std::memory_order_relaxed)) return;
+
+        // Parked by the push above. Same situation as the retry path at the top
+        // of fire_once — and the same lost-wakeup race, which that path closes
+        // and this one did not. A space callback that fired while queued_ was
+        // still up got swallowed by try_submit's CAS, and the resubmit below
+        // cannot cover it: this firing consumed its input, so inputs are empty
+        // and on_input_ready() will not resubmit. The node would then hold its
+        // value forever while its consumer waits for exactly that value and its
+        // producer parks on an input channel that never drains. Re-check now
+        // that the flag is down.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                    try_submit(0.5f);
+                return;                    // parked
+            }
+        }
+
         if constexpr (input_count == 0) try_submit(0.5f);
         else                            on_input_ready();
     }
@@ -692,26 +1035,32 @@ private:
     }
 
     template<std::size_t... Is>
+    /// Pushes what it can and parks the rest. `pending_done_` marks the elements
+    /// already taken, so a retry never re-pushes one — a duplicate is as wrong as
+    /// a drop and harder to notice.
     void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
-        (push_one_out<Is>(std::get<Is>(std::move(result))), ...);
+        bool all = true;
+        ((pending_done_[Is] = pending_done_[Is] ||
+            push_one_out<Is>(std::get<Is>(std::move(result))),
+          all = all && pending_done_[Is]), ...);
+        if (all) { pending_.reset(); pending_done_.fill(false); }
+        else       pending_ = std::move(result);
     }
+    /// Returns false when the ring was full and the value was NOT taken; the
+    /// caller must keep it and retry after the channel signals space.
     template<std::size_t I>
-    void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
+    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
-        if (!ch) return;
+        if (!ch) return true;
         // Sentinels (EOF) must never be dropped: a lost token wedges every
         // downstream pop() forever. Deliver them out-of-band (push_sentinel),
         // which never overflows and never blocks this node's worker thread.
         if (is_sentinel_value(val)) {
             ch->push_sentinel(std::move(val));
-            return;
+            return true;
-        }
-        try {
-            ch->push(std::move(val));
-        } catch (const ChannelOverflowError&) {
-            throw ChannelOverflowError(ch->capacity(),
-                "pool node '" + name_ + "'");
         }
+        // See the note on the typed overload above: park rather than block.
+        return ch->try_push(val);
     }
 
     Obj&                                     obj_;
@@ -722,8 +1071,20 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{false};
+    /// A wake that arrived while queued_ was up. See try_submit.
+    std::atomic<bool>                        wake_pending_{false};
+
+    /// The hidden one-slot output buffer (see push_outputs). Holding the value
+    /// here is what lets a node stop running without dropping it or occupying a
+    /// scheduler worker. One slot suffices because at most one fire_once() runs
+    /// per node at a time — with concurrent firing it would have to hold a whole
+    /// FIFO's worth.
+    std::conditional_t<std::is_void_v<return_raw>, std::monostate,
+                       std::optional<return_tuple>> pending_{};
+    std::array<bool, (output_count ? output_count : 1)> pending_done_{};
     NodeStats                                stats_;
     NodeErrorHandler                         error_handler_;
+    NodeErrorHandler                         net_error_handler_;
     std::chrono::milliseconds                max_exec_time_{0};
     std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
     std::array<NodeEventCallback, 2>         closed_callbacks_{};

include/kpn/static_network.hpp

@@ -122,6 +122,10 @@ public:
                     [this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); });
             }
         }
+        if (error_handler_) {
+            for (auto* node : user_nodes_topo_)
+                node->set_network_error_callback(error_handler_);
+        }
         for (auto* n : user_nodes_topo_)  n->start();
         for (auto* n : fanout_nodes_ptr_) n->start();
 #ifdef KPN_WEB_DEBUG
@@ -181,6 +185,14 @@ public:
 
     void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
 
+    /// 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. Without a listener the exception is discarded
+    /// and only a Closed event survives, which reports that a node stopped
+    /// but not why.
+    void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
+
 #ifdef KPN_WEB_DEBUG
     void set_web_debug_port(uint16_t port) { web_debug_port_ = port; }
     // Called by DebugHub::register_network() so the hub owns the debug server.
@@ -276,6 +288,7 @@ private:
     std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
     std::vector<std::pair<std::string, IPoolProbe*>>     pool_probes_;
     EventHandler                                        event_handler_;
+    NodeErrorHandler                                    error_handler_;
     clock_t::time_point                                 start_time_;
 #ifdef KPN_WEB_DEBUG
     uint16_t                                            web_debug_port_{9090};

tests/CMakeLists.txt

@@ -34,6 +34,7 @@ add_executable(kpn_tests
     test_static_network.cpp
     test_shared_resource.cpp
     test_pool_node.cpp
+    test_backpressure_deadlock.cpp
     test_scheduler.cpp
 )
 

tests/test_backpressure_deadlock.cpp

@@ -0,0 +1,171 @@
+// Regression: a blocking push must not park a pool worker.
+//
+// Node outputs use push_blocking so a full channel costs time rather than data
+// (a dropped frame does not degrade a downstream result, it silently changes
+// one). But push_blocking sleeps *inside* fire_once, which runs on a pool
+// worker — and nodes are pinned to workers by index. Park enough workers in
+// that retry loop and there is nobody left to run the consumer that would drain
+// the channel, so the whole chain wedges.
+//
+// This is the failure channel.hpp:174 already warns about for sentinels
+// ("a blocking push would park that thread and stop it draining its own input,
+// cascading into a hold-and-wait deadlock under backpressure"). The warning
+// applies to data pushes too.
+//
+// Observed in the field as an intermittent hang: frame_source, camera_pos,
+// face_detector and face_aligner all asleep in push_blocking at once.
+#include <catch2/catch_test_macros.hpp>
+#include <kpn/kpn.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <thread>
+
+namespace {
+
+struct Produce {
+    static constexpr std::string_view label() { return "produce"; }
+    int n{0};
+    int operator()() { return n++; }
+};
+
+struct Relay {
+    static constexpr std::string_view label() { return "relay"; }
+    int operator()(int v) { return v; }
+};
+
+// Deliberately slower than the producer, so the channels between them fill.
+struct SlowSink {
+    static constexpr std::string_view label() { return "slow_sink"; }
+    std::atomic<int>* seen;
+    void operator()(int) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(2));
+        seen->fetch_add(1, std::memory_order_relaxed);
+    }
+};
+
+}  // namespace
+
+TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") {
+    std::atomic<int> seen{0};
+
+    Produce  p_fn;
+    Relay    r1_fn, r2_fn, r3_fn;
+    SlowSink s_fn{&seen};
+
+    // Small channels so they saturate immediately, and a chain longer than a
+    // modest pool — the shape that starves workers.
+    kpn::ObjectNode<Produce,  kpn::in<>,        kpn::out<"a">, "produce",   0> p (p_fn,  2);
+    kpn::ObjectNode<Relay,    kpn::in<"a">,     kpn::out<"b">, "relay1",    0> r1(r1_fn, 2);
+    kpn::ObjectNode<Relay,    kpn::in<"b">,     kpn::out<"c">, "relay2",    0> r2(r2_fn, 2);
+    kpn::ObjectNode<Relay,    kpn::in<"c">,     kpn::out<"d">, "relay3",    0> r3(r3_fn, 2);
+    kpn::ObjectNode<SlowSink, kpn::in<"d">,     kpn::out<>,    "slow_sink", 0> s (s_fn,  2);
+
+    auto net = kpn::make_network(
+        kpn::edge(p.output<"a">(),  r1.input<"a">()),
+        kpn::edge(r1.output<"b">(), r2.input<"b">()),
+        kpn::edge(r2.output<"c">(), r3.input<"c">()),
+        kpn::edge(r3.output<"d">(), s.input<"d">())
+    );
+    net.start();
+
+    // The sink is the slowest stage at 2 ms/item, so 40 items is ~80 ms of real
+    // work. Anything approaching the timeout means the chain stopped draining
+    // rather than merely running slowly.
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(20);
+    while (seen.load(std::memory_order_relaxed) < 40 &&
+           std::chrono::steady_clock::now() < deadline)
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+
+    const int got = seen.load(std::memory_order_relaxed);
+    net.stop();
+
+    INFO("items drained: " << got << " of 40");
+    CHECK(got >= 40);
+}
+
+// Regression: a saturated chain must never stall.
+//
+// push_outputs parks from two places: the retry at the top of fire_once, and
+// the ordinary push after the node function returns. Both release the worker,
+// so both face the same lost wakeup — a space callback firing while queued_ is
+// still up is swallowed by try_submit's CAS. Only the retry path re-checked for
+// space afterwards. The normal path fell through to on_input_ready(), which
+// resubmits only if inputs are ready — and the firing that just parked had
+// consumed its input, so they are not.
+//
+// The strand is permanent under saturation: the node holds its value, its
+// consumer waits for exactly that value, and its producer fills the node's
+// input channel and parks too. Nothing moves again.
+//
+// The test above cannot catch it — 40 items drain before any strand occurs.
+// This one runs the chain saturated and watches for progress to *freeze*, which
+// is the signature of the deadlock. It deliberately does not assert a total:
+// capacity-1 channels are slow, and "slow" must never be reported as "wedged".
+namespace {
+
+struct FreeRun {
+    static constexpr std::string_view label() { return "free_run"; }
+    int n{0};
+    int operator()() { return n++; }
+};
+
+struct CountingSink {
+    static constexpr std::string_view label() { return "counting_sink"; }
+    std::atomic<int>* seen;
+    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
+};
+
+}  // namespace
+
+TEST_CASE("a saturated chain never stalls", "[backpressure][deadlock]") {
+    std::atomic<int> seen{0};
+
+    FreeRun      p_fn;
+    Relay        r1_fn, r2_fn;
+    CountingSink s_fn{&seen};
+
+    // Capacity 1 everywhere: every push contends, so the park path is taken
+    // constantly and the race window is sampled millions of times.
+    kpn::ObjectNode<FreeRun,      kpn::in<>,     kpn::out<"a">, "free_run", 0> p (p_fn,  1);
+    kpn::ObjectNode<Relay,        kpn::in<"a">,  kpn::out<"b">, "relay1",   0> r1(r1_fn, 1);
+    kpn::ObjectNode<Relay,        kpn::in<"b">,  kpn::out<"c">, "relay2",   0> r2(r2_fn, 1);
+    kpn::ObjectNode<CountingSink, kpn::in<"c">,  kpn::out<>,    "sink",     0> s (s_fn,  1);
+
+    auto net = kpn::make_network(
+        kpn::edge(p.output<"a">(),  r1.input<"a">()),
+        kpn::edge(r1.output<"b">(), r2.input<"b">()),
+        kpn::edge(r2.output<"c">(), s.input<"c">())
+    );
+    net.start();
+
+    // A live chain moves thousands of items a second, so 3 s with no movement
+    // at all is a wedge, not a slow patch. Sampling for 25 s gives the race
+    // ample opportunity: the pipeline hit it roughly twice in 30 runs.
+    const auto  giveup    = std::chrono::steady_clock::now() + std::chrono::seconds(25);
+    int         last      = 0;
+    auto        last_move = std::chrono::steady_clock::now();
+    bool        stalled   = false;
+    int         stall_at  = 0;
+
+    while (std::chrono::steady_clock::now() < giveup) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        const int now_seen = seen.load(std::memory_order_relaxed);
+        if (now_seen != last) {
+            last      = now_seen;
+            last_move = std::chrono::steady_clock::now();
+        } else if (std::chrono::steady_clock::now() - last_move >
+                   std::chrono::seconds(3)) {
+            stalled  = true;
+            stall_at = now_seen;
+            break;
+        }
+    }
+
+    net.stop();
+
+    INFO("chain stalled after " << stall_at << " items");
+    CHECK_FALSE(stalled);
+    // Guard against the test passing because nothing ever ran.
+    CHECK(last > 1000);
+}

tests/test_pool_node.cpp

@@ -243,7 +243,13 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") {
 
 // ── Overflow callback ─────────────────────────────────────────────────────────
 
-TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") {
+// NOTE: output overflow is no longer reachable on the data path. A node whose
+// output channel is full now PARKS — it keeps the value in a hidden one-slot
+// buffer, releases its scheduler worker, and is re-submitted when the consumer
+// frees a slot. The overflow callback survives for other producers (a direct
+// Channel::push by non-node code still throws), but a pool node cannot trigger
+// it, so these cases assert the stronger property instead: nothing is dropped.
+TEST_CASE("pool node parks instead of overflowing a full output", "[pool_node][overflow]") {
     auto pool = std::make_shared<ThreadPool>(2);
     pool->start();
 
@@ -264,10 +270,13 @@ TEST_CASE("pool node overflow callback fires on full output channel", "[pool_nod
     node.stop();
     pool->stop();
 
-    REQUIRE(overflow_count.load() > 0);
+    // Parked, not overflowed: the value is still owned by the node.
+    REQUIRE(overflow_count.load() == 0);
+    // And it was never handed downstream, so nothing was lost or duplicated.
+    REQUIRE(full_ch.size() == 1);
 }
 
-TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") {
+TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") {
     auto pool = std::make_shared<ThreadPool>(2);
     pool->start();
 
@@ -296,7 +305,10 @@ TEST_CASE("pool node overflow callback is independent per instance", "[pool_node
     nodeB.stop();
     pool->stop();
 
-    REQUIRE(a_overflows.load() > 0);
+    // Neither overflows now: A parks on its full output, B runs normally. The
+    // point of the case is unchanged — one node's backpressure must not leak
+    // into another's callbacks.
+    REQUIRE(a_overflows.load() == 0);
     REQUIRE(b_overflows.load() == 0);
 }
 
@@ -412,7 +424,9 @@ TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]")
     node.stop();
     pool->stop();
 
-    REQUIRE(net_overflows.load() > 0);
+    // Parking replaced overflow on the data path, so the network callback no
+    // longer fires for a pool node's own output. See the note above.
+    REQUIRE(net_overflows.load() == 0);
 }
 
 TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") {
@@ -459,6 +473,8 @@ TEST_CASE("per-node and network overflow callbacks both fire independently", "[p
     node.stop();
     pool->stop();
 
-    REQUIRE(per_node.load() > 0);
+    // Both zero now: the node parks rather than overflowing. The case still
-    REQUIRE(network.load() > 0);
+    // guards that the two callbacks are wired independently.
+    REQUIRE(per_node.load() == 0);
+    REQUIRE(network.load() == 0);
 }