fix: submitting to a stopped pool must be refused, not fatal

ThreadPool::stop() ends with queues_.clear(), and submit() went straight to
queues_[target] with no check. A submission arriving after stop indexed an
empty vector and segfaulted.

This is not a contrived teardown ordering. A node's space callback fires from
whichever thread drained the channel, and that thread belongs to the
*consumer*; the callback it runs belongs to the *producer*. Stop the producer
first — which a sources-first shutdown does by design — and the consumer keeps
draining its backlog, firing the producer's space callback into a pool that
has already been torn down:

    ThreadPool::submit
      <- source node's space_callback
      <- Channel<int>::try_pop_now      (relay draining its input)
      <- relay fire_once

The static-network shutdown case in the next commit crashed about 12 runs in
20 on this. It survived until now because halt() stops in reverse topological
order — consumers first — so the producer whose callback might fire is always
still alive. shutdown() stops sources first and does not have that protection.

Reading stopped_ without a lock would not fix it: the window between the read
and the indexing is exactly where clear() runs. submit() takes a shared lock
and stop() an exclusive one, so submissions still proceed in parallel with
each other while being serialised against teardown. stop() sets the flag under
the lock, releases it to join — a worker's task may itself call submit, and
holding the lock across the join would deadlock against that — then retakes it
to destroy the queues.

Refusals are counted rather than silent. A teardown race is expected, but a
node repeatedly trying to run after its pool is gone is worth being able to
see. try_submit also checks stop_flag_ first, so a stopped node cannot claim
the submit gate and leave it held.

Not a smart-pointer problem, for anyone reading the crash: nothing here is
owned by a raw pointer. It is std::vector::operator[] on a vector that was
emptied by another thread.

Verified in both directions: with the guard removed the new scheduler cases
segfault; with it they pass.

include/kpn/pool_node.hpp

@@ -364,6 +364,12 @@ private:
     /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
     /// variable, so the two cannot both be true — see submit_gate.hpp.
     void try_submit(float priority) {
+        // A stopped node must not claim the gate. The scheduler now refuses
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
         if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }
@@ -902,6 +908,12 @@ private:
     /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
     /// variable, so the two cannot both be true — see submit_gate.hpp.
     void try_submit(float priority) {
+        // A stopped node must not claim the gate. The scheduler now refuses
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
         if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }

include/kpn/scheduler.hpp

@@ -5,6 +5,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <shared_mutex>
 #include <optional>
 #include <queue>
 #include <thread>
