diff --git a/mlc/repo_action.py b/mlc/repo_action.py index b4a17c73d..2e13223a7 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 @@ -13,6 +16,172 @@ 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 + + +# 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.""" + + +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. + """ + # 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(os.path.dirname(absolute)), + os.path.basename(absolute)) + ".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.""" + 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. + + 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. + """ + 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 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: + 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 +303,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 @@ -199,6 +369,101 @@ def _validate_extra_git_args(extra_args): ) return None + # 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.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'], + capture_output=True, text=True) + 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): """ #################################################################################################################### @@ -329,6 +594,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 +606,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 +618,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 +785,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 +859,72 @@ 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. - repo_lock_file = repo_path + ".lock" - with FileLock(repo_lock_file, timeout=300): + # + # 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_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 + # 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}...") + # 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 +932,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 +1141,44 @@ 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: + # 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 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 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. 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 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 + # only surface a bare "[Errno 13]". + return { + 'return': 1, + 'error': ( + 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." ) } except Exception as e: @@ -1004,9 +1384,44 @@ 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) - 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()): + # 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 { + '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): @@ -1015,9 +1430,20 @@ 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. + + 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. + """ 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)) @@ -1073,14 +1499,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..c3ddca8a7 100644 --- a/tests/test_pull_repo_thread_safety.py +++ b/tests/test_pull_repo_thread_safety.py @@ -1,22 +1,23 @@ import json import os +import shutil +import stat import subprocess import tempfile import threading 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 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 +55,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 +158,746 @@ 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: + # _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=target + "\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"; + # 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, 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 + + 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_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" + 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 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') + + # 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), + (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, "", "") + + 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" + + 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 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"))) + + # 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 + 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_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.""" + 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) + + 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" + + 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: + with patch.object(Index, 'remove_repo_from_index', spy): + 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") + # 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): + 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.""" + + 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__":