Address Copilot review: leak fix + group drain + doc cleanup

* TaskQueue/MPSCQueue::link_next_block: hold the freshly allocated Block in
  a std::unique_ptr so a lost CAS race no longer leaks the block.
* Group: drain the wait_buckets and call unregister_external_waiter for
  every parked entry in ~Group so the Pool::external_waiters counter is
  balanced on Scheduler teardown (per declaration order Pools outlive
  Groups, so the raw Pool* pointers in WaitEntry remain valid).
* Group.hpp: drop the orphan try_submit doc block that was sitting above
  try_acquire_running_lock and add the rule-of-five deletes that come with
  declaring an explicit destructor.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: TrentHouliston

src/threading/scheduler/Group.cpp

@@ -151,6 +151,27 @@ namespace threading {
         Group::Group(std::shared_ptr<const util::GroupDescriptor> descriptor)
             : descriptor(std::move(descriptor)), tokens(this->descriptor->concurrency) {}
 
+        Group::~Group() {
+            // Drain any waiters still parked in the fast-path buckets so the external_waiters
+            // counter on each Pool is balanced back to zero. If we let the wait_buckets just go
+            // out of scope, the WaitEntry destructors would silently drop the tasks but never
+            // call unregister_external_waiter, and the matching Pool worker would loop forever
+            // in get_task() waiting for waiters that no longer exist.
+            //
+            // Per the Scheduler field declaration order (`pools` declared before `groups`),
+            // Groups are destroyed before Pools, so every WaitEntry::pool pointer is still
+            // valid here.
+            WaitEntry entry;
+            for (auto& bucket : wait_buckets) {
+                while (bucket.try_dequeue(entry)) {
+                    if (entry.pool != nullptr) {
+                        entry.pool->unregister_external_waiter();
+                    }
+                    entry = WaitEntry{};
+                }
+            }
+        }
+
         std::unique_ptr<Lock> Group::try_acquire_running_lock() {
             if (slow_pending.load(std::memory_order_acquire) > 0) {
                 return nullptr;

src/threading/scheduler/Group.hpp

@@ -179,17 +179,18 @@ namespace threading {
             explicit Group(std::shared_ptr<const util::GroupDescriptor> descriptor);
 
             /**
-             * Try to submit a task through the lock-free fast path.
-             *
-             * If a group token is available the task is submitted to the pool immediately.
-             * Otherwise the task is queued until a token is released.
-             *
-             * @param task       the reaction task to submit
-             * @param pool       the pool to submit to when runnable
-             * @param clear_idle if true, clear idle state on submission
-             *
-             * @return true if the task was submitted immediately
+             * Destroy the Group object. Drains any parked waiters in the fast-path buckets so the
+             * `external_waiters` counter on every Pool referenced by a queued WaitEntry is balanced
+             * back to zero; otherwise a Pool worker could spin forever in `get_task()` waiting for
+             * waiters that will never be drained.
              */
+            ~Group();
+
+            Group(const Group&)            = delete;
+            Group(Group&&)                 = delete;
+            Group& operator=(const Group&) = delete;
+            Group& operator=(Group&&)      = delete;
+
             /**
              * Try to acquire a token for inline execution without submitting to a pool.
              *

src/threading/scheduler/queue/MPSCQueue.hpp

@@ -26,6 +26,7 @@
 #include <array>
 #include <atomic>
 #include <cstddef>
+#include <memory>
 #include <new>
 #include <thread>
 #include <type_traits>
@@ -102,10 +103,17 @@ namespace threading {
                 }
 
                 bool link_next_block(Block* block) {
+                    // Hold the new block in a unique_ptr so that if the CAS fails (another producer
+                    // linked the next block first) we don't leak the freshly allocated Block.
+                    // Function arguments are unconditionally evaluated in C++, so the previous form
+                    // `compare_exchange_strong(expected, allocate_block(), ...)` leaked one Block per
+                    // contended overflow.
                     Block* expected = nullptr;
+                    std::unique_ptr<Block> candidate(allocate_block());
                     if (block->next.compare_exchange_strong(expected,
-                                                            allocate_block(),
+                                                            candidate.get(),
                                                             std::memory_order_acq_rel)) {
+                        candidate.release();
                         return true;
                     }
                     return expected != nullptr;

src/threading/scheduler/queue/TaskQueue.hpp

@@ -26,6 +26,7 @@
 #include <array>
 #include <atomic>
 #include <cstddef>
+#include <memory>
 #include <new>
 #include <thread>
 #include <type_traits>
@@ -99,8 +100,17 @@ namespace threading {
                 }
 
                 bool link_next_block(Block* block) {
+                    // Hold the new block in a unique_ptr so that if the CAS fails (another producer
+                    // linked the next block first) we don't leak the freshly allocated Block.
+                    // Function arguments are unconditionally evaluated in C++, so the previous form
+                    // `compare_exchange_strong(expected, allocate_block(), ...)` leaked one Block per
+                    // contended overflow.
                     Block* expected = nullptr;
-                    if (block->next.compare_exchange_strong(expected, allocate_block(), std::memory_order_acq_rel)) {
+                    std::unique_ptr<Block> candidate(allocate_block());
+                    if (block->next.compare_exchange_strong(expected,
+                                                            candidate.get(),
+                                                            std::memory_order_acq_rel)) {
+                        candidate.release();
                         return true;
                     }
                     return expected != nullptr;