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).
-    bool push_sentinel(T value) {
+    //
+    // The slot holds exactly one undelivered token. A second offered before the
+    // 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);
+}