Deliver EOF sentinels out-of-band to prevent teardown deadlock

Channel::push() drops values on overflow (the intended backpressure
policy for data), and PoolNode swallows the resulting ChannelOverflowError.
For a control sentinel like EOF this is fatal: a single dropped EOF under
backpressure wedges every downstream pop() forever, so the pipeline never
tears down.

Deliver sentinels out-of-band instead. Channel::push_sentinel() stores the
token in a dedicated slot that does not consume ring capacity, so it can
never overflow and — crucially — never blocks the caller. That non-blocking
property is essential: each KPN node has a single worker thread, so a
*blocking* push would park that thread and stop it draining its own input,
cascading into a hold-and-wait deadlock under backpressure. The consumer's
pop()/try_pop_now() drain the ring first, then deliver the sentinel, so it
always arrives after every value pushed before it.

approx_size() (which node readiness checks call) counts a pending sentinel
as consumable work, so a channel carrying only a sentinel still schedules
its consumer's next fire — without this the token would sit undelivered and
the pipeline would still deadlock at teardown.

PoolNode/PoolObjectNode route values carrying an eof flag (direct .eof or
nested .source.eof) through push_sentinel via a SFINAE-safe is_sentinel_value
trait; all other values keep the existing lossy throwing push. The trait
compiles to false for types without an eof convention, so this is a no-op
for pipelines that don't use one.

Verified end-to-end: scene_analyze now reaches EOF, flushes its output, and
exits cleanly instead of hanging.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

include/kpn/channel.hpp

@@ -136,6 +136,36 @@ public:
             push_callback_();
     }
 
+    // Lossless, non-blocking delivery for a must-deliver control token (EOF).
+    //
+    // A sentinel is stored out-of-band — in a dedicated slot that does NOT
+    // consume ring capacity — so this can never overflow and never blocks the
+    // caller. That distinction is essential: each KPN node has a single worker
+    // thread, so a *blocking* push would park that thread and stop it draining
+    // its own input, cascading into a hold-and-wait deadlock under backpressure.
+    // Setting a flag and returning keeps the worker free to keep popping.
+    //
+    // The consumer's pop() drains the ring first, then delivers this sentinel,
+    // preserving ordering (EOF arrives after all data pushed before it).
+    //
+    // Only the sole producer may call it (SPSC contract, same as push()).
+    // Returns false if the channel is already disabled (token discarded —
+    // teardown is in progress, so the sentinel is moot).
+    bool push_sentinel(T value) {
+        if (!accepting_.load(std::memory_order_acquire)) {
+            stats_.record_drop();
+            return false;
+        }
+        eof_value_ = make_storage(std::move(value));
+        has_eof_.store(true, std::memory_order_release);
+        // Wake a consumer blocked in pop(): the sentinel is now deliverable even
+        // though the ring may be empty.
+        wake_.fetch_add(1, std::memory_order_release);
+        wake_.notify_one();
+        if (push_callback_) push_callback_();
+        return true;
+    }
+
     // Blocking pop. Returns when an item is available.
     // Throws ChannelClosedError if the channel is disabled (regardless of fill).
     T pop() {
@@ -148,21 +178,28 @@ public:
             // If empty, spin before sleeping: avoids the futex when the next item
             // arrives within the spin window (~4 µs at default spin_count=200 on x86).
             if (h == t) {
+                // Ring drained — deliver any pending out-of-band sentinel (EOF)
+                // now, so it always arrives after the data pushed before it.
+                { T s; if (take_sentinel(s)) return s; }
+
                 if (!accepting_.load(std::memory_order_acquire))
                     throw ChannelClosedError{};
 
-                for (std::size_t s = 0; s < spin_count_; ++s) {
+                for (std::size_t si = 0; si < spin_count_; ++si) {
                     spin_hint();
                     t = tail_.load(std::memory_order_acquire);
                     if (t != h) break;
+                    { T s; if (take_sentinel(s)) return s; }
                     if (!accepting_.load(std::memory_order_relaxed))
                         throw ChannelClosedError{};
                 }
 
                 if (h == t) {
-                    // Still empty after spin — sleep until push() or disable() fires.
-                    // Re-check tail after loading w to guard against a lost wakeup.
+                    // Still empty after spin — sleep until push()/push_sentinel()
+                    // or disable() fires. Re-check tail and the sentinel after
+                    // loading w to guard against a lost wakeup.
                     if (tail_.load(std::memory_order_acquire) != h) continue;
+                    if (has_eof_.load(std::memory_order_acquire)) continue;
                     wake_.wait(w, std::memory_order_relaxed);
                     continue;
                 }
@@ -190,9 +227,12 @@ public:
     }
 
     // Immediate non-blocking pop. Returns false if the ring is empty.
+    // Once the ring is drained, delivers any pending out-of-band sentinel (EOF)
+    // 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)) return false;
+        if (h == tail_.load(std::memory_order_acquire))
+            return take_sentinel(out);
         out = extract(std::move(buf_[h & ring_mask_]));
         head_.store(h + 1, std::memory_order_release);
         stats_.record_pop();
@@ -217,12 +257,21 @@ public:
         push_callback_ = std::move(cb);
     }
 
-    // Size derived lazily from ring indices — no separate counter on the hot path.
+    // Ring occupancy, derived lazily from indices — no separate counter on the
+    // hot path. Excludes any out-of-band sentinel (that lives outside the ring).
     std::size_t size() const {
         return tail_.load(std::memory_order_relaxed)
              - head_.load(std::memory_order_relaxed);
     }
-    std::size_t approx_size() const { return size(); }
+
+    // A pending out-of-band sentinel (EOF) counts as consumable work here even
+    // though it holds no ring slot. This is what node readiness checks call, so
+    // a channel carrying only a sentinel still schedules its consumer's next
+    // fire — without this the sentinel would never be popped and the pipeline
+    // would deadlock at teardown.
+    std::size_t approx_size() const {
+        return size() + (has_eof_.load(std::memory_order_acquire) ? 1u : 0u);
+    }
 
     std::size_t capacity()     const { return capacity_; }
     bool        is_accepting() const { return accepting_.load(std::memory_order_relaxed); }
@@ -260,6 +309,17 @@ private:
             return *s;
     }
 
