Use the newer committed or artifact data in staging

.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/

.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:

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()

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()