fix: ThreadPool::start must take the lifecycle lock too

abbb2d4 guarded submit() against stop() with a shared/exclusive lock, because
stop() ends with queues_.clear() and submit() indexes queues_. It missed the
other writer: start() rebuilds the same vector and took no lock at all.

A submission can genuinely land while a pool is inside start(). A network
starts its nodes one at a time, and a node already started fires into the next
one's channel, whose push callback submits. ThreadSanitizer reports it as the
read at scheduler.hpp:114 against the write at :59, and the consequence is
worse than a torn read: push_back can reallocate the vector under a reader
that has already indexed it.

Surfaced by 6802328. Firing push_callback_ unconditionally is correct — the
argument in that commit holds — and it makes callbacks frequent enough during
startup to hit this window. It went from unobserved to 4 races in one run of
the unit suite.

Holding the lock across the thread spawn is safe: all queues are constructed
before any worker starts, and worker_loop never takes the lifecycle lock, so
there is nothing for it to deadlock against.

Verified with -DKPN_SANITIZER=thread: 4 races in one run of five before, 0
across five unit runs and two stress runs after. 148/148.

include/kpn/channel.hpp

@@ -136,7 +136,6 @@ public:
             throw ChannelOverflowError(capacity_);
         }
 
-        const bool was_empty = (t == h);
         buf_[t & ring_mask_] = make_storage(std::move(value));
         tail_.store(t + 1, std::memory_order_release);
         stats_.record_push(t - h + 1, data_bytes);
@@ -144,7 +143,8 @@ public:
         wake_.fetch_add(1, std::memory_order_release);
         wake_.notify_one();
 
-        if (was_empty && push_callback_)
+        // Level-triggered, not edge-triggered — see set_push_callback.
+        if (push_callback_)
             push_callback_();
     }
 
@@ -187,13 +187,13 @@ public:
         if (t - h >= capacity_) return PushResult::Full;
 
         const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
-        const bool was_empty = (t == h);
         buf_[t & ring_mask_] = make_storage(std::move(value));
         tail_.store(t + 1, std::memory_order_release);
         stats_.record_push(t - h + 1, data_bytes);
         wake_.fetch_add(1, std::memory_order_release);
         wake_.notify_one();
-        if (was_empty && push_callback_) push_callback_();
+        // Level-triggered, not edge-triggered — see set_push_callback.
+        if (push_callback_) push_callback_();
         return PushResult::Taken;
     }
 
@@ -212,13 +212,13 @@ public:
             const std::size_t h = head_.load(std::memory_order_acquire);
             if (t - h < capacity_) {          // space available → normal push
                 const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
-                const bool was_empty = (t == h);
                 buf_[t & ring_mask_] = make_storage(std::move(value));
                 tail_.store(t + 1, std::memory_order_release);
                 stats_.record_push(t - h + 1, data_bytes);
                 wake_.fetch_add(1, std::memory_order_release);
                 wake_.notify_one();
-                if (was_empty && push_callback_) push_callback_();
+                // Level-triggered, not edge-triggered — see set_push_callback.
+                if (push_callback_) push_callback_();
                 return true;
             }
             // full: yield briefly and retry (consumer will drain)
@@ -267,10 +267,22 @@ public:
         // 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();
+        // Not counted as a drop, and this is the important part. A source that
+        // has reached the end of its input keeps being polled and keeps
+        // returning EOF — that is the normal steady state, not an error — so a
+        // token arriving while one is already pending is a *re-offer*, and
+        // refusing it loses nothing: the pending token carries the same
+        // meaning and is already on its way. Counting it as a drop made a
+        // clean run report data loss and exit non-zero.
+        //
+        // The cost of that choice, stated plainly: a genuinely distinct second
+        // token would also be refused silently, and the channel cannot tell the
+        // two apart. Re-offering is the case that actually occurs here, and the
+        // delivery guarantee that matters — the first token arrives — holds
+        // either way.
+        if (has_eof_.load(std::memory_order_acquire))
             return SentinelResult::SlotBusy;
