diff --git a/docs/network-v2.md b/docs/network-v2.md index 1a4b7b8..00f6356 100644 --- a/docs/network-v2.md +++ b/docs/network-v2.md @@ -192,7 +192,10 @@ uv run --frozen hypertrain-miner run-v2 --config miner.toml --round 0 This follows accept, authenticated inputs, exact-N launcher, one island commit, original artifacts, regional custody, master acceptance and delta submission. Tokens derive from assigned samples times sequence length, not self-reported GPU -count. `UPLOADED` is not audit success, settlement or payment. Unlike v1 `run`, +count. Rerunning a round that is already `UPLOADED` exits 0 with `UPLOADED`: the +master returns its first acceptance for the same receipt without a new relay +retrieval, and answers 409 `RELAY_RECEIPT_CONFLICT` for a different receipt. +`UPLOADED` is not audit success, settlement or payment. Unlike v1 `run`, `run-v2` has no `--rounds`/`--start` option. Preserve the work directory; do not claim the CLI is a complete unattended production scheduler. diff --git a/scripts/mine_multi_round_v2.py b/scripts/mine_multi_round_v2.py index e31d7c0..671ed60 100644 --- a/scripts/mine_multi_round_v2.py +++ b/scripts/mine_multi_round_v2.py @@ -282,10 +282,11 @@ def count(kind: str, identity: str | None = None) -> int: "relay_receipts_unchanged": count("relay-receipt") == receipts, } if not ( - probe["after_upload"]["delta_unchanged"] + probe["after_upload"]["rc"] == 0 + and probe["after_upload"]["delta_unchanged"] and probe["after_upload"]["relay_receipts_unchanged"] ): - raise RuntimeError(f"rerun after UPLOADED changed the accepted delta: {probe}") + raise RuntimeError(f"rerun after UPLOADED not a clean no-op: {probe}") w = open_round(svc, n, ROUNDS + 1, settle(svc, n, w)).w for i, conf in enumerate(configs[1:], 1): @@ -312,6 +313,8 @@ def count(kind: str, identity: str | None = None) -> int: probe["kill_mid_round"] = { "killed_after_job_staged": alive and bool(count("island-job", f"{w}:{hot}")) } + if not probe["kill_mid_round"]["killed_after_job_staged"]: + raise RuntimeError("kill probe did not interrupt a staged job") try: run_v2(stages, configs[0], w, 0) probe["kill_mid_round"]["resume"] = "UPLOADED" diff --git a/src/hypertrain/challenge/store.py b/src/hypertrain/challenge/store.py index b725ed6..2b84ca5 100644 --- a/src/hypertrain/challenge/store.py +++ b/src/hypertrain/challenge/store.py @@ -9929,6 +9929,25 @@ def upload_grant_v2( self._put_record_v2(run_id, "chunk-manifest", grant.digest(), chunk.body()) return signed + def _accepted_receipt_v2( + self, run_id: str, w: int, hotkey: str, receipt_hash: str + ) -> dict[str, Any] | None: + """Stored acceptance of receipt_hash for (w, hotkey); 409 if another was accepted.""" + accepted: list[dict[str, Any]] = [ + json.loads(stored) + for (stored,) in self._db.execute( + "SELECT data FROM records_v2 WHERE run_id=? AND kind='master-acceptance' " + "AND json_extract(data,'$.body.w')=? AND json_extract(data,'$.body.hotkey')=?", + (run_id, w, hotkey), + ) + ] + for stored in accepted: + if stored["body"]["receipt_hash"] == receipt_hash: + return stored + if accepted: + raise ChallengeError(409, "RELAY_RECEIPT_CONFLICT") + return None + async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]: import base64 @@ -9959,6 +9978,12 @@ async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]: grant=grant, received_before=deadline, ) + # A rerun must not challenge the relay again: it refuses a second live + # retrieval of the same object, and the first acceptance is authoritative. + if ( + stored := self._accepted_receipt_v2(run_id, grant.w, grant.hotkey, receipt.digest()) + ) is not None: + return stored assignment = self._record_v2(run_id, "relay-assignment", f"{grant.w}:{grant.hotkey}")[ "body" ] @@ -9989,18 +10014,30 @@ async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]: capability = base64.b64encode( canonicalize(self._record_v2(run_id, "grant", grant.digest())) ).decode() - custody_response = await client.get( - spec.https_url + f"/v1/uploads/{grant.digest()}/acks", - headers={"X-Upload-Grant": capability}, - ) - custody_response.raise_for_status() - custody = custody_response.json() - relay = RelayClient(registry, client) - with spool() as payload: - response = await relay.retrieval(request_raw, payload, relay_public=env.signer) - payload.seek(0) - original = payload.read() + import httpx + + try: + custody_response = await client.get( + spec.https_url + f"/v1/uploads/{grant.digest()}/acks", + headers={"X-Upload-Grant": capability}, + ) + custody_response.raise_for_status() + custody = custody_response.json() + relay = RelayClient(registry, client) + with spool() as payload: + response = await relay.retrieval(request_raw, payload, relay_public=env.signer) + payload.seek(0) + original = payload.read() + except httpx.HTTPStatusError as exc: + raise ChallengeError( + 502, f"relay refused master retrieval: HTTP {exc.response.status_code}" + ) from exc with self._tx(): + # A concurrent submission may have been accepted while this one awaited the relay. + if ( + stored := self._accepted_receipt_v2(run_id, grant.w, grant.hotkey, receipt.digest()) + ) is not None: + return stored current = self._fresh_v2( manifest, f"relay-accept:{receipt.digest()}:{self._now(self._db)}" ) diff --git a/tests/challenge/test_capacity_outer_v2.py b/tests/challenge/test_capacity_outer_v2.py index 6c4bc7f..4e7a604 100644 --- a/tests/challenge/test_capacity_outer_v2.py +++ b/tests/challenge/test_capacity_outer_v2.py @@ -936,6 +936,8 @@ def test_prepared_profile_opening_metadata_without_tensor_calls( tiny, tmp_path, monkeypatch, attack, metadata_genesis ): """Signed hermetic metadata/state oracle; opening never gains kernel approval.""" + # Store imports the worker lazily; a first import under the trap would bind forbidden. + import hypertrain.auditor.worker # noqa: F401 from hypertrain.aggregator.core import OuterState from hypertrain.auditor.replay import AnchorCache from hypertrain.challenge.store import ChallengeError, ChallengeStore diff --git a/tests/challenge/test_relay_receipt_idempotent.py b/tests/challenge/test_relay_receipt_idempotent.py new file mode 100644 index 0000000..4c3cddc --- /dev/null +++ b/tests/challenge/test_relay_receipt_idempotent.py @@ -0,0 +1,332 @@ +"""A miner rerunning a finished round resubmits its relay receipt: clean answer, never a 500.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import anyio +import httpx +import pytest + +from hypertrain.challenge.store import ChallengeError +from hypertrain.data.store import LocalFSStore +from hypertrain.data.stream_store import StreamStore +from hypertrain.protocol import envelope_v2, relay_envelope +from hypertrain.protocol.hashing import sha256_hex +from hypertrain.protocol.jcs import canonicalize +from hypertrain.protocol.relay_messages import ( + MasterAcceptanceV2, + RelayAssignment, + RelayNetworkManifest, + RelayReceipt, + RetrievalRequest, + UploadChunk, + UploadChunkManifest, + UploadGrant, +) +from hypertrain.relay.app import create_app as relay_app +from hypertrain.relay.core import Relay + +spec = importlib.util.spec_from_file_location( + "relay_receipt_idempotent_fixture", Path(__file__).parent / "test_service_network_v2.py" +) +assert spec is not None and spec.loader is not None +fixtures = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = fixtures +spec.loader.exec_module(fixtures) +network = fixtures.network +COORD, RELAY, HOT = fixtures.COORD, fixtures.RELAY, fixtures.HOT[0].ss58 +PAYLOAD = b"rerun delta" + + +@pytest.fixture +def uploaded(network, monkeypatch, tmp_path): + """Real master app and real relay app; the delta is uploaded and its receipt signed.""" + run_id = network.manifest.run_id() + registry = network.store._registry_v2(network.manifest) + net = RelayNetworkManifest( + run_id=run_id, + base_manifest_hash=network.manifest.training.training_hash(), + registry_hash=registry.digest(), + specs_hash=sha256_hex(canonicalize([s.body() for s in registry.specs])), + assignment_policy="master-observed-median3-v1", + ) + chunks = UploadChunkManifest( + size=len(PAYLOAD), + chunks=[UploadChunk(index=0, off=0, len=len(PAYLOAD), chunk_sha256=sha256_hex(PAYLOAD))], + ) + assignment = RelayAssignment( + w=0, + hotkey=HOT, + primary_id="local", + fallback_ids=[], + assignment_epoch=registry.epoch, + manifest_hash=net.digest(), + exp_drand=network.now + 20, + ) + grant = UploadGrant( + w=0, + hotkey=HOT, + relay_id="local", + assignment_epoch=registry.epoch, + delta_hash=sha256_hex(PAYLOAD), + size=len(PAYLOAD), + chunk_manifest_hash=chunks.chunk_manifest_hash(), + retain_until=network.now + 200, + nonce="ad" * 32, + exp_drand=assignment.exp_drand, + ) + signed_grant = relay_envelope.seal(COORD, "UploadGrant", run_id, grant, grant.exp_drand) + signed_assignment = relay_envelope.seal( + COORD, "RelayAssignment", run_id, assignment, assignment.exp_drand + ) + opened = { + "w": 0, + **{ + k: "12" * 32 + for k in ( + "prev_final_hash", + "theta_hash", + "outer_state_hash", + "center_hash", + "roster_hash", + "honeypot_commit", + "start_state_index_hash", + ) + }, + "d_open": 1, + "d_assign": 2, + "d_commit": network.now + 10, + "d_audit": network.now + 11, + "d_upload": grant.exp_drand, + "d_final": network.now + 30, + "contract_version": 2, + "policy_hashes": { + k: getattr(network.manifest.network, k) + for k in ( + "admission_policy_hash", + "economics_policy_hash", + "aggregation_policy_hash", + "dispute_policy_hash", + "audit_policy_hash", + ) + }, + "registry_epoch": registry.epoch, + "audit_mode": "anchored-full", + "roster": [], + } + # Accepted master journal boundary with original signatures; no training round. + with network.store._tx(): + network.store._put_record_v2(run_id, "grant", grant.digest(), signed_grant) + network.store._put_record_v2(run_id, "chunk-manifest", grant.digest(), chunks.body()) + network.store._put_record_v2(run_id, "relay-assignment", "0:" + HOT, signed_assignment) + network.store._put_record_v2(run_id, "round", "0", {"body": opened}) + real_client = httpx.AsyncClient + backing = real_client() + + async def now(): + return network.now + + relay = Relay( + run_id=run_id, + master=COORD.ss58, + registry=registry, + network_manifest_hash=net.digest(), + observers={a.ss58 for a in fixtures.AUDITORS}, + relay_id="local", + region="local", + keys={"k1": RELAY}, + active_key="k1", + backing=StreamStore(LocalFSStore(tmp_path / "relay-backing"), backing), + now=now, + ) + + async def upload(): + await relay.accept( + relay_envelope.parse_envelope(signed_grant), + chunks, + relay_envelope.parse_envelope(signed_assignment), + envelope_v2.parse_envelope(network.signed(COORD, "RoundOpenV2", opened)), + ) + + async def source(): + yield PAYLOAD + + await relay.chunk(grant.digest(), 0, source()) + return await relay.complete(grant.digest()) + + receipt = anyio.run(upload) + app = relay_app(relay, "x" * 32) + monkeypatch.setattr( + httpx, + "AsyncClient", + lambda *a, **kw: real_client(*a, **{**kw, "transport": httpx.ASGITransport(app=app)}), + ) + yield relay, grant, receipt + anyio.run(backing.aclose) + + +def submit(network, receipt): + return network.client.post( + "/v2/relay-receipts", content=canonicalize(receipt.model_dump(mode="json")) + ) + + +def count(network, kind): + return network.store._db.execute( + "SELECT COUNT(*) FROM records_v2 WHERE kind=?", (kind,) + ).fetchone()[0] + + +def test_second_relay_receipt_for_same_key_is_idempotent(network, uploaded): + _, grant, receipt = uploaded + first = submit(network, receipt) + assert first.status_code == 200, first.text + assert first.json()["body"]["delta_hash"] == grant.delta_hash + second = submit(network, receipt) + assert second.status_code == 200, second.text + assert second.json() == first.json() + # The first acceptance stays authoritative: no second retrieval or acceptance. + assert count(network, "master-acceptance") == 1 + assert count(network, "retrieval") == 1 + + +def test_relay_receipt_different_payload_same_key_is_409(network, uploaded): + _, grant, receipt = uploaded + first = submit(network, receipt) + assert first.status_code == 200, first.text + # A genuinely relay-signed receipt for the same grant, but a different body. + other = RelayReceipt.model_validate(receipt.body).model_copy( + update={"received_drand": RelayReceipt.model_validate(receipt.body).received_drand + 1} + ) + resigned = relay_envelope.parse_envelope( + relay_envelope.seal(RELAY, "RelayReceipt", receipt.run_id, other, grant.retain_until) + ) + assert other.digest() != RelayReceipt.model_validate(receipt.body).digest() + refused = submit(network, resigned) + assert refused.status_code == 409, refused.text + assert refused.json()["detail"] == "RELAY_RECEIPT_CONFLICT" + assert count(network, "master-acceptance") == 1 + assert count(network, "retrieval") == 1 + + +def test_relay_400_on_retrieval_is_a_clean_error_not_500(network, uploaded): + relay, grant, receipt = uploaded + run_id = network.manifest.run_id() + # Another live challenge on the same object makes the real relay answer 400. + request = RetrievalRequest( + request_id="ab" * 32, + nonce="ac" * 32, + relay_id="local", + key_id="k1", + assignment_hash=sha256_hex( + canonicalize(network.store._record_v2(run_id, "relay-assignment", "0:" + HOT)["body"]) + ), + grant_hash=grant.digest(), + custody_ack_hashes=[], + receipt_hash=RelayReceipt.model_validate(receipt.body).digest(), + retention_hash=RelayReceipt.model_validate(receipt.body).digest(), + object_or_chunk_hash=grant.delta_hash, + size=grant.size, + requested_beacon=network.now, + deadline_beacon=network.now + 10, + ) + anyio.run( + relay.retrieve, + relay_envelope.parse_envelope( + relay_envelope.seal(COORD, "RetrievalRequest", run_id, request, network.now + 10) + ), + ) + refused = submit(network, receipt) + assert refused.status_code == 502, refused.text + assert "400" in refused.json()["detail"] + assert count(network, "master-acceptance") == 0 + + +def race(network, monkeypatch, receipt, finish_a): + """B passes the first transaction and the relay; A is accepted before B writes.""" + from hypertrain.relay.client import RelayClient + + real = RelayClient.retrieval + pending = [finish_a] + + async def retrieval(self, *args, **kwargs): + response = await real(self, *args, **kwargs) + if pending: + await pending.pop()() + return response + + monkeypatch.setattr(RelayClient, "retrieval", retrieval) + store = network.store.for_run(network.manifest.run_id()) + + async def b(): + async with httpx.AsyncClient() as client: + try: + return await store.relay_receipt_v2( + canonicalize(receipt.model_dump(mode="json")), client + ) + except ChallengeError as exc: + return exc + + outcome = anyio.run(b) + assert not pending + return outcome + + +def test_second_transaction_does_not_write_a_second_acceptance_same_receipt( + network, uploaded, monkeypatch +): + _, _, receipt = uploaded + store = network.store.for_run(network.manifest.run_id()) + start = store._now(store._db) + accepted = [] + + async def finish_a(): + # A arrives after B's relay challenge expired and is accepted one beacon before B writes. + network.push(start + 11) + async with httpx.AsyncClient() as client: + accepted.append( + await store.relay_receipt_v2(canonicalize(receipt.model_dump(mode="json")), client) + ) + network.push(start + 12) + + outcome = race(network, monkeypatch, receipt, finish_a) + assert count(network, "master-acceptance") == 1 + assert outcome == accepted[0] + + +def test_second_transaction_conflicts_with_acceptance_of_another_receipt( + network, uploaded, monkeypatch +): + _, grant, receipt = uploaded + run_id = network.manifest.run_id() + store = network.store.for_run(run_id) + other = RelayReceipt.model_validate(receipt.body).model_copy( + update={"received_drand": RelayReceipt.model_validate(receipt.body).received_drand + 1} + ) + + async def finish_a(): + # The real relay only serves its own receipt, so A's acceptance is written as A would. + acceptance = MasterAcceptanceV2( + w=grant.w, + hotkey=grant.hotkey, + delta_hash=grant.delta_hash, + receipt_hash=other.digest(), + received_drand=store._now(store._db), + ) + with store._tx(): + store._put_record_v2( + run_id, + "master-acceptance", + acceptance.digest(), + relay_envelope.seal( + COORD, "MasterAcceptanceV2", run_id, acceptance, grant.retain_until + ), + ) + + outcome = race(network, monkeypatch, receipt, finish_a) + assert count(network, "master-acceptance") == 1 + assert isinstance(outcome, ChallengeError) + assert (outcome.status, outcome.detail) == (409, "RELAY_RECEIPT_CONFLICT") diff --git a/tests/test_mine_multi_round_v2.py b/tests/test_mine_multi_round_v2.py new file mode 100644 index 0000000..bbfa4fe --- /dev/null +++ b/tests/test_mine_multi_round_v2.py @@ -0,0 +1,49 @@ +import importlib.util +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +spec = importlib.util.spec_from_file_location( + "mine_multi_round_v2", ROOT / "scripts" / "mine_multi_round_v2.py" +) +assert spec is not None and spec.loader is not None +mine_multi_round_v2 = importlib.util.module_from_spec(spec) +spec.loader.exec_module(mine_multi_round_v2) + + +def test_kill_probe_guard_fails_when_job_not_staged( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """run-v2 exits before its island job is staged: no kill happened, so no resume was tested.""" + m = mine_multi_round_v2 + reruns: list[int] = [] + monkeypatch.setattr(m, "open_round", lambda svc, n, w, prev: SimpleNamespace(w=w)) + monkeypatch.setattr(m, "settle", lambda svc, n, w: "final") + monkeypatch.setattr(m, "run_v2", lambda stages, conf, w, i: reruns.append(i)) + monkeypatch.setattr(m.smoke, "reap", lambda marker: []) + monkeypatch.setattr( + subprocess, "run", lambda *a, **kw: SimpleNamespace(returncode=0, stdout="", stderr="") + ) + monkeypatch.setattr( + subprocess, "Popen", lambda *a, **kw: SimpleNamespace(poll=lambda: 0, wait=lambda: 0) + ) + # No record of any kind: the island-job row never appears. + db = SimpleNamespace(execute=lambda sql, args: SimpleNamespace(fetchone=lambda: (0,))) + n: Any = SimpleNamespace( + manifest=SimpleNamespace(run_id=lambda: "run"), + store=SimpleNamespace(_db=db, _record_v2=lambda rid, kind, ident: {}), + ) + svc = SimpleNamespace(HOT=[SimpleNamespace(ss58="hot0")]) + probe: dict[str, Any] = {} + configs = [tmp_path / f"miner-{i}.toml" for i in range(4)] + with pytest.raises(RuntimeError, match="kill probe did not interrupt a staged job"): + m.restart_probes([], svc, n, configs, "prev", tmp_path, probe) + assert probe["kill_mid_round"]["killed_after_job_staged"] is False + # The rerun of miner 0 in the kill round never ran: it would have reported UPLOADED. + assert reruns == [0, 1, 2, 3, 1, 2, 3]