From e4bf57d7554ddfcad5f76ee667ec3567f5955673 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:04:16 +0000 Subject: [PATCH 1/3] fix(miner): sized object PUT, bounded 5xx retry, parallel part fetch - _put_object sends a bytes body (Content-Length) instead of a one-shot iterator (chunked): the miner token proxy and similar fronts reject chunked uploads. Content-addressed PUT, so transport errors and 5xx are retried (1+5 attempts, ~31 s backoff); 4xx propagate at once. - _get_object and Api.call GET retry the same way; signed POSTs are never replayed. - fetch_to(workers=N) prefetches a bounded sliding window (4N parts) of 1 MiB parts concurrently, written in index order with unchanged per-part and whole-object verification; queued GETs are cancelled on error. The miner uses 16 workers for job inputs (~5.7 GB per island). --- src/hypertrain/data/store.py | 38 +++++++++-- src/hypertrain/miner/core.py | 93 ++++++++++++++++++--------- tests/data/test_parts.py | 44 +++++++++++++ tests/miner/test_client_network_v2.py | 80 +++++++++++++++++++++++ 4 files changed, 219 insertions(+), 36 deletions(-) diff --git a/src/hypertrain/data/store.py b/src/hypertrain/data/store.py index eac8032..e8201dd 100644 --- a/src/hypertrain/data/store.py +++ b/src/hypertrain/data/store.py @@ -15,7 +15,10 @@ import re import stat import tempfile -from collections.abc import Callable, Mapping +from collections import deque +from collections.abc import Callable, Iterator, Mapping +from concurrent.futures import ThreadPoolExecutor +from itertools import islice from pathlib import Path from typing import Protocol from urllib.parse import parse_qsl, quote, urlsplit @@ -344,15 +347,33 @@ def chunks() -> bytes: return _publish(chunks, put) -def _stream(sha: str, get: Get, sink: Callable[[bytes], object]) -> None: +def _parts(get: Get, shas: list[str], workers: int) -> Iterator[bytes]: + """Verified parts in index order; workers > 1 prefetches a bounded window concurrently + (at most 4 * workers parts = 4 * workers MiB in flight).""" + if workers <= 1: + yield from (_get(get, s) for s in shas) + return + pending = iter(shas) + with ThreadPoolExecutor(workers) as pool: + try: + inflight = deque(pool.submit(_get, get, s) for s in islice(pending, 4 * workers)) + while inflight: + part = inflight.popleft().result() + for s in islice(pending, 1): + inflight.append(pool.submit(_get, get, s)) + yield part + finally: # error/cancel/early close: drop queued GETs instead of draining them + pool.shutdown(wait=True, cancel_futures=True) + + +def _stream(sha: str, get: Get, sink: Callable[[bytes], object], workers: int = 1) -> None: raw = _get(get, sha) index = parse_index(raw) if index is None: sink(raw) return whole, size = hashlib.sha256(), 0 - for i, part_sha in enumerate(index.parts): - part = _get(get, part_sha) + for i, part in enumerate(_parts(get, list(index.parts), workers)): want = PART_BYTES if i < len(index.parts) - 1 else index.size - size if len(part) != want: raise CorruptObjectError(f"part {i} has length {len(part)}, expected {want}") @@ -369,13 +390,16 @@ def get_large(sha: str, get: Get) -> bytes: return bytes(out) -def fetch_to(sha: str, get: Get, dest: Path) -> None: - """Verified reassembly into `dest` via same-directory temp + atomic replace.""" +def fetch_to(sha: str, get: Get, dest: Path, workers: int = 1) -> None: + """Verified reassembly into `dest` via same-directory temp + atomic replace. + + With workers > 1 parts are fetched concurrently (written in order), so `get` must be + thread-safe.""" dest = Path(dest) fd, tmp = tempfile.mkstemp(dir=dest.parent, prefix=f".{dest.name}.", suffix=".part") try: with os.fdopen(fd, "wb") as f: - _stream(sha, get, f.write) + _stream(sha, get, f.write, workers) f.flush() os.fsync(f.fileno()) os.replace(tmp, dest) diff --git a/src/hypertrain/miner/core.py b/src/hypertrain/miner/core.py index 2461aea..2ce57f6 100644 --- a/src/hypertrain/miner/core.py +++ b/src/hypertrain/miner/core.py @@ -49,6 +49,25 @@ log = logging.getLogger("hypertrain.miner") EXP_ROUNDS = 2**40 # envelope expiry far beyond any round; replay protection is per (w, type) +# Backoff between attempts of an idempotent request (object PUT/GET, API GET) that hit a +# transport error or a 5xx; the last failure propagates. ~31 s worst case per request. +RETRY_DELAYS: tuple[float, ...] = (1, 2, 4, 8, 16) +FETCH_WORKERS = 16 # concurrent 1 MiB part GETs per job input (bounded window, ordered sink) + + +def _retrying[T](what: str, fn: Callable[[], T]) -> T: + """Run an idempotent request, retrying transport errors and 5xx responses only.""" + for delay in (*RETRY_DELAYS, None): + try: + return fn() + except (httpx.TransportError, httpx.HTTPStatusError, ChallengeHTTPError) as error: + status = getattr(getattr(error, "response", None), "status_code", None) + status = getattr(error, "status", status) + if delay is None or (status is not None and status < 500): + raise + log.warning("%s failed (%s); retry in %ss", what, error, delay) + time.sleep(delay) + raise AssertionError("unreachable") class MinerError(RuntimeError): @@ -219,6 +238,11 @@ def __init__(self, client: httpx.Client, base: str) -> None: self.c, self.base = client, base.rstrip("/") def call(self, method: str, path: str, **kw: Any) -> Any: + if method == "GET": # reads are idempotent; signed POSTs are never replayed here + return _retrying("GET " + path, lambda: self._call(method, path, **kw)) + return self._call(method, path, **kw) + + def _call(self, method: str, path: str, **kw: Any) -> Any: r = self.c.request(method, self.base + path, timeout=120, **kw) if r.status_code >= 400: try: @@ -819,20 +843,26 @@ def put(data: bytes) -> str: return put def _put_object(self, data: bytes) -> str: - """One signed v2 object PUT (plain object, part or part index; each <= 1 MiB).""" + """One signed v2 object PUT (plain object, part or part index; each <= 1 MiB). + + Bytes body, so httpx sends Content-Length (proxies reject chunked uploads); the PUT is + content-addressed, so transport errors and 5xx are retried.""" digest = sha256_hex(data) - response = self.api.c.put( - self.api.base + self.url + "/objects/" + digest, - content=iter([data]), # one-shot body: httpx keeps no request.content copy - headers={ - "X-Hotkey": self.kp.ss58, - "X-Object-Signature": self.kp.sign( - f"hypertrain/object/2|{self.run_id}|{digest}".encode() - ).hex(), - }, - ) - response.raise_for_status() - return digest + headers = { + "X-Hotkey": self.kp.ss58, + "X-Object-Signature": self.kp.sign( + f"hypertrain/object/2|{self.run_id}|{digest}".encode() + ).hex(), + } + + def put() -> str: + response = self.api.c.put( + self.api.base + self.url + "/objects/" + digest, content=data, headers=headers + ) + response.raise_for_status() + return digest + + return _retrying("PUT object " + digest, put) def _fetch_inputs(self, job: IslandJobV1, objects: Mapping[str, str], directory: Path) -> None: """objects: name -> stored ref (plain or ht-parts-v1 index); the job pins whole SHAs. @@ -852,7 +882,7 @@ def _fetch_inputs(self, job: IslandJobV1, objects: Mapping[str, str], directory: for name, digest in objects.items(): dest = directory / job.object_paths[name] try: - fetch_to(digest, self._get_object, dest) + fetch_to(digest, self._get_object, dest, workers=FETCH_WORKERS) except StoreError as e: raise MinerError(f"job input {name} corrupt: {e}") from None if name in whole: @@ -867,21 +897,26 @@ def _get_object(self, digest: str) -> bytes: from hypertrain.protocol.messages_v2 import MAX_MANIFEST_BYTES, PART_BYTES limit = max(PART_BYTES, MAX_MANIFEST_BYTES) - with self.api.c.stream( - "GET", - self.api.base + self.url + "/objects/" + digest, - headers={"Accept-Encoding": "identity"}, - ) as response: - response.raise_for_status() - # httpx would decode gzip/br before the bound check; a hostile server ignores headers. - if response.headers.get("Content-Encoding", "identity").lower() != "identity": - raise MinerError("job input object has a non-identity content encoding") - body = bytearray() - for chunk in response.iter_bytes(chunk_size=64 << 10): - if len(body) + len(chunk) > limit: - raise MinerError("job input object exceeds the 1 MiB part bound") - body.extend(chunk) - return bytes(body) + + def get() -> bytes: + with self.api.c.stream( + "GET", + self.api.base + self.url + "/objects/" + digest, + headers={"Accept-Encoding": "identity"}, + ) as response: + response.raise_for_status() + # httpx would decode gzip/br before the bound check; a hostile server ignores + # headers. + if response.headers.get("Content-Encoding", "identity").lower() != "identity": + raise MinerError("job input object has a non-identity content encoding") + body = bytearray() + for chunk in response.iter_bytes(chunk_size=64 << 10): + if len(body) + len(chunk) > limit: + raise MinerError("job input object exceeds the 1 MiB part bound") + body.extend(chunk) + return bytes(body) + + return _retrying("GET object " + digest, get) def proof(self, document: Path) -> dict[str, Any]: from hypertrain.protocol.envelope_v2 import load_json diff --git a/tests/data/test_parts.py b/tests/data/test_parts.py index a1d26ec..b584974 100644 --- a/tests/data/test_parts.py +++ b/tests/data/test_parts.py @@ -211,3 +211,47 @@ def chunks() -> bytes: with pytest.raises(StoreError): _publish(chunks, rec.put) assert all(parse_index(b) is None for b in rec.puts) + + +def test_parallel_fetch_writes_parts_in_order_despite_out_of_order_completion( + tmp_path: Path, +) -> None: + import random + import threading + import time + + # Given: a 9-part object whose GETs finish in random order on several threads. + rec, data, index = _setup(tmp_path, 8 * PART_BYTES + 77) + ref = hashlib.sha256(rec.puts[-1]).hexdigest() + threads: set[int] = set() + + def get(sha: str) -> bytes: + threads.add(threading.get_ident()) + time.sleep(random.random() / 200) + return rec.store.get(sha) + + out = tmp_path / "out" + out.mkdir() + # When: fetched with 4 workers. + fetch_to(ref, get, out / "dest", workers=4) + # Then: byte-identical reassembly, really concurrent. + assert (out / "dest").read_bytes() == data and len(threads) > 1 + + +def test_parallel_fetch_failing_part_leaves_no_temp_and_stops_queueing(tmp_path: Path) -> None: + rec, _, index = _setup(tmp_path, 8 * PART_BYTES + 1) + ref = hashlib.sha256(rec.puts[-1]).hexdigest() + asked: list[str] = [] + + def get(sha: str) -> bytes: + asked.append(sha) + if sha == index.parts[0]: + raise ObjectNotFound(sha) + return rec.store.get(sha) + + out = tmp_path / "out" + out.mkdir() + with pytest.raises(ObjectNotFound): + fetch_to(ref, get, out / "dest", workers=2) + # Window is 4 * workers = 8 parts: the 9th is never requested. + assert os.listdir(out) == [] and index.parts[-1] not in asked diff --git a/tests/miner/test_client_network_v2.py b/tests/miner/test_client_network_v2.py index b88f70a..f08bfea 100644 --- a/tests/miner/test_client_network_v2.py +++ b/tests/miner/test_client_network_v2.py @@ -326,3 +326,83 @@ def gzip_route(request: httpx.Request) -> httpx.Response: with pytest.raises(MinerError, match="content encoding"): miner._get_object("ab" * 32) assert seen == ["identity"] and served == [] + + +def test_object_put_sends_content_length_and_retries_5xx_and_transport(tmp_path: Path, monkeypatch): + import hypertrain.miner.core as core + + # Given: a proxy that drops the first PUT, answers 503 to the second, then stores. + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0)) + miner, store, puts = _parts_miner(tmp_path) + route, seen = miner.api.c._transport.handler, [] + + def flaky(request: httpx.Request) -> httpx.Response: + seen.append( + (request.headers.get("Content-Length"), request.headers.get("Transfer-Encoding")) + ) + if len(seen) == 1: + raise httpx.ConnectError("reset", request=request) + return httpx.Response(503) if len(seen) == 2 else route(request) + + miner.api.c._transport.handler = flaky + # When: one object PUT. + digest = miner._put_object(b"payload") + # Then: stored after two retries, every attempt a sized (non-chunked) body. + assert store.get(digest) == b"payload" and len(puts) == 1 + assert seen == [("7", None)] * 3 + + +def test_object_put_does_not_retry_4xx_and_gives_up_after_bounded_5xx(tmp_path: Path, monkeypatch): + import hypertrain.miner.core as core + + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0)) + miner, _, _ = _parts_miner(tmp_path) + calls = [] + + def answer(status: int): + def handler(request: httpx.Request) -> httpx.Response: + calls.append(status) + return httpx.Response(status) + + return handler + + # 403: refused at once. + miner.api.c._transport.handler = answer(403) + with pytest.raises(httpx.HTTPStatusError): + miner._put_object(b"x") + assert calls == [403] + # 500 forever: 1 attempt + len(RETRY_DELAYS) retries, then the error propagates. + calls.clear() + miner.api.c._transport.handler = answer(500) + with pytest.raises(httpx.HTTPStatusError): + miner._put_object(b"x") + assert calls == [500] * 3 + + +def test_object_get_and_api_get_retry_5xx_but_signed_post_is_never_replayed( + tmp_path: Path, monkeypatch +): + import hypertrain.miner.core as core + from hypertrain.miner.core import ChallengeHTTPError + + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0)) + miner, store, _ = _parts_miner(tmp_path) + sha = store.put(b"part") + route, hits = miner.api.c._transport.handler, [] + + def flaky(request: httpx.Request) -> httpx.Response: + hits.append(request.method) + if len(hits) % 2 == 1: + return httpx.Response(502, json={"detail": "bad gateway"}) + if request.method == "POST": + return httpx.Response(200, json={"ok": True}) + return route(request) + + miner.api.c._transport.handler = flaky + assert miner._get_object(sha) == b"part" and hits == ["GET", "GET"] + hits.clear() + assert "manifest_envelope" in miner.api.call("GET", "/v2/runs/x") and len(hits) == 2 + hits.clear() + with pytest.raises(ChallengeHTTPError, match="502"): + miner.api.call("POST", "/v2/runs/x/commit", json={}) + assert hits == ["POST"] From acec6baadd4df11cf6bbaa0860601b855049ea78 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:07:39 +0000 Subject: [PATCH 2/3] feat(scripts): one-command 8x RTX 5090 CUDA run + miner launch scripts/launch_cuda_run.sh chains, for one TAG (manifest/job/prep set): preflight -> beacon pusher -> Vast admit -> deps/seed staging -> CUDA qualification -> rescue -> roots -> backend bootstrap -> POST /v2/admin/runs (verifies the run froze on backend cuda) -> dispatch.json + master reload -> auditors -> relays -> miner host bundle/config -> round open -> miner retry loop -> hold the host lease to the cutoff. - --dry-run is read-only (no provider call): checks the inputs, the 8x RTX 5090 / 170 SM profile, Vast base_url, master health, the miner proxy, refuses a run already frozen on cpu, and builds the miner bundle (master code + this branch's transport fixes) to prove the patch applies. - Each stage writes /.done and is skipped on re-run; every failure prints `FAIL : ` with the log path. - Miner host is checked for 8x RTX 5090, an allowlisted driver and torch 2.14.0+cu130; loop.sh exports HT_IMAGE_DIGEST and retries run-v2 until d_assign, bounded by the cutoff. --- scripts/launch_cuda_run.sh | 316 ++++++++++++++++++++++++++++++++++ tests/test_launch_cuda_run.py | 30 ++++ 2 files changed, 346 insertions(+) create mode 100755 scripts/launch_cuda_run.sh create mode 100644 tests/test_launch_cuda_run.py diff --git a/scripts/launch_cuda_run.sh b/scripts/launch_cuda_run.sh new file mode 100755 index 0000000..734a632 --- /dev/null +++ b/scripts/launch_cuda_run.sh @@ -0,0 +1,316 @@ +#!/usr/bin/env bash +# One-shot CUDA run launch on 8x RTX 5090 (Vast): admit -> deps/seed staging -> CUDA qualification +# -> rescue -> roots -> backend bootstrap -> POST /v2/admin/runs (backend cuda) -> dispatch.json +# -> master reload -> auditors -> relays -> miner host -> round open -> miner loop -> hold. +# +# scripts/launch_cuda_run.sh --dry-run # preflight + miner bundle build; no provider call +# TAG=v1 MINER_SSH="ssh5.vast.ai 12345" setsid nohup scripts/launch_cuda_run.sh --yes-spend \ +# > /data/role-dispatch/launch-v1.log 2>&1 < /dev/null & +# +# Idempotent: each stage writes $STATE/.done and is skipped on re-run; failures print +# "FAIL : " and exit non-zero. The network-v2 supervisor binds the rented hosts to +# THIS process and deletes them when it exits, so admit..rescue cannot resume under a new parent: +# a crash there needs a fresh TAG. After launch the process holds until the cutoff so the hosts +# stay up for master-side CUDA audit replay. +set -euo pipefail +umask 077 + +TAG=${TAG:-v1} +RD=${RD:-/data/role-dispatch} +T=${TOOLS:-$RD/tools} +E2E=${E2E:-/root/e2e-deploy-wt/hypertrain-e2e-b/.e2e} +OUT=${OUT:-$RD/manifest/out-$TAG} +JOB=${JOB:-$RD/manifest/job-$TAG} +PREP=${PREP:-$RD/prep-$TAG} +RUN=${RUN:-$RD/run-$TAG} +STATE=${STATE:-$RD/launch-$TAG} +MASTER=${MASTER:-cortex-challenge-hypertrain} +PY=${PY:-$RD/runtime/venv/bin/python} +MINER_SRC=${MINER_SRC:-/data/pydeps/hypertrain} +REPO=${REPO:-$(cd "$(dirname "$0")/.." && pwd)} +FIX_BASE=${FIX_BASE:-1a8f91e} +MINER_SSH=${MINER_SSH:-} +MINER_KEY=${MINER_KEY:-$RD/vast-m5/id_ed25519} +HOTKEY_SEED=${HOTKEY_SEED:-$RD/keys/uid247-hotkey.seed} +COORD_KEY=${COORD_KEY:-/root/.config/cortex-secrets/s100-master/challenges/hypertrain/coord.key} +PROXY_TOKEN=${PROXY_TOKEN:-/root/.config/cortex-secrets/od-large-e2e-miners/proxy.token} +PUBLIC=${PUBLIC:-185.8.107.43} +W=${W:-0} +# Tighter offsets fail with "no future replay budget": opening builds the start state on CPU. +OFFSETS=${OFFSETS:-150,300,900,1000,1020,1100} +GENESIS=1692803367 +PINS="fastapi==0.142.2 uvicorn==0.54.0 httpx==0.28.1 pydantic==2.13.5 py-sr25519-bindings==0.2.4 \ +numpy==2.5.3 safetensors==0.8.0 py-ecc==8.0.0 pyarrow==25.0.1 tokenizers==0.23.2" + +MODE=${1:-} +case "$MODE" in --dry-run | --yes-spend) ;; *) echo "usage: $0 --dry-run | --yes-spend" >&2; exit 2 ;; esac + +log() { echo "$(date -u +%FT%TZ) $*"; } +die() { log "FAIL $STAGE: $*"; exit 1; } +done_() { : > "$STATE/$STAGE.done"; log "STAGE_OK $STAGE"; } +stage() { + STAGE=$1 + if [ -e "$STATE/$STAGE.done" ]; then log "STAGE_SKIP $STAGE"; return 1; fi + log "STAGE $STAGE" +} +as_runtime() { setpriv --reuid=65532 --regid=65532 --clear-groups "$@"; } +J() { "$PY" -c "import json,sys;d=json.load(open(sys.argv[1]));print(eval(sys.argv[2],{'d':d}))" "$@"; } +beacon_round() { echo $((($(date +%s) - GENESIS) / 3 + 1)); } +mexec() { docker exec -i "$MASTER" "$@"; } +healthy() { mexec python -c 'import httpx;assert httpx.get("http://127.0.0.1:8000/health").status_code==200' 2> /dev/null; } +rescued() { "$PY" -c "import json,sys;sys.exit(0 if any(r.get('kind')=='network_rescued' and r.get('role')==sys.argv[2] for r in map(json.loads,open(sys.argv[1]))) else 1)" "$RUN/journal.jsonl" "$1"; } +mssh() { + ssh -i "$MINER_KEY" -o IdentitiesOnly=yes -o BatchMode=yes -o ConnectTimeout=20 \ + -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile="$STATE/miner_known_hosts" \ + -p "${MINER_SSH#* }" "root@${MINER_SSH%% *}" "$@" +} + +export PYTHONHOME=$RD/runtime/local CUDA_VISIBLE_DEVICES='' HOME=$RD +export PYTHONPATH=/data/hypertrain-tree/src:/data/hypertrain-tree/scripts:/data/hypertrain-tree:/data/pydeps:$RD/runtime/venv/lib/python3.12/site-packages +mkdir -p "$STATE" + +STAGE=preflight +for f in "$T"/{prepare_record.py,task8-deps.py,rescue_parallel.py,task8-materialize-roots.py} \ + "$T"/{backend_bootstrap.py,live_operator.py,auditors.sh} \ + "$E2E/beacon/pusher.py" "$E2E/relays/build_relay_configs.py" "$E2E/relays/deploy_relay.sh" \ + "$OUT/manifest-envelope.json" "$OUT/manifest.json" "$OUT/profile.json" "$JOB/job.json" \ + "$PREP/base/private/review-base.json" "$PREP/base/private/lifecycle-config.json" \ + "$PREP/context/join.json" "$PREP/context/seed_context.json" "$PREP/evidence/evidence.json" \ + "$RD/keys/vast.key" "$RD/keys/owner.seed" "$HOTKEY_SEED" "$PROXY_TOKEN" "$COORD_KEY" "$RD/dispatch.json"; do + [ -e "$f" ] || die "missing $f" +done +RID=$(J "$OUT/manifest-envelope.json" "d['run_id']") +[ "$(J "$JOB/job.json" "d['run_id']")" = "$RID" ] || die "job-$TAG binds another run than out-$TAG" +[ "$(J "$OUT/profile.json" "(d['admission']['gpu_name'],d['admission']['gpus_per_host'],d['admission']['sm_count'])")" \ + = "('RTX 5090', 8, 170)" ] || die "profile is not 8x RTX 5090 / 170 SM" +[ "$(J "$OUT/manifest.json" "d['training']['reference_spec']['layout']['n_gpus']")" = 8 ] \ + || die "manifest layout is not 8 GPUs" +IMAGE=$(J "$OUT/manifest.json" "d['training']['reference_spec']['image_digest']") +ALLOW=" $(J "$OUT/manifest.json" "' '.join(d['training']['reference_spec']['driver_allowlist'])") " +BASE_URL=$(J "$PREP/base/private/lifecycle-config.json" "d['base_url']") +HOTKEY=$(as_runtime "$PY" -c 'import pathlib,sys;from hypertrain.miner.core import load_keyfile;print(load_keyfile(pathlib.Path(sys.argv[1])).ss58)' "$HOTKEY_SEED") \ + || die "cannot load miner hotkey seed $HOTKEY_SEED" +[ "$BASE_URL" = https://console.vast.ai ] || die "lifecycle base_url is $BASE_URL, expected Vast" +JD=$(J "$JOB/job.json" "d['deadline']") +healthy || die "master $MASTER unhealthy" +ss -ltn | grep -q ':18990 ' || die "miner proxy :18990 not listening" +rc=0 +mexec python - "$RID" << 'EOF' || rc=$? +import pathlib, sqlite3, sys +p = pathlib.Path("/data/runs") / sys.argv[1] / "challenge.db" +row = p.exists() and sqlite3.connect(f"file:{p}?mode=ro", uri=True).execute( + "SELECT data FROM records_v2 WHERE kind='execution-backend' AND id='run'").fetchone() +sys.exit(0 if not row else 3 if '"cuda"' in row[0] else 4) +EOF +case $rc in 0) BACKEND=unregistered ;; 3) BACKEND=cuda ;; 4) die "run ${RID:0:8} is frozen on cpu: use a new TAG" ;; *) die "backend probe rc=$rc" ;; esac + +# Miner code = what the master runs + this repo's transport fixes (Content-Length PUT, 5xx retry, +# parallel part fetch); built locally so --dry-run proves the patch applies. +BUNDLE=$STATE/bundle +rm -rf "$BUNDLE" && mkdir -p "$BUNDLE" && cp -a "$MINER_SRC" "$BUNDLE/hypertrain" +find "$BUNDLE" -name __pycache__ -prune -exec rm -rf {} + +if ! grep -q '_retrying("PUT object' "$BUNDLE/hypertrain/miner/core.py"; then + git -C "$REPO" diff "$FIX_BASE" HEAD -- src/hypertrain/miner/core.py src/hypertrain/data/store.py \ + | patch -p3 --forward --quiet -d "$BUNDLE/hypertrain" || die "transport fixes do not apply onto $MINER_SRC" +fi +grep -q '_retrying("PUT object' "$BUNDLE/hypertrain/miner/core.py" || die "bundle lacks the PUT retry fix" +grep -q 'def _parts' "$BUNDLE/hypertrain/data/store.py" || die "bundle lacks the parallel part fetch" +log "PREFLIGHT_OK run=$RID image=$IMAGE job_deadline=$JD backend=$BACKEND miner=${MINER_SSH:-unset}" +if [ "$MODE" = --dry-run ]; then log "DRY_RUN_DONE no provider call, nothing spent"; exit 0; fi +[ -n "$MINER_SSH" ] || die 'MINER_SSH="host port" of the rented 8x5090 miner host is required' + +if [ ! -e "$STATE/window" ]; then + A=$(date +%s); CUT=$((A + 13200)); [ "$CUT" -le "$JD" ] || CUT=$JD + echo "$A $CUT" > "$STATE/window" +fi +read -r A CUT < "$STATE/window" + +STAGE=beacon +# The master clock only advances on POST /v1/admin/beacon; the pusher dies on ReadTimeout while +# the master is busy building start states, so it runs in a restart loop until the cutoff. +if ! { [ -s "$STATE/beacon.pid" ] && kill -0 "$(cat "$STATE/beacon.pid")" 2> /dev/null; }; then + setsid bash -c "while [ \$(date +%s) -lt $CUT ]; do docker exec -i $MASTER python - --cutoff $CUT \ + < '$E2E/beacon/pusher.py' >> '$STATE/beacon.log' 2>&1; sleep 3; done" < /dev/null > /dev/null 2>&1 & + echo $! > "$STATE/beacon.pid" +fi +log "STAGE_OK beacon pid=$(cat "$STATE/beacon.pid") cutoff=$CUT" + +CTX=(--file "join_h0=$PREP/context/join.json" --file "join_h1=$PREP/context/join.json" + --file "seed_context_h0=$PREP/context/seed_context.json" + --file "seed_context_h1=$PREP/context/seed_context.json" + --file "evidence=$PREP/evidence/evidence.json") +COMMON=(--base "$PREP/base/private/review-base.json" --parent-pid "$$" --owner-seed "$RD/keys/owner.seed" + --run-id "$RID" --provider-key-file "$RD/keys/vast.key" --run-dir "$RUN" + --source-message "Owner order 2026-10-14: 8x RTX 5090 Vast CUDA run, admission to mining authorized" + --live --cutoff-unix "$CUT" --admitted-unix "$A") +if stage admit; then + [ ! -e "$RUN/journal.jsonl" ] || die "$RUN already holds a transaction of another parent: use a new TAG" + [ $((A + 9600 + 1800)) -le "$CUT" ] || die "window too short A=$A CUT=$CUT (job deadline $JD)" + "$PY" "$T/prepare_record.py" "${COMMON[@]}" --action admit --beacon "$(beacon_round)" \ + --output "$STATE/review-admit.json" "${CTX[@]}" > "$STATE/admit.out" 2>&1 || die "see $STATE/admit.out" + done_ +fi +if stage deps; then + "$PY" "$T/task8-deps.py" "$RUN" "$PREP/base/private/review-base.json" > "$STATE/deps.out" 2>&1 \ + || die "see $STATE/deps.out" + done_ +fi +if stage qualify; then + "$PY" "$T/prepare_record.py" "${COMMON[@]}" --action qualification --beacon "$(beacon_round)" \ + --output "$STATE/review-qual.json" "${CTX[@]}" > "$STATE/qual.out" 2>&1 || die "see $STATE/qual.out" + done_ +fi +if stage rescue; then + pids=() + for r in h0 h1; do + rescued "$r" && continue + port=18560; [ "$r" = h1 ] && port=18570 + as_runtime "$PY" "$T/rescue_parallel.py" --run-dir "$RUN" --role "$r" --via range --workers 16 \ + --public-ip "$PUBLIC" --base-port "$port" > "$STATE/rescue-$r.out" 2>&1 & + pids+=($!) + done + for p in "${pids[@]}"; do wait "$p" || die "rescue failed, see $STATE/rescue-h*.out"; done + done_ +fi +if stage roots; then + as_runtime "$PY" "$T/task8-materialize-roots.py" "$RUN" "$PREP/base/private/review-base.json" h0 h1 \ + > "$STATE/materialize.out" 2>&1 || true + [ "$(grep -c '^MATERIALIZED ' "$STATE/materialize.out")" = 2 ] || die "see $STATE/materialize.out" + done_ +fi +if stage bootstrap; then + RES=() + for r in h0 h1; do RES+=("$(J "/data/hypertrain-$r/qualification.json" "d['output']")/result.json"); done + mexec python -B "$T/backend_bootstrap.py" --manifest-envelope "$OUT/manifest-envelope.json" \ + --owner-seed "$RD/keys/owner.seed" \ + --config /data/hypertrain-h0/qualification.json /data/hypertrain-h1/qualification.json \ + --result "${RES[@]}" > "$STATE/bootstrap.out" 2>&1 || die "see $STATE/bootstrap.out" + done_ +fi +if stage register; then + for p in "$OUT"/policy-*.json; do + mexec python -c 'import pathlib,sys;from hypertrain.data.store import LocalFSStore;LocalFSStore(pathlib.Path("/data/objects")).put(sys.stdin.buffer.read())' \ + < "$p" || die "policy object $p" + done + mexec python -c ' +import os, sys, httpx +tok = open(os.environ["CHALLENGE_ADMIN_TOKEN_FILE"]).read().strip() +r = httpx.post("http://127.0.0.1:8000/v2/admin/runs", content=sys.stdin.buffer.read(), timeout=300, + headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}) +print("POST_STATUS", r.status_code, r.text[:400]) +sys.exit(0 if r.status_code in (200, 201) or (r.status_code == 409 and "already exists" in r.text) else 1) +' < "$OUT/manifest-envelope.json" > "$STATE/register.out" 2>&1 || die "see $STATE/register.out" + mexec python -c ' +import sqlite3, sys +db = sqlite3.connect(f"file:/data/runs/{sys.argv[1]}/challenge.db?mode=ro", uri=True) +row = db.execute("SELECT data FROM records_v2 WHERE kind=\"execution-backend\" AND id=\"run\"").fetchone() +sys.exit(0 if row and "\"cuda\"" in row[0] else 1) +' "$RID" || die "registered run is not frozen on the CUDA backend" + done_ +fi +if stage dispatch; then + "$PY" - "$RD/dispatch.json" "$RUN" "$RID" "$HOTKEY" << 'EOF' || die "dispatch.json rewrite" +import json, os, sys +path, run, rid, hotkey = sys.argv[1:] +cfg = json.load(open(path)) +cfg.update(run_dir=run, run_id=rid, roles={hotkey: "h0"}) +tmp = path + ".tmp" +with open(tmp, "w") as f: + json.dump(cfg, f, separators=(",", ":")) +os.chown(tmp, 65532, 65532) +os.chmod(tmp, 0o600) +os.replace(tmp, path) +EOF + # create_app loads HYPERTRAIN_ROLE_DISPATCH once; the new binding needs a process restart. + docker restart "$MASTER" > /dev/null + for _ in $(seq 90); do healthy && break; sleep 2; done + healthy || die "master not healthy after reload" + done_ +fi +if stage auditors; then + { "$T/auditors.sh" stop && "$T/auditors.sh" start "$RID" && "$T/auditors.sh" check "$RID"; } \ + > "$STATE/auditors.out" 2>&1 || die "see $STATE/auditors.out" + done_ +fi +if stage relays; then + R=$STATE/relays + mkdir -p "$R/prep" + cp "$OUT/manifest.json" "$OUT/policy-relay_registry_hash.json" "$R/prep/" + if [ ! -e "$R/out/relay-de.json" ]; then + "$PY" "$E2E/relays/build_relay_configs.py" --run-id "$RID" --prep "$R/prep" --coord-key "$COORD_KEY" \ + --master-url https://gateway.cortex.foundation/challenge/hypertrain --out-dir "$R/out" \ + > "$R/build.out" 2>&1 || die "see $R/build.out" + fi + for pair in de:185.191.119.141 us:185.8.107.22 sg:88.216.191.197; do + region=${pair%%:*} ip=${pair#*:} cfg=$R/out/relay-$region.json + rs=(ssh -i /root/.ssh/ascii_box_ed25519 -o BatchMode=yes -o ConnectTimeout=15 -o IdentitiesOnly=yes "root@$ip") + if [ "$("${rs[@]}" sha256sum /etc/hypertrain/relay/relay.json | cut -c1-64)" = "$(sha256sum "$cfg" | cut -c1-64)" ] \ + && "${rs[@]}" docker ps -q --filter 'name=^hypertrain-relay$' | grep -q .; then continue; fi + "${rs[@]}" docker rm -f hypertrain-relay > /dev/null 2>&1 || true + bash "$E2E/relays/deploy_relay.sh" "$region" "$ip" "$cfg" >> "$R/deploy.out" 2>&1 \ + || die "relay $region, see $R/deploy.out" + done + done_ +fi +if stage miner; then + mssh "nvidia-smi --query-gpu=name,driver_version --format=csv,noheader" > "$STATE/miner-gpus.txt" \ + || die "miner ssh to $MINER_SSH" + [ "$(grep -c 'RTX 5090' "$STATE/miner-gpus.txt")" = 8 ] \ + || die "miner host is not 8x RTX 5090: $(sort -u "$STATE/miner-gpus.txt" | tr '\n' ';')" + DRV=$(head -1 "$STATE/miner-gpus.txt" | awk -F', ' '{print $2}') + case "$ALLOW" in *" $DRV "*) ;; *) die "miner driver $DRV not in allowlist$ALLOW" ;; esac + tar -C "$BUNDLE" -czf - hypertrain \ + | mssh 'rm -rf /opt/ht && mkdir -p /opt/ht /opt/miner/work && tar -C /opt/ht -xzf -' || die "bundle upload" + mssh 'install -m 0600 /dev/stdin /opt/miner/hotkey.seed' < "$HOTKEY_SEED" || die "hotkey upload" + API="http://$PUBLIC:18990/$(cat "$PROXY_TOKEN")" + mssh 'cat > /opt/miner/miner.toml' << EOF || die "miner.toml upload" +api = "$API" +keyfile = "/opt/miner/hotkey.seed" +workdir = "/opt/miner/work" +state_source = "$API/v2/runs/$RID/objects" +image_digest = "$IMAGE" +run_id = "$RID" +owner_hotkey = "$(J "$OUT/manifest-envelope.json" "d['signer']")" +device = "cuda" +EOF + # island_worker refuses to train unless HT_IMAGE_DIGEST names the manifest image; run-v2 exits + # "authenticated assignment not ready" until d_assign, hence the retry loop up to the cutoff. + mssh 'cat > /opt/miner/loop.sh' << EOF || die "loop.sh upload" +cd /opt/ht +export PYTHONPATH=/opt/ht CUBLAS_WORKSPACE_CONFIG=:4096:8 CUDA_DISABLE_PTX_JIT=1 HT_IMAGE_DIGEST=$IMAGE +rc=1 +while [ \$(date +%s) -lt $CUT ]; do + /venv/main/bin/python -m hypertrain.miner.cli run-v2 --config /opt/miner/miner.toml --round $W >> /opt/miner/run$W.log 2>&1; rc=\$? + echo "LOOP rc=\$rc \$(date -u +%T)" >> /opt/miner/run$W.log + [ \$rc -eq 0 ] && break + sleep 10 +done +echo "LOOP_END rc=\$rc" >> /opt/miner/run$W.log +EOF + mssh "cd /opt/ht && { PYTHONPATH=/opt/ht /venv/main/bin/python -c 'import hypertrain.miner.cli' 2> /dev/null \ + || /venv/main/bin/python -m pip install -q $PINS; }" > "$STATE/miner-pip.out" 2>&1 || die "see $STATE/miner-pip.out" + mssh "cd /opt/ht && PYTHONPATH=/opt/ht /venv/main/bin/python -c 'import torch, hypertrain.miner.cli; \ +assert torch.__version__ == \"2.14.0+cu130\" and torch.cuda.device_count() == 8, torch.__version__'" \ + || die "miner host needs torch 2.14.0+cu130 in /venv/main with 8 visible GPUs" + done_ +fi +if stage open; then + if ! mexec python -c 'import httpx,sys;sys.exit(httpx.get(f"http://127.0.0.1:8000/v2/runs/{sys.argv[1]}/rounds/{sys.argv[2]}").status_code!=200)' \ + "$RID" "$W"; then + mexec sh -c "python -B $T/live_operator.py --master http://127.0.0.1:8000 --run-id $RID \ + --state-dir /data/runs/$RID --coord-key \$HYPERTRAIN_COORD_KEY_FILE \ + --admin-token-file \$CHALLENGE_ADMIN_TOKEN_FILE open --w $W --hotkey $HOTKEY --offsets $OFFSETS" \ + > "$STATE/open.out" 2>&1 || die "see $STATE/open.out" + fi + done_ +fi +if stage minerstart; then + mssh 'pgrep -f /opt/miner/loop.sh > /dev/null || { setsid nohup bash /opt/miner/loop.sh > /dev/null 2>&1 < /dev/null & }' \ + || die "miner loop start" + done_ +fi + +log "LAUNCHED run=$RID round=$W miner=$MINER_SSH; holding the host lease until cutoff $CUT" +n=$(date +%s) +if [ "$CUT" -gt "$n" ]; then sleep $((CUT - n)); fi +log PARENT_RELEASED diff --git a/tests/test_launch_cuda_run.py b/tests/test_launch_cuda_run.py new file mode 100644 index 0000000..57a3f2e --- /dev/null +++ b/tests/test_launch_cuda_run.py @@ -0,0 +1,30 @@ +"""launch_cuda_run.sh refuses without an explicit mode and fails closed in preflight.""" + +import os +import shutil +import subprocess +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).parents[1] / "scripts/launch_cuda_run.sh" + + +def _run(tmp_path: Path, *args: str) -> subprocess.CompletedProcess[str]: + env = dict(os.environ, RD=str(tmp_path), STATE=str(tmp_path / "state"), PY="python3") + return subprocess.run( + ["bash", str(SCRIPT), *args], capture_output=True, text=True, env=env, timeout=60 + ) + + +def test_requires_explicit_mode(tmp_path: Path) -> None: + result = _run(tmp_path) + assert result.returncode == 2 and "--dry-run | --yes-spend" in result.stderr + + +@pytest.mark.skipif(shutil.which("setpriv") is None, reason="util-linux") +def test_preflight_names_the_first_missing_input_and_spends_nothing(tmp_path: Path) -> None: + result = _run(tmp_path, "--yes-spend") + assert result.returncode == 1 + assert "FAIL preflight: missing " + str(tmp_path / "tools/prepare_record.py") in result.stdout + assert not (tmp_path / "state" / "window").exists() From 9c49a89fd8a8cbb5a2c9962e08e723ca2c693f17 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:17:54 +0000 Subject: [PATCH 3/3] fix(launch): start the miner loop (pgrep self-match), keep proxy token out of retry logs pgrep -f /opt/miner/loop.sh matched the remote bash -c running it, so the loop never started while the stage reported success; use [/]opt pattern and verify the loop runs. Retry warnings log the error type and status, not the URL (which carries the proxy token). The parallel-fetch test proves concurrency with an event instead of random sleeps. --- scripts/launch_cuda_run.sh | 4 +++- src/hypertrain/miner/core.py | 5 ++++- tests/data/test_parts.py | 20 +++++++++++--------- tests/test_launch_cuda_run.py | 4 ---- 4 files changed, 18 insertions(+), 15 deletions(-) diff --git a/scripts/launch_cuda_run.sh b/scripts/launch_cuda_run.sh index 734a632..f7d126b 100755 --- a/scripts/launch_cuda_run.sh +++ b/scripts/launch_cuda_run.sh @@ -305,8 +305,10 @@ if stage open; then done_ fi if stage minerstart; then - mssh 'pgrep -f /opt/miner/loop.sh > /dev/null || { setsid nohup bash /opt/miner/loop.sh > /dev/null 2>&1 < /dev/null & }' \ + # [/] keeps pgrep from matching the remote shell whose own argv contains the pattern. + mssh "pgrep -f '[/]opt/miner/loop.sh' > /dev/null || { setsid nohup bash /opt/miner/loop.sh > /dev/null 2>&1 < /dev/null & }" \ || die "miner loop start" + mssh "sleep 2; pgrep -f '[/]opt/miner/loop.sh' > /dev/null" || die "miner loop is not running on $MINER_SSH" done_ fi diff --git a/src/hypertrain/miner/core.py b/src/hypertrain/miner/core.py index 2ce57f6..ae3afa2 100644 --- a/src/hypertrain/miner/core.py +++ b/src/hypertrain/miner/core.py @@ -65,7 +65,10 @@ def _retrying[T](what: str, fn: Callable[[], T]) -> T: status = getattr(error, "status", status) if delay is None or (status is not None and status < 500): raise - log.warning("%s failed (%s); retry in %ss", what, error, delay) + # Type + status only: request URLs can carry the proxy token. + log.warning( + "%s failed (%s %s); retry in %ss", what, type(error).__name__, status, delay + ) time.sleep(delay) raise AssertionError("unreachable") diff --git a/tests/data/test_parts.py b/tests/data/test_parts.py index b584974..0f7fec0 100644 --- a/tests/data/test_parts.py +++ b/tests/data/test_parts.py @@ -216,26 +216,28 @@ def chunks() -> bytes: def test_parallel_fetch_writes_parts_in_order_despite_out_of_order_completion( tmp_path: Path, ) -> None: - import random import threading - import time - # Given: a 9-part object whose GETs finish in random order on several threads. + # Given: a 9-part object; part 0 is only served once part 1 has been fetched, so a serial + # fetch would deadlock (bounded by the timeout) and in-order output needs real reordering. rec, data, index = _setup(tmp_path, 8 * PART_BYTES + 77) ref = hashlib.sha256(rec.puts[-1]).hexdigest() - threads: set[int] = set() + second_done = threading.Event() def get(sha: str) -> bytes: - threads.add(threading.get_ident()) - time.sleep(random.random() / 200) - return rec.store.get(sha) + if sha == index.parts[0]: + assert second_done.wait(timeout=10), "parts were not fetched concurrently" + body = rec.store.get(sha) + if sha == index.parts[1]: + second_done.set() + return body out = tmp_path / "out" out.mkdir() # When: fetched with 4 workers. fetch_to(ref, get, out / "dest", workers=4) - # Then: byte-identical reassembly, really concurrent. - assert (out / "dest").read_bytes() == data and len(threads) > 1 + # Then: byte-identical reassembly. + assert (out / "dest").read_bytes() == data def test_parallel_fetch_failing_part_leaves_no_temp_and_stops_queueing(tmp_path: Path) -> None: diff --git a/tests/test_launch_cuda_run.py b/tests/test_launch_cuda_run.py index 57a3f2e..fd65461 100644 --- a/tests/test_launch_cuda_run.py +++ b/tests/test_launch_cuda_run.py @@ -1,12 +1,9 @@ """launch_cuda_run.sh refuses without an explicit mode and fails closed in preflight.""" import os -import shutil import subprocess from pathlib import Path -import pytest - SCRIPT = Path(__file__).parents[1] / "scripts/launch_cuda_run.sh" @@ -22,7 +19,6 @@ def test_requires_explicit_mode(tmp_path: Path) -> None: assert result.returncode == 2 and "--dry-run | --yes-spend" in result.stderr -@pytest.mark.skipif(shutil.which("setpriv") is None, reason="util-linux") def test_preflight_names_the_first_missing_input_and_spends_nothing(tmp_path: Path) -> None: result = _run(tmp_path, "--yes-spend") assert result.returncode == 1