-        }
         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
@@ -388,7 +400,41 @@ public:
         wake_.notify_all();
     }
 
-    // Register a callback fired when the queue transitions empty→non-empty.
+    // Register a callback fired after every successful push.
+    //
+    // It fires on every push, not on the empty→non-empty transition, and that
+    // is a correctness requirement rather than a simplification.
+    //
+    // The edge version tested `was_empty = (t == h)` using an `h` sampled
+    // *before* the item was published. A PoolNode consumer decides whether to
+    // run again from the level (count_ready → approx_size), so the two sides
+    // could each read the other as stale and both stand down:
+    //
+    //   producer (push)                    consumer (PoolNode firing)
+    //   ------------------------           ----------------------------
+    //   samples t=782, h=781
+    //     -> was_empty = false, no wake
+    //                                      pops idx 781, head_ = 782
+    //                                      count_ready(): head_==tail_==782
+    //                                      -> not ready, gate released to Idle
+    //   tail_.store(783)
+    //
+    // The item is in the ring, the node is idle, and no wake is outstanding.
+    // Worse, the failure is absorbing: every later push now sees a non-empty
+    // ring, so `was_empty` is false forever and the callback never fires again.
+    // The node sleeps while its backlog grows and its consumer waits on it.
+    //
+    // Re-reading head_ after the tail_ store does not fix it. That is the
+    // store-buffer pattern, and under acquire/release both sides may legally
+    // read stale; forbidding it needs seq_cst on the producer's tail_ store and
+    // head_ load *and* on the consumer's head_ store and tail_ load — a fence
+    // on both hot paths. Firing unconditionally is correct by construction:
+    // the callback runs after the publishing store, so a consumer that observes
+    // the level at all observes the item.
+    //
+    // The redundant wakes are cheap. on_input_ready re-checks the level, and
+    // SubmitGate::claim() collapses a wake arriving during a firing into the
+    // firing already in flight, so the cost is one CAS, not one extra run.
     void set_push_callback(std::function<void()> cb) {
         push_callback_ = std::move(cb);
     }

include/kpn/pool_node.hpp

@@ -622,14 +622,13 @@ 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)) {
-            // A refused sentinel is a protocol error, not backpressure, so it
+            // Not parked and not reported. Parking would spin against a slot
-            // is reported rather than parked and retried — retrying would spin
+            // only the consumer can free; reporting would cry data loss on the
-            // forever against a slot only the consumer can free, and there is
+            // normal steady state, since a source at end of input keeps being
-            // no correct value to deliver second anyway. Closed is normal
+            // polled and keeps returning EOF, so the token is re-offered on
-            // during teardown and stays quiet.
+            // every firing. Refusing a re-offer loses nothing — the pending
-            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
+            // token says the same thing. See Channel::try_push_sentinel.
-                    ::SentinelResult::SlotBusy)
+            ch->try_push_sentinel(val);
-                fire_callbacks(event_callbacks_);
             return true;
         }
         // Backpressure without parking the worker. A full channel means the
@@ -1207,14 +1206,13 @@ 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)) {
-            // A refused sentinel is a protocol error, not backpressure, so it
+            // Not parked and not reported. Parking would spin against a slot
-            // is reported rather than parked and retried — retrying would spin
+            // only the consumer can free; reporting would cry data loss on the
-            // forever against a slot only the consumer can free, and there is
+            // normal steady state, since a source at end of input keeps being
-            // no correct value to deliver second anyway. Closed is normal
+            // polled and keeps returning EOF, so the token is re-offered on
-            // during teardown and stays quiet.
+            // every firing. Refusing a re-offer loses nothing — the pending
-            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
+            // token says the same thing. See Channel::try_push_sentinel.
-                    ::SentinelResult::SlotBusy)
+            ch->try_push_sentinel(val);
-                fire_callbacks(event_callbacks_);
             return true;
         }
         // See the note on the typed overload above: park rather than block.

