fix: make shell hotfix startup atomic and fail closed

CHANGELOG.md

@@ -4,6 +4,10 @@
 
 - **Multi-key API auth via `DSPARK_API_KEYS` (default empty = unchanged, unauthenticated)**, with fail-closed startup-log redaction for keyed starts. vLLM receives all space-separated keys through exactly one `--api-key` flag; `VLLM_API_KEY` remains the mutually exclusive legacy single-key option, and all four auth contexts exit 2 before side effects when both are set. CR/LF/VT/FF and backslashes are rejected unconditionally before empty classification, dash-leading tokens get a fixed diagnostic that never echoes token bytes, and the launcher rejects a process-only or mismatched ambient `DSPARK_API_KEYS`. Keyed entrypoints apply and verify the redaction patch outside the optional performance-hotfix loop and fail before `exec vllm` on missing, drifted, partial, or failed status; keyless boot remains unchanged. The pinned runtime leaves every route outside `/v1`, `/v2`, `/inference` keyless, including compute-capable `POST /invocations` and `POST /generative_scoring`, so network-level access control remains required. Maintainer round-3 review hardening added the fail-closed gate, truthful status/partial-state handling, launcher preflight, ambient guard, complete route disclosure, secret-free diagnostics, and behavioral regression coverage while preserving the contributor implementation.
 