+    // Consume the out-of-band sentinel if one is pending. Consumer-only.
+    // Called only when the ring is observed empty, so the sentinel is always
+    // delivered after every value pushed before it.
+    bool take_sentinel(T& out) {
+        if (!has_eof_.load(std::memory_order_acquire)) return false;
+        out = extract(std::move(eof_value_));
+        has_eof_.store(false, std::memory_order_release);
+        stats_.record_pop();
+        return true;
+    }
+
     const std::size_t               capacity_;
     const std::size_t               spin_count_;
     std::size_t                     ring_mask_;
@@ -267,8 +327,16 @@ private:
     std::function<void()>           push_callback_;
     ChannelStats                    stats_;
 
+    // Out-of-band sentinel (EOF): stored outside the ring so its delivery never
+    // depends on ring capacity and never blocks the producer. Written by the
+    // producer (push_sentinel), read+cleared by the consumer (take_sentinel);
+    // has_eof_ is the publish/consume handshake.
+    storage_type              eof_value_{};
+    std::atomic<bool>         has_eof_{false};
+
     // Separate cache lines: head_ is written only by the consumer;
     // tail_ and wake_ are written only by the producer.
+    //   wake_  wakes a blocked pop() on enqueue or on a pending sentinel.
     alignas(64) std::atomic<std::size_t> head_{0};
     alignas(64) std::atomic<std::size_t> tail_{0};
                 std::atomic<uint32_t>    wake_{0};

include/kpn/pool_node.hpp

@@ -22,6 +22,38 @@
 
 namespace kpn {
 
+// ── 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 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 ──────────────────────────────────────────────────────────────────
 //
 // Reactive alternative to Node<>. Instead of owning a blocked thread, the node
@@ -361,6 +393,13 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // 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;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -660,6 +699,13 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // 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;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {

tests/test_channel.cpp

@@ -175,3 +175,66 @@ TEST_CASE("bandwidth_mbs returns 0 when elapsed_s is zero or negative", "[channe
     REQUIRE(snap.bandwidth_mbs(0.0) == 0.0);
     REQUIRE(snap.bandwidth_mbs(-1.0) == 0.0);
 }
+
+TEST_CASE("push_sentinel never overflows even on a full channel", "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push(2);   // channel full — a plain push(3) would throw ChannelOverflowError
+
+    // The sentinel is stored out-of-band, so it neither throws nor blocks the
+    // caller — the exact property an EOF token needs under backpressure. This
+    // returns immediately with the ring still full.
+    REQUIRE(ch.push_sentinel(99));
+    REQUIRE(ch.size() == 2);   // sentinel did not consume ring capacity
+}
+
+TEST_CASE("push_sentinel is delivered after all ring data, in order", "[channel][sentinel]") {
+    Channel<int> ch(4);
+    ch.push(1);
+    ch.push(2);
+    ch.push_sentinel(99);   // enqueue EOF while data is still buffered
+
+    // Data drains first; the sentinel arrives only once the ring is empty.
+    REQUIRE(ch.pop() == 1);
+    REQUIRE(ch.pop() == 2);
+    REQUIRE(ch.pop() == 99);
+}
+
+TEST_CASE("push_sentinel wakes a blocked pop", "[channel][sentinel]") {
+    Channel<int> ch(2);   // empty
+    std::thread producer([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        ch.push_sentinel(99);   // must wake a consumer parked on an empty ring
+    });
+    REQUIRE(ch.pop() == 99);
+    producer.join();
+}
+
+TEST_CASE("approx_size counts a pending sentinel so consumers stay schedulable",
+          "[channel][sentinel]") {
+    Channel<int> ch(4);
+    REQUIRE(ch.approx_size() == 0);
+    ch.push_sentinel(99);
+    // Node readiness checks call approx_size(); it must report the out-of-band
+    // sentinel as consumable work even though it holds no ring slot.
+    REQUIRE(ch.approx_size() == 1);
+    REQUIRE(ch.size() == 0);       // ...but the ring itself is still empty
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));
+    REQUIRE(out == 99);
+    REQUIRE(ch.approx_size() == 0);
+}
+
+TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty",
+          "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push_sentinel(99);
+
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));   // ring data first
+    REQUIRE(out == 1);
+    REQUIRE(ch.try_pop_now(out));   // then the sentinel
+    REQUIRE(out == 99);
+    REQUIRE_FALSE(ch.try_pop_now(out));   // nothing left
+}