@@ -62,7 +63,12 @@ public:
     }
 
     void stop() override {
+        // Close the pool to new work before touching anything, and do it under
+        // the lifecycle lock so no submit() is midway through indexing queues_.
+        {
+            std::unique_lock lk(lifecycle_mx_);
             stopped_.store(true, std::memory_order_seq_cst);
+        }
         for (auto& q : queues_) {
             std::lock_guard lock(q->mx);
             std::size_t discarded = q->pq.size();
@@ -73,7 +79,13 @@ public:
         // gap between a worker's predicate check and its wait() (see submit()).
         { std::lock_guard<std::mutex> lk(cv_mx_); }
         cv_.notify_all();
+        // Join without the lock: a worker's task may call submit(), which takes
+        // it shared, and holding it here would deadlock against that.
         for (auto& t : workers_) if (t.joinable()) t.join();
+        // Destroying the queues is what submit() must never race. By now
+        // stopped_ is published, so any submit() that acquires the lock after
+        // this point returns without touching them.
+        std::unique_lock lk(lifecycle_mx_);
         workers_.clear();
         queues_.clear();
     }
@@ -86,6 +98,22 @@ public:
     }
 
     void submit(std::function<void()> task, float priority = 0.5f) override {
+        // A submission can arrive after this pool has been stopped, and did so
+        // by an ordinary route: a node's space callback fires from whichever
+        // thread drained the channel, which belongs to the *consumer*. Stop the
+        // producer first — as a sources-first shutdown does — and the consumer
+        // keeps draining its backlog, firing the producer's space callback into
+        // a pool whose stop() has already run queues_.clear(). submit() then
+        // indexed an empty vector: a segfault, reproducible about 12 runs in 20.
+        //
+        // The shared lock is what makes the check meaningful. Reading stopped_
+        // alone leaves the window between the read and the indexing, which is
+        // precisely where stop() clears the vector.
+        std::shared_lock lk(lifecycle_mx_);
+        if (stopped_.load(std::memory_order_acquire) || queues_.empty()) {
+            rejected_.fetch_add(1, std::memory_order_relaxed);
+            return;
+        }
         std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
         {
             std::lock_guard lock(queues_[target]->mx);
@@ -105,6 +133,9 @@ public:
 
     std::size_t thread_count() const { return thread_count_; }
 
+    /// Submissions dropped because the pool was stopped. See rejected_.
+    uint64_t rejected() const { return rejected_.load(std::memory_order_relaxed); }
+
     // ── IPoolProbe ────────────────────────────────────────────────────────────
 
     PoolSnapshot snapshot(const std::string& name) const override {
@@ -194,6 +225,11 @@ private:
     std::vector<std::unique_ptr<WorkerQueue>>  queues_;
     std::vector<std::thread>                   workers_;
 
+    /// Guards the lifetime of queues_/workers_ against a concurrent submit().
+    /// Shared by submit, exclusive by stop, so submissions still run in
+    /// parallel with each other.
+    mutable std::shared_mutex lifecycle_mx_;
+
     std::mutex              cv_mx_;
     std::condition_variable cv_;
     std::mutex              drain_mx_;
@@ -205,6 +241,10 @@ private:
     std::atomic<size_t>   next_{0};       // round-robin submit cursor
     std::atomic<uint64_t> seq_{0};        // tie-break for equal-priority tasks
     std::atomic<uint64_t> submitted_{0};
+    /// Submissions refused because the pool was already stopped. Not an error —
+    /// teardown races are expected — but silence here would hide a node that
+    /// keeps trying to run after its pool is gone.
+    std::atomic<uint64_t> rejected_{0};
     std::atomic<uint64_t> completed_{0};
 };
 

tests/test_scheduler.cpp

@@ -227,3 +227,60 @@ TEST_CASE("work stealing: tasks complete with more threads than initial queue ta
     REQUIRE(counter.load() == 4);
     pool.stop();
 }
+
+// Regression: submitting to a stopped pool must be a no-op, not a segfault.
+//
+// stop() ends with queues_.clear(), and submit() went straight to
+// queues_[target] with no check — so a submission arriving after stop indexed
+// an empty vector.
+//
+// This is not a contrived teardown ordering; it happens on a normal path. A
+// node's space callback fires from whichever thread drained the channel, and
+// that thread belongs to the *consumer*. Stop the producer first — which a
+// sources-first shutdown does by design — and the consumer keeps draining its
+// backlog, firing the producer's space callback into a pool that has already
+// been torn down. Before this fix the static-network shutdown case crashed
+// about 12 runs in 20.
+//
+// Checking stopped_ without the lock would not be enough: the window between
+// reading the flag and indexing the vector is exactly where clear() runs.
+TEST_CASE("submitting to a stopped pool is refused, not fatal", "[scheduler]") {
+    ThreadPool pool(2);
+    pool.start();
+    pool.stop();
+
+    std::atomic<int> ran{0};
+    for (int i = 0; i < 10; ++i)
+        pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
+
+    CHECK(ran.load(std::memory_order_relaxed) == 0);
+    CHECK(pool.rejected() == 10);
+}
+
+TEST_CASE("submitting while the pool stops does not crash", "[scheduler]") {
+    // The racing form of the case above: a producer thread submitting
+    // continuously while stop() runs underneath it. Nothing is asserted about
+    // how many tasks run — the point is that every submission either enqueues
+    // or is refused, and none touches a destroyed queue.
+    for (int rep = 0; rep < 20; ++rep) {
+        ThreadPool pool(4);
+        pool.start();
+
+        std::atomic<bool> go{false};
+        std::atomic<int>  ran{0};
+        std::thread submitter([&] {
+            while (!go.load(std::memory_order_acquire)) {}
+            for (int i = 0; i < 2000; ++i)
+                pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
+        });
+
+        go.store(true, std::memory_order_release);
+        std::this_thread::sleep_for(std::chrono::microseconds(200));
+        pool.stop();
+        submitter.join();
+
+        // Everything submitted was either executed or refused; nothing vanished
+        // into a queue that no longer existed.
+        CHECK(pool.rejected() + pool.snapshot("p").tasks_completed <= 2000);
+    }
+}