review round: pointer written+read-back before prune, verify retried, latest rolled to final, stale-pointer scan on resume, ddp_timeout 4h, chunked batched commits, manifest self-consistency gate, full content verification after upload

build/publish_mix.py

@@ -120,21 +120,29 @@ def attribution(man):
 def remote_digests(api, repo):
     """path -> content sha256 for what is already on the Hub, so a re-run resumes instead of resending.
 
-    Only LFS-tracked files expose a content sha256 (`lfs.oid`, sha256 by construction). Small non-LFS
-    files expose only a git blob sha, which is not a content sha256, so they are reported unknown and
-    re-uploaded -- a handful of json/md/u8 files, which costs nothing. `lfs` arrives as a dict from
-    huggingface_hub, so attribute access alone silently returns None for every shard and turns a resume
-    into a full 2.3 GB re-upload (Gate 2f).
+    Only LFS-tracked files expose a content sha256 (`lfs.oid` is the LFS OID, so sha256 of the payload by
+    construction). Small non-LFS files expose only a git blob sha, which is not a content sha256, so they
+    are reported None (unknown) and verified by hashing the bytes back from `resolve` instead. `lfs` arrives
+    as a dict from huggingface_hub, so attribute access alone silently returns None for every shard and
+    turns a resume into a full 2.3 GB re-upload (Gate 2f).
+
+    A digest that is not 64 hex characters is reported unknown rather than as a mismatch: the field has
+    appeared both bare and `sha256:`-prefixed across Hub versions, and reading a format change as a content
+    disagreement would re-send everything *and* fail a repo that is actually correct.
     """
     out = {}
     for ent in api.dataset_info(repo, repo_type="dataset", files_metadata=True).siblings or []:
         lfs = getattr(ent, "lfs", None)
         if isinstance(lfs, dict):
-            out[ent.rfilename] = lfs.get("sha256") or lfs.get("oid")
+            raw = lfs.get("sha256") or lfs.get("oid")
         elif lfs is not None:
-            out[ent.rfilename] = getattr(lfs, "sha256", None) or getattr(lfs, "oid", None)
+            raw = getattr(lfs, "sha256", None) or getattr(lfs, "oid", None)
         else:
-            out[ent.rfilename] = None
+            raw = None
+        if isinstance(raw, str):
+            raw = raw.split(":")[-1].strip().lower()
+        ok = isinstance(raw, str) and len(raw) == 64 and all(c in "0123456789abcdef" for c in raw)
+        out[ent.rfilename] = raw if ok else None
     return out
 
 
@@ -187,6 +195,18 @@ def main():
                         f"{str(aud.get('mix_bytes_sha256'))[:16]}, this manifest is {want_digest[:16]}")
     if int(aud.get("val_documents_audited") or 0) <= 0:
         problems.append("the audit never scanned the held-out validation shards")
+    # A manifest whose records do not add up is not a mix, and every consumer downstream trusts it: the
+    # reader builds its corpus from man["shards"], the audit digests it, and the report quotes its totals.
+    # Nothing else compares the record list against the claimed counts, so a rebuild that truncated the
+    # list while leaving `total_tokens` alone would certify 1.11 B tokens of a 1.03 B corpus (review).
+    rec_tokens = sum(int(s.get("tokens") or 0) for s in man.get("shards") or [])
+    tot_claim = man.get("total_tokens")
+    if len(man.get("shards") or []) != int(man.get("n_shards") or -1):
+        problems.append(f"manifest carries {len(man.get('shards') or [])} shard records but claims "
+                        f"n_shards={man.get('n_shards')}")
+    if rec_tokens != int(tot_claim or -1):
+        problems.append(f"the shard records sum to {rec_tokens} tokens but total_tokens says "
+                        f"{tot_claim}")
     if problems:
         print("FATAL: contamination gate refuses publication:", flush=True)
         for p in problems:
@@ -262,35 +282,43 @@ def main():
         print(f"  (no resumable listing: {type(e).__name__}; uploading everything)", flush=True)
 
     todo = []
+    digests = {}
     for np, lp in to_upload:
         h = hashlib.sha256()
         with open(lp, "rb") as fh:
             for chunk in iter(lambda: fh.read(1 << 20), b""):
                 h.update(chunk)
-        local_sha = h.hexdigest()
+        local_sha = digests[np] = h.hexdigest()
         if remote_sha.get(np) == local_sha:
             print(f"  = {np} already on the Hub with this sha256, skipped", flush=True)
             continue
         todo.append((np, lp, local_sha))
     print(f"  {len(todo)} of {len(to_upload)} files still need uploading", flush=True)
 
