Fix resubmit burn crash recovery and concurrent-push safety

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.
 
 ---
 

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

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,