pgw#1002/#1003/#1004/#1005: a publish failure costs the UPLOAD, not the CAST (th#1653 client half)

changelog.d/pgw1002-pgw1005.md

@@ -0,0 +1,62 @@
+- **pgw#1002/#1003/#1004/#1005 (th#1653 upload audit, client half): a publish
+  failure now costs the UPLOAD, not the CAST.** Four defects on one exit path,
+  which together turned a transient at the end of a 2h16m fp8 cast into total
+  loss.
+
+  *pgw#1002 — honor the hub's own classification.* `_map_exception` had no
+  `HubPublishError` branch at all, so a failure the hub tagged
+  `retryable: true` reported `JOB_STATUS_FATAL` and the orchestrator's
+  five-attempt budget was never spent on the final artifact of a two-hour job.
+  It is mapped from the th#1301 `retryable` bit (`True` -> RETRYABLE; `False`
+  and "the hub named nothing" stay FATAL), with the hub's `code` leading the
+  detail so refusals group by a stable token. Separately, the exception handler
+  used to `DELETE` the publish session on ANY exception, and that DELETE
+  reclaims the staging prefix hub-side — 37 GB already on the wire, thrown away
+  to tidy a session row. The abort now fires only for a refusal the hub itself
+  classified terminal.
+
+  *pgw#1003 — the artifact survives.* `IncrementalSafetensorsWriter` writes to
+  a same-directory temp and reaches its real name only via fsync ->
+  `os.replace` -> fsync(dir), so a hard-killed pod can no longer leave a
+  truncated cast output (and an incomplete tensor set is never committed). A
+  new `convert/publish_journal.py` records the in-flight `publish_id`, the
+  declared object set and the producer's own recovery state NEXT TO the
+  produced bytes, before the first PUT; `publish_v2` re-adopts a journalled
+  session instead of declaring a fresh one, `clone.py` retains a workdir that
+  still has a resumable session (reaped by the existing scratch sweeper), and a
+  tree a predecessor finished casting is re-published rather than re-cast. This
+  is a SAME-POD durability record by construction — the disk dies with the pod
+  — and it is built to exploit th#1654's staging-aware re-plan the moment it
+  lands hub-side.
+
+  *pgw#1004 — the chunk data plane grows up.* `_put_one` had five attempts with
+  no `time.sleep` anywhere: four threads hammering a store that had just
+  answered 429. It now classifies before charging an attempt and backs off on
+  the repo's one decorrelated-jitter helper (`_upload_transport.backoff_sleep_s`,
+  now public and shared). `UploadGrant` carries `expires_at` — on the wire since
+  th#1303 and read by nobody — so a grant that is expired, or too close to it to
+  start a 64 MiB PUT, re-plans (the CAS path's own presign re-mint) instead of
+  spending a terminal-classification pass; an expired 403 and a substituted
+  claim are finally distinguishable. Every completed object feeds
+  `upload:bytes` + `note_progress()`, so a healthy multi-GB publish stops
+  looking exactly like a wedge to the 10-minute activity stall window, and
+  `convert/publish.py` emits per-leg `convert_publish` events. Concurrency 4 ->
+  8 under a 16-PUT process ceiling, and the slot is taken BEFORE the span is
+  read so it bounds buffers rather than sockets. `_concurrent_upload.py`'s
+  claim to own a `parallel_map_uploads` that has never existed is deleted.
+
+  *pgw#1005 — verified digests and real tests.* `upload_file_with_grant`
+  returned the caller's unverified claim; it now proves the digest from the
+  bytes, refuses an expired scoped credential (`expires_at` was parsed and
+  never read), and no longer multiplies four outer attempts by botocore's ten.
+  `_upload_transport` gets its first functional coverage — re-open-per-attempt
+  at the true offset, fresh pool per retry, `discard_connections` on failure,
+  every classifier branch, the 2xx-without-ETag refusal. The four adversarial
+  injectors `tests/convert/fake_hub.py` has shipped since the day it was
+  written, which no test ever switched on, are wired to the v2 chunk surface
+  and driven: mid-PUT reset, 5xx-then-success (asserting the delay was taken,
+  by injection — no wall clock), expired presign, terminal repudiation. Plus
+  one row at the real 64 MiB chunk size, which no upload test had ever
+  allocated.
+
+  No version bump: this rides the next train.

src/gen_worker/_upload_transport.py

@@ -35,9 +35,11 @@
 ``presigned_upload.py``.
 
 Public API: ``upload_part_to_presigned_url(url, file_path, offset,
-length, pool=None)`` -> ``etag``, plus ``PutPool``. Caller owns the
-part-level fan-out (``presigned_upload.py``) and the file-level fan-out
-(``_concurrent_upload.py``). This module is a pure transport leaf.
+length, pool=None)`` -> ``etag``, ``PutPool``, and ``backoff_sleep_s`` —
+the ONE decorrelated-jitter backoff every worker-side upload loop uses
+(pgw#1004 wired the chunk-CAS data plane onto it rather than growing a
+fourth implementation). Caller owns the part-level fan-out
+(``presigned_upload.py``). This module is a pure transport leaf.
 """
 
 from __future__ import annotations
@@ -247,7 +249,7 @@ def _classify_transport_exception(exc: BaseException) -> TransportError:
     return TransportError(f"S3 part upload unexpected error: {exc!r}", retryable=False)
 
 
-def _backoff_sleep_s(attempt: int, base_s: float = _BACKOFF_BASE_S, cap_s: float = _BACKOFF_CAP_S) -> float:
+def backoff_sleep_s(attempt: int, base_s: float = _BACKOFF_BASE_S, cap_s: float = _BACKOFF_CAP_S) -> float:
     """Decorrelated-jitter backoff (AWS Architecture Blog: 'Exponential Backoff And Jitter').
 
     Attempt is 1-indexed; attempt=1 returns U[base, base*3], attempt=2