-    # ONE commit for the whole batch, not one commit per file. Gate 2f proved why: 139 sequential
-    # `upload_file` commits succeeded and then the Hub began rejecting every request -- including 30-byte
-    # files -- with a bare HfHubHTTPError, which is a commit-rate limit, not a network fault. A single
-    # upload_folder call is one commit and handles LFS itself.
+    # Commit in batches of at most 120 files, not one commit per file and not one commit for everything.
+    # Gate 2f proved the first is refused: 139 sequential `upload_file` commits succeeded and then the Hub
+    # began rejecting every request -- including 30-byte files -- with a bare HfHubHTTPError, a commit-rate
+    # limit rather than a network fault. The second is limited too: `upload_folder` above a ~150-file
+    # threshold wants to split into multiple commits, and a cold publish of all 173 files would hit that.
     failures = []
-    if todo:
+    CHUNK = 120
+    for i in range(0, len(todo), CHUNK):
+        batch = todo[i:i + CHUNK]
+        paths = [t[0] for t in batch]
+        mb = sum(os.path.getsize(p) for _, p, _ in batch) / 1e6
         try:
-            api.upload_folder(repo_id=a.repo, repo_type="dataset", folder_path=a.root, path_in_repo="",
-                              allow_patterns=[t[0] for t in todo], max_workers=2,
-                              commit_message=f"publish mix-v1: {len(todo)} file(s)",
-                              commit_description="single batched commit; see audit.json for the "
-                                                 "contamination verdict bound to these bytes")
-            print(f"  + batched commit with {len(todo)} file(s)", flush=True)
+            ci = api.upload_folder(repo_id=a.repo, repo_type="dataset", folder_path=a.root,
+                                   path_in_repo="", allow_patterns=paths, max_workers=2,
+                                   commit_message=f"publish mix-v1: batch {i // CHUNK + 1} "
+                                                  f"of {len(batch)} file(s)",
+                                   commit_description="batched commits; see audit.json for the "
+                                                      "contamination verdict bound to these bytes")
+            print(f"  + batch {i // CHUNK + 1}: {len(batch)} file(s), {mb:.0f} MB, "
+                  f"commit {getattr(ci, 'oid', None) or ci}", flush=True)
         except Exception as e:
             print(f"  batched commit failed ({type(e).__name__}: {str(e)[:160]}); "
                   "falling back to per-file uploads with retries", flush=True)
-            for np, lp, local_sha in todo:
+            for np, lp, local_sha in batch:
                 for attempt in (1, 2, 3):
                     try:
                         with open(lp, "rb") as fh:
@@ -313,33 +341,66 @@ def main():
         sys.exit(11)
 
     # Independent verification from the Hub, not from the client that just wrote it. A batched commit can
-    # return without raising and still not carry every file, so presence is checked per path and the sha
-    # is re-required wherever the Hub exposes one (LFS).
-    remote = set(api.list_repo_files(repo_id=a.repo, repo_type="dataset"))
-    try:
-        after = remote_digests(api, a.repo)
-    except Exception:
-        after = {}
+    # return without raising and still not carry every file, so presence is checked per path; and because
+    # the reader that consumes this repo downloads it anonymously, every path is content-checked -- by the
+    # Hub's own digest where it publishes one (LFS), by hashing the bytes back over `resolve` where it does
+    # not (the json/md/u8 files, all of which are small). A failed listing is never allowed to read as "no
+    # mismatches", which is what `except Exception: after = {}` used to do (review finding).
+    after = {}
+    for attempt in (1, 2, 3):
+        try:
+            after = remote_digests(api, a.repo)
+            break
+        except Exception as e:
+            print(f"  verification listing attempt {attempt}: {type(e).__name__}: {str(e)[:140]}",
+                  flush=True)
+            __import__("time").sleep(20 * attempt)
+    remote = set()
+    for attempt in (1, 2, 3):
+        try:
+            remote = set(api.list_repo_files(repo_id=a.repo, repo_type="dataset"))
+            break
+        except Exception as e:
+            print(f"  file listing attempt {attempt}: {type(e).__name__}: {str(e)[:140]}", flush=True)
+            __import__("time").sleep(20 * attempt)
+    if not remote:
+        print("FATAL: the repo cannot be listed after the upload -- refusing to call it published",
+              flush=True)
+        sys.exit(15)
+
     want = {np for np, _ in to_upload} | {".gitattributes"}
     local_by_path = {np: lp for np, lp in to_upload}
-    sha_bad = []
-    for np, rsha in after.items():
-        if rsha and np in local_by_path:
-            h = hashlib.sha256()
-            with open(local_by_path[np], "rb") as fh:
-                for chunk in iter(lambda: fh.read(1 << 20), b""):
-                    h.update(chunk)
-            if h.hexdigest() != rsha:
+    base_url = f"https://huggingface.co/datasets/{a.repo}/resolve/main/"
+    sha_bad, unresolved = [], []
+    for np, lp in to_upload:
+        rsha = after.get(np)
+        if rsha is None:
+            if os.path.getsize(lp) > 8 << 20:
+                unresolved.append(f"{np} (too large to read back, no Hub digest)")
+                continue
+            try:
+                body = urllib.request.urlopen(base_url + np, timeout=300).read()
+            except Exception as e:
+                unresolved.append(f"{np} ({type(e).__name__})")
+                continue
+            if hashlib.sha256(body).hexdigest() != digests[np]:
                 sha_bad.append(np)
