fix: the sentinel slot holds one token and refuses a second

push_sentinel wrote eof_value_ unconditionally. Offering a second token
before the first was taken did two wrong things at once.

It lost the first silently, and a lost EOF is not a lost frame — it is the
token every downstream node is waiting for in order to shut down, so losing
it wedges the pipeline.

And it wrote the storage while the consumer could be moving the previous
value out of it. I expected that to be a stale read; ThreadSanitizer shows
it is worse. On the shared_ptr storage that non-trivial types use, the
racing write tears the refcount, and the stress case added here reports
heap-use-after-free in extract() alongside the data race.

try_push_sentinel now refuses when the slot is occupied, which turns the
slot into a correct SPSC handshake: the producer is the only writer of
eof_value_ and the only one that sets has_eof_, the consumer is the only one
that clears it, so observing it false is what licenses the write. Refusal is
recorded as a drop, and PoolNode reports it through the overflow event
callback, because a refused control token going unnoticed is the failure
this commit exists to stop.

Refusing rather than queueing is deliberate. Two control tokens on one
channel means the stream ended twice, which is a caller protocol error and
not backpressure; parking and retrying would spin against a slot only the
consumer can free, and there is no sensible second value to deliver after
the end of a stream. The non-consuming try_push_sentinel exists so a refused
token is still the caller's to report — the consuming push_sentinel cannot
offer that, since the value has already been moved into its parameter.

Single-shot EOF is what every current caller does, so this is latent for
them today. It stops being latent the moment a pipeline is reused for a
second input, which is what the persistent-pipeline work in 4b6e498 sets up.

Verified in both directions under -DKPN_SANITIZER=thread: the new contended
case reports three data races and a heap-use-after-free against the old
overwrite, and is clean with the handshake. Full suite 137/137, TSan clean
across unit and stress suites.

include/kpn/channel.hpp

@@ -226,12 +226,37 @@ public:
     // 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).
+    // The slot holds exactly one undelivered token. A second offered before the
-    bool push_sentinel(T value) {
+    // first is taken is refused, not queued and not overwritten: two control
+    // tokens on one channel means the stream ended twice, which is a caller
+    // protocol error rather than backpressure, and silently coalescing them
+    // would hide it.
+    /// Outcome of offering a sentinel. SlotBusy is a protocol error, not
+    /// backpressure: it means a second control token was offered while the
+    /// first was still undelivered, and a channel carries at most one.
+    enum class SentinelResult { Taken, Closed, SlotBusy };
+
+    /// Non-consuming form. `value` is left untouched unless the result is
+    /// Taken, so a refused token is still the caller's to report.
+    SentinelResult try_push_sentinel(T& value) {
         if (!accepting_.load(std::memory_order_acquire)) {
             stats_.record_drop();
-            return false;
+            return SentinelResult::Closed;
+        }
+        // Refuse rather than overwrite. Overwriting lost the first token
+        // silently, and worse, wrote eof_value_ while the consumer could be
+        // moving the previous one out of it — a data race on the storage, which
+        // for a shared_ptr payload is a torn refcount rather than a stale read.
+        //
+        // Checking here is what makes the slot a correct SPSC handshake: the
+        // producer is the only writer of eof_value_ and the only one that sets
+        // has_eof_, the consumer is the only one that clears it, so observing
+        // false here means the consumer has finished with the storage and will
+        // not touch it again until this store publishes the next token.
+        if (has_eof_.load(std::memory_order_acquire)) {
+            stats_.record_drop();
+            return SentinelResult::SlotBusy;
         }
         eof_value_ = make_storage(std::move(value));
         has_eof_.store(true, std::memory_order_release);
@@ -240,7 +265,13 @@ public:
         wake_.fetch_add(1, std::memory_order_release);
         wake_.notify_one();
         if (push_callback_) push_callback_();
-        return true;
+        return SentinelResult::Taken;
+    }
+
+    /// Consuming convenience form. Returns false when the token was not stored,
+    /// whether because the channel is closed or because one is already pending.
+    bool push_sentinel(T value) {
+        return try_push_sentinel(value) == SentinelResult::Taken;
     }
 
     // Blocking pop. Returns when an item is available.

include/kpn/pool_node.hpp

@@ -570,7 +570,14 @@ private:
         // 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));
+            // A refused sentinel is a protocol error, not backpressure, so it
+            // is reported rather than parked and retried — retrying would spin
+            // forever against a slot only the consumer can free, and there is
+            // no correct value to deliver second anyway. Closed is normal
+            // during teardown and stays quiet.
+            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
+                    ::SentinelResult::SlotBusy)
+                fire_callbacks(event_callbacks_);
             return true;
         }
         // Backpressure without parking the worker. A full channel means the
@@ -1087,7 +1094,14 @@ private:
         // 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));
+            // A refused sentinel is a protocol error, not backpressure, so it
+            // is reported rather than parked and retried — retrying would spin
+            // forever against a slot only the consumer can free, and there is
+            // no correct value to deliver second anyway. Closed is normal
+            // during teardown and stays quiet.
+            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
+                    ::SentinelResult::SlotBusy)
+                fire_callbacks(event_callbacks_);
             return true;
         }
         // See the note on the typed overload above: park rather than block.

tests/test_channel.cpp