@@ -349,7 +351,7 @@ def upload_part_to_presigned_url(
             last_err = err
             if not err.retryable or attempt >= max_attempts:
                 raise err
-            sleep_s = _backoff_sleep_s(attempt)
+            sleep_s = backoff_sleep_s(attempt)
             logger.info(
                 "presigned_part_retry attempt=%d/%d sleep_s=%.2f err=%s",
                 attempt, max_attempts, sleep_s, err,
@@ -387,7 +389,7 @@ def upload_part_to_presigned_url(
         last_err = status_err
         if not status_err.retryable or attempt >= max_attempts:
             raise status_err
-        sleep_s = _backoff_sleep_s(attempt)
+        sleep_s = backoff_sleep_s(attempt)
         logger.info(
             "presigned_part_retry attempt=%d/%d sleep_s=%.2f status=%d",
             attempt, max_attempts, sleep_s, status_err.status_code or 0,

src/gen_worker/convert/clone.py

@@ -26,6 +26,7 @@
 from gen_worker.api.errors import ValidationError
 
 from .hub import HubClient, files_from_tree
+from .publish_journal import JOURNAL_NAME, PublishJournal
 from .keepalive import HubKeepalive
 from .ingest import (
     IngestedSource,
@@ -543,6 +544,51 @@ def _preflight_disk(workdir: Path, plan: Any, specs: list[OutputSpec]) -> None:
             f"{_DISK_MARGIN_BYTES / gib:.0f} GiB margin), have {free / gib:.1f} GiB "
             f"at {workdir}")
 
+def _reusable_flavor_tree(
+    workdir: Path, spec_label: str, flavor_dir: Path,
+) -> Optional[dict[str, str]]:
+    """The flavor attrs of a retained tree this run may re-publish as-is, or None.
+
+    pgw#1003. A publish that failed on a retryable error leaves its session
+    live and its journal entry behind, and ``run_clone``'s ``finally`` then
+    RETAINS the workdir instead of deleting it. This is the other half: on the
+    retry, a tree the predecessor already finished casting is re-published
+    rather than re-cast — which is the whole point, since the cast is the $10
+    and the two hours.
+
+    Three conditions, all cheap and all necessary:
+      * a journal entry exists for this spec label (so the predecessor got as
+        far as DECLARING, which happens only after the tree was complete and
+        every file hashed — a run that died mid-cast leaves no entry);
+      * the tree on disk still yields exactly the file set that entry declared;
+      * the entry carries the flavor attrs, so nothing has to be re-derived.
+
+    Bytes are re-proven for free downstream: ``publish_v2`` re-hashes every
+    file and adopts the journalled session only if the artifact key matches,
+    so a same-shape-but-corrupted tree publishes fresh instead of resuming
+    wrong.
+    """
+    if not flavor_dir.is_dir():
+        return None
+    journal = PublishJournal.open(workdir / JOURNAL_NAME)
+    entry = journal.for_producer(spec_label=str(spec_label), tree=str(flavor_dir))
+    if entry is None:
+        return None
+    attrs = entry.producer_state.get("attrs")
+    if not isinstance(attrs, dict):
+        return None
+    if not entry.declares([f.path for f in files_from_tree(flavor_dir)]):
+        logger.info(
+            "flavor-%s: retained tree no longer matches publish %s's declaration; "
+            "rebuilding", spec_label, entry.publish_id)
+        return None
+    logger.warning(
+        "flavor-%s: REUSING the retained cast output for publish %s — "
+        "re-uploading rather than re-casting (pgw#1003)",
+        spec_label, entry.publish_id)
+    return {str(k): str(v) for k, v in attrs.items()}
+
+
 def _sweep_stale_workdirs(base: Path, *, keep: Optional[Path] = None) -> None:
     """Remove clone scratch left by crashed predecessors: dirs whose flock is
     free and that have been idle past COZY_CONVERT_SCRATCH_TTL_S (default 1h).
@@ -854,18 +900,24 @@ def _dl_progress(done: int, total: Optional[int]) -> None:
                             target_dtype=spec.dtype,
                         )
                 else:
-                    # Wipe any partial flavor tree from a prior failed run —
-                    # only the downloaded source is resumable.
                     flavor_dir = workdir / f"flavor-{spec.label}"
-                    shutil.rmtree(flavor_dir, ignore_errors=True)
-                    shutil.rmtree(workdir / f"flavor-{spec.label}.__repack__",
-                                  ignore_errors=True)
-                    tree, attrs = build_flavor_tree(
-                        source, spec, flavor_dir,
-                        quantize_components=quantize_components,
-                        objective=objective_fact,
-                        distilled=distilled_fact,
-                    )
+                    # pgw#1003: a tree a predecessor already CAST and DECLARED
+                    # is re-published, not re-cast. Anything else is wiped —
+                    # a partial tree from a run that died mid-cast is not
+                    # resumable, and never was.
+                    reused = _reusable_flavor_tree(workdir, spec.label, flavor_dir)
+                    if reused is not None:
+                        tree, attrs = flavor_dir, reused
+                    else:
+                        shutil.rmtree(flavor_dir, ignore_errors=True)
+                        shutil.rmtree(workdir / f"flavor-{spec.label}.__repack__",
+                                      ignore_errors=True)
+                        tree, attrs = build_flavor_tree(
+                            source, spec, flavor_dir,
+                            quantize_components=quantize_components,
+                            objective=objective_fact,
+                            distilled=distilled_fact,
+                        )
                     # dtype="source" resolves to the detected on-disk dtype.
                     flavor_label = str(attrs.get("dtype") or spec.dtype)
                 # Hub flavor tokens are [a-z0-9][a-z0-9._-]{0,63}: the gguf
@@ -956,6 +1008,18 @@ def _dl_progress(done: int, total: Optional[int]) -> None:
                 metadata=metadata,
                 provenance=provenance,
                 repo_spec=source.repo_spec,
+                # pgw#1003: the session id lands beside the produced bytes
+                # before the first PUT, so a retry re-uploads instead of
+                # re-cloning. Kept at the workdir root, out of every flavor
+                # tree `files_from_tree` walks.
+                journal_path=workdir / JOURNAL_NAME,
+                # What `_reusable_flavor_tree` reads back on the retry so the
+                # cast does not have to run again.
+                journal_state={
+                    "spec_label": str(spec.label),
+                    "tree": str(tree),
+                    "attrs": {str(k): str(v) for k, v in attrs.items()},
+                },
             )
             result.published.append({
                 "flavor": flavor_label,
@@ -996,16 +1060,34 @@ def _dl_progress(done: int, total: Optional[int]) -> None:
                     keepalive.probes, keepalive.longest_outage_s,
                     keepalive.reachable)
         # gw#462: a long-running worker must not leak scratch — the workdir
-        # goes after EVERY job. Cross-run resume lives in the publish bank
-        # (th#592) + CAS dedup, not in retained local bytes.
+        # goes after every job that has nothing left to resume.
         # pgw#929: COZY_CONVERT_RETAIN_WORKDIR deleted. Retaining a failed
         # job's scratch is a DEBUGGING ACTION taken against one run, not a
         # deployment mode a pod is booted in — and as an env it could only ever
         # be set fleet-wide and forgotten, which is how a long-running worker
         # leaks scratch (the gw#462 defect this cleanup exists for).
-        shutil.rmtree(workdir, ignore_errors=True)
-        if not succeeded:
-            logger.warning("clone failed; workdir %s removed", workdir)
+        #
+        # pgw#1003 makes ONE exception, and it is not a debugging one: a
+        # publish that failed with its session still live has a journal entry
+        # naming it, and the produced tree is the only copy of bytes that cost
+        # hours of GPU. Deleting it means the retry re-runs the cast to redo an
+        # upload. So the tree survives exactly as long as there is a session to
+        # resume it into — a real, machine-checkable condition, not a mode.
+        # The disk budget is the one that already exists: `_sweep_stale_workdirs`
+        # reaps any unlocked workdir past COZY_CONVERT_SCRATCH_TTL_S (1 h) at
+        # the start of the next clone, and the hub's own staging lifecycle
+        # (th#1319) bounds how long resuming is possible anyway.
+        resumable = 0 if succeeded else len(
+            PublishJournal.open(workdir / JOURNAL_NAME).entries)
+        if resumable:
+            logger.warning(
+                "clone failed with %d publish session(s) still resumable; "
+                "RETAINING %s so a retry re-uploads instead of re-cloning "
+                "(swept after COZY_CONVERT_SCRATCH_TTL_S)", resumable, workdir)
+        else:
+            shutil.rmtree(workdir, ignore_errors=True)
+            if not succeeded:
+                logger.warning("clone failed; workdir %s removed", workdir)
         os.close(lock_fd)  # releases the flock