publish_mix: resume listing must read lfs.oid as a dict (getattr on a dict returned None for every shard, so a 'resume' re-sent all 2.3 GB); hard-exit 11/12 when a batched commit leaves a path absent or a remote sha that disagrees with the local bytes

build/publish_mix.py

@@ -117,6 +117,27 @@ def attribution(man):
     return ATTRIBUTION + "\n".join(lines) + "\n"
 
 
+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).
+    """
+    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")
+        elif lfs is not None:
+            out[ent.rfilename] = getattr(lfs, "sha256", None) or getattr(lfs, "oid", None)
+        else:
+            out[ent.rfilename] = None
+    return out
+
+
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--root", default="/kaggle/working/mixroot")
@@ -235,30 +256,29 @@ def main():
     # because a bare upload_file has no retry and no place to resume from).
     remote_sha = {}
     try:
-        for ent in api.list_repo_files(repo_id=a.repo, repo_type="dataset"):
-            remote_sha[ent] = None
-        for ent in api.dataset_info(a.repo, files_metadata=True).siblings or []:
-            remote_sha[ent.rfilename] = getattr(ent, "lfs", None) and getattr(ent.lfs, "sha256", None) \
-                or getattr(ent, "sha256", None)
+        remote_sha = remote_digests(api, a.repo)
+        print(f"  {len(remote_sha)} file(s) already on the Hub", flush=True)
     except Exception as e:
         print(f"  (no resumable listing: {type(e).__name__}; uploading everything)", flush=True)
 
-    failures = []
     todo = []
     for np, lp in to_upload:
+        h = hashlib.sha256()
         with open(lp, "rb") as fh:
-            payload = fh.read()
-        local_sha = hashlib.sha256(payload).hexdigest()
+            for chunk in iter(lambda: fh.read(1 << 20), b""):
+                h.update(chunk)
+        local_sha = 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, payload))
+        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.
+    failures = []
     if todo:
         try:
             api.upload_folder(repo_id=a.repo, repo_type="dataset", folder_path=a.root, path_in_repo="",
@@ -270,9 +290,11 @@ def main():
         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, payload in todo:
+            for np, lp, local_sha in todo:
                 for attempt in (1, 2, 3):
                     try:
+                        with open(lp, "rb") as fh:
+                            payload = fh.read()
                         api.upload_file(path_or_fileobj=payload, path_in_repo=np, repo_id=a.repo,
                                         repo_type="dataset",
                                         commit_message=f"add {np} ({local_sha[:12]})")
@@ -290,11 +312,30 @@ def main():
               "resumes from the ones that did", flush=True)
         sys.exit(11)
 
-    # Independent verification from the Hub, not from the client that just wrote it.
+    # 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 = {}
     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:
+                sha_bad.append(np)
     R = {"repo": a.repo, "files_expected": len(want), "files_present": len(want & remote),
-         "missing": sorted(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)
     first = man["shards"][0]["file"]
     url = f"https://huggingface.co/datasets/{a.repo}/resolve/main/{first}"
     try:
@@ -307,11 +348,19 @@ 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 R["anonymous_shard_read"].get(
-        "sha256_matches_manifest"))
+    R["published_verified"] = bool(not R["missing"] and not R["sha_mismatch_on_hub"]
+                                   and R["anonymous_shard_read"].get("sha256_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)
+        sys.exit(11)
 
 
 if __name__ == "__main__":