fix(worker): deterministic pool teardown; EditorState::drop signals workers

Every EditorState leaked its entire worker pool (cores-1 threads, each
waking every 100ms): the Rc<AsyncRuntime> is cloned into dozens of Lua
closures, and registries those closures capture store mlua::Function
values -- reference cycles through the Lua VM that keep the Rc from
ever reaching zero, so WorkerPool::Drop never ran. Harmless for one
editor per process; in the m4 acceptance suite (54 editor-building
tests) it accumulated 1000+ live threads (observed: 60 complete
pmacs-worker-0..14 pools at once) plus ~9k spurious wakeups/second.

Fix: WorkerPool::signal_shutdown() -- set the shutdown flag + wake the
parkers from a shared reference; parked workers exit within their
100ms park timeout. Exposed as AsyncRuntime::shutdown_workers(), called
from a new impl Drop for EditorState. Deliberately signal-only, NO
join on the drop path: a worker can be blocked publishing its reply
onto the bus only the main thread drains, and a first cut that joined
in Drop deadlocked the m4 suite at teardown (fake-LSP tests wedged
2h+). WorkerPool::shutdown() (signal + join) remains for owners with
no bus consumer to deadlock against; Drop delegates to it as before.

Measured on the m4 suite: peak live threads 1026 -> 122.
Regression: tests/worker_shutdown_acceptance.rs (thread-count probe is
/proc-based, Linux-gated; the idempotence test runs everywhere).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/async_runtime.rs

@@ -602,6 +602,20 @@ impl AsyncRuntime {
         Self::with_pool(WorkerPool::new(size))
     }
 
+    /// Signal the worker-pool threads to exit (see
+    /// [`crate::worker::WorkerPool::signal_shutdown`]). Idempotent,
+    /// non-blocking. Called from [`crate::editor::EditorState`]'s
+    /// `Drop`: the runtime's `Rc` is captured into Lua-VM reference
+    /// cycles and never reaches a zero refcount, so without this
+    /// explicit teardown every editor instance leaks its whole pool
+    /// --- one leaked pool per test in a suite that builds real
+    /// editors. Signal-only (no join): a worker blocked handing its
+    /// reply to the main thread must not deadlock the main thread's
+    /// drop.
+    pub fn shutdown_workers(&self) {
+        self.pool.signal_shutdown();
+    }
+
     /// Returns the number of in-flight or settled-but-not-yet-taken
     /// pending entries.
     #[must_use]

src/editor.rs

@@ -102,6 +102,32 @@ pub struct EditorState {
     mouse_click: Option<MouseClickState>,
 }
 
