from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
import tempfile
from pathlib import Path
REPOSITORY = Path(__file__).resolve().parent.parent
LOCKFILE = REPOSITORY / "Cargo.lock"
MANIFEST = REPOSITORY / "Cargo.toml"
FIRST_PARTY_SOURCE = re.compile(r'^source = "git\+https://github\.com/ryancinsight/', re.M)
FIRST_PARTY_GIT = "git+https://github.com/ryancinsight/"
def run_outside_the_overlay(arguments: list[str]) -> subprocess.CompletedProcess[str]:
with tempfile.TemporaryDirectory() as neutral_directory:
return subprocess.run(
["cargo", *arguments, "--manifest-path", str(MANIFEST)],
cwd=neutral_directory,
capture_output=True,
encoding="utf-8",
errors="replace",
check=False,
)
def declared_first_party_dependencies() -> int | None:
completed = run_outside_the_overlay(["metadata", "--no-deps", "--format-version", "1"])
if completed.returncode != 0:
return None
packages = json.loads(completed.stdout)["packages"]
return sum(
1
for package in packages
for dependency in package["dependencies"]
if (dependency.get("source") or "").startswith(FIRST_PARTY_GIT)
)
def check() -> int:
if not LOCKFILE.is_file():
print(f"error: {LOCKFILE} does not exist", file=sys.stderr)
return 1
sources = len(FIRST_PARTY_SOURCE.findall(LOCKFILE.read_text(encoding="utf-8")))
declared = declared_first_party_dependencies()
if sources == 0 and declared != 0:
print(
"error: Cargo.lock contains no first-party git sources.\n"
"\n"
"It was regenerated with the Atlas stack overlay active, which\n"
"resolves those dependencies to local paths and drops their git\n"
"sources. CI has no overlay and will fail every --locked job.\n"
"\n"
"Fix: scripts/lockfile.py --regenerate",
file=sys.stderr,
)
return 1
completed = run_outside_the_overlay(
["metadata", "--locked", "--format-version", "1", "--all-features"]
)
if completed.returncode != 0:
print(
f"error: the committed Cargo.lock does not resolve under --locked "
f"({sources} first-party git sources present, so it is stale rather "
f"than flattened).\n"
f"\n"
f"The pinned first-party revisions no longer satisfy the manifests'\n"
f"version requirements, so cargo must re-resolve and --locked\n"
f"refuses. This is what blocks the benchmark baseline alignment.\n"
f"\n"
f"Fix: scripts/lockfile.py --regenerate\n"
f"\n"
f"cargo said:\n{completed.stderr.strip()}",
file=sys.stderr,
)
return 1
if declared == 0:
print("Cargo.lock resolves under --locked; no first-party dependencies declared.")
else:
print(f"Cargo.lock resolves under --locked; {sources} first-party git sources.")
return 0
def check_staged() -> int:
staged = subprocess.run(
["git", "diff", "--cached", "--name-only", "--", "Cargo.lock"],
capture_output=True,
encoding="utf-8",
errors="replace",
check=False,
)
if staged.returncode != 0 or not staged.stdout.strip():
return 0
blob = subprocess.run(
["git", "show", ":Cargo.lock"],
capture_output=True,
encoding="utf-8",
errors="replace",
check=False,
)
if blob.returncode != 0:
return 0
if len(FIRST_PARTY_SOURCE.findall(blob.stdout)) > 0:
return 0
if declared_first_party_dependencies() == 0:
return 0
print(
"error: the staged Cargo.lock contains no first-party git sources.\n"
"\n"
"A cargo command run against a tree under the Atlas stack root rewrote\n"
"it with the overlay active, which resolves those dependencies to local\n"
"paths and drops their git sources. Committing it now is what turns a\n"
"working branch into one that can never be pushed.\n"
"\n"
"Fix: scripts/lockfile.py --regenerate, then stage the result.\n"
"To commit anyway: SKIP_LOCKFILE_CHECK=1 git commit",
file=sys.stderr,
)
return 1
def regenerate() -> int:
completed = run_outside_the_overlay(["generate-lockfile"])
if completed.returncode != 0:
print(f"error: regeneration failed:\n{completed.stderr.strip()}", file=sys.stderr)
return 1
print("Cargo.lock regenerated outside the overlay.")
return check()
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--check", action="store_true", help="verify the committed lock")
mode.add_argument("--regenerate", action="store_true", help="rewrite the lock correctly")
mode.add_argument(
"--check-staged",
action="store_true",
help="fast structural check of the staged lock, for pre-commit",
)
arguments = parser.parse_args()
if arguments.regenerate:
return regenerate()
if arguments.check_staged:
return check_staged()
return check()
if __name__ == "__main__":
raise SystemExit(main())