@@ -1,3 +1,4 @@
+#include <string>
 #include <catch2/catch_test_macros.hpp>
 #include <catch2/catch_approx.hpp>
 #include <kpn/channel.hpp>
@@ -238,3 +239,62 @@ TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty",
     REQUIRE(out == 99);
     REQUIRE_FALSE(ch.try_pop_now(out));   // nothing left
 }
+
+// Regression: the sentinel slot holds one token and refuses a second.
+//
+// push_sentinel used to write eof_value_ unconditionally. Offering a second
+// token before the first was taken therefore did two wrong things at once: it
+// lost the first silently — and a lost EOF wedges every downstream pop forever
+// — and it wrote the storage while the consumer could be moving the previous
+// value out of it. For the shared_ptr storage that non-trivial types use, that
+// is a torn refcount, not merely a stale read.
+//
+// Refusing is correct rather than queueing: two control tokens on one channel
+// means the stream ended twice, which is a caller protocol error. Coalescing
+// them would hide it, and there is no second value that could sensibly follow
+// the end of a stream.
+TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]") {
+    Channel<int> ch(4);
+
+    REQUIRE(ch.push_sentinel(1));
+    // Slot occupied: the first token is still undelivered.
+    REQUIRE_FALSE(ch.push_sentinel(2));
+
+    // The first survives intact — the overwrite is what used to lose it.
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));
+    CHECK(out == 1);
+
+    // And the slot is reusable once drained.
+    REQUIRE(ch.push_sentinel(3));
+    REQUIRE(ch.try_pop_now(out));
+    CHECK(out == 3);
+}
+
+TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") {
+    // Visibility matters more here than for a dropped value: the refusal means
+    // a control token went nowhere, and the only alternative to a counter is
+    // for it to vanish.
+    Channel<int> ch(4);
+    REQUIRE(ch.push_sentinel(1));
+    const auto before = ch.stats().drops.load();
+    REQUIRE_FALSE(ch.push_sentinel(2));
+    CHECK(ch.stats().drops.load() == before + 1);
+}
+
+TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") {
+    // The non-consuming form exists so a refused token is still the caller's to
+    // report. The consuming push_sentinel cannot offer that, since the value is
+    // already moved into its parameter.
+    Channel<std::string> ch(4);
+    std::string first = "eof-1", second = "eof-2";
+
+    REQUIRE(ch.try_push_sentinel(first) == Channel<std::string>::SentinelResult::Taken);
+    REQUIRE(ch.try_push_sentinel(second) == Channel<std::string>::SentinelResult::SlotBusy);
+    CHECK(second == "eof-2");          // not moved from
+
+    ch.disable();
+    std::string third = "eof-3";
+    CHECK(ch.try_push_sentinel(third) == Channel<std::string>::SentinelResult::Closed);
+    CHECK(third == "eof-3");
+}

tests/test_channel_stress.cpp

@@ -19,6 +19,7 @@
 // Channel<T> is SPSC: exactly one producer thread and one consumer thread per
 // channel. Every scenario below honours that contract.
 
+#include <string>
 #include <catch2/catch_test_macros.hpp>
 #include <atomic>
 #include <chrono>
@@ -279,3 +280,51 @@ TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)",
         REQUIRE(ch.approx_size() == 0);
     }
 }
+
+// Contended: a producer offering sentinels while the consumer takes them.
+//
+// The old push_sentinel wrote eof_value_ with no regard for whether the
+// consumer was reading it, so a second offer racing a take was a data race on
+// the storage — for the shared_ptr form used by non-trivial types, on the
+// refcount. Under TSan the old code reports it; the handshake added alongside
+// this test makes the producer's write conditional on observing the slot free,
+// which is what serialises the two.
+//
+// Payload is a std::string so the storage is the shared_ptr path rather than
+// the trivially-copyable one, and each token carries its own identity so a torn
+// value shows up as a mismatch rather than as a plausible-looking result.
+TEST_CASE("SPSC: offering sentinels concurrently with takes is race-free",
+          "[channel][stress][sentinel]") {
+    constexpr int kRounds = 20000;
+    Channel<std::string> ch(4);
+
+    std::atomic<int> taken{0};
+    std::atomic<bool> torn{false};
+    std::atomic<bool> done{false};
+
+    std::thread consumer([&] {
+        std::string out;
+        while (!done.load(std::memory_order_acquire) || ch.approx_size() > 0) {
+            if (ch.try_pop_now(out)) {
+                if (out.rfind("eof-", 0) != 0) torn.store(true, std::memory_order_relaxed);
+                taken.fetch_add(1, std::memory_order_relaxed);
+            }
+        }
+    });
+
+    int accepted = 0;
+    for (int i = 0; i < kRounds; ++i) {
+        std::string tok = "eof-" + std::to_string(i);
+        if (ch.try_push_sentinel(tok) == Channel<std::string>::SentinelResult::Taken)
+            ++accepted;
+    }
+    done.store(true, std::memory_order_release);
+    consumer.join();
+
+    INFO("accepted " << accepted << " taken " << taken.load());
+    CHECK_FALSE(torn.load(std::memory_order_relaxed));
+    // Every accepted token must be delivered: the slot is refused while full,
+    // so acceptance and delivery are one-to-one.
+    CHECK(taken.load(std::memory_order_relaxed) == accepted);
+    CHECK(accepted > 0);
+}