+### Changed
+
+- **Shell hotfix boot is now fail-closed, and all seven multi-hunk DSV4 backports apply transactionally**: the enabled issue22, spin-wait, and seven-script DSV4 hotfix train now aborts before `vllm serve` when a script is missing or exits nonzero; the existing `DSPARK_SKIP_ISSUE22_HOTFIX`, `DSPARK_SKIP_SPIN_WAIT_HOTFIX`, and `DSPARK_SKIP_HOTFIX` escape hatches are unchanged. Each multi-hunk script validates every target and anchor before writing, publishes through same-directory atomic renames, preserves file modes, verifies committed bytes, and restores every published target on commit or verification failure. The CPU regression suite covers all 35 hunks, idempotence, injected commit/rollback/interrupt failures, and real Compose exec blocking.
+
 ### Fixed
 
 - **Start normalizes BOM/CRLF env files and atomically publishes the worker copy ([PR #98](https://github.com/MiaAI-Lab/DeepSeek-v4-Flash-DSpark-2x-DGX-Spark/pull/98), reported and initially implemented by [@hecisaza](https://github.com/hecisaza))**: the operator file stays byte-identical while one private `0600` snapshot feeds the head, Compose, and worker. Worker credentials are staged privately and renamed atomically, so a failed transfer cannot expose or truncate the previous env file. The resolved-profile banner now reports the actual `MAX_NUM_SEQS=6` default.

docker-compose.dspark.yml

@@ -216,9 +216,9 @@ services:
         if [ -z "$${ENCODING_SOURCE}" ]; then for candidate in /cache/huggingface/hub/models--deepseek-ai--DeepSeek-V4-Flash-0731/snapshots/*/encoding/encoding_dsv4.py; do if [ -f "$${candidate}" ]; then ENCODING_SOURCE="$${candidate}"; break; fi; done; fi;
         if [ -z "$${ENCODING_SOURCE}" ] && [ -f /models/deepseek-ai/DeepSeek-V4-Flash-0731/encoding/encoding_dsv4.py ]; then ENCODING_SOURCE=/models/deepseek-ai/DeepSeek-V4-Flash-0731/encoding/encoding_dsv4.py; fi;
         if [ -n "$${ENCODING_SOURCE}" ] && [ -f "$${ENCODING_SOURCE}" ]; then cp "$${ENCODING_SOURCE}" /usr/local/lib/python3.12/dist-packages/vllm/tokenizers/deepseek_v4_encoding.py; python3 -c 'from pathlib import Path; p=Path("/usr/local/lib/python3.12/dist-packages/vllm/tokenizers/deepseek_v4.py"); s=p.read_text(); old="elif reasoning_effort in (\"max\", \"xhigh\"):\n                reasoning_effort = \"max\"\n            else:\n                reasoning_effort = \"high\""; new="elif reasoning_effort in (\"max\", \"xhigh\"):\n                reasoning_effort = \"max\"\n            elif reasoning_effort == \"high\":\n                reasoning_effort = \"high\"\n            else:\n                reasoning_effort = \"low\""; updated=s.replace(old,new); assert new in updated; p.write_text(updated)'; python3 /opt/hotfix-encoding-dsv4-issue21.py; else echo "WARN: encoding_dsv4.py not found (set DSPARK_ENCODING_FILE or check the model snapshot) - encoder copy, reasoning-effort mapping and issue21 skipped" >&2; fi; if [ "$${DSPARK_ENABLE_ISSUE31_GPU_HOTFIX:-0}" = "1" ]; then python3 /opt/hotfix-dsv4-issue31-v2-thinking-budget-gpu.py || exit 1; fi; python3 /opt/hotfix-dsv4-issue55-tool-truncation.py;
-        if [ "$${DSPARK_SKIP_ISSUE22_HOTFIX:-0}" != "1" ] && [ -f /opt/dspark-patches/hotfix-nvfp4-ds-mla-issue22.sh ]; then bash /opt/dspark-patches/hotfix-nvfp4-ds-mla-issue22.sh || true; fi;
-        if [ "$${DSPARK_SKIP_SPIN_WAIT_HOTFIX:-0}" != "1" ] && [ -f /opt/dspark-patches/hotfix-gb10-spin-wait.sh ]; then bash /opt/dspark-patches/hotfix-gb10-spin-wait.sh || true; fi;
-        if [ "$${DSPARK_SKIP_HOTFIX:-0}" != "1" ]; then for _hf in hotfix-dsv4-mtp-buffer-50312.sh hotfix-dsv4-adaptive-topk-50004.sh hotfix-dsv4-skip-topk-49486.sh hotfix-dsv4-dense-prefill-indexer-48407.sh hotfix-dsv4-skip-empty-c128-48957.sh hotfix-dsv4-flashmla-workspace-50298.sh hotfix-dsv4-grammar-advance.sh; do if [ -f /opt/dspark-patches/$${_hf} ]; then bash /opt/dspark-patches/$${_hf} || true; fi; done; fi;
+        if [ "$${DSPARK_SKIP_ISSUE22_HOTFIX:-0}" != "1" ]; then if [ ! -f /opt/dspark-patches/hotfix-nvfp4-ds-mla-issue22.sh ]; then echo "FATAL: /opt/dspark-patches/hotfix-nvfp4-ds-mla-issue22.sh missing" >&2; exit 1; fi; bash /opt/dspark-patches/hotfix-nvfp4-ds-mla-issue22.sh || exit 1; fi;
+        if [ "$${DSPARK_SKIP_SPIN_WAIT_HOTFIX:-0}" != "1" ]; then if [ ! -f /opt/dspark-patches/hotfix-gb10-spin-wait.sh ]; then echo "FATAL: /opt/dspark-patches/hotfix-gb10-spin-wait.sh missing" >&2; exit 1; fi; bash /opt/dspark-patches/hotfix-gb10-spin-wait.sh || exit 1; fi;
+        if [ "$${DSPARK_SKIP_HOTFIX:-0}" != "1" ]; then for _hf in hotfix-dsv4-mtp-buffer-50312.sh hotfix-dsv4-adaptive-topk-50004.sh hotfix-dsv4-skip-topk-49486.sh hotfix-dsv4-dense-prefill-indexer-48407.sh hotfix-dsv4-skip-empty-c128-48957.sh hotfix-dsv4-flashmla-workspace-50298.sh hotfix-dsv4-grammar-advance.sh; do if [ ! -f /opt/dspark-patches/$${_hf} ]; then echo "FATAL: /opt/dspark-patches/$${_hf} missing" >&2; exit 1; fi; bash /opt/dspark-patches/$${_hf} || exit 1; done; fi;
         if [ "$${_dspark_keys_set}" = "1" ] || [ -n "$${VLLM_API_KEY:-}" ]; then bash /opt/dspark-patches/hotfix-vllm-redact-api-key-log.sh || exit 1; bash /opt/dspark-patches/hotfix-vllm-redact-api-key-log.sh --status || exit 1; fi;
         python3 /opt/hotfix-dsv4-issue27-partial-prefill-concurrency.py; python3 /opt/hotfix-dsv4-issue43-decode-fairness-and-diag.py; python3 /opt/hotfix-dsv4-issue26-hybrid-swa-min.py; if [ "$${DSPARK_SKIP_SUPPRESS_STOPS_HOTFIX:-0}" != "1" ]; then python3 /opt/hotfix-dsv4-suppress-stops-in-reasoning.py; fi; VLLM_QUANTIZATION_ARGS="";
         if [ "$${DSPARK_ENABLE_ASSISTANT_FINAL_HOTFIX:-0}" = "1" ]; then python3 /opt/hotfix-dsv4-assistant-final-continuation.py || exit 1; fi;

patches/hotfix-dsv4-adaptive-topk-50004.sh

@@ -199,32 +199,56 @@ echo "=== Hotfix: DSV4 adaptive C128A topk width (upstream #50004 backport) ==="
 echo "vLLM root: $VLLM_ROOT  image: 0.25.2.dev0+g752a3a504.d20260714"
 
 python3 <<PYEOF
+import os
+import stat
 import sys
+import tempfile
 from pathlib import Path
 
 root = Path("$VLLM_ROOT")
 applied = 0
 skipped = 0
 errors = []
 
+# ---- whole-script transaction ------------------------------------------------
+# Hunks validate against an in-memory staged view; nothing touches the tree
+# until the last hunk has validated. originals/modes hold the per-run bytes and
+# permission bits of every target so a failed commit rolls back byte-exactly.
+originals: dict[str, bytes] = {}
+modes: dict[str, int] = {}
+staged: dict[str, str] = {}
+
+
+def _stage(path: str) -> str:
+    if path not in staged:
+        p = root / path
+        originals[path] = p.read_bytes()
+        modes[path] = stat.S_IMODE(p.stat().st_mode)
+        staged[path] = originals[path].decode()
+    return staged[path]
+
+
 def patch(path: str, old: str, new: str, label: str, expect: int = 1) -> None:
     global applied, skipped
     p = root / path
     if not p.exists():
         errors.append(f"File not found: {path}")
         return
-    text = p.read_text()
-    if new in text:
+    text = _stage(path)
+    n_old = text.count(old)
+    n_new = text.count(new)
+    if n_new == expect and n_old == new.count(old) * expect:
         print(f"  [skip] {label} (already applied)")
         skipped += 1
         return
-    n = text.count(old)
-    if n == 0 or (expect and n != expect):
-        errors.append(f"[ERR] anchor x{n} (expect {expect}) for {label} in {path}")
+    if n_old != expect or n_new != 0:
+        errors.append(
+            f"[ERR] ambiguous state for {label} in {path}: "
+            f"old x{n_old} (expect {expect}), new x{n_new}"
+        )
         return
-    text = text.replace(old, new)
-    p.write_text(text)
-    print(f"  [OK]   {label} (replaced {n})")
+    staged[path] = text.replace(old, new)
+    print(f"  [stage] {label} (prepared {n_old})")
     applied += 1
 
 
@@ -313,14 +337,124 @@ patch(
 )
 
 
-print(f"\nApplied: {applied}, Skipped: {skipped}, Errors: {len(errors)}")
+print(f"\nStaged: {applied}, Skipped: {skipped}, Errors: {len(errors)}")
 for e in errors:
     print(f"  {e}", file=sys.stderr)
 
 if errors:
-    print("\nWARNING: Some patches could not be applied. Nothing was left half-applied.")
+    print("\nWARNING: Some patches could not be applied. Nothing was written.")
     sys.exit(1)
 
+
+# ---- atomic commit -----------------------------------------------------------
+# Every hunk validated against the staged view. Publish changed targets in
+# deterministic order via same-directory temp files + os.replace (mode kept,
+# file fsynced, directory fsynced best-effort), then re-read to verify the
+# written bytes. A target is tracked as published the moment its rename lands,
+# so any later failure — including interrupts and verification faults — rolls
+# back every published target from the per-run originals in reverse order
+# through the same safe writer and re-verifies byte identity; a failed
+# rollback is loud and exits nonzero.
+
+def _fsync_dir(d: Path) -> None:
+    try:
+        fd = os.open(str(d), os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+    except OSError:
+        pass
+
+
+def _safe_write(dest: Path, data: bytes, mode: int, done: list[str] | None = None, rel: str | None = None) -> None:
+    fd, tmp_name = tempfile.mkstemp(
+        dir=str(dest.parent), prefix="." + dest.name + ".", suffix=".tmp"
+    )
+    tmp = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            fh.write(data)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.chmod(tmp, mode)
+        os.replace(tmp, dest)
+    except BaseException:
+        try:
+            tmp.unlink()
+        except FileNotFoundError:
+            pass
+        raise
+    if done is not None:
+        done.append(rel)
+    _fsync_dir(dest.parent)
+
+
+def _rollback(done: list[str], why: str) -> None:
+    print(f"\nERROR: {why}", file=sys.stderr)
+    print(f"Restoring {len(done)} published file(s) in reverse order...", file=sys.stderr)
+    for rel in reversed(done):
+        try:
+            _safe_write(root / rel, originals[rel], modes[rel])
+            if (root / rel).read_bytes() != originals[rel]:
+                raise RuntimeError("restored bytes differ from the originals")
+        except Exception as exc:
+            print(f"FATAL: rollback of {rel} failed: {exc}", file=sys.stderr)
+            print("FATAL: targets above may not match their pre-run state; inspect manually.", file=sys.stderr)
+            sys.exit(2)
+        print(f"  [restored] {rel}", file=sys.stderr)
+
+
+pending = [rel for rel in sorted(staged) if staged[rel].encode() != originals[rel]]
+if pending:
+    done: list[str] = []
+
+    def _track_published(rel: str) -> None:
+        # A failure may race the rename (e.g. an interrupt delivered after
+        # the syscall completed): if the destination already holds the staged
+        # bytes, the target is published and must join the rollback set.
+        try:
+            published = (root / rel).read_bytes() == staged[rel].encode()
+        except OSError:
+            published = True
+        if published and rel not in done:
+            done.append(rel)
+
+    try:
+        for rel in pending:
+            try:
+                _safe_write(root / rel, staged[rel].encode(), modes[rel], done, rel)
+            except Exception as exc:
+                _track_published(rel)
+                raise RuntimeError(f"commit failed for {rel}: {exc}") from exc
+            except BaseException as exc:
+                _track_published(rel)
+                raise
+
+        try:
+            bad = [
+                rel
+                for rel in done
+                if (root / rel).read_bytes() != staged[rel].encode()
+            ]
+        except Exception as exc:
+            raise RuntimeError(f"post-commit verification failed: {exc}") from exc
+        if bad:
+            raise RuntimeError(
+                "post-commit verification failed for " + ", ".join(bad)
+            )
+    except Exception as exc:
+        _rollback(done, str(exc))
+        sys.exit(1)
+    except BaseException as exc:
+        _rollback(done, f"transaction interrupted: {exc}")
+        raise
+
+    print(f"\nCommitted: {applied} hunk(s) across {len(done)} file(s).")
+
 if applied == 0 and skipped > 0:
     print("Patch already applied. No changes needed.")
 elif applied > 0: