fix: a filter or router must not drop on a full output

RouterNode and FilterNode were the last nodes on a data path still using the
throwing push() and swallowing the result:

    try { out_ch_->push(val); } catch (const ChannelOverflowError&) {}

6595e6e made node outputs lossless, 28e0667 stopped them parking a worker,
a8cfe73 did the same for FanoutNode. These two were in none of them.

For ordinary values that is the familiar silent-loss problem: a dropped item
does not degrade a downstream result, it silently changes one, and the
consumer cannot tell it happened.

For a sentinel it is a hang. EOF is what tells every downstream node to shut
down and there is nothing after it to retry, so a filter that passes EOF by
predicate but drops it by backpressure produces a pipeline that never
terminates. scene-actor-extraction's decimator is exactly that shape —
`if (f.eof) return true;` in the predicate, feeding a chain whose slowest
node is an ONNX embedder, so the output is reliably full when EOF arrives.
Everything downstream then waits forever for a token that was discarded, and
the run has to be killed.

Both now route sentinels out-of-band via push_sentinel, which consumes no
ring capacity and cannot overflow, and retry ordinary values until taken.
Like FanoutNode and unlike a pool node, these own a private thread, so
waiting costs no scheduler worker and needs no space-callback park;
stop_flag_ is rechecked every pass so teardown cannot hang on a full output.
Time spent parked is charged to blocked rather than exec, so a held-up node
does not report as busy. An out-of-range router selector still drops by
design — that item was routed nowhere, which is not the same as lost.

is_sentinel_value moves from pool_node.hpp to traits.hpp. Every node type
that forwards a value needs it; these two not having it is the bug.

Verified in both directions. On c73edff the new case delivers 6 of 40 values
and never sets saw_eof; here it delivers 40 and terminates. EOF is emitted
exactly once, as a real source does — a test source that re-offered it would
mask the bug, since a later attempt could find the channel drained.

Note for downstream: the decimator is now a backpressure point rather than a
relief valve, so the source throttles to the face branch instead of quietly
thinning it. That is the intended behaviour, but it changes the shape of a
loaded run and is worth a benchmark comparison on a known clip.

include/kpn/branch.hpp

@@ -7,12 +7,70 @@
 
 #include <array>
 #include <atomic>
+#include <chrono>
 #include <functional>
 #include <memory>
 #include <thread>
 
 namespace kpn {
 
+// ── Lossless single-output delivery ───────────────────────────────────────────
+//
+// Shared by RouterNode and FilterNode, which each deliver a value to exactly one
+// channel. Both previously did
+//
+//     try { ch->push(val); } catch (const ChannelOverflowError&) {}
+//
+// which discards the value whenever the consumer is behind. 6595e6e made node
+// outputs lossless, 28e0667 stopped them parking a worker, and a8cfe73 did the
+// same for FanoutNode — these two were in none of them, and were the last
+// remaining users of the throwing push() on a data path.
+//
+// A dropped item does not degrade a downstream result, it silently changes one.
+// Worse, a dropped *sentinel* wedges the pipeline outright: EOF is what tells
+// every downstream node to shut down, and there is nothing after it to retry.
+// A filter that passes EOF by predicate but drops it by backpressure is a
+// pipeline that never terminates.
+//
+// So sentinels go out-of-band via push_sentinel (a dedicated slot that consumes
+// no ring capacity and cannot overflow), and everything else is retried until
+// taken. Like FanoutNode and unlike a pool node, these own a private thread, so
+// waiting here costs no scheduler worker and needs no space-callback park.
+// stop_flag_ is rechecked every pass so teardown cannot hang on a full output.
+//
+// `parked` receives the time spent waiting, which the caller charges to blocked
+// rather than exec — a parked node is idle, and charging it to exec reports the
+// node as busy exactly when it is the one being held up.
+//
+// Returns false if stopped with the value undelivered.
+template<typename T>
+bool deliver_one(Channel<T>* ch, T& val, const std::atomic<bool>& stop_flag,
+                 duration_t& parked) {
+    if (is_sentinel_value(val)) {
+        ch->push_sentinel(std::move(val));
+        return true;
+    }
+    const auto park_from = clock_t::now();
+    for (;;) {
+        if (ch->try_push(val)) {
+            parked = duration_t(clock_t::now() - park_from);
+            return true;
+        }
+        if (stop_flag.load(std::memory_order_relaxed)) {
+            // Teardown with work in hand. One last throwing push, purely so the
+            // channel's own stats record the loss (drop if it is disabled,
+            // overflow if it is merely full). The point of the lossless path is
+            // that a loss is never invisible, and a silent return here would
+            // reintroduce exactly the hole this function exists to close.
+            try { ch->push(std::move(val)); }
+            catch (const ChannelOverflowError&) {}
+            parked = duration_t(clock_t::now() - park_from);
+            return false;
+        }
+        std::this_thread::sleep_for(std::chrono::microseconds(50));
+    }
+}
+
 // ── RouterNode ────────────────────────────────────────────────────────────────
 //
 // Reads one item and pushes it to exactly one of N output channels, chosen by
@@ -126,15 +184,20 @@ private:
                 auto        t1  = clock_t::now();
                 auto        cpu0 = NodeStats::cpu_now();
 
+                // An out-of-range selector still drops by design (documented on
+                // the class): the item was routed nowhere, not lost to a full
+                // channel. Only the latter is what deliver_one exists to stop.
                 std::size_t idx = selector_(val);
-                if (idx < N && out_channels_[idx]) {
+                duration_t  parked{0};
-                    try { out_channels_[idx]->push(val); }
+                bool        delivered = true;
-                    catch (const ChannelOverflowError&) {}
+                if (idx < N && out_channels_[idx])
-                }
+                    delivered = deliver_one(out_channels_[idx], val, stop_flag_, parked);
 
                 auto cpu1 = NodeStats::cpu_now();
                 auto t2   = clock_t::now();
-                stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
+                stats_.record_exec(duration_t(t2 - t1) - parked,
+                                   duration_t(t1 - t0) + parked, cpu0, cpu1);
+                if (!delivered) break;
             } catch (const ChannelClosedError&) {
                 break;
             }
