From 373da3aea33190891082b92f886106857305588e Mon Sep 17 00:00:00 2001 From: anandhu-eng Date: Fri, 14 Aug 2026 11:59:52 +0530 Subject: [PATCH 1/4] Address review feedback on pull repo thread safety The lock gives mutual exclusion but not atomicity: nothing rolled back partial on-disk state, and the next process misread that state as healthy. Six review comments, all in mlc/repo_action.py. 1. Partial clone poisoned the path permanently (blocking). A clone killed mid-flight leaves a .git with an origin but no HEAD; the existence check then took the "already exists" branch and every later pull failed with "no tracking information". Clone now goes to a .tmp-clone sibling and is os.rename'd into place inside the lock, with cleanup on BaseException so KeyboardInterrupt also tidies up. repo_path is therefore only ever absent or complete. 2. The existing-repo branch could not tell a healthy checkout from a half-cloned one. git status exits 0 with empty stdout on the wreckage and 128 with empty stdout on a non-git directory, so both read as "clean". Added _is_valid_git_repo(), which probes `rev-parse HEAD`; anything that fails it is removed and re-cloned rather than pulled. 3. repos.json writes were non-atomic and readers never take the lock: Action.load_repos_and_meta() and Action.load_repos() both json.load it bare, so a read landing in the truncate-then-rewrite window sees a truncated file. Added _atomic_write_json() (temp file + os.replace), used by register_repo and unregister_repo. This makes the existing lock-free readers safe without touching action.py. 4. The 300s repo lock timeout was too short for a cold clone of a large repo and failed hard. Now MLC_REPO_LOCK_TIMEOUT, default 1800s, and on Timeout we re-check the repo: if the holder left a valid checkout the work is already done, so return success instead of a spurious error. Added an explicit PermissionError branch, since creating .lock needs write access to repos_path itself and the generic handler only surfaced a bare "[Errno 13]". 5. Documented the lock ordering invariant (repo lock -> repos.json lock) at all three acquisition sites, so a future path taking them in the opposite order is caught in review rather than as a mystery hang. 6. The post-register re-read: moving it inside the lock as suggested is NOT safe. load_repos_and_meta() calls rm_repo() for vanished entries, and rm_repo -> unregister_repo takes the same repos.json lock on a new FileLock instance; filelock is only reentrant per instance, so this self-deadlocks for the full 60s on every pull whose repos.json holds one stale entry. Verified empirically. Kept the read outside the lock, documented why, and closed the reported symptom -- the lookup missing the repo just registered -- by falling back to the meta already in hand. Tests: rewrote test_concurrent_pull_repo_clone_called_once, which patched subprocess.run inside each thread, so the first thread to finish restored the real one while the others were still in pull_repo and could shell out to real git. The patch is now applied once in the main thread, a barrier makes the threads actually contend, and the assertions check the returned dicts -- pull_repo returns error dicts rather than raising, so the old assertEqual(errors, []) would have stayed green on {'return': 1}. Added coverage for each new behaviour, and extracted _RepoActionTestBase so the new classes do not re-run the existing class's cases. Verified by mutation: neutering _is_valid_git_repo, making the repos.json write non-atomic, and removing the clone rollback each fail their tests; full suite is 59 passed, stable over 5 consecutive runs. Co-Authored-By: Claude Opus 5 --- mlc/repo_action.py | 176 ++++++++++++++-- tests/test_pull_repo_thread_safety.py | 279 +++++++++++++++++++++++--- 2 files changed, 420 insertions(+), 35 deletions(-) diff --git a/mlc/repo_action.py b/mlc/repo_action.py index b4a17c73d..0e9f50a3d 100644 --- a/mlc/repo_action.py +++ b/mlc/repo_action.py @@ -13,6 +13,59 @@ from .index import Index from filelock import FileLock, Timeout +# How long to wait for another process's per-repo lock before giving up. +# A cold clone of a large repo on a throttled link can run well past five +# minutes, so the default is generous and can be raised further. +REPO_LOCK_TIMEOUT_ENV = "MLC_REPO_LOCK_TIMEOUT" +DEFAULT_REPO_LOCK_TIMEOUT = 1800 + + +def _atomic_write_json(file_path, data): + """Write JSON so a concurrent reader never observes a partial file. + + open(path, 'w') truncates and json.dump rewrites incrementally, so a + reader landing in that window sees a truncated file and raises + JSONDecodeError. Action.load_repos_and_meta() and Action.load_repos() + both read repos.json with a bare json.load and no lock, so the writer's + lock alone does not protect them. Writing to a sibling temp file and + os.replace()-ing it -- atomic on POSIX and Windows -- makes those + lock-free readers safe without having to change them. + + If this process dies mid-write, os.replace never runs: the original file + is left intact and only the temp file is orphaned. + """ + tmp_path = f"{file_path}.tmp" + try: + with open(tmp_path, 'w') as f: + json.dump(data, f, indent=2) + os.replace(tmp_path, file_path) + except BaseException: + try: + os.remove(tmp_path) + except OSError: + pass + raise + + +def _get_repo_lock_timeout(): + """Seconds to wait for a per-repo lock, overridable via the environment.""" + raw = os.environ.get(REPO_LOCK_TIMEOUT_ENV, "") + if not raw: + return DEFAULT_REPO_LOCK_TIMEOUT + try: + timeout = float(raw) + except (TypeError, ValueError): + logger.warning( + f"Ignoring invalid {REPO_LOCK_TIMEOUT_ENV}={raw!r}; " + f"using {DEFAULT_REPO_LOCK_TIMEOUT}s.") + return DEFAULT_REPO_LOCK_TIMEOUT + if timeout <= 0: + logger.warning( + f"Ignoring non-positive {REPO_LOCK_TIMEOUT_ENV}={raw!r}; " + f"using {DEFAULT_REPO_LOCK_TIMEOUT}s.") + return DEFAULT_REPO_LOCK_TIMEOUT + return timeout + class RepoAction(Action): """ @@ -134,7 +187,8 @@ def _checkout_pull_branch(self, repo_path, branch): f"Initial checkout of '{branch}' failed in {repo_path}. " "After fetching from origin, creating a tracking branch also failed. " f"Initial error: {checkout_error_text}. " - f"Tracking branch error: {self._subprocess_error_message(tracking_error)}. " + f"Tracking branch error: { + self._subprocess_error_message(tracking_error)}. " "Check that the branch name is correct and that your local checkout can track origin." ) from tracking_error @@ -199,6 +253,26 @@ def _validate_extra_git_args(extra_args): ) return None + @staticmethod + def _is_valid_git_repo(repo_path): + """True only if repo_path is a git checkout with a resolvable HEAD. + + `git status` is not sufficient here: on the directory left behind by + an interrupted clone it exits 0 with empty stdout, which reads as + "clean", and on a non-git directory it exits 128 with empty stdout, + which reads the same way. `rev-parse HEAD` distinguishes both cases + because a half-cloned repo has no HEAD to resolve. + """ + if not os.path.isdir(repo_path): + return False + try: + result = subprocess.run( + ['git', '-C', repo_path, 'rev-parse', 'HEAD'], + capture_output=True, text=True) + except (OSError, subprocess.SubprocessError): + return False + return result.returncode == 0 and bool(result.stdout.strip()) + def add(self, run_args): """ #################################################################################################################### @@ -329,6 +403,10 @@ def register_repo(self, repo_path, repo_meta, ignore_on_conflict=False): repos_file_path = os.path.join(self.repos_path, 'repos.json') try: + # LOCK ORDERING: repos.json.lock is the *inner* lock -- pull_repo + # already holds .lock when it calls this. Never acquire + # a per-repo lock while holding this one, or the two orders will + # deadlock until their timeouts expire. with FileLock(_repos_lock_file(repos_file_path), timeout=60): with open(repos_file_path, 'r') as f: repos_list = json.load(f) @@ -337,9 +415,8 @@ def register_repo(self, repo_path, repo_meta, ignore_on_conflict=False): repos_list.append(repo_path) logger.info(f"Added new repo path: {repo_path}") - with open(repos_file_path, 'w') as f: - json.dump(repos_list, f, indent=2) - logger.info(f"Updated repos.json at {repos_file_path}") + _atomic_write_json(repos_file_path, repos_list) + logger.info(f"Updated repos.json at {repos_file_path}") except Timeout: return { 'return': 1, @@ -350,11 +427,24 @@ def register_repo(self, repo_path, repo_meta, ignore_on_conflict=False): ) } + # Deliberately outside the lock. load_repos_and_meta() calls rm_repo() + # for entries whose path has vanished (mlc/action.py), and rm_repo -> + # unregister_repo takes this same repos.json lock on a fresh FileLock + # instance. filelock is only reentrant per instance, so re-entering it + # here would block the process against itself for the full 60s timeout + # on every pull whose repos.json holds one stale entry. + # + # The window this leaves is that a concurrent writer may have replaced + # repos.json before the reload, so the lookup below can miss the repo + # just registered. Falling back to the meta already in hand closes + # that without widening the lock. self.repos = self.load_repos_and_meta() repo_obj = next( (r for r in self.repos if r.path == repo_path), None ) + if repo_obj is None: + repo_obj = Repo(path=repo_path, meta=repo_meta) if repo_obj: index = Action.get_index(self) @@ -504,6 +594,8 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, pat=None, ssh=None, ignore_on_conflict=False, repo_path=None, force=False, shallow=False, depth=None, extra_git_args=None, fast_forward_only=False): + repo_lock_timeout = _get_repo_lock_timeout() + # Determine the checkout path from environment or default repo_base_path = self.repos_path # either the value will be from 'MLC_REPOS' # Ensure the directory exists @@ -576,13 +668,39 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, # Lock file sits next to the repo directory; left on disk but # harmless. + # + # LOCK ORDERING: the per-repo lock is always acquired *before* + # repos.json.lock -- register_repo() is called from inside this + # block and takes that second lock. Any new code path that needs + # both must acquire them in this same order, or the two will + # deadlock until their timeouts expire. repo_lock_file = repo_path + ".lock" - with FileLock(repo_lock_file, timeout=300): + with FileLock(repo_lock_file, timeout=repo_lock_timeout): + # A directory left behind by an interrupted clone is not a + # usable repo: it has a .git with an origin but no HEAD, and + # every later pull fails against it. Treat anything that is + # not a valid checkout as absent and clone afresh. + if os.path.exists(repo_path) and not self._is_valid_git_repo( + repo_path): + logger.warning( + f"{repo_path} exists but is not a usable git checkout " + "(likely a previously interrupted clone). Removing it " + "and cloning again.") + shutil.rmtree(repo_path, ignore_errors=True) + # If the directory doesn't exist, clone it if not os.path.exists(repo_path): logger.info( f"Cloning repository {repo_url} to {repo_path}...") + # Clone into a sibling temp path and rename into place, so + # repo_path is only ever absent or complete. Without this, + # a clone killed part-way (SIGKILL/OOM/dropped link) leaves + # a half-repo that the existence check above would have to + # clean up on the *next* run. + tmp_clone_path = repo_path + ".tmp-clone" + shutil.rmtree(tmp_clone_path, ignore_errors=True) + # Build clone command clone_command = ['git', 'clone'] if branch: @@ -590,9 +708,15 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, if clone_depth is not None: clone_command += ['--depth', str(clone_depth)] clone_command += extra_args - clone_command += [repo_url, repo_path] + clone_command += [repo_url, tmp_clone_path] - subprocess.run(clone_command, check=True) + try: + subprocess.run(clone_command, check=True) + os.rename(tmp_clone_path, repo_path) + except BaseException: + # BaseException so KeyboardInterrupt also cleans up. + shutil.rmtree(tmp_clone_path, ignore_errors=True) + raise else: logger.info( @@ -793,12 +917,37 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, except subprocess.CalledProcessError as e: return {'return': 1, 'error': f"Git command failed: {e}"} except Timeout: + # Waiting out the timeout is not itself a failure: the holder may + # simply have been doing a slow cold clone and finished. If the + # repo is now a valid checkout, the work this call wanted done is + # done, so report success rather than a spurious error. + if self._is_valid_git_repo(repo_path): + logger.info( + f"Lock for {repo_path} was held by another mlc process, " + "which has left a valid checkout in place. Nothing to do.") + return {'return': 0} return { 'return': 1, 'error': ( - f"Could not acquire lock for {repo_path} after 300 seconds. " - "Another mlc process may be cloning or pulling this repo. " - "Try again once the other operation completes." + f"Could not acquire lock for {repo_path} after " + f"{repo_lock_timeout} seconds, and no valid checkout was " + "left behind. Another mlc process may still be cloning or " + "pulling this repo. Try again once it completes, or raise " + f"the timeout via {REPO_LOCK_TIMEOUT_ENV}." + ) + } + except PermissionError as e: + # Creating .lock needs write permission on the repos + # directory itself, not just on the repo. On a shared MLC_REPOS + # this is the usual cause, and the generic handler below would + # only surface a bare "[Errno 13]". + return { + 'return': 1, + 'error': ( + f"Permission denied while pulling {repo_path}: {e}. " + "Creating the lock file requires write access to " + f"{self.repos_path}; check the permissions on that " + "directory if MLC_REPOS is shared between users." ) } except Exception as e: @@ -1073,14 +1222,17 @@ def unregister_repo(repo_path, repos_file_path): logger.info(f"Unregistering the repo in path {repo_path}") try: + # LOCK ORDERING: repos.json.lock is the *inner* lock -- callers such as + # pull_repo may already hold .lock. Never acquire a per-repo + # lock while holding this one, or the two orders will deadlock until + # their timeouts expire. with FileLock(_repos_lock_file(repos_file_path), timeout=60): with open(repos_file_path, 'r') as f: repos_list = json.load(f) if repo_path in repos_list: repos_list.remove(repo_path) - with open(repos_file_path, 'w') as f: - json.dump(repos_list, f, indent=2) + _atomic_write_json(repos_file_path, repos_list) logger.info(f"Path: {repo_path} has been removed.") else: logger.info( diff --git a/tests/test_pull_repo_thread_safety.py b/tests/test_pull_repo_thread_safety.py index 66bf50296..319117640 100644 --- a/tests/test_pull_repo_thread_safety.py +++ b/tests/test_pull_repo_thread_safety.py @@ -12,11 +12,9 @@ from mlc import utils -class RegisterRepoThreadSafetyTest(unittest.TestCase): - """ - Verifies that concurrent calls to register_repo / unregister_repo do not - corrupt repos.json (no entries lost, no duplicates). - """ +class _RepoActionTestBase(unittest.TestCase): + """Shared MLC temp-environment setup. Deliberately holds no tests itself, + so subclassing it does not re-run another class's cases.""" def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() @@ -54,6 +52,13 @@ def _make_fake_repo_dir(self, name): yaml.dump(meta, f) return repo_dir, meta + +class RegisterRepoThreadSafetyTest(_RepoActionTestBase): + """ + Verifies that concurrent calls to register_repo / unregister_repo do not + corrupt repos.json (no entries lost, no duplicates). + """ + def test_concurrent_register_repo_no_data_loss(self): """ N threads each call RepoAction.register_repo with a unique repo path; @@ -150,45 +155,273 @@ def test_concurrent_pull_repo_clone_called_once(self): repo_url = "https://github.com/example/test-repo.git" repo_path = os.path.join(self.repos_path, "example@test-repo") clone_call_count = [] + count_lock = threading.Lock() errors = [] + results = [] + results_lock = threading.Lock() + # Release all threads at once so they genuinely contend for the lock. + start_together = threading.Barrier(n_threads, timeout=60) original_subprocess_run = subprocess.run + def completed(cmd, returncode=0, stdout="", stderr=""): + return subprocess.CompletedProcess(cmd, returncode, stdout, stderr) + def fake_subprocess_run(cmd, *args, **kwargs): - if cmd[0] == 'git' and len(cmd) > 1 and cmd[1] == 'clone': - clone_call_count.append(1) - # Simulate the clone by creating the directory + meta.yaml - os.makedirs(repo_path, exist_ok=True) + """Model the git calls pull_repo makes, against a fake checkout. + + The clone must create whatever destination it was handed -- that is + now a '.tmp-clone' sibling that pull_repo renames into place, not + repo_path itself -- so the destination is read from the command. + """ + if not (isinstance(cmd, (list, tuple)) + and cmd and cmd[0] == 'git'): + return original_subprocess_run(cmd, *args, **kwargs) + + if 'clone' in cmd: + with count_lock: + clone_call_count.append(1) + destination = cmd[-1] + os.makedirs(destination, exist_ok=True) + # .git marks it as a checkout for the rev-parse probe below. + os.makedirs(os.path.join(destination, '.git'), exist_ok=True) meta = { 'uid': utils.get_new_uid()['uid'], 'alias': 'example@test-repo'} - with open(os.path.join(repo_path, 'meta.yaml'), 'w') as f: + with open(os.path.join(destination, 'meta.yaml'), 'w') as f: yaml.dump(meta, f) - result = MagicMock() - result.returncode = 0 - return result - # Pass-through for any other subprocess calls - return original_subprocess_run(cmd, *args, **kwargs) + return completed(cmd) + + if 'rev-parse' in cmd: + # _is_valid_git_repo's probe: healthy only once .git exists. + target = cmd[cmd.index('-C') + 1] + if os.path.isdir(os.path.join(target, '.git')): + return completed(cmd, stdout="0" * 40 + "\n") + return completed( + cmd, returncode=128, + stderr="fatal: not a git repository\n") + + # status reports a clean tree; pull/checkout succeed silently. + return completed(cmd) def do_pull(): try: ra = self._make_repo_action() - with patch('mlc.repo_action.subprocess.run', side_effect=fake_subprocess_run): - ra.pull_repo(repo_url) + start_together.wait() + result = ra.pull_repo(repo_url) + with results_lock: + results.append(result) except Exception as exc: errors.append(exc) - threads = [threading.Thread(target=do_pull) for _ in range(n_threads)] - for t in threads: - t.start() - for t in threads: - t.join() + # Patch once, in the main thread. Patching inside each thread would let + # the first thread to finish restore the real subprocess.run while the + # others are still inside pull_repo, which would shell out to real git. + with patch('mlc.repo_action.subprocess.run', + side_effect=fake_subprocess_run): + threads = [threading.Thread(target=do_pull) + for _ in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() self.assertEqual(errors, [], msg=f"Threads raised errors: {errors}") + # pull_repo returns error dicts rather than raising, so an empty + # `errors` list on its own would not tell us the calls succeeded. + self.assertEqual(len(results), n_threads, + msg=f"Expected {n_threads} results, got {results}") + failed = [r for r in results if r.get('return', 1) != 0] + self.assertEqual( + failed, [], msg=f"pull_repo returned errors: {failed}") + self.assertEqual( len(clone_call_count), 1, - msg=f"git clone was called {len(clone_call_count)} times; expected exactly 1" + msg=f"git clone was called { + len(clone_call_count)} times; expected exactly 1" ) + self.assertTrue( + os.path.isdir(repo_path), + msg=f"{repo_path} should exist after the clone was renamed into place") + self.assertFalse( + os.path.exists(repo_path + ".tmp-clone"), + msg="temporary clone directory should not survive a successful pull") + + +class PullRepoPartialCloneTest(_RepoActionTestBase): + """Recovery from clones that died part-way through. + + The lock gives mutual exclusion but not atomicity: a clone killed by + SIGKILL/OOM/a dropped link leaves a directory with a .git but no HEAD, + which every later pull would otherwise treat as a healthy checkout. + """ + + def _fake_git(self, clone_calls, fail_clone=False): + original_subprocess_run = subprocess.run + + def fake(cmd, *args, **kwargs): + if not (isinstance(cmd, (list, tuple)) + and cmd and cmd[0] == 'git'): + return original_subprocess_run(cmd, *args, **kwargs) + if 'clone' in cmd: + clone_calls.append(cmd) + if fail_clone: + # Model an abrupt death: the destination is left behind + # half-written rather than cleaned up by git. + os.makedirs(os.path.join(cmd[-1], '.git'), exist_ok=True) + raise subprocess.CalledProcessError(128, cmd) + os.makedirs(os.path.join(cmd[-1], '.git'), exist_ok=True) + with open(os.path.join(cmd[-1], 'meta.yaml'), 'w') as f: + yaml.dump( + {'uid': utils.get_new_uid()['uid'], + 'alias': 'example@test-repo'}, f) + return subprocess.CompletedProcess(cmd, 0, "", "") + if 'rev-parse' in cmd: + target = cmd[cmd.index('-C') + 1] + # Only a directory carrying our HEAD marker is "healthy". + if os.path.exists(os.path.join(target, '.git', 'HEAD_OK')): + return subprocess.CompletedProcess( + cmd, 0, "0" * 40 + "\n", "") + return subprocess.CompletedProcess(cmd, 128, "", "fatal\n") + return subprocess.CompletedProcess(cmd, 0, "", "") + + return fake + + def test_interrupted_clone_is_discarded_and_recloned(self): + """A leftover half-clone must be removed and cloned again, not pulled.""" + repo_url = "https://github.com/example/test-repo.git" + repo_path = os.path.join(self.repos_path, "example@test-repo") + + # The wreckage an interrupted clone leaves: a .git, but no HEAD. + os.makedirs(os.path.join(repo_path, '.git'), exist_ok=True) + poison_marker = os.path.join(repo_path, 'left-over-from-interruption') + with open(poison_marker, 'w') as f: + f.write('x') + + clone_calls = [] + fake = self._fake_git(clone_calls) + + def healthy_clone(cmd, *args, **kwargs): + result = fake(cmd, *args, **kwargs) + if 'clone' in cmd: + open(os.path.join(cmd[-1], '.git', 'HEAD_OK'), 'w').close() + return result + + ra = self._make_repo_action() + with patch('mlc.repo_action.subprocess.run', + side_effect=healthy_clone): + result = ra.pull_repo(repo_url) + + self.assertEqual(result.get('return'), 0, msg=str(result)) + self.assertEqual(len(clone_calls), 1, + msg="the broken checkout should have been re-cloned") + self.assertFalse( + os.path.exists(poison_marker), + msg="leftovers from the interrupted clone were not removed") + + def test_failed_clone_leaves_no_partial_directory(self): + """A clone that dies must leave repo_path absent, not half-populated.""" + repo_url = "https://github.com/example/test-repo.git" + repo_path = os.path.join(self.repos_path, "example@test-repo") + + clone_calls = [] + ra = self._make_repo_action() + with patch('mlc.repo_action.subprocess.run', + side_effect=self._fake_git(clone_calls, fail_clone=True)): + result = ra.pull_repo(repo_url) + + self.assertNotEqual(result.get('return'), 0, + msg="a failed clone must report failure") + self.assertFalse( + os.path.exists(repo_path), + msg=f"{repo_path} must not exist after a failed clone") + self.assertFalse( + os.path.exists(repo_path + ".tmp-clone"), + msg="the temporary clone directory must be cleaned up") + + +class ReposJsonAtomicWriteTest(_RepoActionTestBase): + """repos.json must never be observable in a truncated state. + + Action.load_repos_and_meta() and Action.load_repos() read it with a bare + json.load and no lock, so the writer's lock alone does not protect them. + """ + + def test_reader_never_observes_a_truncated_file(self): + from mlc import repo_action as repo_action_module + + with open(self.repos_file) as f: + original = json.load(f) + observed = [] + real_dump = repo_action_module.json.dump + + def dump_then_peek(data, fp, *args, **kwargs): + real_dump(data, fp, *args, **kwargs) + fp.flush() + # Mid-write: a concurrent reader hitting repos.json right now must + # still see the previous complete file, not a truncated one. + with open(self.repos_file) as reader: + observed.append(json.load(reader)) + + with patch.object(repo_action_module.json, 'dump', dump_then_peek): + repo_action_module._atomic_write_json( + self.repos_file, original + ["/tmp/atomic-write-probe"]) + + self.assertEqual( + observed, [original], + msg="a reader saw something other than the previous complete file") + with open(self.repos_file) as f: + self.assertIn("/tmp/atomic-write-probe", json.load(f)) + + def test_failed_write_leaves_original_intact(self): + from mlc import repo_action as repo_action_module + + with open(self.repos_file) as f: + original = json.load(f) + + # A set is not JSON-serialisable, so json.dump raises part-way. + with self.assertRaises(TypeError): + repo_action_module._atomic_write_json( + self.repos_file, original + [{"unserialisable"}]) + + with open(self.repos_file) as f: + self.assertEqual(json.load(f), original) + self.assertFalse(os.path.exists(self.repos_file + ".tmp"), + msg="temp file left behind after a failed write") + + +class RepoLockTimeoutTest(unittest.TestCase): + """The per-repo lock timeout must be overridable for slow cold clones.""" + + def setUp(self): + self.previous = os.environ.get("MLC_REPO_LOCK_TIMEOUT") + self.addCleanup(self._restore) + + def _restore(self): + if self.previous is None: + os.environ.pop("MLC_REPO_LOCK_TIMEOUT", None) + else: + os.environ["MLC_REPO_LOCK_TIMEOUT"] = self.previous + + def test_default_and_override(self): + from mlc.repo_action import ( + _get_repo_lock_timeout, DEFAULT_REPO_LOCK_TIMEOUT) + + os.environ.pop("MLC_REPO_LOCK_TIMEOUT", None) + self.assertEqual(_get_repo_lock_timeout(), DEFAULT_REPO_LOCK_TIMEOUT) + + os.environ["MLC_REPO_LOCK_TIMEOUT"] = "7200" + self.assertEqual(_get_repo_lock_timeout(), 7200) + + def test_invalid_values_fall_back_to_default(self): + from mlc.repo_action import ( + _get_repo_lock_timeout, DEFAULT_REPO_LOCK_TIMEOUT) + + for bad in ("not-a-number", "-5", "0", ""): + os.environ["MLC_REPO_LOCK_TIMEOUT"] = bad + self.assertEqual( + _get_repo_lock_timeout(), DEFAULT_REPO_LOCK_TIMEOUT, + msg=f"{bad!r} should have fallen back to the default") if __name__ == "__main__": From 717bc1993a8563cc2641fee7eba2b6064812aa38 Mon Sep 17 00:00:00 2001 From: anandhu-eng Date: Fri, 14 Aug 2026 12:27:36 +0530 Subject: [PATCH 2/4] Fix defects found reviewing the previous commit Three critical, three medium, four minor. CRITICAL - the package did not parse on Python < 3.12. Running autopep8 under 3.13 rewrote a pre-existing line into a PEP 701 multi-line f-string expression, which is a syntax error on 3.11 and older. pyproject declares requires-python >=3.7 and CI runs 3.8/3.9/3.11, so this broke import of the whole package -- including the format workflow itself, which runs on 3.9. Reverted both occurrences and added a syntax gate to my workflow; verified mlc/ now parses on 3.10, 3.11 and 3.13. CRITICAL - _is_valid_git_repo returned False for *every* failure, so a git that could not be executed, or that refused (e.g. "detected dubious ownership" on a shared MLC_REPOS), was read as "this is junk" and pull_repo deleted the user's checkout, uncommitted work included. Replaced with _git_repo_state() returning VALID/INVALID/UNKNOWN; only INVALID is removed, UNKNOWN aborts with an explanation. CRITICAL - the Timeout handler reported success whenever any checkout existed, skipping the work actually requested: with --tag=v2 it returned {'return': 0} while the tree sat at v1, unregistered and unindexed. For MLPerf that is a silent wrong-version run, worse than the spurious error it replaced. The shortcut is now limited to the one safe case -- no branch/checkout/tag/force requested, repo already registered, checkout valid -- and errors otherwise. MEDIUM - lock ordering was not total. register_repo recurses into pull_repo for deps while holding the parent's repo lock, so the real order is .lock -> .lock -> repos.json.lock, and a self- or cyclic dependency blocked for the whole timeout (now 1800s). Added thread-local reentrancy so a lock already held by this thread is not re-acquired. Two threads with crossed deps remain a genuine inversion; that is documented, and it now surfaces as a Timeout error rather than hiding behind the false success above. MEDIUM - `git -C` searches upwards, so a plain directory nested inside another checkout answered for the enclosing repo and was classified valid; mlc would then run pull/checkout against that unrelated workspace. Now compares `rev-parse --show-toplevel` against the path itself. This also fixes a false negative: a freshly cloned empty repo has an unborn HEAD, so the old `rev-parse HEAD` probe rejected a perfectly good checkout and re-cloned it on every run. MEDIUM - rmtree(ignore_errors=True) hid failed removals and fell through to the "already exists -> pull" branch, i.e. straight back into the broken state being cleaned up. Removal is now checked and reported, and handles a file or symlink sitting where the repo should be (rmtree raises NotADirectoryError on those). MINOR - _atomic_write_json now uses mkstemp instead of a fixed .tmp name and copies the original mode across, since os.replace installs a new inode and would otherwise drop group access on a shared MLC_REPOS. PermissionError handling narrowed to lock-file creation only, via RepoLockPermissionError, rather than wrapping the whole 250-line body. rm_repo now takes the same per-repo lock as pull_repo, closing the gap where `mlc rm repo X` could delete the tree mid-clone. Tests: added coverage for each new guard, and sharpened two that did not isolate what they claimed. The clone-rollback test only checked the aftermath, so cloning straight into repo_path still passed it; it now asserts repo_path is absent *while* the clone runs. The timeout tests were gated by the registration check, so the revision-request guard was untested; they now register the repo first. All mutants killed: tmp-clone -> repo_path, UNKNOWN treated as INVALID, Timeout -> unconditional success, and dropping the toplevel comparison each fail their tests. 69 passed, stable over 5 consecutive runs. Co-Authored-By: Claude Opus 5 --- mlc/repo_action.py | 285 ++++++++++++++++++++++---- tests/test_pull_repo_thread_safety.py | 245 +++++++++++++++++++++- 2 files changed, 478 insertions(+), 52 deletions(-) diff --git a/mlc/repo_action.py b/mlc/repo_action.py index 0e9f50a3d..b96313161 100644 --- a/mlc/repo_action.py +++ b/mlc/repo_action.py @@ -1,6 +1,9 @@ from .action import Action +import contextlib import os import subprocess +import tempfile +import threading import re import shlex import yaml @@ -20,6 +23,55 @@ DEFAULT_REPO_LOCK_TIMEOUT = 1800 +# Per-repo lock paths currently held by this thread. register_repo() can +# recurse back into pull_repo() for a dependency while the parent's lock is +# still held, so without this a repo that (transitively) depends on itself +# would block against its own lock for the full timeout. +# +# This does NOT rescue two *threads* pulling repos with crossed dependencies +# (A deps B, B deps A): that is a genuine lock-order inversion between two +# per-repo locks and still resolves only by timing out. It surfaces as a +# Timeout error rather than silently, which is why the Timeout handler in +# pull_repo must not report success indiscriminately. +_held_repo_locks = threading.local() + + +def _repo_locks_held_by_this_thread(): + held = getattr(_held_repo_locks, "paths", None) + if held is None: + held = set() + _held_repo_locks.paths = held + return held + + +class RepoLockPermissionError(Exception): + """Raised only when the per-repo lock file itself cannot be created.""" + + +@contextlib.contextmanager +def _repo_lock(repo_lock_file, timeout): + """Acquire a per-repo lock, tolerating same-thread re-entry.""" + held = _repo_locks_held_by_this_thread() + if repo_lock_file in held: + # Already ours further up the call stack (dependency recursion). + yield + return + try: + lock = FileLock(repo_lock_file, timeout=timeout) + lock.acquire() + except PermissionError as e: + # Narrow: only failures creating the lock file. Catching + # PermissionError around the whole pull would mislabel an EACCES + # from git, rmtree or reading meta.yaml as a lock problem. + raise RepoLockPermissionError(str(e)) from e + held.add(repo_lock_file) + try: + yield + finally: + held.discard(repo_lock_file) + lock.release() + + def _atomic_write_json(file_path, data): """Write JSON so a concurrent reader never observes a partial file. @@ -34,10 +86,22 @@ def _atomic_write_json(file_path, data): If this process dies mid-write, os.replace never runs: the original file is left intact and only the temp file is orphaned. """ - tmp_path = f"{file_path}.tmp" + directory = os.path.dirname(file_path) or '.' + # A unique temp name rather than a fixed ".tmp": callers happen to + # hold a lock today, but a helper named "atomic write" should not depend + # on that to avoid two writers trampling the same scratch file. + fd, tmp_path = tempfile.mkstemp( + dir=directory, prefix=os.path.basename(file_path) + '.', suffix='.tmp') try: - with open(tmp_path, 'w') as f: + with os.fdopen(fd, 'w') as f: json.dump(data, f, indent=2) + # os.replace installs a new inode, so mode/ownership would otherwise + # be whatever mkstemp chose (0600) rather than the file's own -- on a + # shared MLC_REPOS that would silently drop group access. + try: + shutil.copymode(file_path, tmp_path) + except OSError: + pass os.replace(tmp_path, file_path) except BaseException: try: @@ -187,8 +251,8 @@ def _checkout_pull_branch(self, repo_path, branch): f"Initial checkout of '{branch}' failed in {repo_path}. " "After fetching from origin, creating a tracking branch also failed. " f"Initial error: {checkout_error_text}. " - f"Tracking branch error: { - self._subprocess_error_message(tracking_error)}. " + f"Tracking branch error: " + f"{self._subprocess_error_message(tracking_error)}. " "Check that the branch name is correct and that your local checkout can track origin." ) from tracking_error @@ -253,25 +317,93 @@ def _validate_extra_git_args(extra_args): ) return None - @staticmethod - def _is_valid_git_repo(repo_path): - """True only if repo_path is a git checkout with a resolvable HEAD. - - `git status` is not sufficient here: on the directory left behind by - an interrupted clone it exits 0 with empty stdout, which reads as - "clean", and on a non-git directory it exits 128 with empty stdout, - which reads the same way. `rev-parse HEAD` distinguishes both cases - because a half-cloned repo has no HEAD to resolve. + # Return values of _git_repo_state(). + GIT_STATE_VALID = "valid" # a git checkout rooted exactly here + GIT_STATE_INVALID = "invalid" # definitely not a checkout; safe to remove + GIT_STATE_UNKNOWN = "unknown" # git could not answer; DO NOT remove + + # git's phrasing when a path is genuinely not a repository. Anything else + # on a non-zero exit (dubious ownership, EACCES, ...) is "unknown". + GIT_NOT_A_REPO_PHRASE = "not a git repository" + + @classmethod + def _git_repo_state(cls, repo_path): + """Classify repo_path as a git checkout, without ever guessing. + + `git status` cannot be used: on the directory left by an interrupted + clone it exits 0 with empty stdout ("clean"), and on a non-git + directory it exits 128 with empty stdout -- indistinguishable. + + `rev-parse --show-toplevel` is used rather than `rev-parse HEAD` for + two reasons: + * a freshly cloned *empty* repo has an unborn HEAD, so `rev-parse + HEAD` fails on a perfectly good checkout; + * `git -C` searches upwards, so a plain directory nested inside + another checkout answers for the *enclosing* repo. Comparing the + reported top level against repo_path rejects that. + + The INVALID/UNKNOWN split matters because callers delete on INVALID. + A git that cannot be executed, or that refuses (e.g. "detected + dubious ownership" on a shared MLC_REPOS), must never be read as + "this is junk, remove it". """ - if not os.path.isdir(repo_path): - return False + if os.path.islink(repo_path) or not os.path.isdir(repo_path): + # A symlink, a plain file, or nothing at all: not a checkout, and + # not something to hand to git. + return (cls.GIT_STATE_INVALID if os.path.lexists(repo_path) + else cls.GIT_STATE_INVALID) try: result = subprocess.run( - ['git', '-C', repo_path, 'rev-parse', 'HEAD'], + ['git', '-C', repo_path, 'rev-parse', '--show-toplevel'], capture_output=True, text=True) - except (OSError, subprocess.SubprocessError): - return False - return result.returncode == 0 and bool(result.stdout.strip()) + except (OSError, subprocess.SubprocessError) as e: + logger.warning( + f"Could not run git to inspect {repo_path}: {e}. " + "Leaving the directory untouched.") + return cls.GIT_STATE_UNKNOWN + + if result.returncode == 0: + toplevel = (result.stdout or "").strip() + if not toplevel: + return cls.GIT_STATE_UNKNOWN + try: + same = os.path.samefile(toplevel, repo_path) + except OSError: + same = (os.path.realpath(toplevel) + == os.path.realpath(repo_path)) + # A non-repo directory sitting inside another checkout reports the + # enclosing repo here; that is not a checkout *at* repo_path. + return cls.GIT_STATE_VALID if same else cls.GIT_STATE_INVALID + + stderr = (result.stderr or "").lower() + if cls.GIT_NOT_A_REPO_PHRASE in stderr: + return cls.GIT_STATE_INVALID + logger.warning( + f"git could not classify {repo_path} " + f"(exit {result.returncode}): {(result.stderr or '').strip()}. " + "Leaving the directory untouched.") + return cls.GIT_STATE_UNKNOWN + + @classmethod + def _is_valid_git_repo(cls, repo_path): + """True only when repo_path is a git checkout rooted exactly there.""" + return cls._git_repo_state(repo_path) == cls.GIT_STATE_VALID + + @staticmethod + def _remove_broken_checkout(path): + """Remove a path that is known not to be a usable checkout. + + Handles the non-directory cases too: a stale symlink or a plain file + sitting where the repo should be would make rmtree raise + NotADirectoryError. + """ + try: + if os.path.islink(path) or os.path.isfile(path): + os.remove(path) + else: + shutil.rmtree(path) + except OSError as e: + logger.warning(f"Failed to remove {path}: {e}") def add(self, run_args): """ @@ -675,21 +807,48 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, # both must acquire them in this same order, or the two will # deadlock until their timeouts expire. repo_lock_file = repo_path + ".lock" - with FileLock(repo_lock_file, timeout=repo_lock_timeout): + with _repo_lock(repo_lock_file, repo_lock_timeout): # A directory left behind by an interrupted clone is not a # usable repo: it has a .git with an origin but no HEAD, and - # every later pull fails against it. Treat anything that is - # not a valid checkout as absent and clone afresh. - if os.path.exists(repo_path) and not self._is_valid_git_repo( - repo_path): - logger.warning( - f"{repo_path} exists but is not a usable git checkout " - "(likely a previously interrupted clone). Removing it " - "and cloning again.") - shutil.rmtree(repo_path, ignore_errors=True) + # every later pull fails against it. Such a directory is + # removed and cloned afresh. + # + # Only GIT_STATE_INVALID is removed. If git could not be run + # or refused to answer we do NOT delete: that path would + # destroy a healthy checkout -- including uncommitted work -- + # over a missing git binary or a "dubious ownership" refusal + # on a shared MLC_REPOS. + if os.path.lexists(repo_path): + repo_state = self._git_repo_state(repo_path) + if repo_state == self.GIT_STATE_UNKNOWN: + return { + 'return': 1, + 'error': ( + f"Could not determine whether {repo_path} is a " + "valid git checkout; refusing to touch it. See " + "the warning above for what git reported." + ) + } + if repo_state == self.GIT_STATE_INVALID: + logger.warning( + f"{repo_path} exists but is not a usable git " + "checkout (likely a previously interrupted " + "clone). Removing it and cloning again.") + self._remove_broken_checkout(repo_path) + # ignore_errors would hide a failed removal and drop us + # into the "already exists -> pull" branch below, which + # is exactly the broken state being cleaned up. + if os.path.lexists(repo_path): + return { + 'return': 1, + 'error': ( + f"Could not remove the unusable checkout at " + f"{repo_path}. Remove it manually and retry." + ) + } # If the directory doesn't exist, clone it - if not os.path.exists(repo_path): + if not os.path.lexists(repo_path): logger.info( f"Cloning repository {repo_url} to {repo_path}...") @@ -917,26 +1076,33 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, except subprocess.CalledProcessError as e: return {'return': 1, 'error': f"Git command failed: {e}"} except Timeout: - # Waiting out the timeout is not itself a failure: the holder may - # simply have been doing a slow cold clone and finished. If the - # repo is now a valid checkout, the work this call wanted done is - # done, so report success rather than a spurious error. - if self._is_valid_git_repo(repo_path): + # A timeout can mean the holder simply finished a slow cold clone + # for us. Reporting success on that basis is only safe when this + # call had nothing version-specific to do -- _is_valid_git_repo + # answers "some checkout exists here", not "it is at the revision + # you asked for". Claiming success while the tree sits at the + # wrong tag would silently run the wrong code, which is far worse + # than a spurious error. + version_specific = any((branch, checkout, tag, force)) + already_registered = repo_path in (self.load_repos() or []) + if (not version_specific and already_registered + and self._is_valid_git_repo(repo_path)): logger.info( f"Lock for {repo_path} was held by another mlc process, " - "which has left a valid checkout in place. Nothing to do.") + "which left a valid registered checkout in place and no " + "specific revision was requested. Nothing to do.") return {'return': 0} return { 'return': 1, 'error': ( f"Could not acquire lock for {repo_path} after " - f"{repo_lock_timeout} seconds, and no valid checkout was " - "left behind. Another mlc process may still be cloning or " - "pulling this repo. Try again once it completes, or raise " - f"the timeout via {REPO_LOCK_TIMEOUT_ENV}." + f"{repo_lock_timeout} seconds. Another mlc process may " + "still be cloning or pulling this repo. Try again once it " + f"completes, or raise the timeout via " + f"{REPO_LOCK_TIMEOUT_ENV}." ) } - except PermissionError as e: + except RepoLockPermissionError as e: # Creating .lock needs write permission on the repos # directory itself, not just on the repo. On a shared MLC_REPOS # this is the usual cause, and the generic handler below would @@ -944,8 +1110,8 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, return { 'return': 1, 'error': ( - f"Permission denied while pulling {repo_path}: {e}. " - "Creating the lock file requires write access to " + f"Permission denied creating the lock file for " + f"{repo_path}: {e}. This requires write access to " f"{self.repos_path}; check the permissions on that " "directory if MLC_REPOS is shared between users." ) @@ -1164,9 +1330,38 @@ def _repos_lock_file(repos_file_path): def rm_repo(repo_path, repos_file_path, force_remove): - logger.info( - "rm command has been called for repo. This would delete the repo folder and unregister the repo from repos.json") + """Remove a repo directory and unregister it. + + Takes the same per-repo lock pull_repo uses. Without it, `mlc rm repo X` + running against a concurrent `mlc pull repo X` would delete the tree while + the pull is mid-clone or mid-rename -- the pull's lock only excluded other + pullers. + + LOCK ORDERING: per-repo lock first, then repos.json.lock (taken by + unregister_repo below). Same order as pull_repo. + """ + try: + with _repo_lock(repo_path + ".lock", _get_repo_lock_timeout()): + return _rm_repo_locked(repo_path, repos_file_path, force_remove) + except Timeout: + return { + 'return': 1, + 'error': ( + f"Could not acquire lock for {repo_path} before removing it. " + "Another mlc process may be cloning or pulling this repo." + ) + } + except RepoLockPermissionError as e: + return { + 'return': 1, + 'error': ( + f"Permission denied creating the lock file for {repo_path}: " + f"{e}. This requires write access to the repos directory." + ) + } + +def _rm_repo_locked(repo_path, repos_file_path, force_remove): repo_name = os.path.basename(repo_path) mlc_repos_path = os.path.abspath(os.path.dirname(repos_file_path)) repo_parent_path = os.path.abspath(os.path.dirname(repo_path)) diff --git a/tests/test_pull_repo_thread_safety.py b/tests/test_pull_repo_thread_safety.py index 319117640..c415d979c 100644 --- a/tests/test_pull_repo_thread_safety.py +++ b/tests/test_pull_repo_thread_safety.py @@ -6,6 +6,7 @@ import unittest import yaml from unittest.mock import patch, MagicMock +from filelock import FileLock from mlc.repo_action import unregister_repo, RepoAction from mlc.action import Action @@ -193,10 +194,11 @@ def fake_subprocess_run(cmd, *args, **kwargs): return completed(cmd) if 'rev-parse' in cmd: - # _is_valid_git_repo's probe: healthy only once .git exists. + # _git_repo_state's probe. --show-toplevel echoes the repo + # root, which the caller compares against repo_path. target = cmd[cmd.index('-C') + 1] if os.path.isdir(os.path.join(target, '.git')): - return completed(cmd, stdout="0" * 40 + "\n") + return completed(cmd, stdout=target + "\n") return completed( cmd, returncode=128, stderr="fatal: not a git repository\n") @@ -237,8 +239,8 @@ def do_pull(): self.assertEqual( len(clone_call_count), 1, - msg=f"git clone was called { - len(clone_call_count)} times; expected exactly 1" + msg=(f"git clone was called {len(clone_call_count)} times; " + "expected exactly 1") ) self.assertTrue( os.path.isdir(repo_path), @@ -278,11 +280,16 @@ def fake(cmd, *args, **kwargs): return subprocess.CompletedProcess(cmd, 0, "", "") if 'rev-parse' in cmd: target = cmd[cmd.index('-C') + 1] - # Only a directory carrying our HEAD marker is "healthy". + # Only a directory carrying our HEAD marker is "healthy"; + # anything else is definitively not a repository, so the + # caller is allowed to remove it. if os.path.exists(os.path.join(target, '.git', 'HEAD_OK')): return subprocess.CompletedProcess( - cmd, 0, "0" * 40 + "\n", "") - return subprocess.CompletedProcess(cmd, 128, "", "fatal\n") + cmd, 0, target + "\n", "") + return subprocess.CompletedProcess( + cmd, 128, "", + "fatal: not a git repository (or any of the parent " + "directories): .git\n") return subprocess.CompletedProcess(cmd, 0, "", "") return fake @@ -319,6 +326,48 @@ def healthy_clone(cmd, *args, **kwargs): os.path.exists(poison_marker), msg="leftovers from the interrupted clone were not removed") + def test_repo_path_is_absent_while_the_clone_is_running(self): + """The clone must land somewhere else and be renamed into place. + + Asserting only on the aftermath is not enough: cloning straight into + repo_path with rollback on failure leaves the same end state, but a + process killed mid-clone (no rollback runs) still poisons the path. + The invariant that actually matters is that repo_path does not exist + *while* the clone is in flight. + """ + repo_url = "https://github.com/example/test-repo.git" + repo_path = os.path.join(self.repos_path, "example@test-repo") + observed_during_clone = {} + + clone_calls = [] + base_fake = self._fake_git(clone_calls) + + def fake(cmd, *args, **kwargs): + if (isinstance(cmd, (list, tuple)) and cmd + and cmd[0] == 'git' and 'clone' in cmd): + observed_during_clone['destination'] = cmd[-1] + observed_during_clone['repo_path_exists'] = os.path.lexists( + repo_path) + result = base_fake(cmd, *args, **kwargs) + if isinstance(cmd, (list, tuple)) and cmd and 'clone' in cmd: + open(os.path.join(cmd[-1], '.git', 'HEAD_OK'), 'w').close() + return result + + ra = self._make_repo_action() + with patch('mlc.repo_action.subprocess.run', side_effect=fake): + result = ra.pull_repo(repo_url) + + self.assertEqual(result.get('return'), 0, msg=str(result)) + self.assertNotEqual( + observed_during_clone.get('destination'), repo_path, + msg="clone wrote directly into repo_path instead of a temp path") + self.assertFalse( + observed_during_clone.get('repo_path_exists', True), + msg="repo_path existed while the clone was still running; an " + "abrupt kill would leave a half-repo behind") + self.assertTrue(os.path.isdir(repo_path), + msg="clone was never renamed into place") + def test_failed_clone_leaves_no_partial_directory(self): """A clone that dies must leave repo_path absent, not half-populated.""" repo_url = "https://github.com/example/test-repo.git" @@ -390,6 +439,188 @@ def test_failed_write_leaves_original_intact(self): msg="temp file left behind after a failed write") +class GitRepoStateTest(unittest.TestCase): + """_git_repo_state must never answer "junk" when it simply cannot tell -- + callers delete on that answer.""" + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + + def _git(self, *args, cwd=None): + return subprocess.run(['git'] + list(args), cwd=cwd, + capture_output=True, text=True, check=True) + + def test_real_git_classification(self): + """Exercised against real git rather than a fake, since the whole + point is matching git's actual behaviour.""" + from mlc.repo_action import RepoAction + + base = self.temp_dir.name + + valid = os.path.join(base, 'valid') + os.makedirs(valid) + self._git('init', '-q', cwd=valid) + self._git('commit', '-q', '--allow-empty', '-m', 'x', cwd=valid) + + # A freshly cloned *empty* repo has an unborn HEAD but is perfectly + # usable -- `rev-parse HEAD` would wrongly reject it. + empty = os.path.join(base, 'empty') + os.makedirs(empty) + self._git('init', '-q', cwd=empty) + + half = os.path.join(base, 'half', '.git') + os.makedirs(half) + + plain = os.path.join(base, 'plain') + os.makedirs(plain) + + # git -C searches upwards, so this answers for `valid` unless the + # reported top level is compared against the path asked about. + nested = os.path.join(valid, 'nested') + os.makedirs(nested) + + dangling = os.path.join(base, 'dangling') + os.symlink(os.path.join(base, 'nowhere'), dangling) + + a_file = os.path.join(base, 'a-file') + with open(a_file, 'w') as f: + f.write('x') + + cases = [ + (valid, RepoAction.GIT_STATE_VALID), + (empty, RepoAction.GIT_STATE_VALID), + (os.path.dirname(half), RepoAction.GIT_STATE_INVALID), + (plain, RepoAction.GIT_STATE_INVALID), + (nested, RepoAction.GIT_STATE_INVALID), + (dangling, RepoAction.GIT_STATE_INVALID), + (a_file, RepoAction.GIT_STATE_INVALID), + ] + for path, expected in cases: + with self.subTest(path=os.path.basename(path)): + self.assertEqual(RepoAction._git_repo_state(path), expected) + + def test_unrunnable_git_is_unknown_not_invalid(self): + """If git cannot be executed we must not conclude "this is junk".""" + from mlc.repo_action import RepoAction + + path = os.path.join(self.temp_dir.name, 'repo') + os.makedirs(path) + with patch('mlc.repo_action.subprocess.run', + side_effect=OSError(2, 'No such file or directory: git')): + self.assertEqual(RepoAction._git_repo_state(path), + RepoAction.GIT_STATE_UNKNOWN) + + def test_refused_git_is_unknown_not_invalid(self): + """git refusing (e.g. dubious ownership on a shared MLC_REPOS) exits + non-zero but is not a statement that the path is not a repo.""" + from mlc.repo_action import RepoAction + + path = os.path.join(self.temp_dir.name, 'repo') + os.makedirs(path) + refusal = subprocess.CompletedProcess( + [], 128, "", + "fatal: detected dubious ownership in repository at '/x'\n") + with patch('mlc.repo_action.subprocess.run', return_value=refusal): + self.assertEqual(RepoAction._git_repo_state(path), + RepoAction.GIT_STATE_UNKNOWN) + + +class PullRepoDestructiveGuardTest(_RepoActionTestBase): + """pull_repo must not delete a checkout it could not classify.""" + + def test_unknown_state_preserves_the_checkout(self): + repo_path = os.path.join(self.repos_path, "example@test-repo") + os.makedirs(repo_path) + precious = os.path.join(repo_path, "UNCOMMITTED_WORK.txt") + with open(precious, 'w') as f: + f.write("do not lose me") + + ra = self._make_repo_action() + with patch('mlc.repo_action.subprocess.run', + side_effect=OSError(2, "No such file or directory: 'git'")): + result = ra.pull_repo("https://github.com/example/test-repo.git") + + self.assertNotEqual(result.get('return'), 0, + msg="an unclassifiable checkout must be an error") + self.assertTrue( + os.path.exists(precious), + msg="pull_repo deleted a checkout it could not classify") + + +class PullRepoTimeoutSemanticsTest(_RepoActionTestBase): + """Timing out must not be reported as success when work was skipped.""" + + def _hold_lock_and_pull(self, registered=False, **pull_kwargs): + repo_path = os.path.join(self.repos_path, "example@test-repo") + # A valid-looking checkout is already in place. + os.makedirs(os.path.join(repo_path, '.git'), exist_ok=True) + + if registered: + # Register it so the "nothing left to do" shortcut is gated only + # by whether a specific revision was requested. + with open(self.repos_file) as f: + entries = json.load(f) + with open(self.repos_file, 'w') as f: + json.dump(entries + [repo_path], f, indent=2) + + def fake_run(cmd, *a, **kw): + if (isinstance(cmd, (list, tuple)) and cmd + and cmd[0] == 'git' and 'rev-parse' in cmd): + target = cmd[cmd.index('-C') + 1] + return subprocess.CompletedProcess(cmd, 0, target + "\n", "") + return subprocess.CompletedProcess(cmd, 0, "", "") + + os.environ["MLC_REPO_LOCK_TIMEOUT"] = "1" + self.addCleanup(os.environ.pop, "MLC_REPO_LOCK_TIMEOUT", None) + + ra = self._make_repo_action() + holder = FileLock(repo_path + ".lock", timeout=30) + holder.acquire() + try: + with patch('mlc.repo_action.subprocess.run', side_effect=fake_run): + return ra.pull_repo( + "https://github.com/example/test-repo.git", **pull_kwargs) + finally: + holder.release() + + def test_timeout_with_requested_tag_is_an_error(self): + """The repo is present AND registered, so only the revision request + stands between this and a false success. _is_valid_git_repo says "a + checkout exists", not "it is at the tag you asked for" -- reporting + success here would silently run the wrong revision.""" + result = self._hold_lock_and_pull(registered=True, tag="v2") + self.assertNotEqual( + result.get('return'), 0, + msg="timing out while a specific tag was requested must not " + "report success") + + def test_timeout_with_requested_branch_is_an_error(self): + result = self._hold_lock_and_pull( + registered=True, branch="some-branch") + self.assertNotEqual(result.get('return'), 0) + + def test_timeout_with_requested_checkout_is_an_error(self): + result = self._hold_lock_and_pull(registered=True, checkout="abc1234") + self.assertNotEqual(result.get('return'), 0) + + def test_timeout_on_unregistered_repo_is_an_error(self): + """Even with no revision requested, the repo still has to be + registered in repos.json for there to be nothing left to do.""" + result = self._hold_lock_and_pull(registered=False) + self.assertNotEqual( + result.get('return'), 0, + msg="repo was never registered, so this was not a no-op") + + def test_timeout_on_registered_repo_with_no_revision_is_a_noop(self): + """The one case where reporting success is legitimate.""" + result = self._hold_lock_and_pull(registered=True) + self.assertEqual( + result.get('return'), 0, + msg="another process left a valid registered checkout and no " + "revision was requested; there was genuinely nothing to do") + + class RepoLockTimeoutTest(unittest.TestCase): """The per-repo lock timeout must be overridable for slow cold clones.""" From 5996ac7bff28bb8afe7c1043fcc5f620cf779261 Mon Sep 17 00:00:00 2001 From: anandhu-eng Date: Fri, 14 Aug 2026 12:55:12 +0530 Subject: [PATCH 3/4] Fix regressions introduced by the previous commit's own fixes A second review pass found that two of the fixes in 717bc19 caused worse problems than the ones they closed. CRITICAL - locking rm_repo broke every mlc command. rm_repo was previously infallible, so action.py's `if res["return"] > 0: return res` was dead code. Adding the lock made it fallible, and load_repos_and_meta() then returns that error *dict* where its callers expect a *list* of Repo objects -- including Action.__init__, so self.repos becomes a dict. The trigger is ordinary: a vanished repos.json entry whose lock is held, which is exactly the state during any first-time clone, since the tmp-clone scheme guarantees repo_path does not exist for the clone's duration. Measured: pulling an unrelated repo stalled for the full timeout (1800s in production) and then failed with "'str' object has no attribute 'meta'", leaving the repo it had successfully cloned unregistered. Locking there was also wrong for two further reasons: it blocks pruning behind an unrelated repo's lock, and filelock creates the lock file's parent, so pruning a vanished path *recreated* its directory -- on an unmounted volume, materialising the mount point. rm_repo is now unlocked and infallible again, with a docstring saying why. The lock moved to RepoAction.rm(), the user-facing `mlc rm repo` path, which is where the delete-during-clone hazard actually lives. MEDIUM - the dangling-symlink fix overshot. Short-circuiting on os.path.islink() before consulting git classified a *live* symlink to a healthy checkout as junk, so relocating a repo to another volume and symlinking it back got the link unlinked and the repo silently re-cloned into MLC_REPOS, stranding the real checkout and any local work in it. Now only a dangling link short-circuits; a live one falls through to git, and samefile() resolves it correctly. MEDIUM - reentrancy traded the deadlock for a recursion storm. It stops pull_repo blocking on its own lock, but nothing stopped the *recursion*: A deps B, B deps A produced 1466 git invocations including 488 pulls, each a network round-trip with real git, and reported success. Added a thread-local cycle guard (_repo_pull_lock) that returns early when a repo is already being pulled further up the same stack. MINOR - _git_repo_state had a conditional whose arms were identical. Rewritten so "absent", "dangling link" and "plain file" each return INVALID explicitly. Lock files are now keyed on a normalised path, since the key is both the cross-process exclusion token and the reentrancy key: "/r" and "/./r" previously took different locks for the same directory, so the exclusion silently did nothing. A test was popping MLC_REPO_LOCK_TIMEOUT instead of restoring it, clobbering any ambient value. Tests: the review found three behaviours from the previous commit with no coverage at all -- reentrancy, the rm lock, and copymode -- each of which survived mutation. Added RepoLockMechanicsTest, RmRepoLockingTest and AtomicWriteMetadataTest, plus a live-symlink case that the classification test was missing. All seven mutants now killed: reentrancy removed, rm lock removed, copymode dropped, live symlink to INVALID, cycle guard removed, rm_repo made fallible, and lock key unnormalised. 77 passed, stable over 4 runs, and mlc/ + tests/ verified parsing on 3.10 and 3.11. Co-Authored-By: Claude Opus 5 --- mlc/repo_action.py | 144 ++++++++++++++------ tests/test_pull_repo_thread_safety.py | 184 +++++++++++++++++++++++++- 2 files changed, 290 insertions(+), 38 deletions(-) diff --git a/mlc/repo_action.py b/mlc/repo_action.py index b96313161..b0e0df8ee 100644 --- a/mlc/repo_action.py +++ b/mlc/repo_action.py @@ -48,6 +48,52 @@ class RepoLockPermissionError(Exception): """Raised only when the per-repo lock file itself cannot be created.""" +def _repo_lock_path(repo_path): + """Lock-file path for a repo, normalised so aliases share one lock. + + The lock file is both the cross-process mutual-exclusion token and the + within-thread reentrancy key, so two spellings of the same repo + ("/r" vs "/./r", or a symlinked MLC_REPOS) must map to one + name -- otherwise two processes take different locks for the same + directory and the exclusion silently does nothing. Only the parent is + resolved, so the lock still sits beside repo_path when repo_path is + itself a symlink. + """ + parent = os.path.dirname(os.path.abspath(repo_path)) + return os.path.join( + os.path.realpath(parent), os.path.basename(repo_path)) + ".lock" + + +# Repos whose pull is already in progress on this thread. register_repo() +# recurses into pull_repo() for each dependency, so a dependency cycle +# (A -> B -> A) would otherwise recurse until the stack blows, running a git +# pull and a repos.json rewrite at every level. The per-repo lock cannot stop +# this on its own: same-thread reentrancy makes the nested acquire a no-op by +# design, which is exactly what turns the old deadlock into a recursion storm. +_pulling_repos = threading.local() + + +@contextlib.contextmanager +def _repo_pull_lock(repo_lock_file, timeout): + """Hold the per-repo lock, yielding False if this repo is already being + pulled further up the same call stack (dependency cycle).""" + active = getattr(_pulling_repos, "paths", None) + if active is None: + active = set() + _pulling_repos.paths = active + + if repo_lock_file in active: + yield False + return + + with _repo_lock(repo_lock_file, timeout): + active.add(repo_lock_file) + try: + yield True + finally: + active.discard(repo_lock_file) + + @contextlib.contextmanager def _repo_lock(repo_lock_file, timeout): """Acquire a per-repo lock, tolerating same-thread re-entry.""" @@ -347,11 +393,18 @@ def _git_repo_state(cls, repo_path): dubious ownership" on a shared MLC_REPOS), must never be read as "this is junk, remove it". """ - if os.path.islink(repo_path) or not os.path.isdir(repo_path): - # A symlink, a plain file, or nothing at all: not a checkout, and - # not something to hand to git. - return (cls.GIT_STATE_INVALID if os.path.lexists(repo_path) - else cls.GIT_STATE_INVALID) + if not os.path.lexists(repo_path): + return cls.GIT_STATE_INVALID # nothing there at all + if os.path.islink(repo_path) and not os.path.exists(repo_path): + return cls.GIT_STATE_INVALID # dangling symlink + if not os.path.isdir(repo_path): + return cls.GIT_STATE_INVALID # a plain file + # NOTE: a *live* symlink to a real checkout must fall through to git. + # Relocating a repo onto a bigger volume and symlinking it back is a + # normal thing to do; short-circuiting on islink here would classify + # it as junk and unlink it, stranding the real checkout and any local + # work in it. os.path.samefile() below follows symlinks, so the + # top-level comparison resolves such a link correctly. try: result = subprocess.run( ['git', '-C', repo_path, 'rev-parse', '--show-toplevel'], @@ -806,8 +859,14 @@ def pull_repo(self, repo_url, branch=None, checkout=None, tag=None, # block and takes that second lock. Any new code path that needs # both must acquire them in this same order, or the two will # deadlock until their timeouts expire. - repo_lock_file = repo_path + ".lock" - with _repo_lock(repo_lock_file, repo_lock_timeout): + repo_lock_file = _repo_lock_path(repo_path) + with _repo_pull_lock(repo_lock_file, + repo_lock_timeout) as proceed: + if not proceed: + logger.debug( + f"{repo_path} is already being pulled further up this " + "call stack (dependency cycle); skipping.") + return {'return': 0} # A directory left behind by an interrupted clone is not a # usable repo: it has a .git with an origin but no HEAD, and # every later pull fails against it. Such a directory is @@ -1321,7 +1380,36 @@ def rm(self, run_args): index = Action.get_index(self) index.remove_repo_from_index(repo_path) - return rm_repo(repo_path, repos_file_path, force_remove) + # Same per-repo lock pull_repo uses: without it, `mlc rm repo X` + # racing a concurrent `mlc pull repo X` deletes the tree while the + # pull is mid-clone or mid-rename, since the pull's lock only excluded + # other pullers. See rm_repo's docstring for why it is here and not + # there. + # + # LOCK ORDERING: per-repo lock first, then repos.json.lock (taken by + # unregister_repo). Same order as pull_repo. + try: + with _repo_lock(_repo_lock_path(repo_path), + _get_repo_lock_timeout()): + return rm_repo(repo_path, repos_file_path, force_remove) + except Timeout: + return { + 'return': 1, + 'error': ( + f"Could not acquire lock for {repo_path} before removing " + "it. Another mlc process may be cloning or pulling this " + "repo." + ) + } + except RepoLockPermissionError as e: + return { + 'return': 1, + 'error': ( + f"Permission denied creating the lock file for " + f"{repo_path}: {e}. This requires write access to the " + "repos directory." + ) + } def _repos_lock_file(repos_file_path): @@ -1332,36 +1420,18 @@ def _repos_lock_file(repos_file_path): def rm_repo(repo_path, repos_file_path, force_remove): """Remove a repo directory and unregister it. - Takes the same per-repo lock pull_repo uses. Without it, `mlc rm repo X` - running against a concurrent `mlc pull repo X` would delete the tree while - the pull is mid-clone or mid-rename -- the pull's lock only excluded other - pullers. - - LOCK ORDERING: per-repo lock first, then repos.json.lock (taken by - unregister_repo below). Same order as pull_repo. + Deliberately NOT locked, and must stay infallible. + Action.load_repos_and_meta() calls this to prune entries whose directory + has vanished and returns whatever this returns straight to its caller -- + where a *list* of Repo objects is expected (mlc/action.py). Making this + fallible turns a lock timeout into `self.repos` being an error dict, which + breaks every mlc command. Locking here would also block pruning behind an + unrelated repo's lock and, since filelock creates the lock file's parent, + recreate directories that have deliberately gone away. + + The user-facing `mlc rm repo` path takes the per-repo lock in + RepoAction.rm(), which is where the delete-during-clone hazard lives. """ - try: - with _repo_lock(repo_path + ".lock", _get_repo_lock_timeout()): - return _rm_repo_locked(repo_path, repos_file_path, force_remove) - except Timeout: - return { - 'return': 1, - 'error': ( - f"Could not acquire lock for {repo_path} before removing it. " - "Another mlc process may be cloning or pulling this repo." - ) - } - except RepoLockPermissionError as e: - return { - 'return': 1, - 'error': ( - f"Permission denied creating the lock file for {repo_path}: " - f"{e}. This requires write access to the repos directory." - ) - } - - -def _rm_repo_locked(repo_path, repos_file_path, force_remove): repo_name = os.path.basename(repo_path) mlc_repos_path = os.path.abspath(os.path.dirname(repos_file_path)) repo_parent_path = os.path.abspath(os.path.dirname(repo_path)) diff --git a/tests/test_pull_repo_thread_safety.py b/tests/test_pull_repo_thread_safety.py index c415d979c..62b23ddd8 100644 --- a/tests/test_pull_repo_thread_safety.py +++ b/tests/test_pull_repo_thread_safety.py @@ -1,5 +1,7 @@ import json import os +import shutil +import stat import subprocess import tempfile import threading @@ -487,8 +489,15 @@ def test_real_git_classification(self): with open(a_file, 'w') as f: f.write('x') + # A repo relocated to another volume and symlinked back is a normal + # setup; classifying the link as junk would unlink it and strand the + # real checkout together with any local work in it. + linked = os.path.join(base, 'linked') + os.symlink(valid, linked) + cases = [ (valid, RepoAction.GIT_STATE_VALID), + (linked, RepoAction.GIT_STATE_VALID), (empty, RepoAction.GIT_STATE_VALID), (os.path.dirname(half), RepoAction.GIT_STATE_INVALID), (plain, RepoAction.GIT_STATE_INVALID), @@ -571,8 +580,17 @@ def fake_run(cmd, *a, **kw): return subprocess.CompletedProcess(cmd, 0, target + "\n", "") return subprocess.CompletedProcess(cmd, 0, "", "") + previous_timeout = os.environ.get("MLC_REPO_LOCK_TIMEOUT") + + def _restore_timeout(): + # Restore, do not delete: popping would clobber an ambient value. + if previous_timeout is None: + os.environ.pop("MLC_REPO_LOCK_TIMEOUT", None) + else: + os.environ["MLC_REPO_LOCK_TIMEOUT"] = previous_timeout + + self.addCleanup(_restore_timeout) os.environ["MLC_REPO_LOCK_TIMEOUT"] = "1" - self.addCleanup(os.environ.pop, "MLC_REPO_LOCK_TIMEOUT", None) ra = self._make_repo_action() holder = FileLock(repo_path + ".lock", timeout=30) @@ -621,6 +639,170 @@ def test_timeout_on_registered_repo_with_no_revision_is_a_noop(self): "revision was requested; there was genuinely nothing to do") +class RepoLockMechanicsTest(unittest.TestCase): + """Reentrancy, cycle-guarding and key normalisation of the repo lock.""" + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + + def test_same_thread_reentry_does_not_self_deadlock(self): + """register_repo recurses into pull_repo for deps while the parent's + repo lock is held. filelock is only reentrant per instance, so without + the thread-local set the nested acquire blocks for the whole timeout.""" + from mlc.repo_action import _repo_lock + + lock_file = os.path.join(self.temp_dir.name, "r.lock") + with _repo_lock(lock_file, 2): + # A fresh FileLock on the same path would block here. + with _repo_lock(lock_file, 2): + pass + + def test_lock_is_released_after_an_exception(self): + from mlc.repo_action import _repo_lock + + lock_file = os.path.join(self.temp_dir.name, "r.lock") + with self.assertRaises(ValueError): + with _repo_lock(lock_file, 2): + raise ValueError("boom") + # Must be acquirable again, and bookkeeping must be clean. + with _repo_lock(lock_file, 2): + pass + + def test_lock_key_is_normalised(self): + """Two spellings of one repo must map to one lock file, or two + processes take different locks for the same directory.""" + from mlc.repo_action import _repo_lock_path + + base = self.temp_dir.name + self.assertEqual( + _repo_lock_path(os.path.join(base, "r")), + _repo_lock_path(os.path.join(base, ".", "r"))) + self.assertEqual( + _repo_lock_path(os.path.join(base, "sub", "..", "r")), + _repo_lock_path(os.path.join(base, "r"))) + + def test_dependency_cycle_does_not_recurse(self): + """A -> B -> A must not re-enter the pull; the per-repo lock cannot + stop that on its own because same-thread reentry is a deliberate + no-op, which would turn the old deadlock into unbounded recursion.""" + from mlc.repo_action import _repo_pull_lock + + lock_file = os.path.join(self.temp_dir.name, "r.lock") + depth = [] + + def pull(remaining): + with _repo_pull_lock(lock_file, 5) as proceed: + if not proceed: + return + depth.append(1) + if remaining: + pull(remaining - 1) + + pull(10) + self.assertEqual( + len(depth), 1, + msg="the cycle guard let the pull re-enter itself") + + +class RmRepoLockingTest(_RepoActionTestBase): + """`mlc rm repo` must not delete a tree a concurrent pull is building, + but pruning of vanished entries must never block or fail.""" + + def test_rm_repo_itself_is_unlocked_and_infallible(self): + """load_repos_and_meta() returns rm_repo's result straight to callers + that expect a list of Repos, so rm_repo must not acquire a lock (it + would block behind an unrelated repo) nor return {'return': 1}.""" + from mlc.repo_action import rm_repo + + vanished = os.path.join(self.repos_path, "gone@repo") + holder = FileLock(vanished + ".lock", timeout=30) + holder.acquire() + try: + result = rm_repo(vanished, self.repos_file, True) + finally: + holder.release() + + self.assertEqual( + result.get("return"), 0, + msg="rm_repo must stay infallible; load_repos_and_meta returns " + "its result where a list is expected") + + def test_pruning_a_vanished_entry_still_returns_a_list(self): + """The end-to-end shape of the above: Action.load_repos_and_meta must + hand back Repo objects, not an error dict.""" + with open(self.repos_file) as f: + entries = json.load(f) + vanished = os.path.join(self.repos_path, "gone@repo") + with open(self.repos_file, 'w') as f: + json.dump(entries + [vanished], f, indent=2) + + holder = FileLock(vanished + ".lock", timeout=30) + holder.acquire() + try: + action = Action() + action.parent = None + repos = action.load_repos_and_meta() + finally: + holder.release() + + self.assertIsInstance( + repos, list, + msg=f"expected a list of Repos, got {type(repos).__name__}: " + f"{repos!r}") + for repo in repos: + self.assertTrue(hasattr(repo, 'path'), + msg=f"not a Repo object: {repo!r}") + + def test_rm_repo_action_takes_the_per_repo_lock(self): + """The user-facing path must serialise against a concurrent pull.""" + repo_dir, meta = self._make_fake_repo_dir("locked@repo") + repo_path = os.path.join(self.repos_path, "locked@repo") + shutil.copytree(repo_dir, repo_path) + + ra = self._make_repo_action() + ra.register_repo(repo_path, meta) + + os.environ["MLC_REPO_LOCK_TIMEOUT"] = "1" + self.addCleanup(os.environ.pop, "MLC_REPO_LOCK_TIMEOUT", None) + + from mlc.repo_action import _repo_lock_path + holder = FileLock(_repo_lock_path(repo_path), timeout=30) + holder.acquire() + try: + result = self._make_repo_action().rm( + {'repo': repo_path, 'f': True}) + finally: + holder.release() + + self.assertNotEqual( + result.get('return'), 0, + msg="rm should have failed to take the lock held by a 'pull'") + self.assertTrue( + os.path.isdir(repo_path), + msg="rm deleted the tree while another process held the lock") + + +class AtomicWriteMetadataTest(_RepoActionTestBase): + def test_file_mode_is_preserved_across_the_replace(self): + """os.replace installs a new inode, so without copymode a + group-writable repos.json on a shared MLC_REPOS silently becomes 0600 + (mkstemp's default) owned by whoever wrote last.""" + from mlc.repo_action import _atomic_write_json + + os.chmod(self.repos_file, 0o664) + before = stat.S_IMODE(os.stat(self.repos_file).st_mode) + + with open(self.repos_file) as f: + entries = json.load(f) + _atomic_write_json(self.repos_file, entries + ["/tmp/mode-probe"]) + + after = stat.S_IMODE(os.stat(self.repos_file).st_mode) + self.assertEqual( + oct(after), oct(before), + msg="file mode was not preserved across the atomic replace") + + class RepoLockTimeoutTest(unittest.TestCase): """The per-repo lock timeout must be overridable for slow cold clones.""" From aacd26777af6df6bd416b2f688f08421ccb7b1ac Mon Sep 17 00:00:00 2001 From: anandhu-eng Date: Fri, 14 Aug 2026 13:14:52 +0530 Subject: [PATCH 4/4] Close the last four low-severity review items A third review pass found nothing above LOW. These are the four it found. _repo_lock_path took basename() from the raw argument while abspath() stripped the trailing slash, so basename("/base/repoA/") was "". That both split one repo across two locks ("repoA" vs "repoA/") and collided unrelated repos onto a single ".lock" -- the exact two failures the helper exists to prevent, and the collision would additionally make _repo_lock's reentrancy treat a nested pull of repoB as already-held while pulling repoA. Trailing slashes reach here from the CLI: rm() uses run_args['repo'] verbatim. basename() is now taken from the normalised path. RepoAction.rm() de-indexed before acquiring the lock, so a concurrent pull could re-index the repo in the window and rm would report success with the index still pointing into a deleted directory. The ordering predates the lock, but the lock is what makes the window reachable. Moved inside; ordering stays repo lock -> index lock, matching pull_repo. The MLC_REPO_LOCK_TIMEOUT save/restore fix from the previous commit was applied to one test class but the newly-added one still popped the variable, clobbering any ambient value. rm_repo's infallibility was pinned only against the per-repo lock. The fallibility that broke every mlc command could also return via unregister_repo's own 60s repos.json timeout, which nothing covered. Testing the de-index ordering took two attempts worth recording: asserting on Index.repos is vacuous, because remove_repo_from_index() mutates Index.indices instead, and asserting on the persisted index files does not work either, since a content-less fixture repo produces no entries. The test now spies on remove_repo_from_index and asserts it is never called when the lock could not be taken. All four mutants killed: basename from the raw path, de-index moved above the lock, rm_repo propagating unregister_repo's Timeout, and the earlier lock-key mutant. 78 passed, stable over 4 runs; mlc/ and tests/ parse on 3.10, 3.11 and 3.13. Co-Authored-By: Claude Opus 5 --- mlc/repo_action.py | 18 +++++-- tests/test_pull_repo_thread_safety.py | 69 +++++++++++++++++++++++++-- 2 files changed, 81 insertions(+), 6 deletions(-) diff --git a/mlc/repo_action.py b/mlc/repo_action.py index b0e0df8ee..2e13223a7 100644 --- a/mlc/repo_action.py +++ b/mlc/repo_action.py @@ -59,9 +59,15 @@ def _repo_lock_path(repo_path): resolved, so the lock still sits beside repo_path when repo_path is itself a symlink. """ - parent = os.path.dirname(os.path.abspath(repo_path)) + # basename() must be taken from the *normalised* path: os.path.abspath + # strips a trailing slash but basename("/base/repoA/") is "", which would + # both split one repo across two locks ("repoA" vs "repoA/") and collide + # unrelated repos onto a single ".lock". Trailing slashes reach here from + # the CLI -- rm() uses run_args['repo'] verbatim. + absolute = os.path.abspath(repo_path) return os.path.join( - os.path.realpath(parent), os.path.basename(repo_path)) + ".lock" + os.path.realpath(os.path.dirname(absolute)), + os.path.basename(absolute)) + ".lock" # Repos whose pull is already in progress on this thread. register_repo() @@ -1378,7 +1384,6 @@ def rm(self, run_args): force_remove = True if run_args.get('f') else False index = Action.get_index(self) - index.remove_repo_from_index(repo_path) # Same per-repo lock pull_repo uses: without it, `mlc rm repo X` # racing a concurrent `mlc pull repo X` deletes the tree while the @@ -1391,6 +1396,13 @@ def rm(self, run_args): try: with _repo_lock(_repo_lock_path(repo_path), _get_repo_lock_timeout()): + # De-index inside the lock. Doing it before acquiring leaves a + # window in which a concurrent pull re-indexes the repo we are + # about to delete, so rm would report success while the index + # still points into a removed directory. Ordering is safe: + # pull_repo already takes repo lock -> index lock, via + # register_repo -> index.add_repo. + index.remove_repo_from_index(repo_path) return rm_repo(repo_path, repos_file_path, force_remove) except Timeout: return { diff --git a/tests/test_pull_repo_thread_safety.py b/tests/test_pull_repo_thread_safety.py index 62b23ddd8..c3ddca8a7 100644 --- a/tests/test_pull_repo_thread_safety.py +++ b/tests/test_pull_repo_thread_safety.py @@ -682,6 +682,18 @@ def test_lock_key_is_normalised(self): _repo_lock_path(os.path.join(base, "sub", "..", "r")), _repo_lock_path(os.path.join(base, "r"))) + # A trailing slash reaches here from the CLI (rm() uses + # run_args['repo'] verbatim). basename() of a raw "…/r/" is "", which + # would both split one repo across two locks and collide unrelated + # repos onto a single ".lock". + self.assertEqual( + _repo_lock_path(os.path.join(base, "r") + os.sep), + _repo_lock_path(os.path.join(base, "r"))) + self.assertNotEqual( + _repo_lock_path(os.path.join(base, "repoA") + os.sep), + _repo_lock_path(os.path.join(base, "repoB") + os.sep), + msg="unrelated repos must not share a lock file") + def test_dependency_cycle_does_not_recurse(self): """A -> B -> A must not re-enter the pull; the per-repo lock cannot stop that on its own because same-thread reentry is a deliberate @@ -728,6 +740,30 @@ def test_rm_repo_itself_is_unlocked_and_infallible(self): msg="rm_repo must stay infallible; load_repos_and_meta returns " "its result where a list is expected") + def test_rm_repo_absorbs_a_repos_json_lock_timeout(self): + """The fallibility that broke every mlc command could also come back + via unregister_repo's own repos.json timeout, not just via a repo + lock. rm_repo must swallow that too.""" + from mlc.repo_action import rm_repo, _repos_lock_file + + vanished = os.path.join(self.repos_path, "gone@repo") + jam = FileLock(_repos_lock_file(self.repos_file), timeout=30) + jam.acquire() + try: + with patch('mlc.repo_action.FileLock') as fake_lock: + # Make unregister_repo's acquire fail immediately rather than + # waiting out its hard-coded 60s. + fake_lock.side_effect = lambda *a, **kw: FileLock( + a[0], timeout=0.1) + result = rm_repo(vanished, self.repos_file, True) + finally: + jam.release() + + self.assertEqual( + result.get("return"), 0, + msg="rm_repo must absorb unregister_repo's Timeout; " + "load_repos_and_meta returns this where a list is expected") + def test_pruning_a_vanished_entry_still_returns_a_list(self): """The end-to-end shape of the above: Action.load_repos_and_meta must hand back Repo objects, not an error dict.""" @@ -763,15 +799,34 @@ def test_rm_repo_action_takes_the_per_repo_lock(self): ra = self._make_repo_action() ra.register_repo(repo_path, meta) + previous_timeout = os.environ.get("MLC_REPO_LOCK_TIMEOUT") + + def _restore_timeout(): + # Restore, do not delete: popping would clobber an ambient value. + if previous_timeout is None: + os.environ.pop("MLC_REPO_LOCK_TIMEOUT", None) + else: + os.environ["MLC_REPO_LOCK_TIMEOUT"] = previous_timeout + + self.addCleanup(_restore_timeout) os.environ["MLC_REPO_LOCK_TIMEOUT"] = "1" - self.addCleanup(os.environ.pop, "MLC_REPO_LOCK_TIMEOUT", None) from mlc.repo_action import _repo_lock_path + from mlc.index import Index + + deindexed = [] + real_remove = Index.remove_repo_from_index + + def spy(self_index, path): + deindexed.append(path) + return real_remove(self_index, path) + holder = FileLock(_repo_lock_path(repo_path), timeout=30) holder.acquire() try: - result = self._make_repo_action().rm( - {'repo': repo_path, 'f': True}) + with patch.object(Index, 'remove_repo_from_index', spy): + result = self._make_repo_action().rm( + {'repo': repo_path, 'f': True}) finally: holder.release() @@ -781,6 +836,14 @@ def test_rm_repo_action_takes_the_per_repo_lock(self): self.assertTrue( os.path.isdir(repo_path), msg="rm deleted the tree while another process held the lock") + # De-indexing must happen *under* the lock. Doing it beforehand + # leaves a window in which a concurrent pull re-indexes the repo we + # then delete, so rm reports success while the index still points + # into a removed directory. Since this rm never got the lock, it must + # not have touched the index at all. + self.assertEqual( + deindexed, [], + msg="rm de-indexed the repo before acquiring the lock") class AtomicWriteMetadataTest(_RepoActionTestBase):