+impl Drop for EditorState {
+    /// Tear down the worker-pool threads.
+    ///
+    /// The `Rc<AsyncRuntime>` is cloned into dozens of Lua closures
+    /// (`pmacs.workers`, LSP request wrappers, ...), and several of
+    /// the registries those closures capture themselves store
+    /// `mlua::Function` values --- reference cycles through the Lua
+    /// VM that keep the `Rc` from ever reaching zero. Harmless for a
+    /// single editor per process (the OS reclaims at exit), but a
+    /// test binary that builds one `EditorState` per test would leak
+    /// one full worker pool (`cores - 1` threads, each waking every
+    /// 100ms) per test --- observed as 1000+ live threads in the m4
+    /// acceptance suite. Dropping the editor reaches the pool through
+    /// its own `Rc` clone and signals the threads down regardless of
+    /// the cycle; parked workers exit within their 100ms wakeup.
+    ///
+    /// Signal-only, NO join: a worker can be blocked publishing its
+    /// reply onto the bus that this (main) thread drains --- joining
+    /// here deadlocked the m4 suite at teardown for hours. A worker
+    /// stuck mid-handoff stays alive (bounded by its job), which is
+    /// still a ~15x improvement over leaking every pool whole.
+    fn drop(&mut self) {
+        self.async_runtime.shutdown_workers();
+    }
+}
+
 #[derive(Copy, Clone)]
 struct MouseClickState {
     frontend_id: FrontendId,

src/worker.rs

@@ -175,7 +175,17 @@ impl PoolShared {
 /// thread).
 pub struct WorkerPool {
     shared: Arc<PoolShared>,
-    workers: Vec<JoinHandle<()>>,
+    /// Join handles, drained exactly once by [`Self::shutdown`]
+    /// (directly or via `Drop`). Behind a `Mutex` so shutdown works
+    /// from a shared reference: the pool's owner is typically an
+    /// `Rc<AsyncRuntime>` cloned into Lua closures, and those clones
+    /// form VM reference cycles that keep the `Rc` from ever
+    /// reaching zero --- an embedder that merely *drops* its handle
+    /// would leak every worker thread. `EditorState::drop` calls
+    /// `shutdown()` explicitly instead.
+    workers: Mutex<Vec<JoinHandle<()>>>,
+    /// Thread count at construction (stable across shutdown).
+    size: usize,
 }
 
 impl WorkerPool {
@@ -194,7 +204,7 @@ impl WorkerPool {
             next_id: AtomicU64::new(0),
             parker: (Mutex::new(()), Condvar::new()),
         });
-        let workers = local_queues
+        let workers: Vec<JoinHandle<()>> = local_queues
             .into_iter()
             .enumerate()
             .map(|(idx, local)| {
@@ -205,7 +215,11 @@ impl WorkerPool {
                     .expect("spawn worker thread")
             })
             .collect();
-        Self { shared, workers }
+        Self {
+            shared,
+            size,
+            workers: Mutex::new(workers),
+        }
     }
 
     /// Build a pool sized at `available_parallelism - 1`, with a
@@ -218,10 +232,48 @@ impl WorkerPool {
         Self::new(cores.saturating_sub(1))
     }
 
-    /// Number of worker threads owned by this pool.
+    /// Number of worker threads this pool was built with.
     #[must_use]
     pub fn size(&self) -> usize {
-        self.workers.len()
+        self.size
+    }
+
+    /// Signal every worker to exit, without joining. Idle (parked)
+    /// workers observe the flag within their 100ms park timeout and
+    /// return; a worker mid-job exits when its job finishes. Queued
+    /// jobs that haven't been picked up are dropped without running;
+    /// jobs dispatched *after* the signal are never picked up.
+    ///
+    /// Exists as an explicit method (not just `Drop`) because the
+    /// pool's owning `Rc<AsyncRuntime>` is captured into Lua-VM
+    /// reference cycles and may never be reclaimed --- callers that
+    /// know the editor is going away (`EditorState::drop`) signal the
+    /// threads down regardless.
+    ///
+    /// Deliberately does NOT join: a worker can be blocked publishing
+    /// its reply onto the message bus that only the *main thread*
+    /// drains, so a main-thread join here is a deadlock (observed as
+    /// the m4 acceptance suite wedging for hours at teardown). Callers
+    /// that own the whole world and want the join use
+    /// [`Self::shutdown`] (or just drop the pool).
+    pub fn signal_shutdown(&self) {
+        self.shared.shutdown.store(true, Ordering::Release);
+        self.shared.notify_all();
+    }
+
+    /// [`Self::signal_shutdown`] plus a join of every worker thread.
+    /// Idempotent: the second call finds no handles and returns
+    /// immediately. Only safe where no worker can be blocked on the
+    /// caller's own thread (see `signal_shutdown`); `Drop` uses it
+    /// because a pool being dropped has no live bus consumer to
+    /// deadlock against in the bare-pool case.
+    pub fn shutdown(&self) {
+        self.signal_shutdown();
+        let handles: Vec<JoinHandle<()>> =
+            std::mem::take(&mut *self.workers.lock().expect("worker pool mutex poisoned"));
+        for handle in handles {
+            let _ = handle.join();
+        }
     }
 
     /// Submit `work` to be run on a worker. Returns a [`JobHandle`]
@@ -263,18 +315,11 @@ impl WorkerPool {
 }
 
 impl Drop for WorkerPool {
-    /// Shutdown semantics: dropping the pool signals every worker
-    /// to exit at its next idle wakeup and joins them. Running
-    /// jobs run to completion (or to their own cancellation
-    /// check); queued jobs that haven't been picked up are dropped
-    /// without running. Tests and callers that want explicit
-    /// shutdown just `drop(pool)`.
+    /// Dropping the pool is an implicit [`Self::shutdown`]: every
+    /// worker is signalled to exit at its next idle wakeup and
+    /// joined. No-op when `shutdown` already ran.
     fn drop(&mut self) {
-        self.shared.shutdown.store(true, Ordering::Release);
-        self.shared.notify_all();
-        for handle in self.workers.drain(..) {
-            let _ = handle.join();
-        }
+        self.shutdown();
     }
 }
 

tests/completion_popup_acceptance.rs

@@ -3,7 +3,7 @@
 //! the Q#C3 partial dispatcher shadow, and Q#C7 validated accept, all
 //! driven through `dispatch_key` exactly as a terminal user would.
 //!
-//! The LSP provider's request path is covered by the m4_5 fake-LSP
+//! The LSP provider's request path is covered by the `m4_5` fake-LSP
 //! suite; these tests run hermetic (dabbrev + custom Lua providers)
 //! so no server binary is needed.
 //!

tests/worker_shutdown_acceptance.rs

@@ -0,0 +1,47 @@
+//! Worker-pool teardown regression: dropping an `EditorState` must
+//! release its worker threads even though the `Rc<AsyncRuntime>` is
+//! trapped in Lua-VM reference cycles and never reaches refcount
+//! zero (`EditorState::drop` → `AsyncRuntime::shutdown_workers`).
+//!
+//! Before the fix, every `EditorState` ever constructed leaked a
+//! full `cores - 1` worker pool: the m4 acceptance suite (54 editor-
+//! building tests) accumulated 1000+ live threads, each waking every
+//! 100ms.
+
+use pmacs::editor::EditorState;
+
+/// Thread-count probe via /proc; Linux-only (macOS CI skips --- the
+/// leak and the fix are platform-independent, the *probe* isn't).
+#[cfg(target_os = "linux")]
+fn live_threads() -> usize {
+    std::fs::read_dir("/proc/self/task").map_or(0, std::iter::Iterator::count)
+}
+
+#[cfg(target_os = "linux")]
+#[test]
+fn editor_state_drop_releases_worker_threads() {
+    let baseline = live_threads();
+    for _ in 0..3 {
+        let s = EditorState::new();
+        drop(s);
+    }
+    // Joins are synchronous in drop; the small sleep only covers
+    // detached per-process reaper threads finishing up.
+    std::thread::sleep(std::time::Duration::from_millis(300));
+    let after = live_threads();
+    assert!(
+        after <= baseline + 2,
+        "worker threads leak across EditorState drop: \
+         baseline {baseline}, after 3 create/drop cycles {after}"
+    );
+}
+
+/// Platform-independent variant: the pool reports itself dead after
+/// an explicit shutdown, and shutdown is idempotent.
+#[test]
+fn explicit_shutdown_is_idempotent() {
+    let s = EditorState::new();
+    s.async_runtime.shutdown_workers();
+    s.async_runtime.shutdown_workers(); // second call must not hang or panic
+    drop(s); // drop runs shutdown a third time
+}