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

train/hubckpt.py

@@ -211,8 +211,29 @@ def ensure_repo(repo, api, token=None, repo_type="dataset"):
             "files_before": len(have)}
 
 
+def prune_local(dirpath, res, min_free_gb=6.0):
+    """Delete a local checkpoint whose Hub copy has been verified, and record the space that freed.
+
+    Separate from `push_and_prune` because the caller has to roll `latest.json` and read it back *between*
+    the verification and the deletion: pruning first would destroy the only copy the pointer should be
+    naming, and a pointer commit that failed silently would then be unrecoverable.
+    """
+    shutil.rmtree(dirpath, ignore_errors=True)
+    res["pruned"] = True
+    res["local_dir_exists_after"] = os.path.exists(dirpath)
+    res["free_after_gb"] = round(free_bytes() / 1e9, 2)
+    res["disk_ok"] = res["free_after_gb"] >= min_free_gb
+    if not res["disk_ok"]:
+        # Free space is a training-blocking metric (§3.13), not a footnote: say so loudly here rather
+        # than letting the next checkpoint fail with an OSError.
+        res["disk_warning"] = (f"only {res['free_after_gb']} GB free after pruning "
+                               f"(floor {min_free_gb} GB)")
+        print(f"   DISK WARNING: {res['disk_warning']}", flush=True)
+    return res
+
+
 def push_and_prune(repo, dirpath, path_in_repo, api, repo_type="dataset", token=None,
-                   prune=True, min_free_gb=6.0):
+                   prune=True, min_free_gb=6.0, verify_tries=3):
     """One full §3.13 cycle. Raises if verification fails and pruning was requested -- the local copy is
     kept in exactly that case, because the Hub copy is not yet proven to exist."""
     files = sorted(os.listdir(dirpath))
@@ -224,26 +245,28 @@ def push_and_prune(repo, dirpath, path_in_repo, api, repo_type="dataset", token=
     up = api.upload_folder(repo_id=repo, folder_path=dirpath, path_in_repo=path_in_repo,
                            repo_type=repo_type, commit_message=f"checkpoint {path_in_repo}",
                            token=token)
-    ver = verify_checkpoint(repo, dirpath, path_in_repo, token=token, repo_type=repo_type)
+    # Retry the verification, not the upload. `verify_checkpoint` reads the tree immediately after the
+    # commit, and E-033 showed this Hub path answers transiently under load; treating a propagation race
+    # as a content problem would kill a billed session over a checkpoint that is actually fine.
+    ver = None
+    for attempt in range(1, verify_tries + 1):
+        ver = verify_checkpoint(repo, dirpath, path_in_repo, token=token, repo_type=repo_type)
+        if ver["ok"]:
+            break
+        if attempt < verify_tries:
+            print(f"   verify attempt {attempt}: {len(ver['missing'])} missing, "
+                  f"{len(ver['mismatch'])} mismatched -- re-checking in {20 * attempt}s", flush=True)
+            __import__("time").sleep(20 * attempt)
     res = {"uploaded_to": getattr(up, "commit_url", str(up)[:120]), "verify": ver,
-           "free_before_gb": round(free_before / 1e9, 2)}
+           "free_before_gb": round(free_before / 1e9, 2), "pruned": False,
+           "free_after_gb": round(free_bytes() / 1e9, 2)}
     if prune:
         if not ver["ok"]:
-            res["pruned"] = False
             res["prune_blocked_reason"] = (
                 f"hub verification failed ({len(ver['missing'])} missing, "
                 f"{len(ver['mismatch'])} mismatched) -- keeping the local copy in {dirpath}")
             raise RuntimeError(res["prune_blocked_reason"])
-        shutil.rmtree(dirpath, ignore_errors=True)
-        res["pruned"] = True
-        res["local_dir_exists_after"] = os.path.exists(dirpath)
-    res["free_after_gb"] = round(free_bytes() / 1e9, 2)
-    res["disk_ok"] = res["free_after_gb"] >= min_free_gb
-    if prune and not res["disk_ok"]:
-        # Free space is a training-blocking metric (§3.13), not a footnote: say so loudly here rather
-        # than letting the next checkpoint fail with an OSError.
-        res["disk_warning"] = (f"only {res['free_after_gb']} GB free after pruning "
-                               f"(floor {min_free_gb} GB)")
+        prune_local(dirpath, res, min_free_gb)
     return res
 
 
@@ -311,10 +334,33 @@ def download_checkpoint(repo, path_in_repo, dest, api, repo_type="dataset", toke
             "dir": dirs.pop(), "requested_dest": dest, "path_in_repo": path_in_repo}
 
 
-def write_latest_pointer(repo, cur, api, repo_type="dataset", token=None):
+def write_latest_pointer(repo, cur, api, repo_type="dataset", token=None, retries=3):
+    """Roll `latest.json` and prove the Hub now serves it.
+
+    The read-back is not ceremony: the caller prunes its local copy immediately afterwards, so a pointer
+    commit that silently failed would leave a Hub-verified checkpoint that nothing points at, and the next
+    session would resume from the previous one and re-read ~100 M tokens with every self-consistency guard
+    passing (weights, cursor, fingerprint and seed all agree with each other -- only the pointer is wrong).
+    """
     p = os.path.join("/tmp", "latest.json")
     with open(p, "w") as f:
         json.dump(cur, f, indent=1, sort_keys=True)
-    api.upload_file(path_or_fileobj=p, path_in_repo="latest.json", repo_id=repo,
-                    repo_type=repo_type, commit_message=f"latest -> {cur.get('step')}", token=token)
-    return p
+    want_step = cur.get("step")
+    last = None
+    for attempt in range(1, retries + 1):
+        try:
+            api.upload_file(path_or_fileobj=p, path_in_repo="latest.json", repo_id=repo,
+                            repo_type=repo_type, commit_message=f"latest -> {want_step}", token=token)
+        except Exception as e:
+            last = f"upload {type(e).__name__}: {str(e)[:120]}"
+        else:
+            back = latest_pointer(repo, repo_type=repo_type, token=token)
+            if back.get("step") == want_step and back.get("path_in_repo") == cur.get("path_in_repo"):
+                return p
+            last = f"hub reports {back.get('step')!r} at {back.get('path_in_repo')!r}, " \
+                   f"we wrote {want_step!r} at {cur.get('path_in_repo')!r} ({str(back)[:120]})"
+        print(f"   latest.json attempt {attempt} failed: {last}; retrying", flush=True)
+        __import__("time").sleep(15 * attempt)
+    raise RuntimeError(f"latest.json never landed at step {want_step} on {repo} after {retries} "
+                       f"attempts: {last}. Refusing to prune the local copy that this pointer should "
+                       "point at.")