From 74eeb6430a6315c3fb645687f823c9ee8f302f93 Mon Sep 17 00:00:00 2001 From: Lukas Wallrich Date: Thu, 10 Sep 2026 19:27:47 +0200 Subject: [PATCH] Select staging data by per-file freshness --- .github/workflows/data-processing.yml | 6 +- .github/workflows/staging-aggregate.yaml | 16 ++- scripts/data_artifact.py | 173 +++++++++++++++++++++++ scripts/tests/test_data_artifact.py | 141 ++++++++++++++++++ 4 files changed, 333 insertions(+), 3 deletions(-) create mode 100644 scripts/data_artifact.py create mode 100644 scripts/tests/test_data_artifact.py diff --git a/.github/workflows/data-processing.yml b/.github/workflows/data-processing.yml index 8b45c49706f..3d24d723f7b 100644 --- a/.github/workflows/data-processing.yml +++ b/.github/workflows/data-processing.yml @@ -62,7 +62,7 @@ jobs: # Checkout the repository code to the runner environment with: # Full history: the generators read per-file commit dates to bootstrap - # `lastmod`, which a shallow clone would report as HEAD for every file. + # `lastmod` and artifact freshness; shallow clones report HEAD for every file. fetch-depth: 0 #====================== @@ -497,12 +497,16 @@ jobs: #======================================== # Upload all processed data files as artifact #======================================== + - name: Record per-file data freshness + run: python3 scripts/data_artifact.py manifest + - name: Upload data artifact id: upload-artifact uses: actions/upload-artifact@v7 with: name: data-artifact path: | + data-artifact-manifest.json content/contributors/tenzing.md scripts/forrt_contribs/contributors_cache.csv content/curated_resources/ diff --git a/.github/workflows/staging-aggregate.yaml b/.github/workflows/staging-aggregate.yaml index f225775f6ee..9b5b26a164a 100644 --- a/.github/workflows/staging-aggregate.yaml +++ b/.github/workflows/staging-aggregate.yaml @@ -285,11 +285,17 @@ jobs: with: workflow: data-processing.yml name: data-artifact - path: . + path: .staging-data-artifact github_token: ${{ secrets.GITHUB_TOKEN }} search_artifacts: true if_no_artifact_found: warn + - name: Select newer committed or artifact data + if: steps.download-artifact.outcome == 'success' + run: | + python3 scripts/data_artifact.py overlay .staging-data-artifact + rm -rf .staging-data-artifact + # ======================= # Data Processing (Fallback) # ======================= @@ -379,10 +385,16 @@ jobs: with: workflow: data-processing.yml name: data-artifact - path: . + path: .staging-data-artifact github_token: ${{ secrets.GITHUB_TOKEN }} run_id: ${{ steps.data-processing.outputs.run_id }} + - name: Select newer data after artifact retry + if: steps.download-artifact-retry.outcome == 'success' + run: | + python3 scripts/data_artifact.py overlay .staging-data-artifact + rm -rf .staging-data-artifact + - name: Run data processing if needed if: steps.download-artifact.outcome == 'failure' && steps.data-processing.outputs.data_processing_triggered != 'true' env: diff --git a/scripts/data_artifact.py b/scripts/data_artifact.py new file mode 100644 index 00000000000..e66a0b66111 --- /dev/null +++ b/scripts/data_artifact.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +"""Record data freshness and overlay the newer version of each staging file. + +Producer: python3 scripts/data_artifact.py manifest +Consumer: python3 scripts/data_artifact.py overlay .staging-data-artifact + +Checkout/download mtimes and artifact upload dates are not content timestamps. +Files carried unchanged from Git retain their last non-merge commit time; +changed/generated outputs use their modification time in the producer workspace. +Synthetic staging merges must not make all PR files appear freshly generated. + +Both checkouts require full Git history. On equal timestamps, committed data +wins. Legacy artifacts without a manifest may add missing files, but cannot +overwrite differing committed files whose relative freshness is unknown. +""" +import argparse +from collections import Counter +import hashlib +import json +from pathlib import Path +import shutil +import subprocess +import time + + +MANIFEST = "data-artifact-manifest.json" +# Match the upload-artifact paths in data-processing.yml. +DATA_PATHS = ( + "content/contributors/tenzing.md", + "scripts/forrt_contribs/contributors_cache.csv", + "content/curated_resources", "content/glossary", "data", "static/data", + "static/partials", "content/contributor-analysis", + "content/publications/citation_chart.webp", +) +EXCLUDED = {"data/partners.json", "data/publications.yaml"} + + +def git(root, *args): + return subprocess.check_output(["git", "-C", str(root), *args]) + + +def git_state(root): + """Return tracked blob hashes and content-change times in two batched reads.""" + if git(root, "rev-parse", "--is-shallow-repository").strip() == b"true": + raise ValueError("Freshness comparison requires checkout fetch-depth: 0") + blobs = {} + for entry in git(root, "ls-tree", "-r", "-z", "HEAD", "--", *DATA_PATHS).split(b"\0"): + if entry: + meta, name = entry.split(b"\t", 1) + blobs[name.decode()] = meta.split()[2].decode() + dates = {} + stamp = None + history = git(root, "log", "--no-merges", "--format=TIME:%ct", "--name-only", "-z", "--", *DATA_PATHS) + for entry in history.split(b"\0"): + if entry.startswith(b"TIME:"): + stamp = int(entry[5:]) + elif entry: + # Git prefixes the first pathname after each commit with a newline. + name = entry.removeprefix(b"\n").decode() + dates.setdefault(name, stamp) + return blobs, dates + + +def blob_hash(content): + return hashlib.sha1(b"blob " + str(len(content)).encode() + b"\0" + content).hexdigest() + + +def data_files(root): + for name in DATA_PATHS: + path = root / name + files = sorted(path.rglob("*")) if path.is_dir() else [path] + for file in files: + if file.is_symlink(): + raise ValueError(f"Unexpected symlink in artifact data: {file}") + if file.is_file() and file.relative_to(root).as_posix() not in EXCLUDED: + yield file + + +def make_manifest(root): + blobs, dates = git_state(root) + records = {} + for file in data_files(root): + name = file.relative_to(root).as_posix() + content = file.read_bytes() + unchanged = blobs.get(name) == blob_hash(content) + updated = dates.get(name) if unchanged else int(file.stat().st_mtime) + if updated is None: + raise ValueError(f"No content-change time for tracked file: {name}") + records[name] = { + "sha256": hashlib.sha256(content).hexdigest(), + "updated_at": updated, + "origin": "git" if unchanged else "generated", + } + manifest = {"version": 1, "source_commit": git(root, "rev-parse", "HEAD").decode().strip(), + "created_at": int(time.time()), "files": records} + (root / MANIFEST).write_text(json.dumps(manifest, indent=2) + "\n") + print(f"Recorded freshness for {len(records)} artifact files") + return manifest + + +def overlay(root, artifact): + manifest_path = artifact / MANIFEST + records = None + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_text()) + if manifest.get("version") != 1: + raise ValueError("Unsupported data artifact manifest version") + records = manifest["files"] + else: + print("::warning::Legacy artifact has no freshness manifest; retaining committed versions of differing files.") + blobs, dates = git_state(root) + plan = [] + for source in data_files(artifact): + name = source.relative_to(artifact).as_posix() + destination = root / name + if not destination.resolve().is_relative_to(root.resolve()): + raise ValueError(f"Destination escapes checkout: {name}") + content = source.read_bytes() + record = records.get(name) if records is not None else None + if records is not None: + if record is None or record["sha256"] != hashlib.sha256(content).hexdigest(): + raise ValueError(f"Artifact freshness manifest mismatch: {name}") + if not isinstance(record["updated_at"], int) or record["updated_at"] < 0: + raise ValueError(f"Invalid freshness timestamp: {name}") + committed_time = dates.get(name) + artifact_time = record["updated_at"] if record else None + if not destination.exists(): + if committed_time is not None and (artifact_time is None or artifact_time <= committed_time): + decision = "committed-deletion-newer-or-unknown" + else: + decision = "artifact-only" + elif destination.read_bytes() == content: + decision = "identical" + elif artifact_time is None: + decision = "committed-unknown-artifact-age" + elif name not in blobs or blob_hash(destination.read_bytes()) != blobs[name]: + raise ValueError(f"Refusing to overwrite locally modified data: {name}") + elif committed_time is None: + raise ValueError(f"No committed content-change time for {name}") + elif artifact_time > committed_time: + decision = "artifact-newer" + else: + decision = "committed-newer-or-equal" + plan.append((source, destination, decision, committed_time, artifact_time)) + # Validate every file before mutating the checkout. + counts = Counter() + for source, destination, decision, committed_time, artifact_time in plan: + counts[decision] += 1 + if decision.startswith("artifact-"): + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source, destination) + if decision != "identical": + print(f"{decision}: {source.relative_to(artifact)} (committed={committed_time}, artifact={artifact_time})") + print("Freshness selection: " + json.dumps(dict(counts), sort_keys=True)) + return counts + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=("manifest", "overlay")) + parser.add_argument("artifact", nargs="?", type=Path) + args = parser.parse_args() + root = Path.cwd() + if args.command == "manifest": + make_manifest(root) + elif args.artifact: + overlay(root, args.artifact.resolve()) + else: + parser.error("overlay requires the downloaded artifact directory") + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/test_data_artifact.py b/scripts/tests/test_data_artifact.py new file mode 100644 index 00000000000..b509bb17b28 --- /dev/null +++ b/scripts/tests/test_data_artifact.py @@ -0,0 +1,141 @@ +"""Run with python3 -m unittest discover -s scripts/tests -p test_data_artifact.py.""" +import contextlib +import io +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import data_artifact as artifact + + +OLD = 1600000000 +NEW = 1700000000 + + +class FreshnessTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) / "repo" + self.root.mkdir() + self.download = Path(self.tmp.name) / "download" + self.download.mkdir() + self.git("init", "-b", "main") + self.git("config", "user.name", "Test") + self.git("config", "user.email", "test@example.org") + self.write(self.root, "data/example.json", '{"value": "committed"}') + self.commit(OLD) + + def git(self, *args, stamp=None): + env = os.environ.copy() + if stamp: + env.update(GIT_AUTHOR_DATE=f"@{stamp} +0000", GIT_COMMITTER_DATE=f"@{stamp} +0000") + return subprocess.check_output(["git", "-C", str(self.root), *args], env=env, stderr=subprocess.STDOUT) + + def commit(self, stamp): + self.git("add", "-A") + self.git("commit", "-m", "Update data", stamp=stamp) + + def write(self, root, name, content): + file = root / name + file.parent.mkdir(parents=True, exist_ok=True) + file.write_text(content) + return file + + def manifest(self): + with contextlib.redirect_stdout(io.StringIO()): + return artifact.make_manifest(self.root) + + def overlay(self): + with contextlib.redirect_stdout(io.StringIO()): + return artifact.overlay(self.root, self.download) + + def package(self, stamp, content="artifact"): + file = self.write(self.root, "data/example.json", content) + os.utime(file, (stamp, stamp)) + self.manifest() + shutil.copytree(self.root / "data", self.download / "data") + shutil.copyfile(self.root / artifact.MANIFEST, self.download / artifact.MANIFEST) + (self.root / artifact.MANIFEST).unlink() + self.git("restore", "data/example.json") + + def test_unchanged_file_uses_commit_time_not_checkout_time(self): + os.utime(self.root / "data/example.json", (NEW, NEW)) + record = self.manifest()["files"]["data/example.json"] + self.assertEqual((record["updated_at"], record["origin"]), (OLD, "git")) + + def test_generated_file_uses_generation_time(self): + file = self.write(self.root, "data/example.json", "generated") + os.utime(file, (NEW, NEW)) + record = self.manifest()["files"]["data/example.json"] + self.assertEqual((record["updated_at"], record["origin"]), (NEW, "generated")) + + def test_newer_artifact_wins_even_with_old_download_mtime(self): + self.package(NEW) + os.utime(self.download / "data/example.json", (OLD - 10, OLD - 10)) + self.assertEqual(self.overlay()["artifact-newer"], 1) + self.assertEqual((self.root / "data/example.json").read_text(), "artifact") + + def test_newer_committed_data_wins(self): + self.package(OLD + 10) + self.write(self.root, "data/example.json", "new commit") + self.commit(NEW) + self.assertEqual(self.overlay()["committed-newer-or-equal"], 1) + self.assertEqual((self.root / "data/example.json").read_text(), "new commit") + + def test_equal_timestamp_prefers_committed(self): + self.package(OLD) + self.assertEqual(self.overlay()["committed-newer-or-equal"], 1) + + def test_synthetic_merge_does_not_refresh_data_age(self): + self.git("checkout", "-b", "feature") + self.write(self.root, "data/example.json", "feature") + self.commit(OLD + 10) + self.git("checkout", "main") + self.git("merge", "--no-ff", "feature", "-m", "Aggregate PR", stamp=NEW) + self.assertEqual(self.manifest()["files"]["data/example.json"]["updated_at"], OLD + 10) + + def test_legacy_artifact_retains_conflicts_but_adds_new_files(self): + self.write(self.download, "data/example.json", "unknown age") + self.write(self.download, "static/data/new.json", "new file") + result = self.overlay() + self.assertEqual(result["committed-unknown-artifact-age"], 1) + self.assertEqual(result["artifact-only"], 1) + + def test_identical_files_need_no_timestamp_decision(self): + self.write(self.download, "data/example.json", (self.root / "data/example.json").read_text()) + self.assertEqual(self.overlay()["identical"], 1) + + def test_committed_deletions_are_not_resurrected(self): + self.package(OLD + 10) + self.git("rm", "data/example.json") + self.commit(NEW) + self.assertEqual(self.overlay()["committed-deletion-newer-or-unknown"], 1) + self.assertFalse((self.root / "data/example.json").exists()) + + def test_corrupt_manifest_fails_before_copying_anything(self): + self.package(NEW) + self.write(self.download, "data/example.json", "tampered") + with self.assertRaisesRegex(ValueError, "manifest mismatch"): + self.overlay() + self.assertIn("committed", (self.root / "data/example.json").read_text()) + + def test_uncommitted_changes_are_not_overwritten(self): + self.package(NEW) + self.write(self.root, "data/example.json", "local edits") + with self.assertRaisesRegex(ValueError, "locally modified"): + self.overlay() + + def test_excluded_files_are_not_packaged(self): + self.write(self.root, "data/partners.json", "partner configuration") + self.assertNotIn("data/partners.json", self.manifest()["files"]) + + +if __name__ == "__main__": + unittest.main()