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