diff --git a/loopx/extensions/lark/event_collector.py b/loopx/extensions/lark/event_collector.py index 87cf42a42e..efed1c9e13 100644 --- a/loopx/extensions/lark/event_collector.py +++ b/loopx/extensions/lark/event_collector.py @@ -8,6 +8,7 @@ import shlex import shutil import subprocess +import time from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any @@ -18,6 +19,7 @@ inspect_lark_event_inbox, load_lark_event_inbox_config, ) +from ..process_runtime import terminate_process_tree from .identity_shapes import LARK_CHAT_ID_PATTERN CONFIG_SCHEMA_VERSION_V0 = "lark_event_collector_config_v0" @@ -40,6 +42,92 @@ Runner = Callable[..., subprocess.CompletedProcess[str]] +# The collector accepts a config path only when Git proves it is ignored and +# untracked. That proof costs three Git reads, and a Git blocked on a stale +# index lock, a network mount or an oversized index would otherwise hold the +# command open with no answer at all. +GIT_PROBE_TIMEOUT_SECONDS = 5.0 + + +def _retire_owned_probe(process: subprocess.Popen[str]) -> None: + """Stop this probe's own process group, then drop its pipes unread. + + Git starts helpers of its own -- an fsmonitor hook is the common one on + ``ls-files`` -- inside the group it runs in, so ending only the Git process + leaves them behind for the next run to inherit. An inherited pipe also + outlives the leader, so the pipes are closed instead of drained. + """ + + try: + terminate_process_tree(process, grace_seconds=0.2) + except (OSError, RuntimeError): + pass + finally: + for stream in (process.stdout, process.stderr): + if stream is not None: + stream.close() + try: + process.wait(timeout=1.0) + except subprocess.TimeoutExpired: + pass + + +def _git_deadline() -> float: + """The moment the whole admission check stops asking Git questions.""" + + return time.monotonic() + GIT_PROBE_TIMEOUT_SECONDS + + +def _run_git_probe( + root: Path, *arguments: str, deadline: float | None = None +) -> subprocess.CompletedProcess[str] | None: + """Ask Git one ownership question, answering ``None`` when it cannot. + + No answer is not a "no" for this caller: acceptance needs proof that the + config is ignored and untracked, so a probe that runs out of budget lands on + the refusing side instead of raising out of the command. The three + questions share one deadline, and a question asked after it has passed is + not started at all. The probe runs in a session of its own so an exhausted + budget can retire the helpers Git started; that also means cancelling the + command reaches this process's group and not the probe, so an interrupted + wait retires the same session before the cancellation keeps travelling. + """ + + budget = ( + GIT_PROBE_TIMEOUT_SECONDS + if deadline is None + else deadline - time.monotonic() + ) + if budget <= 0: + return None + argv = ["git", "-C", str(root), *arguments] + try: + process = subprocess.Popen( + argv, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + start_new_session=os.name == "posix", + ) + except OSError: + return None + try: + stdout, _stderr = process.communicate(timeout=budget) + except subprocess.TimeoutExpired: + _retire_owned_probe(process) + return None + except BaseException: + # The operator cancelled the command. An exhausted budget keeps + # refusing the config, while Ctrl-C retires its helpers and stays the + # cancellation the caller raised. + _retire_owned_probe(process) + raise + return subprocess.CompletedProcess( + args=argv, returncode=process.returncode, stdout=stdout, stderr="" + ) + def _project_config_path(project: str | Path, raw: str | Path) -> Path: root = Path(project).expanduser().resolve() @@ -50,11 +138,10 @@ def _project_config_path(project: str | Path, raw: str | Path) -> Path: relative = path.relative_to(root) except ValueError as exc: raise ValueError("lark collector config must stay inside the project") from exc - git_probe = subprocess.run( - ["git", "-C", str(root), "rev-parse", "--show-toplevel"], - capture_output=True, - check=False, - ) + deadline = _git_deadline() + git_probe = _run_git_probe(root, "rev-parse", "--show-toplevel", deadline=deadline) + if git_probe is None: + raise ValueError("lark collector config needs a Git answer the probe could not give") if git_probe.returncode != 0: has_git_boundary = any( (candidate / ".git").exists() for candidate in (root, *root.parents) @@ -65,16 +152,18 @@ def _project_config_path(project: str | Path, raw: str | Path) -> Path: "under .loopx/config in a non-Git project" ) return path - tracked = subprocess.run( - ["git", "-C", str(root), "ls-files", "--error-unmatch", "--", str(relative)], - capture_output=True, - check=False, + tracked = _run_git_probe( + root, "ls-files", "--error-unmatch", "--", str(relative), deadline=deadline ) - ignored = subprocess.run( - ["git", "-C", str(root), "check-ignore", "--quiet", "--", str(relative)], - capture_output=True, - check=False, + if tracked is None: + # Acceptance already failed; asking the next question would only spend + # what is left of the same budget on the same hang. + raise ValueError("lark collector config needs a Git answer the probe could not give") + ignored = _run_git_probe( + root, "check-ignore", "--quiet", "--", str(relative), deadline=deadline ) + if ignored is None: + raise ValueError("lark collector config needs a Git answer the probe could not give") if tracked.returncode == 0 or ignored.returncode != 0: raise ValueError("lark collector config must be ignored and untracked") return path diff --git a/tests/test_lark_event_collector_git_budget.py b/tests/test_lark_event_collector_git_budget.py new file mode 100644 index 0000000000..7628b71d6d --- /dev/null +++ b/tests/test_lark_event_collector_git_budget.py @@ -0,0 +1,245 @@ +"""The Lark collector's config admission asks Git three questions. + +Each of those reads used to be unbounded, so a Git blocked on a stale index +lock, a network mount or an oversized index held the command open with no +answer. These tests pin the shared budget, the early refusal, and the process +tree a blocked read would otherwise leave behind. +""" + +from __future__ import annotations + +import os +import shlex +import shutil +import signal +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from loopx.extensions.lark import event_collector +from loopx.extensions.lark.event_collector import _project_config_path + + +def _git_project(tmp_path: Path) -> Path: + project = tmp_path / "project" + project.mkdir() + subprocess.run(["git", "init", "-q", str(project)], check=True) + (project / ".gitignore").write_text(".loopx/\n", encoding="utf-8") + return project + + +def _config(project: Path) -> Path: + target = project / ".loopx" / "config" / "collector.json" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("{}", encoding="utf-8") + return target + + +def _fake_git(tmp_path: Path, script: str, monkeypatch: pytest.MonkeyPatch) -> Path: + bin_dir = tmp_path / "fakebin" + bin_dir.mkdir(exist_ok=True) + fake = bin_dir / "git" + fake.write_text(script, encoding="utf-8") + fake.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + return bin_dir + + +def test_a_git_that_cannot_start_refuses_the_config( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + project = _git_project(tmp_path) + _config(project) + + def cannot_start(command, **kwargs): + raise OSError("git is not installed") + + monkeypatch.setattr(event_collector.subprocess, "Popen", cannot_start) + + with pytest.raises(ValueError, match="needs a Git answer"): + _project_config_path(project, ".loopx/config/collector.json") + + +@pytest.mark.skipif(os.name != "posix", reason="the fake git is a POSIX script") +def test_the_next_question_is_not_asked_after_one_goes_unanswered( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + project = _git_project(tmp_path) + _config(project) + asked = tmp_path / "asked.txt" + asked.write_text("", encoding="utf-8") + _fake_git( + tmp_path, + "#!/bin/sh\n" + f'echo "$*" >> {shlex.quote(str(asked))}\n' + 'for argument in "$@"; do\n' + ' case "$argument" in *--error-unmatch*) sleep 30 ;; esac\n' + "done\n" + 'if [ "$3" = "rev-parse" ]; then printf "%s\\n" "$2"; exit 0; fi\n' + "exit 0\n", + monkeypatch, + ) + monkeypatch.setattr(event_collector, "GIT_PROBE_TIMEOUT_SECONDS", 0.5) + + started = time.monotonic() + with pytest.raises(ValueError, match="needs a Git answer"): + _project_config_path(project, ".loopx/config/collector.json") + elapsed = time.monotonic() - started + + commands = asked.read_text(encoding="utf-8") + assert "rev-parse" in commands + assert "--error-unmatch" in commands + # The ignore flag of an unproven config cannot change the answer, so the + # third read is not started at all. + assert "check-ignore" not in commands + assert elapsed < 10, f"the admission spent {elapsed:.1f}s on two questions" + + +@pytest.mark.skipif(os.name != "posix", reason="the fake git is a POSIX script") +def test_a_git_that_never_answers_stays_inside_the_budget( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + project = _git_project(tmp_path) + _config(project) + _fake_git(tmp_path, "#!/bin/sh\nsleep 30\n", monkeypatch) + monkeypatch.setattr(event_collector, "GIT_PROBE_TIMEOUT_SECONDS", 0.3) + + started = time.monotonic() + with pytest.raises(ValueError, match="needs a Git answer"): + _project_config_path(project, ".loopx/config/collector.json") + elapsed = time.monotonic() - started + + assert elapsed < 5.0, f"the probe held the command open for {elapsed:.2f}s" + + +@pytest.mark.skipif(os.name != "posix", reason="core.fsmonitor needs a POSIX hook") +def test_a_timed_out_probe_takes_the_helpers_it_started( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Git starts an fsmonitor hook for `ls-files`, so the budget owns it too.""" + + project = _git_project(tmp_path) + _config(project) + subprocess.run(["git", "-C", str(project), "config", "user.email", "f@e.invalid"], check=True) + subprocess.run(["git", "-C", str(project), "config", "user.name", "Fixture"], check=True) + (project / "tracked.txt").write_text("original\n", encoding="utf-8") + subprocess.run(["git", "-C", str(project), "add", "tracked.txt"], check=True) + subprocess.run(["git", "-C", str(project), "commit", "-qm", "fixture"], check=True) + + heartbeats = tmp_path / "heartbeats" + heartbeats.write_text("", encoding="utf-8") + hook = project / ".git" / "blocking-fsmonitor" + hook.write_text( + "#!/bin/sh\n" + f"while :; do echo tick >> {shlex.quote(str(heartbeats))}; sleep 0.1; done\n", + encoding="utf-8", + ) + hook.chmod(0o755) + subprocess.run( + ["git", "-C", str(project), "config", "core.fsmonitor", str(hook)], check=True + ) + monkeypatch.setattr(event_collector, "GIT_PROBE_TIMEOUT_SECONDS", 0.8) + + started = time.monotonic() + with pytest.raises(ValueError, match="needs a Git answer"): + _project_config_path(project, ".loopx/config/collector.json") + assert time.monotonic() - started < 6 + + retired = heartbeats.stat().st_size + assert retired > 0, "the fixture hook never ran, so this proves nothing" + time.sleep(0.6) + assert heartbeats.stat().st_size == retired, ( + "the fsmonitor hook and its child outlived the probe budget" + ) + + +def test_an_ignored_untracked_config_still_loads(tmp_path: Path) -> None: + project = _git_project(tmp_path) + target = _config(project) + + assert _project_config_path(project, ".loopx/config/collector.json") == target.resolve() + + +@pytest.mark.skipif(os.name != "posix", reason="a terminal Ctrl-C is a POSIX group signal") +def test_a_cancelled_command_takes_the_git_it_started(tmp_path: Path) -> None: + """Ctrl-C during the admission retires the session the read owns. + + `collector-plan` dispatches to `plan_lark_event_collector`, so the child + below runs that entry in its own process, as the CLI would, with a `git` + that answers everything through real Git except the ownership read: that + one blocks, the shape a real `core.fsmonitor` hook has while Git waits on + it. The interrupt then goes to the child's own process group, which is the + scope a terminal Ctrl-C reaches. Git runs in a session of its own, so the + interrupt never reaches it: only the waiting call can retire it. + """ + + project = _git_project(tmp_path) + _config(project) + real_git = shutil.which("git") + assert real_git is not None + heartbeats = tmp_path / "heartbeats" + heartbeats.write_text("", encoding="utf-8") + stop = tmp_path / "stop" + bin_dir = tmp_path / "blocking-bin" + bin_dir.mkdir() + fake = bin_dir / "git" + fake.write_text( + f"#!{sys.executable}\n" + "import os, sys, time\n" + "from pathlib import Path\n" + "if 'ls-files' not in sys.argv:\n" + f" os.execv({real_git!r}, [{real_git!r}] + sys.argv[1:])\n" + f"heartbeats, stop = Path({str(heartbeats)!r}), Path({str(stop)!r})\n" + "while not stop.exists():\n" + " with heartbeats.open('a', encoding='utf-8') as stream:\n" + " stream.write('tick\\n')\n" + " time.sleep(0.1)\n", + encoding="utf-8", + ) + fake.chmod(0o755) + environment = { + key: value + for key, value in os.environ.items() + if not key.startswith(("PYTEST", "COVERAGE", "COV_CORE")) + } + environment.update({ + "PYTHONPATH": str(Path(__file__).resolve().parents[1]), + "LOOPX_USAGE_PING": "0", + "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}", + }) + command = subprocess.Popen( + [ + sys.executable, "-c", + "import sys\n" + "from loopx.extensions.lark.event_collector import plan_lark_event_collector\n" + "plan_lark_event_collector(project=sys.argv[1], config_path=sys.argv[2])\n", + str(project), ".loopx/config/collector.json", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=environment, + start_new_session=True, + ) + try: + deadline = time.monotonic() + 25 + while heartbeats.stat().st_size == 0 and time.monotonic() < deadline: + time.sleep(0.05) + assert heartbeats.stat().st_size > 0, "the fixture never blocked, so this proves nothing" + assert command.poll() is None, "the command had already returned" + os.killpg(os.getpgid(command.pid), signal.SIGINT) + command.communicate(timeout=25) + assert command.returncode != 0 + retired = heartbeats.stat().st_size + time.sleep(0.6) + assert heartbeats.stat().st_size == retired, ( + "the blocked Git and its helper outlived the cancellation" + ) + finally: + stop.write_text("stop\n", encoding="utf-8") + if command.poll() is None: + os.killpg(os.getpgid(command.pid), signal.SIGKILL) + command.communicate(timeout=25)