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/pool_node.hpp

@@ -265,6 +265,8 @@ private:
         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);
     }
@@ -353,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 ─────────────────────────────────────────────────────────────
@@ -379,7 +403,7 @@ private:
         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);
+                release_and_recheck();
                 if (pending_) {
                     // Close the lost-wakeup race: a space_callback that fired
                     // between the failed push and clearing queued_ was
@@ -408,7 +432,7 @@ private:
         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);
+                release_and_recheck();
                 if (pending_) {
                     // Close the lost-wakeup race: a space_callback that fired
                     // between the failed push and clearing queued_ was
@@ -438,7 +462,7 @@ private:
         // 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);
+                release_and_recheck();
                 on_input_ready();   // data may have arrived while we checked
                 return;
             }
@@ -483,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);
@@ -574,6 +615,8 @@ 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
@@ -753,6 +796,8 @@ private:
         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);
     }
@@ -834,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() {
@@ -856,7 +923,7 @@ private:
         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);
+                release_and_recheck();
                 if (pending_) {
                     // Close the lost-wakeup race: a space_callback that fired
                     // between the failed push and clearing queued_ was
@@ -884,7 +951,7 @@ private:
         // 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);
+                release_and_recheck();
                 on_input_ready();
                 return;
             }
@@ -926,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();
     }
@@ -986,6 +1071,8 @@ 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

tests/test_backpressure_deadlock.cpp

@@ -83,3 +83,89 @@ TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") {
     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);
+}