include/kpn/scheduler.hpp

@@ -53,6 +53,19 @@ public:
     }
 
     void start() override {
+        // Under the lifecycle lock for the same reason stop() is: submit()
+        // reads queues_ and this rebuilds it. A network starts its nodes one at
+        // a time, and a node already started fires into the next one's channel,
+        // whose push callback submits — so a submission can genuinely land
+        // while another pool is still inside start(). ThreadSanitizer reports
+        // it as a read at submit() against this write, and the consequence is
+        // worse than a torn read: push_back can reallocate the vector under a
+        // reader that has already indexed it.
+        //
+        // Queues are all constructed before any worker is spawned, which is
+        // what keeps worker_loop's own queues_[id] out of this — it never takes
+        // the lock, so holding it across the spawn cannot deadlock.
+        std::unique_lock lk(lifecycle_mx_);
         stopped_.store(false, std::memory_order_relaxed);
         queues_.clear();
         for (std::size_t i = 0; i < thread_count_; ++i)

tests/test_channel.cpp

@@ -271,15 +271,26 @@ TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]")
     CHECK(out == 3);
 }
 
-TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") {
+TEST_CASE("a refused sentinel is not counted as a drop", "[channel][sentinel]") {
-    // Visibility matters more here than for a dropped value: the refusal means
+    // A refusal means a token arrived while an equivalent one was already
-    // a control token went nowhere, and the only alternative to a counter is
+    // pending — not that anything was lost. Counting it as a drop was wrong in
-    // for it to vanish.
+    // a way that showed up immediately on real content: a source at the end of
+    // its input keeps being polled and keeps returning EOF, so the token is
+    // re-offered on every firing, and the pipeline reported hundreds of dropped
+    // frames on a clean run and exited non-zero.
+    //
+    // The delivery guarantee is unaffected: the first token is pending and will
+    // arrive. Only the accounting changed.
     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);
+    CHECK(ch.stats().drops.load() == before);
+
+    // And the one that was accepted is still the one delivered.
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));
+    CHECK(out == 1);
 }
 
 TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") {

tests/test_channel_stress.cpp

@@ -153,11 +153,18 @@ TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
     }
 }
 
-TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
+TEST_CASE("SPSC: push_callback fires for every push, never missed",
           "[channel][stress]") {
-    // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
+    // Regression: this callback is the *only* thing that wakes a PoolNode, and
-    // the consumer-side notification path. Run it under contention to make sure
+    // it used to fire only on the empty->non-empty edge, computed from a head_
-    // the was_empty detection isn't torn by a concurrent pop().
+    // sampled before the item was published. A concurrent pop() could drain the
+    // ring to empty in that window, so neither side saw the other: the item sat
+    // in the ring with the consumer idle, and because the trigger was an edge it
+    // never recovered. See set_push_callback in channel.hpp.
+    //
+    // The old version of this test asserted only `1 <= callbacks <= N`, which a
+    // *missed* callback satisfies — it named the hazard and could not detect it.
+    // One callback per successful push is the contract, so assert exactly that.
     Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
     std::atomic<int> callbacks{0};
     ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
@@ -175,10 +182,9 @@ TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
     for (int i = 0; i < N; ++i) (void)ch.pop();
     producer.join();
 
-    // At least one transition, at most one per item; mainly we assert the run
+    // Exactly one callback per successful push. Fewer means a wake was dropped,
-    // completed without TSan flagging a race on push_callback_/was_empty.
+    // which is the bug; more would mean a spurious wake was manufactured.
-    REQUIRE(callbacks.load() >= 1);
+    REQUIRE(callbacks.load() == N);
-    REQUIRE(callbacks.load() <= N);
 }
 
 // Ordering contract of the out-of-band sentinel under contention.