fix: park nodes on a full output instead of blocking the worker

push_blocking parked a scheduler worker inside the push. Nodes own a
private single-thread pool, so the parked thread was the only one that
could drain that node's own input — hold-and-wait, and under sustained
backpressure four nodes of a five-node chain slept in nanosleep at once.
channel.hpp already warned about this for sentinels; it applies just as
much to data pushes.

The scheduler was purely input-driven: on_input_ready() wakes a node when
input arrives, with no counterpart for "my output has room". Lacking that
signal, blocking the thread was the only way to handle a full output.
This adds the missing half.

- Channel::try_push + has_space + set_space_callback; the callback fires
  from both pop() and try_pop_now().
- PoolNode/PoolObjectNode keep a one-slot pending_ buffer with per-element
  done flags, so a retry cannot duplicate an already-accepted element. One
  slot suffices because queued_ admits at most one fire_once per node.
- The re-check after clearing queued_ closes the lost-wakeup race where a
  space callback fires while the flag is still up and is swallowed.

Two bugs surfaced once nodes actually parked, both fixed here:

- pop_one reports an *empty* channel as ChannelClosedError, which is also
  the node's "upstream finished, self-stop" signal. A node woken by output
  space with empty inputs therefore killed itself. fire_once now releases
  the worker when its inputs are not ready rather than falling through.
- The drained-park path resubmitted unconditionally instead of via
  on_input_ready(), firing nodes with nothing to read.

compute_priority is now output-aware: mean output fill is deducted from
mean input fill, mapped as 0.5·(1 + in - out). Input fill alone asks only
"how much work is waiting for me"; a node whose outputs are already full
cannot deliver, so running it just parks it again and wastes the slot
while the node that would drain that channel waits behind it. The
scheduler now favours whoever is furthest downstream of a bottleneck.

Also adds a network-level error listener. A node's exception was discarded
at the node boundary and survived only as a Closed event, which reports
that a node stopped but not why — that missing detail is what made the
above slow to diagnose. INode::set_network_error_callback plus
StaticNetwork::set_error_handler forward it to the application.

Tests: 121/121. test_backpressure_deadlock drives a five-node chain with
capacity-2 channels against a slow sink and fails on the old code. The
four test_pool_node overflow tests now assert parking rather than the
removed drop-and-report behaviour.

Known-incomplete: a rare hang remains, roughly 1 run in 20 against a 300s
timeout, down from every run failing. Committed because the fix is a large
strict improvement and the residual case needs its own reproduction.

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,6 +247,20 @@ 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>{});
@@ -280,11 +298,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>
@@ -315,6 +373,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>{});
+                queued_.store(false, std::memory_order_release);
+                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>{});
+                queued_.store(false, std::memory_order_release);
+                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) {
+                queued_.store(false, std::memory_order_release);
+                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 +469,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_);
@@ -385,30 +518,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;
         }
-        // Backpressure, not loss. A full downstream channel means the consumer
+        // Backpressure without parking the worker. A full channel means the
-        // is behind, and the correct response is for this producer to run
+        // consumer is behind; the value is kept and this node stops running
-        // slower — not to discard a value. A dropped frame does not degrade a
+        // until the channel signals space (set_space_callback re-submits it).
-        // result, it silently changes one, and the caller has no way to tell.
         //
-        // Safe here because sentinels are handled above, out-of-band: this
+        // Blocking here instead would sleep inside a scheduler worker, and
-        // blocks only on data, so the EOF token that unwinds the network can
+        // nodes are pinned to workers — park enough of them and nothing is left
-        // always overtake a stalled data path.
+        // to run the consumer that would drain the channel. That is the
-        ch->push_blocking(std::move(val));
+        // 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>
@@ -430,8 +574,18 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{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_{};
@@ -499,6 +653,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); }
@@ -570,6 +725,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(); }), ...);
     }
 
@@ -578,6 +735,20 @@ 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>{});
@@ -609,11 +780,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...>) {
@@ -639,6 +850,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>{});
+                queued_.store(false, std::memory_order_release);
+                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) {
+                queued_.store(false, std::memory_order_release);
+                on_input_ready();
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1   = clock_t::now();
@@ -662,7 +913,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();
@@ -695,22 +950,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;
         }
-        // See the note on the typed overload above: block rather than drop.
+        // See the note on the typed overload above: park rather than block.
-        ch->push_blocking(std::move(val));
+        return ch->try_push(val);
     }
 
     Obj&                                     obj_;
@@ -721,8 +986,18 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{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,85 @@
+// 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);
+}

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);
 }