+        elif rsha != digests[np]:
+            sha_bad.append(np)
+    missing = sorted(want - remote)
+    # A shard on the Hub that the manifest does not name is not "extra disk", it is a shard some reader
+    # globbing `mix/*.bin` will train on while the audit never saw it.
+    extra_bin = sorted(p for p in (remote - want) if p.endswith(".bin"))
     R = {"repo": a.repo, "files_expected": len(want), "files_present": len(want & remote),
-         "missing": sorted(want - remote), "sha_mismatch_on_hub": sorted(sha_bad)[:20],
-         "sha_verified_files": sum(1 for v in after.values() if v)}
-    print("after upload: %d/%d present, %d missing, %d sha mismatch" % (
-        R["files_present"], R["files_expected"], len(R["missing"]), len(sha_bad)), flush=True)
+         "missing": missing, "sha_mismatch_on_hub": sorted(sha_bad)[:20],
+         "extra_shards_on_hub": extra_bin[:20], "unverified": unresolved[:20],
+         "sha_verified_via_hub": sum(1 for k, v in after.items() if v and k in local_by_path)}
+    print("after upload: %d/%d present, %d missing, %d sha mismatch, %d extra shards, %d unverified" % (
+        R["files_present"], R["files_expected"], len(missing), len(sha_bad), len(extra_bin),
+        len(unresolved)), flush=True)
     first = man["shards"][0]["file"]
-    url = f"https://huggingface.co/datasets/{a.repo}/resolve/main/{first}"
     try:
-        with urllib.request.urlopen(url, timeout=300) as r:
+        with urllib.request.urlopen(base_url + first, timeout=300) as r:
             body = r.read()
         hv = struct.unpack("<IIII", body[:16])
         R["anonymous_shard_read"] = {"bytes": len(body), "header": list(hv),
@@ -348,19 +409,36 @@ def main():
                                      == man["shards"][0]["sha256"]}
     except Exception as e:
         R["anonymous_shard_read"] = {"error": f"{type(e).__name__}: {str(e)[:160]}"}
-    R["published_verified"] = bool(not R["missing"] and not R["sha_mismatch_on_hub"]
-                                   and R["anonymous_shard_read"].get("sha256_matches_manifest"))
+    R["published_verified"] = bool(not missing and not sha_bad and not extra_bin and not unresolved
+                                   and R["anonymous_shard_read"].get("sha256_matches_manifest")
+                                   and R["anonymous_shard_read"].get("n_tokens_matches_manifest"))
     print("PUBLISH_JSON_BEGIN")
     print(json.dumps(R, indent=1, default=str))
     print("PUBLISH_JSON_END")
-    if not R["missing"] and R["sha_mismatch_on_hub"]:
-        print(f"FATAL: {len(R['sha_mismatch_on_hub'])} file(s) on the Hub do not match the local bytes",
-              flush=True)
-        sys.exit(12)
-    if R["missing"]:
-        print(f"FATAL: {len(R['missing'])} file(s) still absent after the commit -- re-running "
-              "publish_mix resumes from the ones that landed", flush=True)
+    # Each of these is a different thing to have gone wrong, so each gets its own exit code: 11 a path
+    # never landed (re-running resumes), 12 the Hub holds bytes that are not ours, 14 a shard the manifest
+    # does not know about, 13 something could not be confirmed at all.
+    if missing:
+        print(f"FATAL: {len(missing)} file(s) still absent after the commit -- re-running publish_mix "
+              "resumes from the ones that landed", flush=True)
         sys.exit(11)
+    if sha_bad:
+        print(f"FATAL: {len(sha_bad)} file(s) on the Hub do not match the local bytes: "
+              f"{sha_bad[:6]}", flush=True)
+        sys.exit(12)
+    if extra_bin:
+        print(f"FATAL: {len(extra_bin)} .bin file(s) on the Hub are not in this manifest: "
+              f"{extra_bin[:6]} -- a reader that globs the directory would train on bytes the audit never "
+              "saw, so they must be deleted before this mix is called published", flush=True)
+        sys.exit(14)
+    if unresolved:
+        print(f"FATAL: {len(unresolved)} file(s) could not be content-verified after upload: "
+              f"{unresolved[:6]}", flush=True)
+        sys.exit(13)
+    if not R["published_verified"]:
+        print("FATAL: the anonymous read-back of the first shard did not match the manifest -- the repo "
+              "is not provably serving the bytes it claims", flush=True)
+        sys.exit(13)
 
 
 if __name__ == "__main__":