Normalize claim remotes and scratch failures

autoform_cli/claims.py

@@ -25,6 +25,11 @@
 CLAIM_TTL_S = 1500
 CLAIM_HEARTBEAT_S = 300
 CLAIM_KEY_RE = re.compile(r"^[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*$")
+_SCP_REPOSITORY_RE = re.compile(r"^(?:[^/@:]+@)?(?:\[[^\]]+\]|[^/:]+):.+$")
+_WINDOWS_DRIVE_PREFIX_RE = re.compile(r"^[A-Za-z]:")
+_WINDOWS_ROOTED_DRIVE_RE = re.compile(r"^[A-Za-z]:[\\/]")
+_SCRATCH_INIT_ATTEMPTS = 10
+_SCRATCH_INIT_DELAY_S = 0.05
 
 _GIT_ENV = {
     "GIT_AUTHOR_NAME": "autoform",
@@ -93,14 +98,28 @@ def author_claim_key(node_id: str) -> str:
     return f"author/{slug}-{digest}"
 
 
+def _repository_is_remote(value: str, *, windows: bool | None = None) -> bool:
+    """Match Git's URL/SCP forms without misclassifying Windows paths."""
+
+    on_windows = os.name == "nt" if windows is None else windows
+    windows_path = (
+        value.startswith("\\")
+        or _WINDOWS_ROOTED_DRIVE_RE.match(value) is not None
+        or (on_windows and _WINDOWS_DRIVE_PREFIX_RE.match(value) is not None)
+    )
+    return not windows_path and (
+        "://" in value or _SCP_REPOSITORY_RE.fullmatch(value) is not None
+    )
+
+
 class ClaimBoard:
     """Lease operations against a Git repository via a local bare object store."""
 
     def __init__(self, repo_url: str | os.PathLike[str], worker_id: str, scratch: str | os.PathLike[str]):
         if not worker_id:
             raise ValueError("worker_id must not be empty")
         raw_repo_url = os.fspath(repo_url)
-        if "://" not in raw_repo_url and not re.match(r"^[^/]+@[^:]+:", raw_repo_url):
+        if not _repository_is_remote(raw_repo_url):
             raw_repo_url = str(Path(raw_repo_url).expanduser().resolve())
         self.repo_url = raw_repo_url
         self.worker_id = worker_id
@@ -131,10 +150,30 @@ def _git(
         return proc
 
     def _ensure_scratch(self) -> None:
-        if (self.scratch / "HEAD").is_file():
-            return
-        self.scratch.mkdir(parents=True, exist_ok=True)
-        self._git(["init", "--bare", "--quiet", "."])
+        last_error: BaseException | None = None
+        for attempt in range(_SCRATCH_INIT_ATTEMPTS):
+            try:
+                if self.scratch.is_dir():
+                    ready = self._git(
+                        ["rev-parse", "--is-bare-repository"],
+                        check=False,
+                    )
+                    if ready.returncode == 0 and ready.stdout.strip() == "true":
+                        return
+                self.scratch.mkdir(parents=True, exist_ok=True)
+            except (OSError, ClaimTransportError) as exc:
+                last_error = exc
+                break
+            try:
+                self._git(["init", "--bare", "--quiet", "."])
+                return
+            except ClaimTransportError as exc:
+                last_error = exc
+            if attempt + 1 < _SCRATCH_INIT_ATTEMPTS:
+                time.sleep(_SCRATCH_INIT_DELAY_S)
+        raise ClaimTransportError(
+            "cannot prepare the local claim-board scratch directory"
+        ) from last_error
 
     @staticmethod
     def _ref(key: str) -> str:

tests/test_claims.py

@@ -43,6 +43,119 @@ def _board(tmp_path: Path, repo: Path, owner: str) -> claims.ClaimBoard:
     return claims.ClaimBoard(repo, owner, tmp_path / f"scratch-{owner}")
 
 
+def test_claim_board_recognizes_scp_remote_without_explicit_user(tmp_path: Path) -> None:
+    board = claims.ClaimBoard(
+        "github-work:org/repo.git",
+        "worker",
+        tmp_path / "scratch",
+    )
+
+    assert board.repo_url == "github-work:org/repo.git"
+
+
+@pytest.mark.parametrize(
+    "remote",
+    [
+        "git@example.com:org/repo.git",
+        "github-work:org/repo.git",
+        "[2001:db8::1]:org/repo.git",
+        "g:repo.git",
+    ],
+)
+def test_repository_remote_grammar_matches_scp_forms(remote: str) -> None:
+    assert claims._repository_is_remote(remote, windows=False)
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        r"C:\projects\claims.git",
+        "C:/projects/claims.git",
+        r"\\?\C:\projects\claims.git",
+        r"\\.\C:\projects\claims.git",
+        r"\\server\share\claims.git",
+    ],
+)
+def test_repository_remote_grammar_rejects_rooted_windows_paths(path: str) -> None:
+    assert not claims._repository_is_remote(path, windows=False)
+
+
+def test_drive_relative_path_is_local_only_on_windows() -> None:
+    assert claims._repository_is_remote("C:claims.git", windows=False)
+    assert not claims._repository_is_remote("C:claims.git", windows=True)
+
+
+def test_claim_board_does_not_treat_windows_drive_as_scp_remote(tmp_path: Path) -> None:
+    raw = r"C:\projects\claims.git"
+    board = claims.ClaimBoard(
+        raw,
+        "worker",
+        tmp_path / "scratch",
+    )
+
+    assert board.repo_url == str(Path(raw).expanduser().resolve())
+    assert Path(board.repo_url).is_absolute()
+
+
+def test_non_directory_scratch_is_a_clean_transport_error(
+    tmp_path: Path,
+    board_repo: Path,
+) -> None:
+    scratch = tmp_path / "scratch"
+    scratch.write_text("not a directory", encoding="utf-8")
+    board = claims.ClaimBoard(board_repo, "worker", scratch)
+
+    with pytest.raises(claims.ClaimTransportError, match="prepare.*scratch"):
+        board.read("author/node")
+
+
+@pytest.mark.skipif(
+    os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() == 0,
+    reason="needs a non-root POSIX user and mode-bit permissions",
+)
+def test_inaccessible_scratch_parent_is_a_clean_transport_error(
+    tmp_path: Path,
+    board_repo: Path,
+) -> None:
+    locked = tmp_path / "locked"
+    locked.mkdir()
+    board = claims.ClaimBoard(board_repo, "worker", locked / "scratch")
+    locked.chmod(0)
+    try:
+        with pytest.raises(claims.ClaimTransportError, match="prepare.*scratch"):
+            board.read("author/node")
+    finally:
+        locked.chmod(0o755)
+
+
+def test_concurrent_first_use_initializes_one_shared_scratch(
+    tmp_path: Path,
+    board_repo: Path,
+) -> None:
+    scratch = tmp_path / "scratch"
+    workers = 12
+    start = threading.Barrier(workers)
+    failures: list[BaseException] = []
+
+    def read_missing_claim() -> None:
+        try:
+            start.wait(timeout=5)
+            board = claims.ClaimBoard(board_repo, "worker", scratch)
+            assert board.read("author/missing") is None
+        except BaseException as error:
+            failures.append(error)
+
+    threads = [threading.Thread(target=read_missing_claim) for _ in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=10)
+
+    assert all(not thread.is_alive() for thread in threads)
+    assert failures == []
+    assert _git("rev-parse", "--is-bare-repository", cwd=scratch) == "true"
+
+
 def _plant_message(repo: Path, key: str, message: str) -> str:
     tree = _git("mktree", cwd=repo, input_text="")
     commit = _git("commit-tree", tree, "-m", message, cwd=repo)