diff --git a/python/robomp/src/natives_cache.py b/python/robomp/src/natives_cache.py index f322475a5..fa6011e3f 100644 --- a/python/robomp/src/natives_cache.py +++ b/python/robomp/src/natives_cache.py @@ -152,6 +152,7 @@ def compute_key(repo_dir: Path, *, target: str | None = None) -> str: capture_output=True, check=True, env=_git_safe_directory_env(repo_dir), + timeout=120.0, ) lines = proc.stdout.splitlines() if len(lines) != len(CACHE_KEY_PATHS): diff --git a/python/robomp/src/sandbox.py b/python/robomp/src/sandbox.py index 4e15b28e7..b048c3a88 100644 --- a/python/robomp/src/sandbox.py +++ b/python/robomp/src/sandbox.py @@ -47,6 +47,7 @@ import shutil import signal import stat import subprocess +import threading from dataclasses import dataclass from pathlib import Path from typing import Any, Protocol @@ -298,16 +299,31 @@ class LocalGitTransport: # ---------- low-level helpers retained for callers expecting old shape ---------- -def _safe_run(cmd: list[str], *, cwd: Path | None = None, **kwargs: Any) -> subprocess.CompletedProcess[str]: +_DEFAULT_SANDBOX_SUBPROCESS_TIMEOUT = 120.0 + + +def _safe_run( + cmd: list[str], + *, + cwd: Path | None = None, + timeout: float | None = _DEFAULT_SANDBOX_SUBPROCESS_TIMEOUT, + **kwargs: Any, +) -> subprocess.CompletedProcess[str]: """Run without raising; caller decides on returncode. Credentials are redacted from any captured output.""" - proc = subprocess.run( - cmd, - cwd=str(cwd) if cwd else None, - check=False, - capture_output=True, - text=True, - **kwargs, - ) + try: + proc = subprocess.run( + cmd, + cwd=str(cwd) if cwd else None, + check=False, + capture_output=True, + text=True, + timeout=timeout, + **kwargs, + ) + except subprocess.TimeoutExpired as exc: + out = redact_credentials(exc.stdout or "") if isinstance(exc.stdout, str) else "" + err = redact_credentials(exc.stderr or "") if isinstance(exc.stderr, str) else "" + return subprocess.CompletedProcess(cmd, 124, out, f"{err}\ntimed out after {timeout:.0f}s") if proc.stdout: proc.stdout = redact_credentials(proc.stdout) if proc.stderr: @@ -315,15 +331,24 @@ def _safe_run(cmd: list[str], *, cwd: Path | None = None, **kwargs: Any) -> subp return proc -def _run(cmd: list[str], *, cwd: Path | None = None) -> subprocess.CompletedProcess[str]: +def _run( + cmd: list[str], + *, + cwd: Path | None = None, + timeout: float | None = _DEFAULT_SANDBOX_SUBPROCESS_TIMEOUT, +) -> subprocess.CompletedProcess[str]: """Legacy raising helper (still used by a sandbox test). Forwards to subprocess.run.""" - proc = subprocess.run( - cmd, - cwd=str(cwd) if cwd else None, - check=False, - capture_output=True, - text=True, - ) + try: + proc = subprocess.run( + cmd, + cwd=str(cwd) if cwd else None, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + raise GitCommandError(cmd, 124, "", f"git timed out after {timeout:.0f}s") from exc if proc.returncode != 0: raise GitCommandError(cmd, proc.returncode, proc.stdout, proc.stderr) return proc @@ -607,8 +632,16 @@ def _chown_workspace(ws_root: Path, slot_uid: int | None) -> None: return uid = slot_uid if slot_uid is not None else os.geteuid() gid = slot_uid if slot_uid is not None else os.getegid() - subprocess.run(["chown", "-R", f"{uid}:{gid}", str(ws_root)], check=True) - subprocess.run(["chmod", "-R", "u=rwX,g=rwX,o=", str(ws_root)], check=True) + subprocess.run( + ["chown", "-R", f"{uid}:{gid}", str(ws_root)], + check=True, + timeout=_DEFAULT_SANDBOX_SUBPROCESS_TIMEOUT, + ) + subprocess.run( + ["chmod", "-R", "u=rwX,g=rwX,o=", str(ws_root)], + check=True, + timeout=_DEFAULT_SANDBOX_SUBPROCESS_TIMEOUT, + ) # ---------- SandboxManager ---------- @@ -634,6 +667,16 @@ class SandboxManager: self.natives_cache = natives_cache root.mkdir(parents=True, exist_ok=True) self.pool.mkdir(parents=True, exist_ok=True) + self._repo_locks: dict[str, threading.RLock] = {} + self._repo_locks_guard = threading.Lock() + + def _repo_lock(self, repo: str) -> threading.RLock: + with self._repo_locks_guard: + lock = self._repo_locks.get(repo) + if lock is None: + lock = threading.RLock() + self._repo_locks[repo] = lock + return lock # ---- pool ---- def pool_path(self, repo: str) -> Path: @@ -696,121 +739,130 @@ class SandboxManager: slot_uid: int | None = None, ) -> Workspace: """Create or resume a per-issue worktree.""" - if pr_head is not None and existing_branch is not None: - raise ValueError("ensure_workspace accepts either pr_head or existing_branch, not both") - pool = self.ensure_clone(repo=repo, clone_url=clone_url, default_branch=default_branch) - ws_root = self.workspace_root(repo, number) - repo_dir = ws_root / "repo" - session_dir = ws_root / ".omp-session" - context_dir = ws_root / "context" - artifacts_dir = ws_root / "artifacts" - for path in (ws_root, session_dir, context_dir, context_dir / "repro", artifacts_dir): - path.mkdir(parents=True, exist_ok=True) + with self._repo_lock(repo): + if pr_head is not None and existing_branch is not None: + raise ValueError("ensure_workspace accepts either pr_head or existing_branch, not both") + pool = self.ensure_clone(repo=repo, clone_url=clone_url, default_branch=default_branch) + ws_root = self.workspace_root(repo, number) + repo_dir = ws_root / "repo" + session_dir = ws_root / ".omp-session" + context_dir = ws_root / "context" + artifacts_dir = ws_root / "artifacts" + for path in (ws_root, session_dir, context_dir, context_dir / "repro", artifacts_dir): + path.mkdir(parents=True, exist_ok=True) - branch = ( - f"review/pr-{pr_head}" - if pr_head is not None - else existing_branch - or make_branch( - issue_number=number, - title=title, - seed=f"{repo}#{number}", + branch = ( + f"review/pr-{pr_head}" + if pr_head is not None + else existing_branch + or make_branch( + issue_number=number, + title=title, + seed=f"{repo}#{number}", + ) ) - ) - repo_exists = (repo_dir / ".git").exists() - workspace_prepared = False - slot_git_kwargs = _slot_subprocess_kwargs(slot_uid) - slot_git_env: dict[str, str] | None = None - if repo_exists: - # Existing workspaces are already slot-owned from the previous run. - # Refresh pool-side group bits, then hand the tree to the current - # slot before running any git command inside the worktree; root's - # uid-0 bypass does not bypass git's safe.directory ownership check. - _share_git_metadata_with_slots(repo_dir, slot_uid) - _provision_runtime_dirs(ws_root) - _chown_workspace(ws_root, slot_uid) - workspace_prepared = True - if not repo_exists: - if pr_head is not None: - self.transport.fetch_pr_head(repo=repo, pool_dir=pool, pr_number=pr_head) - _run(["git", "worktree", "add", "--detach", str(repo_dir), "FETCH_HEAD"], cwd=pool) - else: - # Make sure the requested start point exists locally (best-effort). - # For follow-ups on an existing PR, `existing_branch` is the remote - # head branch we need to amend; starting from default would silently - # lose the PR's current commits if the local pool branch is absent. - self.transport.fetch_base_ref(repo=repo, pool_dir=pool, ref=existing_branch or default_branch) - check = _safe_run(["git", "rev-parse", "--verify", f"refs/heads/{branch}"], cwd=pool) - if check.returncode == 0: - _run(["git", "worktree", "add", str(repo_dir), branch], cwd=pool) + repo_exists = (repo_dir / ".git").exists() + workspace_prepared = False + slot_git_kwargs = _slot_subprocess_kwargs(slot_uid) + slot_git_env: dict[str, str] | None = None + if repo_exists: + # Existing workspaces are already slot-owned from the previous run. + # Refresh pool-side group bits, then hand the tree to the current + # slot before running any git command inside the worktree; root's + # uid-0 bypass does not bypass git's safe.directory ownership check. + _share_git_metadata_with_slots(repo_dir, slot_uid) + _provision_runtime_dirs(ws_root) + _chown_workspace(ws_root, slot_uid) + workspace_prepared = True + if not repo_exists: + if pr_head is not None: + self.transport.fetch_pr_head(repo=repo, pool_dir=pool, pr_number=pr_head) + _run(["git", "worktree", "add", "--detach", str(repo_dir), "FETCH_HEAD"], cwd=pool) else: - start_point = f"origin/{default_branch}" - if existing_branch: - remote = _safe_run( - ["git", "rev-parse", "--verify", f"refs/remotes/origin/{existing_branch}"], + # Make sure the requested start point exists locally (best-effort). + # For follow-ups on an existing PR, `existing_branch` is the remote + # head branch we need to amend; starting from default would silently + # lose the PR's current commits if the local pool branch is absent. + self.transport.fetch_base_ref(repo=repo, pool_dir=pool, ref=existing_branch or default_branch) + probe = ["git", "rev-parse", "--verify", f"refs/heads/{branch}"] + check = _safe_run(probe, cwd=pool) + if check.returncode == 124: + # A timed-out probe is indeterminate, not "branch absent". + # Falling through would create the worktree from the wrong + # start point; fail instead so the event retries. + raise GitCommandError(probe, check.returncode, check.stdout, check.stderr) + if check.returncode == 0: + _run(["git", "worktree", "add", str(repo_dir), branch], cwd=pool) + else: + start_point = f"origin/{default_branch}" + if existing_branch: + remote_probe = ["git", "rev-parse", "--verify", f"refs/remotes/origin/{existing_branch}"] + remote = _safe_run(remote_probe, cwd=pool) + if remote.returncode == 124: + # Same: an indeterminate probe must not silently fall + # back to origin/default and lose the PR's commits. + raise GitCommandError(remote_probe, remote.returncode, remote.stdout, remote.stderr) + if remote.returncode == 0: + start_point = f"origin/{existing_branch}" + _run( + [ + "git", + "worktree", + "add", + "-b", + branch, + str(repo_dir), + start_point, + ], cwd=pool, ) - if remote.returncode == 0: - start_point = f"origin/{existing_branch}" - _run( - [ - "git", - "worktree", - "add", - "-b", + else: + slot_git_env = _git_env_for_repo(repo_dir) + current = _safe_run( + ["git", "symbolic-ref", "--quiet", "--short", "HEAD"], + cwd=repo_dir, + env=slot_git_env, + **slot_git_kwargs, + ) + if current.returncode == 0 and current.stdout.strip(): + branch = current.stdout.strip() + if existing_branch is not None and existing_branch != branch: + log.warning( + "workspace branch mapping %r differs from checked-out branch %r; using checkout", + existing_branch, branch, - str(repo_dir), - start_point, - ], - cwd=pool, - ) - else: - slot_git_env = _git_env_for_repo(repo_dir) - current = _safe_run( - ["git", "symbolic-ref", "--quiet", "--short", "HEAD"], - cwd=repo_dir, - env=slot_git_env, - **slot_git_kwargs, - ) - if current.returncode == 0 and current.stdout.strip(): - branch = current.stdout.strip() - if existing_branch is not None and existing_branch != branch: - log.warning( - "workspace branch mapping %r differs from checked-out branch %r; using checkout", - existing_branch, - branch, - ) - if not workspace_prepared: + ) + if not workspace_prepared: + _share_git_metadata_with_slots(repo_dir, slot_uid) + _provision_runtime_dirs(ws_root) + _chown_workspace(ws_root, slot_uid) + if slot_git_env is None: + slot_git_env = _git_env_for_repo(repo_dir) + # Identity is set on the worktree's shared config; idempotent. Run as + # the slot after the chown so git never trips over safe.directory. + for command in (["git", "config", "user.email", author_email], ["git", "config", "user.name", author_name]): + proc = _safe_run(command, cwd=repo_dir, env=slot_git_env, **slot_git_kwargs) + if proc.returncode != 0: + raise GitCommandError(command, proc.returncode, proc.stdout, proc.stderr) _share_git_metadata_with_slots(repo_dir, slot_uid) - _provision_runtime_dirs(ws_root) - _chown_workspace(ws_root, slot_uid) - if slot_git_env is None: - slot_git_env = _git_env_for_repo(repo_dir) - # Identity is set on the worktree's shared config; idempotent. Run as - # the slot after the chown so git never trips over safe.directory. - for command in (["git", "config", "user.email", author_email], ["git", "config", "user.name", author_name]): - proc = _safe_run(command, cwd=repo_dir, env=slot_git_env, **slot_git_kwargs) - if proc.returncode != 0: - raise GitCommandError(command, proc.returncode, proc.stdout, proc.stderr) - _share_git_metadata_with_slots(repo_dir, slot_uid) - workspace = Workspace( - root=ws_root, - repo_dir=repo_dir, - session_dir=session_dir, - context_dir=context_dir, - artifacts_dir=artifacts_dir, - branch=branch, - repo_full_name=repo, - issue_number=number, - ) - # Best-effort: hardlink pre-built natives in if we've cached this - # source state before. Runs AFTER the slot chown so the cache inode - # keeps its `root:omp` ownership (the slot reads through group `omp`); - # write-temp + rename in the napi build replaces with a new inode if - # the agent rebuilds, so the cached file is never mutated. - self._populate_natives_cache(workspace, slot_uid=slot_uid) - return workspace + workspace = Workspace( + root=ws_root, + repo_dir=repo_dir, + session_dir=session_dir, + context_dir=context_dir, + artifacts_dir=artifacts_dir, + branch=branch, + repo_full_name=repo, + issue_number=number, + ) + # Best-effort: hardlink pre-built natives in if we've cached this + # source state before. Runs AFTER the slot chown so the cache inode + # keeps its `root:omp` ownership (the slot reads through group `omp`); + # write-temp + rename in the napi build replaces with a new inode if + # the agent rebuilds, so the cached file is never mutated. + self._populate_natives_cache(workspace, slot_uid=slot_uid) + return workspace def _populate_natives_cache(self, workspace: Workspace, *, slot_uid: int | None = None) -> None: """Try to hardlink cached pi-natives artifacts into the worktree. @@ -838,7 +890,7 @@ class SandboxManager: # produces natives, so creating the dir is correct. try: key = natives_compute_key(workspace.repo_dir) - except (subprocess.CalledProcessError, RuntimeError, OSError) as exc: + except (subprocess.SubprocessError, RuntimeError, OSError) as exc: log.debug( "natives_cache key compute failed", extra={"workspace": workspace.workspace_key, "err": redact_credentials(str(exc))}, @@ -895,15 +947,22 @@ class SandboxManager: ) def remove_workspace(self, *, repo: str, number: int) -> None: - ws_root = self.workspace_root(repo, number) - repo_dir = ws_root / "repo" - if repo_dir.exists(): - pool = self.pool_path(repo) - _safe_run(["git", "worktree", "remove", "--force", str(repo_dir)], cwd=pool) + with self._repo_lock(repo): + ws_root = self.workspace_root(repo, number) + repo_dir = ws_root / "repo" if repo_dir.exists(): - shutil.rmtree(repo_dir, ignore_errors=True) - if ws_root.exists(): - shutil.rmtree(ws_root, ignore_errors=True) + pool = self.pool_path(repo) + _safe_run(["git", "worktree", "remove", "--force", str(repo_dir)], cwd=pool) + if repo_dir.exists(): + # `git worktree remove` did not clean up (nonzero exit, incl. a + # 124 timeout). Delete the checkout ourselves, then prune the + # pool's now-dangling worktree registration so a later + # `git worktree add` for the same path does not trip on stale + # metadata. + shutil.rmtree(repo_dir, ignore_errors=True) + _safe_run(["git", "worktree", "prune"], cwd=pool) + if ws_root.exists(): + shutil.rmtree(ws_root, ignore_errors=True) __all__ = [ diff --git a/python/robomp/src/tasks.py b/python/robomp/src/tasks.py index 5597d5550..08e7b1a64 100644 --- a/python/robomp/src/tasks.py +++ b/python/robomp/src/tasks.py @@ -2,9 +2,10 @@ from __future__ import annotations +import asyncio import logging -from collections.abc import Mapping -from typing import Any +from collections.abc import Callable, Mapping +from typing import Any, TypeVar from robomp import persona from robomp.config import Settings @@ -23,6 +24,38 @@ from robomp.worker import DirectiveInfo, TaskInputs, ThreadMessage, run_task log = logging.getLogger(__name__) +_T = TypeVar("_T") + + +async def _run_workspace_op(func: Callable[..., _T], /, **kwargs: object) -> _T: + """Offload a blocking sandbox workspace op to a worker thread, uncancellably. + + Workspace setup/teardown (git clone/fetch, worktree add/remove, chown) is + blocking, so it runs off the event loop via a thread. Unlike a bare + ``await asyncio.to_thread(...)``, cancelling the awaiting coroutine here does + NOT detach the still-running thread: the thread holds the per-repo lock and + owns the slot mid-setup, and ``_run_event``'s ``finally`` reaps/releases that + slot on cancellation. If the await detached, the reaped slot could be reused + while the thread is still touching it. So on cancellation we drain the thread + to completion before propagating, gating the caller's ``finally`` behind the + thread. The inner subprocesses are timeout-bounded, so completion is assured. + """ + inner = asyncio.ensure_future(asyncio.to_thread(func, **kwargs)) + try: + return await asyncio.shield(inner) + except asyncio.CancelledError: + # A repeated cancel can interrupt even a shielded await, so loop until + # the thread is actually done; swallow the inner's own outcome and + # re-raise the cancellation the caller expects. + while not inner.done(): + try: + await asyncio.shield(inner) + except asyncio.CancelledError: + continue + except BaseException: + break + raise + def _comment_from_payload(payload: Mapping[str, Any]) -> CommentInfo: c = payload.get("comment") or {} @@ -267,7 +300,8 @@ async def triage_issue( return db.upsert_issue(key=key, repo=repo.full_name, number=issue.number, state="reproducing") clone_url = repo.clone_url - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -341,7 +375,8 @@ async def review_pr( ) db.upsert_issue(key=key, repo=repo.full_name, number=pr_number, state="reviewing", pr_number=pr_number) - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=pr_number, title=issue.title, @@ -406,7 +441,8 @@ async def handle_comment( # first and executes the directive in the same RPC turn. log.info("directive bootstrap", extra={"key": key, "author": directive.author}) db.upsert_issue(key=key, repo=repo.full_name, number=issue.number, state="reproducing") - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -456,9 +492,10 @@ async def handle_comment( # Maintainer reopen: tear down stale workspace, reset state, branch # afresh from default. The old branch may have been merged/deleted. log.info("directive reopen", extra={"key": key, "from_state": existing.state, "author": directive.author}) - sandbox.remove_workspace(repo=repo.full_name, number=issue.number) + await _run_workspace_op(sandbox.remove_workspace, repo=repo.full_name, number=issue.number) db.upsert_issue(key=key, repo=repo.full_name, number=issue.number, state="reproducing") - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -493,7 +530,8 @@ async def handle_comment( await run_task(task_kind="handle_comment", inputs=inputs, comment=comment, directive=directive) return - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -567,7 +605,8 @@ async def handle_review( log.warning("review fetch failed", extra={"err": str(exc)}) return clone_url = repo.clone_url - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -677,7 +716,7 @@ async def handle_pr_conversation( "directive reopen (pr)", extra={"key": issue_row.key, "from_state": issue_row.state, "author": directive.author}, ) - sandbox.remove_workspace(repo=issue_row.repo, number=issue_row.number) + await _run_workspace_op(sandbox.remove_workspace, repo=issue_row.repo, number=issue_row.number) db.upsert_issue(key=issue_row.key, repo=issue_row.repo, number=issue_row.number, state="reproducing") issue_row = db.get_issue(issue_row.key) or issue_row # Bare @mention with no request body — the route stashes an empty @@ -713,7 +752,8 @@ async def handle_pr_conversation( if existing_branch is None and not (directive and issue_row.state == "reproducing"): log.info("skip: pr-conversation PR missing branch mapping", extra={"repo": repo_full, "pr": pr_number}) return - workspace = sandbox.ensure_workspace( + workspace = await _run_workspace_op( + sandbox.ensure_workspace, repo=repo.full_name, number=issue.number, title=issue.title, @@ -797,7 +837,7 @@ async def cleanup_workspace( issue_row = db.get_issue(issue_key(repo_full, number)) if issue_row is None: return - sandbox.remove_workspace(repo=issue_row.repo, number=issue_row.number) + await _run_workspace_op(sandbox.remove_workspace, repo=issue_row.repo, number=issue_row.number) db.set_issue_state(issue_row.key, target_state) log.info("cleanup", extra={"key": issue_row.key, "state": target_state}) diff --git a/python/robomp/tests/test_sandbox.py b/python/robomp/tests/test_sandbox.py index 7dfd5f52d..071d63141 100644 --- a/python/robomp/tests/test_sandbox.py +++ b/python/robomp/tests/test_sandbox.py @@ -5,7 +5,9 @@ import platform import signal import stat import subprocess +import threading from pathlib import Path +from types import SimpleNamespace import pytest @@ -508,7 +510,7 @@ def test_chown_workspace_runs_chown_and_chmod_as_root_on_linux(tmp_path: Path, m monkeypatch.setattr("robomp.sandbox.os.geteuid", lambda: 0) monkeypatch.setattr( "robomp.sandbox.subprocess.run", - lambda cmd, *, check: calls.append((cmd, check)), + lambda cmd, *, check, timeout=None: calls.append((cmd, check)), ) _chown_workspace(tmp_path, 2001) @@ -530,7 +532,7 @@ def test_chown_workspace_makes_workspace_slot_owned(tmp_path: Path, monkeypatch: file_path.chmod(0o777) owned: dict[Path, tuple[int, int]] = {} - def fake_run(cmd: list[str], *, check: bool) -> None: + def fake_run(cmd: list[str], *, check: bool, timeout: float | None = None) -> None: assert check if cmd[:2] == ["chown", "-R"]: uid_text, gid_text = cmd[2].split(":", 1) @@ -1104,6 +1106,41 @@ def test_remove_workspace(tmp_path: Path, upstream_repo: Path) -> None: assert not ws.root.exists() +def test_remove_workspace_prunes_pool_after_failed_worktree_remove(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mgr = SandboxManager(tmp_path) + # Create a real repo_dir on disk so `repo_dir.exists()` is True on entry. + ws_root = mgr.workspace_root("o/r", 7) + repo_dir = ws_root / "repo" + repo_dir.mkdir(parents=True, exist_ok=True) + pool = mgr.pool_path("o/r") + + calls: list[tuple[list[str], object]] = [] + + def fake_safe_run(cmd, **k): + calls.append((list(cmd), k.get("cwd"))) + # The `worktree remove` "times out" (124) and does NOT delete repo_dir, + # so the code must fall back to rmtree + prune. `prune` succeeds (0). + if cmd[:3] == ["git", "worktree", "remove"]: + return subprocess.CompletedProcess(cmd, 124, "", "timed out") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr("robomp.sandbox._safe_run", fake_safe_run) + + mgr.remove_workspace(repo="o/r", number=7) + + cmds = [c for c, _ in calls] + # The dangling registration must have been pruned, in the correct pool. + assert ["git", "worktree", "prune"] in cmds, ( + "remove_workspace did not prune pool metadata after a failed worktree remove" + ) + prune_idx = next(i for i, (c, _) in enumerate(calls) if c == ["git", "worktree", "prune"]) + assert calls[prune_idx][1] == pool, "prune did not run in the repo's pool dir" + # And the remove was attempted first (ordering: remove before prune). + remove_idx = next(i for i, (c, _) in enumerate(calls) if c[:3] == ["git", "worktree", "remove"]) + assert remove_idx < prune_idx + # The real checkout dir was cleaned up. + assert not repo_dir.exists() + def test_redact_credentials_strips_userinfo() -> None: from robomp.sandbox import redact_credentials @@ -1301,9 +1338,7 @@ def test_run_git_injects_safe_directory_and_subprocess_identity( assert captured["umask"] == 0o002 -def test_run_git_scopes_token_and_scrubs_parent_auth_env( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_run_git_scopes_token_and_scrubs_parent_auth_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: from robomp.git_ops import AUTH_ENV_VAR, _run_git captured: dict[str, object] = {} @@ -1649,3 +1684,248 @@ def test_ensure_workspace_cache_miss_is_silent_noop(tmp_path: Path, upstream_rep # Cache is empty so the workspace ends up identical to the no-cache case. assert ws.repo_dir.is_dir() assert not (ws.repo_dir / "packages" / "natives" / "native").exists() + + +def test_repo_lock_is_per_repo_identity(tmp_path: Path) -> None: + mgr = SandboxManager(tmp_path) + assert mgr._repo_lock("o/r") is mgr._repo_lock("o/r") + assert mgr._repo_lock("o/r") is not mgr._repo_lock("o/r2") + + +def test_repo_lock_serializes_same_repo(tmp_path: Path) -> None: + # Deterministic: while one thread holds the repo lock, a probe from a + # DIFFERENT thread must fail to acquire the SAME repo's lock non-blockingly. + mgr = SandboxManager(tmp_path) + held = threading.Event() + release = threading.Event() + probe_result: dict[str, bool] = {} + + def holder() -> None: + with mgr._repo_lock("o/r"): + held.set() + assert release.wait(2.0), "probe never completed" + + def probe() -> None: + assert held.wait(2.0), "holder never acquired the lock" + lock = mgr._repo_lock("o/r") + got = lock.acquire(blocking=False) + probe_result["acquired"] = got + if got: + lock.release() + release.set() + + th = threading.Thread(target=holder) + tp = threading.Thread(target=probe) + th.start() + tp.start() + th.join(3.0) + tp.join(3.0) + # Same repo, held cross-thread -> non-blocking acquire MUST fail. + assert probe_result.get("acquired") is False, ( + "same-repo lock was acquirable from another thread while held (did not serialize)" + ) + + +def test_repo_lock_allows_distinct_repos_to_overlap(tmp_path: Path) -> None: + # Deterministic: holding one repo's lock must NOT block acquiring a + # DIFFERENT repo's lock from another thread. + mgr = SandboxManager(tmp_path) + held = threading.Event() + release = threading.Event() + probe_result: dict[str, bool] = {} + + def holder() -> None: + with mgr._repo_lock("o/a"): + held.set() + assert release.wait(2.0), "probe never completed" + + def probe() -> None: + assert held.wait(2.0), "holder never acquired the lock" + lock = mgr._repo_lock("o/b") + got = lock.acquire(blocking=False) + probe_result["acquired"] = got + if got: + lock.release() + release.set() + + th = threading.Thread(target=holder) + tp = threading.Thread(target=probe) + th.start() + tp.start() + th.join(3.0) + tp.join(3.0) + # Distinct repos -> the other lock is free -> non-blocking acquire succeeds. + assert probe_result.get("acquired") is True, ( + "distinct-repo lock was NOT acquirable while an unrelated repo's lock was held" + ) + + +def test_ensure_workspace_acquires_repo_lock(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mgr = SandboxManager(tmp_path) + mgr.natives_cache = None + real = threading.RLock() + events: list[str | tuple[str, str]] = [] + + class Rec: + def __enter__(self) -> Rec: + events.append("acquire") + real.acquire() + return self + + def __exit__(self, *a: object) -> bool: + real.release() + events.append("release") + return False + + def fake_lock(repo: str) -> Rec: + events.append(("lock", repo)) + return Rec() + + monkeypatch.setattr(mgr, "_repo_lock", fake_lock) + mgr.transport = SimpleNamespace( + clone_pool=lambda **k: None, + fetch_pool=lambda **k: None, + fetch_base_ref=lambda **k: None, + fetch_pr_head=lambda **k: None, + ) # type: ignore + ok = subprocess.CompletedProcess(["x"], 0, "", "") + monkeypatch.setattr("robomp.sandbox._run", lambda *a, **k: ok) + monkeypatch.setattr("robomp.sandbox._safe_run", lambda *a, **k: ok) + monkeypatch.setattr("robomp.sandbox._chown_workspace", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._share_git_metadata_with_slots", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._provision_runtime_dirs", lambda *a, **k: None) + + ws = mgr.ensure_workspace( + repo="o/r", + number=1, + title="t", + clone_url="https://x/o/r.git", + default_branch="main", + author_name="n", + author_email="e@e", + slot_uid=None, + ) + assert ("lock", "o/r") in events + assert events.count("acquire") == 1 + assert events.count("release") == 1 + assert ws.repo_full_name == "o/r" + + +def test_remove_workspace_acquires_repo_lock(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mgr = SandboxManager(tmp_path) + events: list[str | tuple[str, str]] = [] + real = threading.RLock() + + class Rec: + def __enter__(self) -> Rec: + events.append("acquire") + real.acquire() + return self + + def __exit__(self, *a: object) -> bool: + real.release() + events.append("release") + return False + + monkeypatch.setattr(mgr, "_repo_lock", lambda repo: (events.append(("lock", repo)), Rec())[1]) + # ws_root does not exist -> remove_workspace just no-ops inside the lock + mgr.remove_workspace(repo="o/r", number=99) + assert ("lock", "o/r") in events + assert events.count("acquire") == 1 and events.count("release") == 1 + + +def test_safe_run_timeout_returns_124(monkeypatch: pytest.MonkeyPatch) -> None: + import robomp.sandbox as s + + def boom(*a: object, **k: object) -> subprocess.CompletedProcess: + cmd = a[0] if a else k.get("args", ["git"]) + raise subprocess.TimeoutExpired(cmd=cmd, timeout=1) # type: ignore + + monkeypatch.setattr("robomp.sandbox.subprocess.run", boom) + r = s._safe_run(["git", "status"]) + assert r.returncode == 124 + + +def test_run_timeout_raises_git_command_error_124(monkeypatch: pytest.MonkeyPatch) -> None: + import robomp.sandbox as s + + def boom(*a: object, **k: object) -> subprocess.CompletedProcess: + cmd = a[0] if a else k.get("args", ["git"]) + raise subprocess.TimeoutExpired(cmd=cmd, timeout=1) # type: ignore + + monkeypatch.setattr("robomp.sandbox.subprocess.run", boom) + with pytest.raises(s.GitCommandError) as exc: + s._run(["git", "status"]) + assert exc.value.returncode == 124 + + +def test_ensure_workspace_raises_when_local_branch_probe_times_out(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mgr = SandboxManager(tmp_path) + mgr.natives_cache = None + mgr.transport = SimpleNamespace( + clone_pool=lambda **k: None, + fetch_pool=lambda **k: None, + fetch_base_ref=lambda **k: None, + fetch_pr_head=lambda **k: None, + ) # type: ignore + + def fake_safe_run(cmd: list[str], **k: object) -> subprocess.CompletedProcess[str]: + if "rev-parse" in cmd and cmd[-1].startswith("refs/heads/"): + return subprocess.CompletedProcess(cmd, 124, "", "timed out") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr("robomp.sandbox._safe_run", fake_safe_run) + monkeypatch.setattr("robomp.sandbox._run", lambda *a, **k: subprocess.CompletedProcess(["x"], 0, "", "")) + monkeypatch.setattr("robomp.sandbox._chown_workspace", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._share_git_metadata_with_slots", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._provision_runtime_dirs", lambda *a, **k: None) + + with pytest.raises(GitCommandError): + mgr.ensure_workspace( + repo="o/r", + number=1, + title="t", + clone_url="https://x/o/r.git", + default_branch="main", + author_name="n", + author_email="e@e", + existing_branch="feature/x", + slot_uid=None, + ) + + +def test_ensure_workspace_raises_when_remote_branch_probe_times_out(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mgr = SandboxManager(tmp_path) + mgr.natives_cache = None + mgr.transport = SimpleNamespace( + clone_pool=lambda **k: None, + fetch_pool=lambda **k: None, + fetch_base_ref=lambda **k: None, + fetch_pr_head=lambda **k: None, + ) # type: ignore + + def fake_safe_run(cmd: list[str], **k: object) -> subprocess.CompletedProcess[str]: + if "rev-parse" in cmd and cmd[-1].startswith("refs/heads/"): + return subprocess.CompletedProcess(cmd, 128, "", "") + if "rev-parse" in cmd and cmd[-1].startswith("refs/remotes/origin/"): + return subprocess.CompletedProcess(cmd, 124, "", "timed out") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr("robomp.sandbox._safe_run", fake_safe_run) + monkeypatch.setattr("robomp.sandbox._run", lambda *a, **k: subprocess.CompletedProcess(["x"], 0, "", "")) + monkeypatch.setattr("robomp.sandbox._chown_workspace", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._share_git_metadata_with_slots", lambda *a, **k: None) + monkeypatch.setattr("robomp.sandbox._provision_runtime_dirs", lambda *a, **k: None) + + with pytest.raises(GitCommandError): + mgr.ensure_workspace( + repo="o/r", + number=1, + title="t", + clone_url="https://x/o/r.git", + default_branch="main", + author_name="n", + author_email="e@e", + existing_branch="feature/x", + slot_uid=None, + ) diff --git a/python/robomp/tests/test_tasks.py b/python/robomp/tests/test_tasks.py new file mode 100644 index 000000000..ed63ba5c9 --- /dev/null +++ b/python/robomp/tests/test_tasks.py @@ -0,0 +1,123 @@ +import asyncio +import threading +from types import SimpleNamespace + +import pytest + +from robomp import tasks +from robomp.github_client import IssueInfo, RepoInfo + + +async def test_triage_issue_keeps_event_loop_live_while_workspace_setup_blocks(db, settings, monkeypatch, tmp_path): + async def _resolve_repo_and_issue(_github, _payload): + repo = RepoInfo( + full_name="octo/widget", + default_branch="main", + clone_url="https://x/octo/widget.git", + private=False, + ) + issue = IssueInfo( + repo="octo/widget", + number=1, + title="bug", + body="b", + state="open", + author="alice", + labels=(), + is_pull_request=False, + ) + return repo, issue + + monkeypatch.setattr(tasks, "_resolve_repo_and_issue", _resolve_repo_and_issue) + + async def _no_closing(*a, **k): + return () + + github = SimpleNamespace(list_closing_pull_requests=_no_closing) + + entered = threading.Event() + release = threading.Event() + captured: dict[str, object] = {} + + def _blocking_ensure(**_kwargs): + entered.set() + # True ONLY if a concurrent coroutine set `release` while we blocked here. + # Blocks a WORKER THREAD (via to_thread) in the fixed code; blocks the + # LOOP itself in the broken code. + captured["release_seen_in_time"] = release.wait(1.0) + return SimpleNamespace(branch="farm/x/y", session_dir=str(tmp_path / "sess")) + + sandbox = SimpleNamespace(natives_cache=None, ensure_workspace=_blocking_ensure) + + async def _noop_run_task(**_kwargs): + return None + + monkeypatch.setattr(tasks, "run_task", _noop_run_task) + + async def _releaser(): + # Waits (off-loop) until ensure_workspace has actually started, then + # releases it. This coroutine can ONLY make progress if the event loop + # is live while ensure_workspace is blocking. + await asyncio.to_thread(entered.wait, 1.0) + assert entered.is_set(), "ensure_workspace never started" + release.set() + + triage_task = asyncio.create_task( + tasks.triage_issue( + settings=settings, + db=db, + github=github, + sandbox=sandbox, + git_transport=SimpleNamespace(), + payload={}, + delivery_id="d1", + ) + ) + releaser_task = asyncio.create_task(_releaser()) + + await asyncio.wait_for(triage_task, timeout=3.0) + await asyncio.wait_for(releaser_task, timeout=1.0) + + assert captured.get("release_seen_in_time") is True, ( + "event loop was frozen during ensure_workspace: the concurrent releaser " + "could not run, so release.wait timed out (this is the pre-fix hang)" + ) + + +async def test_run_workspace_op_drains_thread_before_propagating_cancel(): + started = threading.Event() + proceed = threading.Event() + finished = threading.Event() + + def slow_op(**_kwargs): + started.set() + # Block on the worker thread until the test releases us. + assert proceed.wait(2.0), "proceed was never set — test bug" + finished.set() + return "done" + + task = asyncio.create_task(tasks._run_workspace_op(slow_op)) + # Wait (off-loop) until the worker thread is actually running. + await asyncio.to_thread(started.wait, 1.0) + assert started.is_set() + + # Cancel the AWAITING coroutine while the thread is mid-flight. + task.cancel() + # Let the loop deliver the cancellation into the helper's drain loop. + await asyncio.sleep(0.05) + + try: + # The thread must NOT have been abandoned: it is still blocked on + # `proceed`, so `finished` is not set and the task has not resolved yet. + assert not finished.is_set(), "thread finished before we released it — impossible unless abandoned" + assert not task.done(), "helper propagated cancel before the thread completed (thread abandoned)" + finally: + # Always release the worker, even if an assert above fails, so a failed + # run cannot leave a blocked thread leaking into later tests. + proceed.set() + + # The helper must now let the thread finish, THEN raise CancelledError. + with pytest.raises(asyncio.CancelledError): + await task + # Deterministic in the fixed helper: the thread completed before the cancel propagated. + assert finished.is_set(), "thread did not complete before cancellation propagated"