@@ -261,12 +324,19 @@ private:
                 auto t1  = clock_t::now();
                 auto cpu0 = NodeStats::cpu_now();
 
+                // A value the predicate rejects is dropped by design and is not
+                // counted as a processed frame. One it accepts is now delivered
+                // losslessly — including a sentinel, which a filter typically
+                // passes unconditionally so downstream can shut down, and which
+                // the old throwing push discarded whenever the output was full.
                 if (pred_(val) && out_ch_) {
-                    try { out_ch_->push(val); }
+                    duration_t parked{0};
-                    catch (const ChannelOverflowError&) {}
+                    const bool delivered = deliver_one(out_ch_, val, stop_flag_, parked);
                     auto cpu1 = NodeStats::cpu_now();
                     auto t2   = clock_t::now();
-                    stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
+                    stats_.record_exec(duration_t(t2 - t1) - parked,
+                                       duration_t(t1 - t0) + parked, cpu0, cpu1);
+                    if (!delivered) break;
                 }
             } catch (const ChannelClosedError&) {
                 break;

include/kpn/pool_node.hpp

@@ -23,37 +23,8 @@
 
 namespace kpn {
 
-// ── Sentinel detection ────────────────────────────────────────────────────────
+// Sentinel detection (has_eof_field / is_sentinel_value) lives in traits.hpp —
-// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// every node type that forwards values needs it, not just pool-scheduled ones.
-// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
-// source Frame) or nested one level under a `.source` member (`v.source.eof`,
-// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
-// the originating Frame). Sentinels are delivered losslessly and non-blockingly
-// via Channel::push_sentinel() instead of the throwing push(), so backpressure
-// can never drop the token that unblocks downstream teardown.
-//
-// Types with neither shape are never treated as sentinels — both traits are
-// SFINAE-safe and the runtime check compiles away to `false` for them, so this
-// stays a no-op for pipelines that don't use an eof convention.
-template<typename T, typename = void>
-struct has_eof_field : std::false_type {};
-template<typename T>
-struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
-    : std::true_type {};
-
-template<typename T, typename = void>
-struct has_source_eof_field : std::false_type {};
-template<typename T>
-struct has_source_eof_field<T,
-    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
-    : std::true_type {};
-
-template<typename T>
-constexpr bool is_sentinel_value(const T& v) {
-    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
-    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
-    else                                               return false;
-}
 
 // ── PoolNode ──────────────────────────────────────────────────────────────────
 //

include/kpn/traits.hpp

@@ -97,4 +97,41 @@ struct repeat_tuple<T, N, std::index_sequence<Is...>> {
 template<typename T, std::size_t N>
 using repeat_tuple_t = typename repeat_tuple<T, N>::type;
 
+// ── Sentinel detection ────────────────────────────────────────────────────────
+// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
+// source Frame) or nested one level under a `.source` member (`v.source.eof`,
+// as on message types that wrap the originating Frame). Sentinels are delivered
+// losslessly and non-blockingly via Channel::push_sentinel() instead of the
+// throwing push(), so backpressure can never drop the token that unblocks
+// downstream teardown.
+//
+// Types with neither shape are never treated as sentinels — both traits are
+// SFINAE-safe and the runtime check compiles away to `false` for them, so this
+// stays a no-op for pipelines that don't use an eof convention.
+//
+// Lives here rather than in pool_node.hpp because every node type that forwards
+// values needs it, not just the pool-scheduled ones. FilterNode and RouterNode
+// not having it is what let an EOF token be dropped on a full output.
+
+template<typename T, typename = void>
+struct has_eof_field : std::false_type {};
+template<typename T>
+struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
+    : std::true_type {};
+
+template<typename T, typename = void>
+struct has_source_eof_field : std::false_type {};
+template<typename T>
+struct has_source_eof_field<T,
+    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
+    : std::true_type {};
+
+template<typename T>
+constexpr bool is_sentinel_value(const T& v) {
+    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
+    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
+    else                                               return false;
+}
+
 } // namespace kpn

tests/test_backpressure_deadlock.cpp

@@ -322,3 +322,100 @@ TEST_CASE("a fanout absorbs an unequal pair by slowing, not dropping",
     // the difference is the loss.
     CHECK(fast_seen - slow_seen < 200);
 }
+
+// Regression: a filter must not drop an EOF sentinel into a full output.
+//
+// RouterNode and FilterNode were the last nodes on a data path still using the
+// throwing push() and swallowing the result:
+//
+//     try { out_ch_->push(val); } catch (const ChannelOverflowError&) {}
+//
+// 6595e6e made node outputs lossless, 28e0667 stopped them parking a worker,
+// a8cfe73 did the same for FanoutNode. These two were in none of them.
+//
+// For ordinary values that is the familiar silent-loss problem. For a sentinel
+// it is a hang. EOF is what tells every downstream node to shut down, and
+// nothing comes after it to retry — so a filter that passes EOF by predicate
+// but drops it by backpressure produces a pipeline that never terminates. The
+// scene-actor-extraction decimator is exactly this shape: `if (f.eof) return
+// true;` in the predicate, feeding a chain whose slowest node is an ONNX
+// embedder, so the output is reliably full at the moment EOF arrives.
+//
+// The test forces that state rather than racing for it: the sink is slow enough
+// that the filter's output channel is saturated for the whole run, so EOF meets
+// a full ring with certainty.
+//
+// Both assertions are needed. `saw_eof` alone would pass on an implementation
+// that dropped every ordinary value and delivered only the sentinel; `count`
+// alone would pass on the broken one, which delivers plenty of values and loses
+// only the token that matters.
+namespace {
+
+struct EofFrame {
+    int  seq{0};
+    bool eof{false};
+};
+
+// EOF is emitted exactly once, as a real source does. Everything after it is a
+// filler frame the predicate rejects, which keeps the node alive without
+// re-offering the sentinel — a source that retried EOF would mask the bug,
+// since a later attempt could find the channel drained.
+struct EofSource {
+    static constexpr std::string_view label() { return "eof_source"; }
+    int  n{0};
+    int  total{0};
+    EofFrame operator()() {
+        if (n > total) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            return {-1, false};          // filler: dropped by the predicate
+        }
+        EofFrame f{n, n == total};
+        ++n;
+        return f;
+    }
+};
+
+struct EofSink {
+    static constexpr std::string_view label() { return "eof_sink"; }
+    std::atomic<int>*  count;
+    std::atomic<bool>* saw_eof;
+    void operator()(EofFrame f) {
+        std::this_thread::sleep_for(std::chrono::microseconds(200));
+        if (f.eof) saw_eof->store(true, std::memory_order_release);
+        else       count->fetch_add(1, std::memory_order_relaxed);
+    }
+};
+
+}  // namespace
+
+TEST_CASE("a filter delivers EOF into a saturated output", "[backpressure][filter]") {
+    std::atomic<int>  count{0};
+    std::atomic<bool> saw_eof{false};
+
+    EofSource src_fn{0, 40};
+    EofSink   sink_fn{&count, &saw_eof};
+
+    // Every real frame passes the predicate, so the only thing between source
+    // and sink is backpressure. Small channels keep the output saturated.
+    auto filt = kpn::make_filter<EofFrame>(
+        [](const EofFrame& f) { return f.seq >= 0; }, 4);
+
+    kpn::ObjectNode<EofSource, kpn::in<>,      kpn::out<"f">, "eof_source", 0> s(src_fn,  4);
+    kpn::ObjectNode<EofSink,   kpn::in<"f">,   kpn::out<>,    "eof_sink",   0> k(sink_fn, 4);
+
+    auto net = kpn::make_network(
+        kpn::edge(s.output<"f">(),      filt.input<0>()),
+        kpn::edge(filt.output<0>(),     k.input<"f">())
+    );
+    net.start();
+
+    // Generous relative to 41 frames at 200 us, and this is a liveness test:
+    // the broken implementation never sets saw_eof no matter how long it runs.
+    for (int i = 0; i < 200 && !saw_eof.load(std::memory_order_acquire); ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    net.stop();
+
+    INFO("values delivered: " << count.load() << " of 40");
+    CHECK(saw_eof.load(std::memory_order_acquire));
+    CHECK(count.load(std::memory_order_relaxed) == 40);
+}