diff --git a/src/java_codebase_rag/cli.py b/src/java_codebase_rag/cli.py index 54a4023..6d63e7d 100644 --- a/src/java_codebase_rag/cli.py +++ b/src/java_codebase_rag/cli.py @@ -1162,8 +1162,20 @@ def _console_script_main() -> None: racy teardown — the command has already done its work and emitted its result. ``main()`` stays return-based so in-process test callers (``cli.main(...)``) keep working. + + ``KeyboardInterrupt`` (Ctrl+C during a long indexing step) is caught here + rather than left to propagate: an uncaught interrupt bypasses this function + and runs full interpreter finalization (traceback + thread teardown), + whereas routing it through the same flush + ``os._exit`` path gives a clean, + immediate exit (code 130) and avoids the finalization-time SIGABRT noted + above for commands that loaded lancedb. """ - rc = main() + try: + rc = main() + except KeyboardInterrupt: + sys.stderr.write("\nInterrupted.\n") + sys.stderr.flush() + rc = 130 sys.stdout.flush() sys.stderr.flush() os._exit(rc) diff --git a/src/java_codebase_rag/jrag.py b/src/java_codebase_rag/jrag.py index bbb43ff..1b47528 100644 --- a/src/java_codebase_rag/jrag.py +++ b/src/java_codebase_rag/jrag.py @@ -4551,9 +4551,19 @@ def _console_script_main() -> None: (SIGABRT, exit -6). Flushing + ``os._exit`` skips that racy teardown - the command has already done its work and emitted its result. ``main()`` stays return-based so in-process test callers keep working. + + ``KeyboardInterrupt`` (Ctrl+C during a long indexing step) is caught here so + it routes through the same flush + ``os._exit`` path — clean, immediate exit + (code 130, no traceback) and no finalization-time SIGABRT — instead of + propagating past this function. """ force_utf8_stdio() - rc = main() + try: + rc = main() + except KeyboardInterrupt: + sys.stderr.write("\nInterrupted.\n") + sys.stderr.flush() + rc = 130 sys.stdout.flush() sys.stderr.flush() os._exit(rc) diff --git a/src/java_codebase_rag/pipeline.py b/src/java_codebase_rag/pipeline.py index 5601ce7..14468b5 100644 --- a/src/java_codebase_rag/pipeline.py +++ b/src/java_codebase_rag/pipeline.py @@ -133,14 +133,60 @@ def drain_err() -> None: t_err = threading.Thread(target=drain_err, name="stream-stderr", daemon=True) t_out.start() t_err.start() + # Wait on the CHILD before joining the drain threads. ``Popen.wait()`` is + # interruptible by Ctrl+C — the underlying ``os.waitpid`` returns EINTR and + # CPython raises ``KeyboardInterrupt`` — whereas ``Thread.join()`` blocks on + # an internal lock whose infinite-timeout acquire CPython never polls for + # signals. Joining *first* therefore made the whole indexing step ignore + # Ctrl+C until the child happened to close its pipes; cocoindex's teardown + # (and any flow-server grandchild it spawned) can hold them open for a long + # time, so an install/reprocess could not be aborted. The daemon drain + # threads keep the child's stdout/stderr pipes empty while we wait, so the + # child never blocks on a full pipe. + try: + code = proc.wait() + except BaseException: + # Ctrl+C or any other abort: best-effort, NON-BLOCKING child teardown + # so a process we spawned does not outlive us, then re-raise WITHOUT + # joining the drain threads. They may be blocked on a pipe the child + # still owns, and a join here would re-introduce the very hang this + # reordering fixes. The threads are daemons, so they vanish at exit. + _abort_child(proc) + raise + # Normal exit: the child is gone, its pipes hit EOF, and the drain threads + # return promptly — safe to join here. (If a pipe-inheriting grandchild + # lingered past the child's exit, these joins would block until it too + # closed its write ends — on Ctrl+C the shared-process-group SIGINT reaches + # it and closes them. This window is the short teardown-after-indexing phase, + # not the long indexing phase the wait-first reorder already made + # interruptible.) t_out.join() t_err.join() if filt is not None: filt.flush() - code = proc.wait() return out_buf.decode(errors="replace"), err_buf.decode(errors="replace"), code +def _abort_child(proc: subprocess.Popen[bytes]) -> None: + """Best-effort, NON-BLOCKING teardown of a spawned child on an abort path. + + Runs when ``proc.wait()`` raised (Ctrl+C, or any other exception). Sends + SIGTERM and returns immediately — this path exists to let the operator exit + *promptly*, and waiting for the child would defeat that. On Ctrl+C the child + already received SIGINT (same process group); this just guarantees teardown + for non-signal aborts, after which the child finishes shutting down on its + own. ``OSError`` (incl. ``ProcessLookupError`` — already-dead / zombie / + reaped) is swallowed — the only caller re-raises regardless. + """ + fn = getattr(proc, "terminate", None) + if fn is None: + return + try: + fn() + except OSError: + pass + + def run_cocoindex_update( env: dict[str, str], *, diff --git a/tests/package/test_java_codebase_rag_cli.py b/tests/package/test_java_codebase_rag_cli.py index c7ad431..dde95bd 100644 --- a/tests/package/test_java_codebase_rag_cli.py +++ b/tests/package/test_java_codebase_rag_cli.py @@ -1392,6 +1392,41 @@ def fake_exit(code: int) -> None: assert result is None, fake_rc +def test_console_script_main_ctrl_c_exits_130_via_os_exit( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Ctrl+C (KeyboardInterrupt out of ``main``) must produce a clean, immediate + exit: code 130 (128 + SIGINT), a short notice on stderr (not stdout), and + routing through the same flush + ``os._exit`` path as the normal rc — never + an uncaught traceback into interpreter finalization. This is the operator- + facing contract for aborting a long indexing step, reachable once + ``pipeline._popen_capturing_stderr`` waits on the child before joining.""" + import os as _os + + from java_codebase_rag import cli as cli + + snapshot: dict[str, object] = {} + + def fake_main() -> int: + raise KeyboardInterrupt + + monkeypatch.setattr(cli, "main", fake_main) + + def fake_exit(code: int) -> None: + snapshot["exit_code"] = code + + monkeypatch.setattr(_os, "_exit", fake_exit) + + result = cli._console_script_main() + + assert snapshot["exit_code"] == 130 + assert result is None + captured = capsys.readouterr() + assert captured.out == "", "Ctrl+C must not write to stdout (machine consumers)" + assert "Interrupted" in captured.err, "Ctrl+C should print a short notice on stderr" + + def test_console_script_entry_point_routes_through_wrapper() -> None: """``[project.scripts]`` must point ``java-codebase-rag`` at ``_console_script_main`` (not ``main``) so the deterministic-exit path is the diff --git a/tests/package/test_pipeline.py b/tests/package/test_pipeline.py index 7d8fb7c..6793ac9 100644 --- a/tests/package/test_pipeline.py +++ b/tests/package/test_pipeline.py @@ -5,10 +5,16 @@ (cocoindex's bulk-update ``merge_insert``) emits ~one deletion-vector + version commit per matched row — O(rows) of tiny file IO that hangs for many minutes on large repos. Drop+recreate is identical output for a full rebuild. + +Also covers ``_popen_capturing_stderr``'s Ctrl+C behavior: it must wait on the +child (interruptible) before joining the drain threads (not interruptible), and +on abort must terminate the child and re-raise WITHOUT joining. """ from __future__ import annotations import subprocess +import sys +import threading from java_codebase_rag import pipeline @@ -101,3 +107,111 @@ def test_drop_preflight_blocker_is_silent(monkeypatch, capsys) -> None: assert seen["update"] == 1 # 127 preflight is expected on graph-only installs and must NOT warn. assert "drop-before-reprocess failed" not in capsys.readouterr().err + + +def test_popen_captures_normal_child_output() -> None: + """Regression guard: a completing child's stdout/stderr/code are still captured + after the wait-before-join reorder (verifies normal operation is unchanged).""" + proc = subprocess.Popen( + [ + sys.executable, + "-c", + "import sys; sys.stdout.write('hello-out'); " + "sys.stderr.write('hello-err'); sys.exit(0)", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=0, + ) + try: + out, err, code = pipeline._popen_capturing_stderr(proc, verbose=True) + finally: + proc.wait() + assert code == 0 + assert out == "hello-out" + assert "hello-err" in err + + +def test_popen_abort_terminates_child_without_joining() -> None: + """Ctrl+C path: when ``proc.wait()`` raises while drain threads are blocked, + ``_popen_capturing_stderr`` must terminate the child and re-raise WITHOUT + joining the drain threads — a join here re-introduces the uninterruptible + hang (``Thread.join()`` on an infinite-timeout lock cannot be interrupted by + SIGINT), which is exactly the bug being fixed. + + Run the helper in a daemon thread with a hard deadline so that a regression + (a join sneaking back onto the abort path) fails the assertion instead of + hanging the whole test session. + """ + release = threading.Event() + + class _BlockingStream: + """``proc.stdout``/``.stderr`` stand-in: ``read()`` blocks until released.""" + + def read(self, _n: int) -> bytes: + release.wait() + return b"" + + class _InterruptedProc: + def __init__(self) -> None: + self.stdout = _BlockingStream() + self.stderr = _BlockingStream() + self.terminated = False + + def wait(self, *_a, **_k) -> int: + # Simulate Ctrl+C interrupting the (interruptible) wait on the child. + raise KeyboardInterrupt + + def terminate(self) -> None: + self.terminated = True + + proc = _InterruptedProc() + outcome: dict = {} + + def runner() -> None: + try: + pipeline._popen_capturing_stderr(proc, verbose=True) + outcome["raised"] = None + except BaseException as exc: # noqa: BLE001 — we WANT every escape + outcome["raised"] = exc + finally: + release.set() + + t = threading.Thread(target=runner, daemon=True) + t.start() + t.join(timeout=5.0) + # Snapshot liveness BEFORE releasing: with the buggy join-first order the + # runner would still be blocked in Thread.join() here (deterministic); with + # the fix it returns in milliseconds. Checking after release.set() would race. + hung = t.is_alive() + release.set() # never leave drain threads blocked, even on a passing run + t.join(timeout=5.0) + + assert not hung, ( + "_popen_capturing_stderr hung on the abort path — it must not join the " + "drain threads when the child wait is interrupted (Ctrl+C regression)" + ) + assert isinstance(outcome.get("raised"), KeyboardInterrupt), ( + "expected the KeyboardInterrupt to propagate out of the abort path" + ) + assert proc.terminated, "the spawned child must be torn down on abort" + + +def test_abort_child_swallows_already_dead() -> None: + """``_abort_child`` must not raise if the child is already gone (zombie / + reaped) — the only caller is on an exception path whose job is to re-raise, + so a teardown failure here must never replace the abort cause. + + ``ProcessLookupError`` is the real-world shape (wait() raised, OS already + reaped the child between wait and terminate); it's a subclass of ``OSError``, + which is what the helper catches. + """ + class _AlreadyDeadProc: + terminated = False + + def terminate(self) -> None: + raise ProcessLookupError(3, "No such process") + + proc = _AlreadyDeadProc() + # Must not raise — caller relies on best-effort, swallow-and-return. + pipeline._abort_child(proc) # type: ignore[arg-type]