feat: add Python script for fast tree manifest generation and update shell script to utilize it

This commit is contained in:
Ivan 2026-07-26 05:57:49 -05:00
parent 0606afd6c8
commit 99b3630e72
No known key found for this signature in database
4 changed files with 228 additions and 0 deletions

Binary file not shown.

View file

@ -63,6 +63,10 @@ tracked_paths() {
}
generate() {
if command -v python3 >/dev/null 2>&1 && [ -f "$ROOT/scripts/ci/tree_manifest_generate.py" ]; then
python3 "$ROOT/scripts/ci/tree_manifest_generate.py"
return $?
fi
printf '%s\n' "$MANIFEST_HEADER"
tracked_paths | while IFS= read -r f; do
[ -n "$f" ] || continue

View file

@ -0,0 +1,138 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: 0BSD
"""Fast git-index tree manifest for meshchatx.rsm signing.
Hashes index blobs (same bytes as git show :path) via git cat-file --batch.
Output format matches scripts/ci/tree-manifest.sh generate.
"""
from __future__ import annotations
import hashlib
import os
import subprocess
import sys
from pathlib import Path
MANIFEST_HEADER = "# meshchatx tree manifest v1"
EXCLUDE_RSM = "meshchatx.rsm"
ALLOWED_MODES = frozenset({"100644", "100755"})
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]
def is_excluded_path(path: str) -> bool:
if path == EXCLUDE_RSM:
return True
if path == "vendor" or path.startswith("vendor/"):
return True
if "/vendor/" in path or path.endswith("/vendor"):
return True
return False
def _parse_ls_files_z(raw: bytes) -> list[tuple[str, str, str]]:
entries: list[tuple[str, str, str]] = []
for chunk in raw.split(b"\0"):
if not chunk:
continue
tab = chunk.find(b"\t")
if tab < 0:
continue
meta = chunk[:tab].decode("ascii", errors="replace")
path = chunk[tab + 1 :].decode("utf-8", errors="surrogateescape")
parts = meta.split()
if len(parts) < 3:
continue
mode, oid, _stage = parts[0], parts[1], parts[2]
entries.append((path, mode, oid))
return entries
def _read_batch_blob(stdout, header: bytes) -> bytes | None:
parts = header.strip().split()
if len(parts) == 2 and parts[1] == b"missing":
return None
if len(parts) != 3:
raise RuntimeError(f"unexpected cat-file header: {header!r}")
_oid, typ, size_s = parts
if typ == b"missing":
return None
size = int(size_s)
data = stdout.read(size)
if len(data) != size:
raise RuntimeError("short read from git cat-file --batch")
if typ == b"blob":
delim = stdout.read(1)
if delim != b"\n":
raise RuntimeError("expected newline delimiter after cat-file blob")
return data
def generate_manifest(root: Path) -> str:
env = os.environ.copy()
env["LC_ALL"] = "C"
raw = subprocess.check_output(
["git", "ls-files", "-s", "-z"],
cwd=root,
env=env,
)
rows: list[tuple[str, str, str]] = []
for path, mode, oid in _parse_ls_files_z(raw):
if not path or is_excluded_path(path):
continue
if mode not in ALLOWED_MODES:
continue
rows.append((path, mode, oid))
rows.sort(key=lambda item: item[0])
lines = [MANIFEST_HEADER]
if not rows:
return "\n".join(lines) + "\n"
proc = subprocess.Popen(
["git", "cat-file", "--batch"],
cwd=root,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
env=env,
)
assert proc.stdin is not None
assert proc.stdout is not None
stdin_buf = "".join(f"{oid}\n" for _path, _mode, oid in rows).encode("ascii")
proc.stdin.write(stdin_buf)
proc.stdin.close()
for path, _mode, _oid in rows:
header = proc.stdout.readline()
if not header:
raise RuntimeError("git cat-file --batch closed stdout early")
blob = _read_batch_blob(proc.stdout, header)
if blob is None:
continue
digest = hashlib.sha256(blob).hexdigest()
lines.append(f"{digest} {path}")
proc.wait()
if proc.returncode not in (0, None):
raise RuntimeError(f"git cat-file --batch exited {proc.returncode}")
return "\n".join(lines) + "\n"
def main() -> int:
root = _repo_root()
try:
sys.stdout.write(generate_manifest(root))
except Exception as exc:
print(f"tree_manifest_generate.py: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,86 @@
# SPDX-License-Identifier: 0BSD
import subprocess
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
def _slow_manifest() -> str:
proc = subprocess.run(
["sh", "scripts/ci/tree-manifest.sh", "generate"],
cwd=REPO,
capture_output=True,
text=True,
env={
**__import__("os").environ,
"PATH": __import__("os").environ.get("PATH", ""),
},
check=False,
)
# Force slow path by temporarily hiding the python helper name is awkward.
# Compare fast output to a golden re-run via inline shell loop instead.
assert proc.returncode == 0, proc.stderr
return proc.stdout
def _fast_manifest() -> str:
proc = subprocess.run(
["python3", "scripts/ci/tree_manifest_generate.py"],
cwd=REPO,
capture_output=True,
text=True,
check=True,
)
return proc.stdout
def _legacy_shell_manifest() -> str:
"""Same rules as tree-manifest.sh generate before the Python fast path."""
header = "# meshchatx tree manifest v1"
lines = [header]
env = {"LC_ALL": "C"}
paths = subprocess.check_output(["git", "ls-files"], cwd=REPO, env=env, text=True)
for f in sorted(paths.splitlines(), key=lambda s: s):
if not f:
continue
if f == "meshchatx.rsm":
continue
if f == "vendor" or f.startswith("vendor/"):
continue
if "/vendor/" in f or f.endswith("/vendor"):
continue
check = subprocess.run(
["git", "cat-file", "-e", f":{f}"],
cwd=REPO,
capture_output=True,
)
if check.returncode != 0:
continue
mode = subprocess.check_output(
["git", "ls-files", "-s", "--", f],
cwd=REPO,
text=True,
).split()[0]
if mode not in ("100644", "100755"):
continue
digest = subprocess.check_output(
["sh", "-c", "git show \":$1\" | sha256sum | awk '{print $1}'", "_", f],
cwd=REPO,
text=True,
).strip()
lines.append(f"{digest} {f}")
return "\n".join(lines) + "\n"
def test_fast_manifest_matches_legacy_shell_algorithm():
legacy = _legacy_shell_manifest()
fast = _fast_manifest()
assert fast == legacy
def test_tree_manifest_sh_uses_fast_generator():
via_sh = _slow_manifest()
fast = _fast_manifest()
assert via_sh == fast