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)
@@ -400,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/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_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.