From 20de9313f3ed7dd94efcf576ef3b2341df473424 Mon Sep 17 00:00:00 2001 From: dylan Date: Tue, 15 Sep 2026 23:36:18 +0000 Subject: [PATCH] fix: journal resubmit burns before broadcast and recover safely --- docs/Discriminative-Mining.md | 6 +- gas/protocol/burn_journal.py | 190 +++++++++++++ gas/protocol/miner_requests.py | 14 + gas/protocol/resubmit_burn.py | 253 +++++++++++++---- neurons/discriminator/push_model.py | 20 +- tests/test_burn_journal.py | 418 ++++++++++++++++++++++++++++ tests/test_resubmit_burn.py | 43 ++- 7 files changed, 858 insertions(+), 86 deletions(-) create mode 100644 gas/protocol/burn_journal.py create mode 100644 tests/test_burn_journal.py diff --git a/docs/Discriminative-Mining.md b/docs/Discriminative-Mining.md index 3e23a32f..b9297938 100644 --- a/docs/Discriminative-Mining.md +++ b/docs/Discriminative-Mining.md @@ -92,7 +92,11 @@ Each registered hotkey gets **one free counted submission** (image, video, or au - A confirmed or superseded model **does** consume the free slot for the life of that registration, for every modality. - A new benchmark version does **not** refill the slot. - To submit another model from the **same** hotkey, run `gascli d push` again. The CLI walks you through burning **0.5 TAO of SN34 alpha**: `burn_alpha` if this hotkey already has enough α, otherwise one `add_stake_burn` of 0.5 TAO. Recycle does not count. There is no second counted model without that burn. -- The burn is spent only when the new model is confirmed. If upload or the entrance exam fails, the next `gascli d push` reuses the same burn. The CLI also keeps the last burn receipt locally so a crash after the extrinsic lands does not ask you to burn again. +- The on-chain burn is irreversible. Its **submission credit** is consumed only when the new model counts; unused credits are reused on upload/exam retries. +- Before confirmation, the CLI shows the exact alpha amount and quoted TAO value, including up to **2% price padding** for `burn_alpha` (about 0.51 TAO worth). `add_stake_burn` spends 0.5 TAO. Transaction fees are additional in both cases. +- Before broadcasting, the CLI atomically saves the signed transaction, hash, chain identity, amount, and intended submission in `$GAS_HOME/resubmit_burns` (default `~/.gas/resubmit_burns`). No private keys are stored. A local lock prevents overlapping pushes for the same hotkey/subnet using that state directory. +- After a crash or timeout, rerun the same command on the same machine with the same `GAS_HOME`. The CLI checks finalized chain history and reuses a successful burn. Malformed state, unavailable history, or an unresolved saved transaction blocks a fresh burn; a saved transaction not found even after expiry requires investigation, not automatic repayment. A confirmed on-chain failure permits a newly confirmed attempt on the next run. +- Preserve the journal and receipt. Deleting state, changing `GAS_HOME`, or using another machine bypasses local protection; this is not a cross-machine exactly-once guarantee. If recovery remains blocked, use the printed transaction hash and saved journal to investigate with the operator before retrying elsewhere. Historical burns made before journaling was introduced cannot be recovered automatically without their saved receipt. --- diff --git a/gas/protocol/burn_journal.py b/gas/protocol/burn_journal.py new file mode 100644 index 00000000..88b56585 --- /dev/null +++ b/gas/protocol/burn_journal.py @@ -0,0 +1,190 @@ +"""Durable local state and process/thread exclusion for irreversible burns. + +Locks deliberately cover all endpoints for a hotkey/netuid. The journal also +binds to the chain genesis hash so changing endpoints cannot replay stale proof. +""" + +from contextlib import contextmanager +import fcntl +import hashlib +import json +import os +from pathlib import Path +import tempfile +import threading + + +class ResubmitBurnError(RuntimeError): + """The interactive burn could not be completed safely.""" + + +_locks = {} +_guard = threading.Lock() +_held = threading.local() + + +def receipt_dir() -> Path: + root = os.environ.get("GAS_HOME") or str(Path.home() / ".gas") + return Path(root) / "resubmit_burns" + + +def state_path(hotkey: str, netuid: int, suffix: str) -> Path: + # Wallet addresses are normally SS58; reject path traversal in callers too. + if not hotkey or not hotkey.isalnum(): + raise ResubmitBurnError("Invalid hotkey for burn state") + return receipt_dir() / f"{int(netuid)}-{hotkey}.{suffix}" + + +def read_state(path: Path): + try: + with path.open() as handle: + data = json.load(handle) + if not isinstance(data, dict): + raise ValueError("expected an object") + return data + except FileNotFoundError: + return None + except (OSError, ValueError) as exc: + raise ResubmitBurnError( + f"Cannot read burn state at {path}; refusing another burn. " + "Preserve this file and recover the existing transaction first." + ) from exc + + +def _sync_directory(path: Path) -> None: + directory = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + + +def _ensure_directory(path: Path) -> None: + if path.is_dir(): + return + _ensure_directory(path.parent) + path.mkdir(exist_ok=True, mode=0o700) + _sync_directory(path.parent) + + +def atomic_write(path: Path, data: dict) -> None: + """Private file, atomic replacement, and fsync of both file and directory.""" + _ensure_directory(path.parent) + fd, temporary = tempfile.mkstemp(prefix=".burn-", dir=path.parent) + try: + with os.fdopen(fd, "w") as handle: + json.dump(data, handle) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + _sync_directory(path.parent) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +@contextmanager +def burn_lock(hotkey: str, netuid: int): + """Nonblocking, reentrant locally; OS lock survives until process exit. + + Keep the lock inode: unlinking a lock file would allow competing owners. + Upload holds this through proof acceptance and receipt cleanup, not just + through transaction submission. + """ + path = state_path(hotkey, netuid, "lock").absolute() + with _guard: + lock = _locks.setdefault(str(path), threading.RLock()) + if not lock.acquire(blocking=False): + raise ResubmitBurnError("Another push/burn is in progress for this hotkey") + try: + held = getattr(_held, "paths", set()) + if str(path) in held: + yield + return + _ensure_directory(path.parent) + fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600) + try: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise ResubmitBurnError( + "Another push/burn is in progress for this hotkey" + ) from exc + _held.paths = held | {str(path)} + try: + yield + finally: + _held.paths = held + finally: + os.close(fd) + finally: + lock.release() + + +def transaction_hash(encoded: str) -> str: + if not isinstance(encoded, str) or not encoded.startswith("0x"): + raise ValueError("missing encoded transaction") + raw = bytes.fromhex(encoded[2:]) + if not raw: + raise ValueError("empty encoded transaction") + return hashlib.blake2b(raw, digest_size=32).hexdigest() + + +def load_journal(hotkey: str, netuid: int): + path = state_path(hotkey, netuid, "pending.json") + data = read_state(path) + if data is None: + return None + try: + if ( + data["version"] != 1 + or data["hotkey"] != hotkey + or data["netuid"] != netuid + or data["status"] not in {"pending", "confirmed", "used", "failed"} + or transaction_hash(data["signed_extrinsic"]) != data["tx_hash"] + or type(data["start_block"]) is not int + or data["start_block"] < 0 + or data["period"] != 64 + or not isinstance(data["genesis_hash"], str) + or not data["genesis_hash"] + ): + raise ValueError("invalid pending transaction") + if data["status"] == "confirmed" and ( + type(data["block_number"]) is not int or data["block_number"] <= 0 + ): + raise ValueError("invalid confirmation") + return data + except (KeyError, TypeError, ValueError) as exc: + raise ResubmitBurnError( + f"Invalid burn journal at {path}; refusing another burn. Preserve it for recovery." + ) from exc + + +def save_journal(hotkey: str, netuid: int, data: dict) -> None: + atomic_write(state_path(hotkey, netuid, "pending.json"), data) + + +def execution_success(events, index: int): + """Require an explicit System event for this exact extrinsic index.""" + outcomes = set() + for event in events: + value = getattr(event, "value", event) + if not isinstance(value, dict): + continue + found_index = value.get("extrinsic_idx") + if found_index is None and isinstance(value.get("phase"), dict): + found_index = value["phase"].get("ApplyExtrinsic") + if isinstance(found_index, str) and found_index.isdecimal(): + found_index = int(found_index) + if type(found_index) is not int or found_index != index: + continue + details = value.get("event") + if not isinstance(details, dict): + details = value + module = details.get("event_module") or details.get("module_id") + name = details.get("event_id") or details.get("event_name") + if module == "System" and name in {"ExtrinsicSuccess", "ExtrinsicFailed"}: + outcomes.add(name == "ExtrinsicSuccess") + # Contradictory RPC events are unknown, not a confirmed failure that could + # authorize another payment. + return next(iter(outcomes)) if len(outcomes) == 1 else None diff --git a/gas/protocol/miner_requests.py b/gas/protocol/miner_requests.py index fe7b6b4e..2aad8713 100644 --- a/gas/protocol/miner_requests.py +++ b/gas/protocol/miner_requests.py @@ -237,6 +237,20 @@ def confirm_upload(wallet: bt.Wallet, upload_endpoint: str, model_id: int, file_ def upload_single_modality( + wallet, file_path: str, modality: str, upload_endpoint: str, + vertical: str = "general", resubmit=None, netuid: int = 34, chain_endpoint=None, +) -> dict: + from gas.protocol.burn_journal import burn_lock + from gas.protocol.resubmit_burn import recover_saved_burn + + with burn_lock(wallet.hotkey.ss58_address, netuid): + recover_saved_burn(wallet, netuid, chain_endpoint) + return _upload_single_modality( + wallet, file_path, modality, upload_endpoint, vertical, resubmit, netuid, + ) + + +def _upload_single_modality( wallet: bt.Wallet, file_path: str, modality: str, diff --git a/gas/protocol/resubmit_burn.py b/gas/protocol/resubmit_burn.py index d032ad72..688cb4b6 100644 --- a/gas/protocol/resubmit_burn.py +++ b/gas/protocol/resubmit_burn.py @@ -2,13 +2,17 @@ from __future__ import annotations -import json -import os +import math import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional +from gas.protocol.burn_journal import ( + ResubmitBurnError, atomic_write, burn_lock, execution_success, + load_journal, read_state, save_journal, state_path, transaction_hash, +) + RESUBMIT_FEE_TAO = 0.5 PRICE_SLACK = 0.02 RAO = 1_000_000_000 @@ -24,10 +28,6 @@ class BurnEvidence: block_number: int -class ResubmitBurnError(RuntimeError): - """The interactive burn could not be completed.""" - - def _detail_text(result: dict) -> str: response = result.get("response") or {} detail = response.get("detail") or response.get("error") or response.get("message") or "" @@ -66,38 +66,37 @@ def is_duplicate_upload(result: dict) -> bool: ) -def receipt_dir() -> Path: - root = os.environ.get("GAS_HOME") or str(Path.home() / ".gas") - return Path(root) / "resubmit_burns" - - def receipt_path(hotkey: str, netuid: int) -> Path: - return receipt_dir() / f"{int(netuid)}-{hotkey}.json" + return state_path(hotkey, netuid, "json") def load_burn_receipt(hotkey: str, netuid: int) -> Optional[BurnEvidence]: - path = receipt_path(hotkey, netuid) + data = read_state(receipt_path(hotkey, netuid)) + if data is None: + return None try: - data = json.loads(path.read_text()) + if type(data["block_number"]) is not int or data["block_number"] <= 0: + raise ValueError("invalid block number") return BurnEvidence( tx_hash=normalize_tx_hash(data.get("tx_hash")), block_number=int(data["block_number"]), ) - except (OSError, KeyError, TypeError, ValueError, ResubmitBurnError): - return None + except (KeyError, TypeError, ValueError, ResubmitBurnError) as exc: + raise ResubmitBurnError("Invalid saved burn receipt; refusing another burn") from exc def save_burn_receipt(hotkey: str, netuid: int, evidence: BurnEvidence) -> None: path = receipt_path(hotkey, netuid) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text( - json.dumps( - {"tx_hash": evidence.tx_hash, "block_number": evidence.block_number} - ) - ) + atomic_write(path, {"tx_hash": evidence.tx_hash, "block_number": evidence.block_number}) def clear_burn_receipt(hotkey: str, netuid: int) -> None: + journal = load_journal(hotkey, netuid) + evidence = load_burn_receipt(hotkey, netuid) + if journal and journal["status"] == "confirmed" and evidence: + if journal["tx_hash"] == evidence.tx_hash: + # Persist API acknowledgement before removing the legacy receipt. + save_journal(hotkey, netuid, {**journal, "status": "used"}) try: receipt_path(hotkey, netuid).unlink() except FileNotFoundError: @@ -105,13 +104,13 @@ def clear_burn_receipt(hotkey: str, netuid: int) -> None: def min_alpha_rao_for_fee(tao_per_alpha: float) -> int: - if tao_per_alpha <= 0: + if not math.isfinite(tao_per_alpha) or tao_per_alpha <= 0: raise ResubmitBurnError("Could not read the SN34 alpha price") return int((RESUBMIT_FEE_TAO / tao_per_alpha) * RAO) def alpha_rao_for_fee(tao_per_alpha: float) -> int: - if tao_per_alpha <= 0: + if not math.isfinite(tao_per_alpha) or tao_per_alpha <= 0: raise ResubmitBurnError("Could not read the SN34 alpha price") return int((RESUBMIT_FEE_TAO / tao_per_alpha) * (1 + PRICE_SLACK) * RAO) + 1 @@ -126,7 +125,7 @@ def normalize_tx_hash(value: Any) -> str: if digest.startswith("0x") or digest.startswith("0X"): digest = digest[2:] digest = digest.lower() - if len(digest) != 64: + if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest): raise ResubmitBurnError("Burn succeeded but the extrinsic hash was invalid") return digest @@ -185,27 +184,182 @@ def execute_resubmit_burn( *, subtensor=None, confirm_fn: Callable[[str], str] = input, + submission: Optional[dict] = None, ) -> BurnEvidence: + with burn_lock(wallet.hotkey.ss58_address, netuid): + import bittensor as bt + + sub = subtensor if subtensor is not None else bt.Subtensor( + network=chain_endpoint or ("finney" if netuid == 34 else "test") + ) + try: + return _execute_resubmit_burn( + wallet, netuid, subtensor=sub, + confirm_fn=confirm_fn, submission=submission, + ) + finally: + if subtensor is None: + sub.close() + + +def recover_saved_burn(wallet, netuid: int, chain_endpoint=None): + """Preflight uploads too: a server-side credit must not erase pending state.""" + journal = load_journal(wallet.hotkey.ss58_address, netuid) + if journal is None: + return import bittensor as bt - network = chain_endpoint or ("finney" if netuid == 34 else "test") - sub = subtensor or bt.Subtensor(network=network) + sub = bt.Subtensor(network=chain_endpoint or ("finney" if netuid == 34 else "test")) + try: + _recover_burn(sub, wallet.hotkey.ss58_address, netuid, wallet.coldkeypub.ss58_address) + finally: + sub.close() + + +def _recover_burn(sub, hotkey: str, netuid: int, coldkey: str): + journal = load_journal(hotkey, netuid) + if journal is None: + return load_burn_receipt(hotkey, netuid) + if ( + journal["genesis_hash"] != sub.substrate.get_block_hash(0) + or journal.get("coldkey") != coldkey + ): + raise ResubmitBurnError( + "Saved burn belongs to a different chain or coldkey; refusing another burn" + ) + if journal["status"] in {"used", "failed"}: + # A crash during cleanup can leave the old receipt behind. + clear_burn_receipt(hotkey, netuid) + return None + if journal["status"] == "pending": + print(f" Recovering pending burn {journal['tx_hash']} from finalized chain history...") + substrate = sub.substrate + try: + finalized = substrate.get_block_number(substrate.get_chain_finalised_head()) + if type(finalized) is not int or finalized < journal["start_block"]: + raise ValueError("finalized head unavailable") + # Mortal transactions are signed with an explicit 64-block era. + # Never infer failure from missing history, even after expiry. + end = min(finalized, journal["start_block"] + journal["period"]) + for number in range(journal["start_block"], end + 1): + block_hash = substrate.get_block_hash(number) + if not block_hash: + raise ValueError("block hash unavailable") + reply = substrate.rpc_request("chain_getBlock", [block_hash]) + extrinsics = reply["result"]["block"]["extrinsics"] + if not isinstance(extrinsics, list): + raise ValueError("block contents unavailable") + for index, encoded in enumerate(extrinsics): + if transaction_hash(encoded) != journal["tx_hash"]: + continue + success = execution_success(substrate.get_events(block_hash), index) + if success is False: + save_journal(hotkey, netuid, {**journal, "status": "failed"}) + raise ResubmitBurnError( + "Previous burn failed on-chain (transaction fees may apply). " + "No new burn was sent. Run again to explicitly authorize a new attempt." + ) + if success is not True: + raise ValueError("execution success unavailable") + journal = {**journal, "status": "confirmed", "block_number": number} + save_journal(hotkey, netuid, journal) + break + if journal["status"] == "confirmed": + break + except ResubmitBurnError: + raise + except Exception as exc: + raise ResubmitBurnError( + f"Cannot reconcile pending burn {journal['tx_hash']}; no new burn will be sent. " + "Keep the burn journal and retry recovery when the chain RPC is available." + ) from exc + if journal["status"] != "confirmed": + raise ResubmitBurnError( + f"Burn {journal['tx_hash']} is unresolved; no new burn will be sent. " + "Retry later; if it remains unresolved, investigate using the saved journal. " + "Do not delete the journal or switch machines to bypass recovery." + ) + evidence = BurnEvidence(journal["tx_hash"], journal["block_number"]) + save_burn_receipt(hotkey, netuid, evidence) + return evidence + + +def _submit_durable_burn(sub, wallet, netuid: int, call, amount_rao: int, kind: str, submission): + hotkey = wallet.hotkey.ss58_address + substrate = sub.substrate + start = substrate.get_block_number(substrate.get_chain_finalised_head()) + genesis = substrate.get_block_hash(0) + if type(start) is not int or start < 0 or not genesis: + raise ResubmitBurnError("Cannot read chain identity/finalized head; no burn was sent") + # Creating/signing does not broadcast. Use a mortal, direct extrinsic for + # both burn calls (not an SDK helper that signs and sends in one step). + extrinsic = substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey, era={"period": 64, "current": start}, + ) + encoded = str(extrinsic.data) + digest = transaction_hash(encoded) + journal = { + "version": 1, "status": "pending", "hotkey": hotkey, "netuid": netuid, + "coldkey": wallet.coldkeypub.ss58_address, "genesis_hash": genesis, + "start_block": start, "period": 64, "signed_extrinsic": encoded, + "tx_hash": digest, "kind": kind, "amount_rao": amount_rao, + "submission": submission, + } + try: + save_journal(hotkey, netuid, journal) + except OSError as exc: + raise ResubmitBurnError("Cannot persist pending burn; no transaction was sent") from exc + try: + receipt = substrate.submit_extrinsic( + extrinsic, wait_for_inclusion=True, wait_for_finalization=True, + ) + if receipt.finalized is not True or receipt.is_success is not True: + raise ValueError("success not confirmed") + if normalize_tx_hash(receipt.extrinsic_hash) != digest: + raise ValueError("receipt transaction mismatch") + number = getattr(receipt, "block_number", None) + if number is None: + number = substrate.get_block_number(receipt.block_hash) + if type(number) is not int or number <= 0: + raise ValueError("missing block number") + save_journal(hotkey, netuid, {**journal, "status": "confirmed", "block_number": number}) + evidence = BurnEvidence(digest, number) + save_burn_receipt(hotkey, netuid, evidence) + except Exception as exc: + raise ResubmitBurnError( + f"Burn {digest} may have executed; its journal is saved. " + "Run again to recover this transaction, not to burn again." + ) from exc + print(f" Burn included in block {evidence.block_number} ({evidence.tx_hash[:12]}…).") + return evidence + + +def _execute_resubmit_burn( + wallet, netuid, *, subtensor, confirm_fn, submission, +) -> BurnEvidence: + sub = subtensor + hotkey = wallet.hotkey.ss58_address + coldkey = wallet.coldkeypub.ss58_address + recovered = _recover_burn(sub, hotkey, netuid, coldkey) + if recovered is not None: + print(f" Reusing existing burn {recovered.tx_hash[:12]}…; no new burn sent.") + return recovered price = sub.get_subnet_price(netuid) tao_per_alpha = float(getattr(price, "tao", 0) or 0) min_rao = min_alpha_rao_for_fee(tao_per_alpha) amount_rao = alpha_rao_for_fee(tao_per_alpha) - amount_alpha = amount_rao / RAO - hotkey = wallet.hotkey.ss58_address - coldkey = wallet.coldkeypub.ss58_address stake = sub.get_stake(coldkey, hotkey, netuid) stake_rao = int(getattr(stake, "rao", 0) or 0) - print( - f" Resubmit fee: {RESUBMIT_FEE_TAO:g} TAO of SN34 alpha " - f"(~{amount_alpha:.4f} α at the current pool price)." - ) + print(f" Resubmit fee: {RESUBMIT_FEE_TAO:g} TAO worth of subnet {netuid} alpha.") print(f" Alpha staked to this hotkey: {stake_rao / RAO:.4f} α") if stake_rao >= min_rao: amount_rao = min(stake_rao, amount_rao) + amount_alpha = amount_rao / RAO + print( + f" Actual burn: {amount_rao} alpha rao ({amount_alpha:.9f} α), " + f"worth ~{amount_alpha * tao_per_alpha:.6f} TAO at the quoted price. " + "Includes up to 2% price padding (plus rounding); transaction fees are additional." + ) if not _confirm( " Burn that alpha now? This cannot be undone. [y/N] ", confirm_fn, @@ -215,13 +369,7 @@ def execute_resubmit_burn( ) print(" Unlock the coldkey if prompted, then wait for the burn to land...") call = _compose_burn(sub, hotkey, netuid, amount_rao) - burned = sub.sign_and_send_extrinsic( - call, - wallet, - sign_with="coldkey", - wait_for_inclusion=True, - wait_for_finalization=True, - ) + kind = "burn_alpha" else: tao_free = sub.get_balance(coldkey) tao_available = float(getattr(tao_free, "tao", 0) or 0) @@ -233,6 +381,7 @@ def execute_resubmit_burn( raise ResubmitBurnError( "Need 0.5 TAO of SN34 alpha, or 0.5 free TAO for add_stake_burn." ) + print(" Actual spend: 0.5 TAO via add_stake_burn; transaction fees are additional.") if not _confirm( " Spend 0.5 TAO via add_stake_burn now? This cannot be undone. [y/N] ", confirm_fn, @@ -241,21 +390,13 @@ def execute_resubmit_burn( "Cannot submit another model without burning 0.5 TAO of SN34 alpha." ) print(" Unlock the coldkey if prompted, then wait for add_stake_burn to land...") - burned = sub.add_stake_burn( - wallet, - netuid, - hotkey, - bt.Balance.from_tao(RESUBMIT_FEE_TAO), - wait_for_inclusion=True, - wait_for_finalization=True, + amount_rao = int(RESUBMIT_FEE_TAO * RAO) + kind = "add_stake_burn" + call = sub.compose_call( + "SubtensorModule", kind, + {"hotkey": hotkey, "netuid": netuid, "amount": amount_rao, "limit": None}, ) - evidence = evidence_from_response(burned, sub) - try: - save_burn_receipt(hotkey, netuid, evidence) - except OSError: - pass - print(f" Burn included in block {evidence.block_number} ({evidence.tx_hash[:12]}…).") - return evidence + return _submit_durable_burn(sub, wallet, netuid, call, amount_rao, kind, submission) def offer_resubmit_burn( @@ -265,6 +406,7 @@ def offer_resubmit_burn( *, confirm_fn: Callable[[str], str] = input, execute_fn=execute_resubmit_burn, + submission: Optional[dict] = None, ) -> BurnEvidence: print() print(" This hotkey already used its free submission.") @@ -283,4 +425,5 @@ def offer_resubmit_burn( netuid, chain_endpoint, confirm_fn=confirm_fn, + submission=submission, ) diff --git a/neurons/discriminator/push_model.py b/neurons/discriminator/push_model.py index e2b4dec9..d6017c23 100644 --- a/neurons/discriminator/push_model.py +++ b/neurons/discriminator/push_model.py @@ -25,6 +25,7 @@ import sys import traceback import time +from pathlib import Path from typing import Optional import bittensor as bt @@ -50,7 +51,7 @@ class Style: from gas.types import DiscriminatorModelId as ModelId from gas.utils.chain_model_metadata_store import ChainModelMetadataStore -from gas.protocol.miner_requests import upload_single_modality +from gas.protocol.miner_requests import upload_single_modality, calculate_file_sha256 MODEL_UPLOAD_ENDPOINT = "https://upload.bitmind.ai/upload" @@ -147,11 +148,15 @@ async def push_separate_models( endpoint = upload_endpoint or MODEL_UPLOAD_ENDPOINT - def resubmit(): + def resubmit(modality, model_path): from gas.protocol.resubmit_burn import ResubmitBurnError, offer_resubmit_burn try: - return offer_resubmit_burn(wallet, netuid, chain_endpoint) + return offer_resubmit_burn( + wallet, netuid, chain_endpoint, + submission={"modality": modality, "file_hash": calculate_file_sha256(Path(model_path)), + "vertical": vertical, "upload_endpoint": endpoint}, + ) except ResubmitBurnError as exc: print_error(str(exc)) return None @@ -173,8 +178,9 @@ def resubmit(): 'image', endpoint, vertical=vertical, - resubmit=resubmit, + resubmit=lambda: resubmit('image', image_model_path), netuid=netuid, + chain_endpoint=chain_endpoint, ) results['image'] = image_result @@ -199,8 +205,9 @@ def resubmit(): 'video', endpoint, vertical=vertical, - resubmit=resubmit, + resubmit=lambda: resubmit('video', video_model_path), netuid=netuid, + chain_endpoint=chain_endpoint, ) results['video'] = video_result @@ -225,8 +232,9 @@ def resubmit(): 'audio', endpoint, vertical=vertical, - resubmit=resubmit, + resubmit=lambda: resubmit('audio', audio_model_path), netuid=netuid, + chain_endpoint=chain_endpoint, ) results['audio'] = audio_result diff --git a/tests/test_burn_journal.py b/tests/test_burn_journal.py new file mode 100644 index 00000000..9d3c9425 --- /dev/null +++ b/tests/test_burn_journal.py @@ -0,0 +1,418 @@ +"""Fault injection only: never connect to a chain or submit a real burn.""" + +import multiprocessing +import stat +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from gas.protocol import burn_journal as journal +from gas.protocol import resubmit_burn as burn +from gas.protocol.miner_requests import upload_single_modality + + +class FakeChain: + def __init__(self): + self.substrate = self + self.head = 100 + self.genesis = "0x" + "11" * 32 + self.encoded = "0x01020304" + self.signed = 0 + self.sent = 0 + self.blocks = {} + self.events = {} + self.error = None + self.stake = 2_000_000_000 + self.calls = [] + self.before_submit = lambda: None + + def get_chain_finalised_head(self): + return f"block-{self.head}" + + def get_block_number(self, block_hash): + return int(block_hash.split("-")[1]) + + def get_block_hash(self, number): + return self.genesis if number == 0 else f"block-{number}" + + def get_subnet_price(self, netuid): + return SimpleNamespace(tao=0.5) + + def get_stake(self, *args): + return SimpleNamespace(rao=self.stake) + + def get_balance(self, *args): + return SimpleNamespace(tao=10) + + def compose_call(self, module, function, params): + self.calls.append((module, function, params)) + return self.calls[-1] + + def create_signed_extrinsic(self, **kwargs): + self.signed += 1 + assert kwargs["era"] == {"period": 64, "current": self.head} + return SimpleNamespace(data=self.encoded) + + def submit_extrinsic(self, extrinsic, **kwargs): + self.before_submit() + self.sent += 1 + assert kwargs == {"wait_for_inclusion": True, "wait_for_finalization": True} + self.head += 1 + self.blocks[self.head] = [self.encoded] + self.events[self.head] = [ + {"extrinsic_idx": 0, "event_module": "System", "event_id": "ExtrinsicSuccess"} + ] + if self.error: + raise self.error + return SimpleNamespace( + finalized=True, is_success=True, extrinsic_hash="0x" + journal.transaction_hash(self.encoded), + block_number=self.head, + ) + + def rpc_request(self, method, params): + assert method == "chain_getBlock" + number = self.get_block_number(params[0]) + return {"result": {"block": {"extrinsics": self.blocks.get(number, [])}}} + + def get_events(self, block_hash): + return self.events.get(self.get_block_number(block_hash), []) + + def close(self): + pass + + +@pytest.fixture +def context(tmp_path, monkeypatch): + monkeypatch.setenv("GAS_HOME", str(tmp_path)) + monkeypatch.setattr(burn.sys.stdin, "isatty", lambda: True) + # Never permit tests to accidentally hit a live network. + monkeypatch.setattr("bittensor.Subtensor", lambda **kw: pytest.fail("unexpected chain connection")) + wallet = SimpleNamespace( + hotkey=SimpleNamespace(ss58_address="5Hot"), + coldkeypub=SimpleNamespace(ss58_address="5Cold"), coldkey=object(), + ) + return wallet, FakeChain() + + +def execute(context, **kwargs): + wallet, chain = context + return burn.execute_resubmit_burn(wallet, 34, subtensor=chain, confirm_fn=lambda _: "y", **kwargs) + + +@pytest.mark.parametrize("stake,kind,amount", [ + (2_000_000_000, "burn_alpha", 1_020_000_001), + (0, "add_stake_burn", 500_000_000), +]) +def test_persist_before_broadcast_and_show_actual_cost(context, stake, kind, amount, capsys): + wallet, chain = context + chain.stake = stake + intent = {"file_hash": "ab" * 32, "modality": "image"} + + def before_submit(): + pending = journal.load_journal("5Hot", 34) + assert pending["signed_extrinsic"] == chain.encoded + assert pending["tx_hash"] == journal.transaction_hash(chain.encoded) + assert pending["amount_rao"] == amount + assert pending["status"] == "pending" + assert pending["submission"] == intent + assert pending["kind"] == kind + + chain.before_submit = before_submit + result = execute(context, submission=intent) + assert result == burn.load_burn_receipt("5Hot", 34) + assert chain.sent == chain.signed == 1 + assert chain.calls[0][1] == kind + assert chain.calls[0][2]["amount"] == amount + text = capsys.readouterr().out + assert "transaction fees are additional" in text + if stake: + assert "~0.510000 TAO" in text + assert "2% price padding" in text + assert stat.S_IMODE(journal.state_path("5Hot", 34, "pending.json").stat().st_mode) == 0o600 + + +@pytest.mark.parametrize("failure", [OSError("disk full"), PermissionError("read only")]) +def test_cannot_persist_never_broadcasts(context, monkeypatch, failure): + def fail(*args): + raise failure + monkeypatch.setattr(burn, "save_journal", fail) + with pytest.raises(burn.ResubmitBurnError, match="no transaction was sent"): + execute(context) + assert context[1].sent == 0 + + +@pytest.mark.parametrize("failure", [TimeoutError("RPC timeout"), KeyboardInterrupt()]) +def test_success_then_timeout_or_process_interruption_recovers_once(context, failure): + wallet, chain = context + chain.error = failure + with pytest.raises((burn.ResubmitBurnError, KeyboardInterrupt)): + execute(context) + assert journal.load_journal("5Hot", 34)["status"] == "pending" + chain.error = None + recovered = burn.execute_resubmit_burn( + wallet, 34, subtensor=chain, + confirm_fn=lambda _: pytest.fail("must not ask for a fresh burn"), + ) + assert recovered.block_number == 101 + assert recovered.tx_hash == journal.transaction_hash(chain.encoded) + assert chain.sent == chain.signed == 1 + + +def test_crash_before_broadcast_blocks_new_burn(context): + _, chain = context + def crash(): + raise KeyboardInterrupt() + chain.before_submit = crash + with pytest.raises(KeyboardInterrupt): + execute(context) + chain.head = 1000 # Even expired/missing transactions do not auto-trigger payment. + with pytest.raises(burn.ResubmitBurnError, match="unresolved"): + execute(context) + assert chain.sent == 0 + assert chain.signed == 1 + + +def test_receipt_save_failure_recovers_from_confirmed_journal(context, monkeypatch): + original = burn.save_burn_receipt + def fail(*args): + raise OSError("disk full") + monkeypatch.setattr(burn, "save_burn_receipt", fail) + with pytest.raises(burn.ResubmitBurnError, match="may have executed"): + execute(context) + monkeypatch.setattr(burn, "save_burn_receipt", original) + execute(context) + assert context[1].sent == context[1].signed == 1 + + +@pytest.mark.parametrize("events", [ + [], + [{"extrinsic_idx": 1, "event_module": "System", "event_id": "ExtrinsicSuccess"}], + [{"extrinsic_idx": 0, "event_module": "Other", "event_id": "ExtrinsicSuccess"}], + [ + {"extrinsic_idx": 0, "event_module": "System", "event_id": "ExtrinsicSuccess"}, + {"extrinsic_idx": 0, "event_module": "System", "event_id": "ExtrinsicFailed"}, + ], +]) +def test_missing_success_fails_closed(context, events): + _, chain = context + chain.error = TimeoutError() + with pytest.raises(burn.ResubmitBurnError): + execute(context) + chain.events[101] = events + with pytest.raises(burn.ResubmitBurnError, match="Cannot reconcile"): + execute(context) + assert chain.sent == chain.signed == 1 + + +def test_explicit_finalized_failure_only_allows_next_authorized_attempt(context): + _, chain = context + chain.error = TimeoutError() + with pytest.raises(burn.ResubmitBurnError): + execute(context) + chain.events[101] = [{"extrinsic_idx": 0, "event_module": "System", "event_id": "ExtrinsicFailed"}] + with pytest.raises(burn.ResubmitBurnError, match="failed on-chain"): + execute(context) + assert chain.sent == 1 + chain.error = None + chain.encoded = "0x05060708" + execute(context) + assert chain.sent == 2 + + +def test_rpc_failure_cannot_allow_new_burn(context, monkeypatch): + _, chain = context + chain.error = TimeoutError() + with pytest.raises(burn.ResubmitBurnError): + execute(context) + monkeypatch.setattr(chain, "rpc_request", lambda *a: {"error": "unavailable"}) + with pytest.raises(burn.ResubmitBurnError, match="Cannot reconcile"): + execute(context) + assert chain.sent == 1 + + +def test_different_chain_cannot_reuse_or_replace_burn(context): + execute(context) + context[1].genesis = "0x" + "22" * 32 + with pytest.raises(burn.ResubmitBurnError, match="different chain"): + execute(context) + assert context[1].sent == 1 + + +@pytest.mark.parametrize("suffix", ["pending.json", "json"]) +def test_corrupt_state_fails_closed(context, suffix): + path = journal.state_path("5Hot", 34, suffix) + path.parent.mkdir(parents=True) + path.write_text("broken json") + with pytest.raises(burn.ResubmitBurnError, match="refusing another burn"): + execute(context) + assert context[1].sent == context[1].signed == 0 + + +def test_unknown_upload_state_stops_before_api(context, tmp_path, monkeypatch): + wallet, chain = context + def crash(): + raise KeyboardInterrupt() + chain.before_submit = crash + with pytest.raises(KeyboardInterrupt): + execute(context) + monkeypatch.setattr("bittensor.Subtensor", lambda **kw: chain) + monkeypatch.setattr("gas.protocol.miner_requests.generate_presigned_url", lambda *a, **kw: pytest.fail("must recover first")) + model = tmp_path / "model.zip" + model.write_bytes(b"model") + with pytest.raises(burn.ResubmitBurnError, match="unresolved"): + upload_single_modality(wallet, str(model), "image", "https://upload.example") + assert chain.sent == 0 + + +def test_api_consumption_is_durable_and_next_submission_can_burn(context): + execute(context) + burn.clear_burn_receipt("5Hot", 34) + assert journal.load_journal("5Hot", 34)["status"] == "used" + context[1].encoded = "0x05060708" + execute(context) + assert context[1].sent == 2 + + +def test_thread_lock_is_reentrant_but_excludes_competing_push(context): + def compete(): + with pytest.raises(burn.ResubmitBurnError, match="in progress"): + with journal.burn_lock("5Hot", 34): + pytest.fail("must not enter") + with journal.burn_lock("5Hot", 34): + with journal.burn_lock("5Hot", 34): + with ThreadPoolExecutor(max_workers=1) as pool: + pool.submit(compete).result() + with journal.burn_lock("5Hot", 34): + pass + + +def _hold_process_lock(ready, release): + with journal.burn_lock("5Hot", 34): + ready.set() + release.wait(10) + + +def test_process_lock_excludes_another_cli(context): + ctx = multiprocessing.get_context("spawn") + ready, release = ctx.Event(), ctx.Event() + process = ctx.Process(target=_hold_process_lock, args=(ready, release)) + process.start() + try: + assert ready.wait(10) + with pytest.raises(burn.ResubmitBurnError, match="in progress"): + with journal.burn_lock("5Hot", 34): + pytest.fail("must not enter") + finally: + release.set() + process.join(10) + if process.is_alive(): + process.terminate() + process.join() + assert process.exitcode == 0 + + +@pytest.mark.parametrize("price", [float("nan"), float("inf"), -1.0, 0.0]) +def test_nonfinite_price_is_rejected(context, price): + context[1].get_subnet_price = lambda _: SimpleNamespace(tao=price) + with pytest.raises(burn.ResubmitBurnError): + execute(context) + assert context[1].sent == 0 + + +def test_execution_event_shapes_and_contradictions_are_unknown(): + assert journal.execution_success([ + {"phase": {"ApplyExtrinsic": 2}, "event": {"module_id": "System", "event_id": "ExtrinsicSuccess"}} + ], 2) is True + assert journal.execution_success([ + {"extrinsic_idx": 2, "event_module": "System", "event_id": "ExtrinsicFailed"}, + {"extrinsic_idx": "2", "event_module": "System", "event_id": "ExtrinsicSuccess"}, + ], 2) is None + + +def test_hash_matches_real_sdk_scale_serialization(): + from scalecodec.base import ScaleBytes + from scalecodec.types import GenericExtrinsic + + extrinsic = GenericExtrinsic(data=ScaleBytes("0x01020304")) + assert journal.transaction_hash(str(extrinsic.data)) == extrinsic.extrinsic_hash.hex() + + +def test_fsync_failure_before_broadcast(context, monkeypatch): + # Create the lock directory first so the injected failure hits the journal. + with journal.burn_lock("5Hot", 34): + pass + def fail(*args): + raise OSError("fsync failed") + monkeypatch.setattr(journal.os, "fsync", fail) + with pytest.raises(burn.ResubmitBurnError, match="no transaction was sent"): + execute(context) + assert context[1].sent == 0 + + +def test_crash_after_api_ack_before_receipt_removal(context, monkeypatch): + execute(context) + from pathlib import Path + original = Path.unlink + def fail(self, *args, **kwargs): + raise OSError("unlink failed") + monkeypatch.setattr(Path, "unlink", fail) + with pytest.raises(OSError): + burn.clear_burn_receipt("5Hot", 34) + assert journal.load_journal("5Hot", 34)["status"] == "used" + monkeypatch.setattr(Path, "unlink", original) + context[1].encoded = "0x05060708" + execute(context) + assert context[1].sent == 2 + + +@pytest.mark.parametrize("field,value", [("finalized", False), ("is_success", None), ("extrinsic_hash", "0x" + "ff" * 32), ("block_number", 0)]) +def test_ambiguous_receipt_keeps_pending_journal(context, field, value): + chain = context[1] + original = chain.submit_extrinsic + def submit(*args, **kwargs): + receipt = original(*args, **kwargs) + setattr(receipt, field, value) + return receipt + chain.submit_extrinsic = submit + with pytest.raises(burn.ResubmitBurnError, match="may have executed"): + execute(context) + assert journal.load_journal("5Hot", 34)["status"] == "pending" + execute(context) + assert chain.sent == 1 + + +def test_full_upload_retry_reuses_durable_burn(context, tmp_path, monkeypatch): + wallet, chain = context + monkeypatch.setattr("bittensor.Subtensor", lambda **kw: chain) + model = tmp_path / "model.zip" + model.write_bytes(b"model") + limit = {"success": False, "status_code": 403, "response": {"detail": "already used its free submission"}} + responses = iter([ + limit, + {"success": False, "status_code": 503, "response": {"detail": "temporary error"}}, + limit, + {"success": True, "response": {"data": {"model_id": 1, "presigned_url": "https://r2.example", "r2_key": "model"}}}, + ]) + proofs = [] + def presign(*args, **kwargs): + proofs.append(kwargs.get("burn_tx_hash")) + return next(responses) + monkeypatch.setattr("gas.protocol.miner_requests.generate_presigned_url", presign) + monkeypatch.setattr("gas.protocol.miner_requests.upload_to_r2", lambda *a: {"success": True}) + monkeypatch.setattr("gas.protocol.miner_requests.confirm_upload", lambda *a: {"success": True}) + first = upload_single_modality( + wallet, str(model), "image", "https://upload.example", + resubmit=lambda: execute(context), + ) + assert first["success"] is False + second = upload_single_modality( + wallet, str(model), "image", "https://upload.example", + resubmit=lambda: pytest.fail("must reuse the first burn"), + ) + assert second["success"] is True + digest = journal.transaction_hash(chain.encoded) + assert proofs == [None, digest, None, digest] + assert chain.sent == chain.signed == 1 + assert journal.load_journal("5Hot", 34)["status"] == "used" + assert burn.load_burn_receipt("5Hot", 34) is None diff --git a/tests/test_resubmit_burn.py b/tests/test_resubmit_burn.py index 3485fd2b..8ecf5123 100644 --- a/tests/test_resubmit_burn.py +++ b/tests/test_resubmit_burn.py @@ -144,13 +144,16 @@ def test_receipt_round_trip(tmp_path, monkeypatch): def test_low_alpha_uses_add_stake_burn(tmp_path, monkeypatch): monkeypatch.setenv("GAS_HOME", str(tmp_path)) monkeypatch.setattr("gas.protocol.resubmit_burn.sys.stdin.isatty", lambda: True) - burned = SimpleNamespace( - success=True, - extrinsic_receipt=SimpleNamespace( - extrinsic_hash="0x" + "ab" * 32, block_number=88 - ), - ) calls = [] + evidence_expected = BurnEvidence(tx_hash="ab" * 32, block_number=88) + + def submit(sub, wallet, netuid, call, amount, kind, submission): + assert kind == "add_stake_burn" + assert amount == 500_000_000 + save_burn_receipt("5Hot", netuid, evidence_expected) + return evidence_expected + + monkeypatch.setattr("gas.protocol.resubmit_burn._submit_durable_burn", submit) class Subtensor: def get_subnet_price(self, netuid): @@ -162,12 +165,11 @@ def get_stake(self, coldkey, hotkey, netuid): def get_balance(self, coldkey): return SimpleNamespace(tao=2.0) - def add_stake_burn(self, wallet, netuid, hotkey, amount, **kwargs): - calls.append((netuid, hotkey, amount)) - return burned - - def compose_call(self, *args, **kwargs): - raise AssertionError("should use add_stake_burn, not burn_alpha") + def compose_call(self, module, function, params): + assert module == "SubtensorModule" + assert function == "add_stake_burn" + calls.append(params) + return "call" wallet = SimpleNamespace( hotkey=SimpleNamespace(ss58_address="5Hot"), @@ -180,22 +182,18 @@ def compose_call(self, *args, **kwargs): confirm_fn=lambda prompt: "y", ) assert evidence.block_number == 88 - assert calls[0][0] == 34 - assert calls[0][1] == "5Hot" - assert calls[0][2].tao == 0.5 + assert calls == [{"netuid": 34, "hotkey": "5Hot", "amount": 500_000_000, "limit": None}] assert load_burn_receipt("5Hot", 34) == evidence def test_exact_fee_alpha_burns_without_slack_padding(tmp_path, monkeypatch): monkeypatch.setenv("GAS_HOME", str(tmp_path)) monkeypatch.setattr("gas.protocol.resubmit_burn.sys.stdin.isatty", lambda: True) - burned = SimpleNamespace( - success=True, - extrinsic_receipt=SimpleNamespace( - extrinsic_hash="0x" + "cd" * 32, block_number=12 - ), - ) amounts = [] + monkeypatch.setattr( + "gas.protocol.resubmit_burn._submit_durable_burn", + lambda *a: BurnEvidence(tx_hash="cd" * 32, block_number=12), + ) class Subtensor: def get_subnet_price(self, netuid): @@ -211,9 +209,6 @@ def compose_call(self, module, function, params): amounts.append(params["amount"]) return "call" - def sign_and_send_extrinsic(self, call, wallet, **kwargs): - return burned - def add_stake_burn(self, *args, **kwargs): raise AssertionError("should burn_alpha, not add_stake_burn")