From 5511ab9a35f690be9ae0f154114f4fa12be8eb06 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 17 Jul 2026 00:10:12 -0700 Subject: [PATCH] fix(session-worker): release the slot claim when Thread.start itself fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If thread creation raised (thread exhaustion, MemoryError), the dispatcher had already claimed the worker slot under ws._lock — but the flag's only clearer is _runner's finally, on a thread that never started. The workstream then looked idle forever (no state change ever fired) while every subsequent dispatch took the reuse path into a queue no worker would drain, until an operator force-cancel. Roll the claim back under the lock (identity-guarded, like _runner's own clear, so a concurrent force-cancel's successor is never clobbered) and re-raise. Re-raise rather than return False: callers' crash paths — the deferred-send drain's per-iteration handler with its backoff — are shaped for exceptions, and a False would masquerade as queue-full backpressure and mislabel the wake gate's refusal log. worker_kind is left stale, as documented (every reader conjoins _worker_running). Affected every dispatch path: sends, wakes, retries, the deferred-send drain, and workstream init. --- CHANGELOG.md | 10 ++++++++ tests/test_session_worker.py | 39 ++++++++++++++++++++++++++++++++ turnstone/core/session_worker.py | 30 +++++++++++++++++++++++- 3 files changed, 78 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 46540321..2949399c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -175,6 +175,16 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. ### Fixed +- **A failed worker-thread spawn no longer wedges the workstream.** If + `Thread.start()` itself raised (thread exhaustion, out-of-memory), the + dispatcher had already claimed the worker slot but the flag's only + clearer lived in the never-started thread — the workstream looked idle + forever while every subsequent message queued behind a worker that + didn't exist, until an operator force-cancel. The claim is now rolled + back under the lock and the error propagates, so the workstream is + dispatchable again as soon as resources recover. Affected every + dispatch path (sends, wakes, retries, deferred-send drain, init). + - **Manual `/compact` from the web UI: no phantom user turn, no frozen server, cancellable.** A slash command typed into the web composer no longer renders as a user chat bubble (it echoes as a distinct command diff --git a/tests/test_session_worker.py b/tests/test_session_worker.py index 0a2a1bc0..5fc69309 100644 --- a/tests/test_session_worker.py +++ b/tests/test_session_worker.py @@ -146,6 +146,45 @@ def test_enqueue_unexpected_exception_returns_false_logged() -> None: assert ws._worker_running is True +def test_spawn_failure_releases_slot_and_reraises(monkeypatch) -> None: + """``Thread.start`` raising (thread exhaustion, MemoryError) must not + wedge the slot: the claim ``(worker_thread, _worker_running)`` taken + under the lock is rolled back and the exception propagates. Without + the rollback the flag's only clearer is a finally on a thread that + never started — the workstream looks idle forever (no state change + ever fired) while every dispatch takes the reuse path into a queue + no worker will drain, until an operator force-cancel.""" + import pytest + + session = _SendSession() + ws = _make_ws(session) + + class _ExhaustedThread(threading.Thread): + def start(self) -> None: + raise RuntimeError("can't start new thread") + + class _ThreadNS: + # Shim only what session_worker.send touches; patching the real + # threading module's Thread attribute would break every other + # test's spawns. + Thread = _ExhaustedThread + current_thread = staticmethod(threading.current_thread) + + monkeypatch.setattr(session_worker, "threading", _ThreadNS) + with pytest.raises(RuntimeError, match="can't start new thread"): + _send_message(ws, session, "doomed") + assert ws._worker_running is False + assert ws.worker_thread is None + assert session.send_calls == [] + + # The slot is genuinely reusable once resources recover. + monkeypatch.setattr(session_worker, "threading", threading) + assert _send_message(ws, session, "recovered") is True + ws.worker_thread.join(timeout=2.0) + assert session.send_calls == ["recovered"] + assert ws._worker_running is False + + def test_closed_workstream_refused_no_spawn() -> None: """Authoritative closed-check: ``close()`` sets ``_closed`` under ``ws._lock``, so a wake (or send) racing it must be refused HERE — diff --git a/turnstone/core/session_worker.py b/turnstone/core/session_worker.py index 89bb3ccb..3bc32ff1 100644 --- a/turnstone/core/session_worker.py +++ b/turnstone/core/session_worker.py @@ -145,6 +145,12 @@ def send( caller surfaces 429) or any other exception (logged). Falling through to spawn a second worker on a full queue would corrupt ChatSession state. + + Raises: + Whatever ``Thread.start()`` raised when the spawn itself fails + (thread exhaustion, ``MemoryError``) — the slot claim is rolled + back first, so the workstream stays dispatchable once resources + recover instead of wedging behind a flag no thread will clear. """ name = thread_name or f"session-worker-{ws.id[:8]}" @@ -233,5 +239,27 @@ def send( # ``t.start()`` may run user code (worker body) before returning; # keep it outside the lock to avoid pinning ``ws._lock`` for the # full thread-creation cost. - t.start() + try: + t.start() + except Exception: + # Thread creation failed (RuntimeError under thread exhaustion, + # MemoryError): the slot was claimed under the lock above, but the + # flag's normal clearer is ``_runner``'s finally — on a thread + # that will never run. Without this release, ``_worker_running`` + # stays True forever on a workstream that LOOKS idle (the worker + # never fired a state change), every future dispatch takes the + # reuse path into a queue with no consumer, and only an operator + # force-cancel unwedges it. Identity-guarded like ``_runner``'s + # clear: a concurrent force-cancel may already have cleared or + # replaced the slot, and this must not clobber a live successor. + # Re-raise rather than return False: callers' crash paths (the + # deferred-send drain's per-iteration handler with its backoff) + # are shaped for exceptions, and a False here would masquerade as + # queue-full backpressure. + with ws._lock: + if ws.worker_thread is t: + ws.worker_thread = None + ws._worker_running = False + log.exception("session_worker.spawn_failed ws=%s — slot released", ws.id[:8]) + raise return True