feat(dataset): admit P2SH failures from proof files, starting with 173886, 174772 and 189498

.github/workflows/sanitycheck.yml

@@ -27,7 +27,7 @@ jobs:
         uses: actions/cache/restore@v6
         with:
           path: .cache/prevouts
-          key: prevouts-v2-${{ hashFiles('blocks/*.bin') }}
+          key: prevouts-v2-${{ hashFiles('blocks/*.bin', 'proofs/*.json') }}
           restore-keys: prevouts-v2-
       - name: validate data and test validator
         run: |

README.md

@@ -13,6 +13,7 @@ This dataset covers blocks that fail those rules, including failures that can be
 - [`docs/schema.md`](docs/schema.md): fields and admission rules.
 - [`docs/notes.md`](docs/notes.md): replay behaviour and incident notes.
 - `blocks/{height}-{hash}.bin`: full block, when available.
+- `proofs/{height}-{hash}.json`: for a P2SH record without a body, the failing transaction and the block's ordered transaction IDs.
 
 Merge-mined recoveries generally provide a header and coinbase rather than a full Bitcoin block.
 
@@ -32,7 +33,7 @@ Use `merge_mining` for child-chain commitments, `p2p` for direct Bitcoin network
 Prefer immutable evidence URLs.
 
 For header rules, observations and full block files are optional.
-Body failures require a complete `.bin` that demonstrates the named failure.
+Body failures require a complete `.bin` that demonstrates the named failure; a P2SH failure may instead supply a proof file whose transaction IDs reproduce the header's merkle root.
 For sigops, CI fetches the referenced previous transactions from public APIs, verifies their transaction IDs, and calculates the cost using their output scripts.
 For `already_confirmed_in_parent`, CI checks that a named non-coinbase transaction also appears in the canonical parent, using a txid list authenticated against the parent's header merkle root.
 The [schema](docs/schema.md#evidence-enforced-by-ci) specifies each rule's evidence contract; observation labels cannot substitute for these checks.

ci/block_evidence.py

@@ -15,9 +15,11 @@
 """
 
 from collections.abc import Mapping, Sequence
+import json
+import re
 from typing import TypeVar
 
-from bitcoin.core import COIN, CBlock, CoreMainParams, CTransaction, Hash as sha256d, MoneyRange, ValidationError, b2lx, lx
+from bitcoin.core import COIN, CBlock, CBlockHeader, CoreMainParams, CTransaction, Hash as sha256d, MoneyRange, ValidationError, b2lx, lx
 from bitcoin.core.scripteval import SCRIPT_VERIFY_P2SH, VerifyScript, VerifySignature
 from bitcoin.core.script import (
     CScript, CScriptInvalidError, CScriptOp, OP_1, OP_16,
@@ -27,6 +29,7 @@
 
 T = TypeVar("T", CBlock, CTransaction)
 
+HEX64 = re.compile(r"[0-9a-fA-F]{64}")
 MAX_MONEY = CoreMainParams.MAX_MONEY
 MAX_BLOCK_SIGOPS_COST = 80_000
 
@@ -117,6 +120,31 @@ def reuses_parent_transaction(block: CBlock, parent_txids: Sequence[str], txid:
             and any(not tx.is_coinbase() and tx.GetTxid() == target for tx in block.vtx[1:]))
 
 
+def checked_txids(txids: object, header: CBlockHeader) -> list[str]:
+    """Require a nonempty list of distinct 64-hex txids whose merkle root is the header's."""
+    if not isinstance(txids, list) or not txids or any(
+            not isinstance(txid, str) or not HEX64.fullmatch(txid) for txid in txids):
+        raise ValueError("txid list must be a nonempty array of 64-hex strings")
+    txids = [txid.lower() for txid in txids]
+    # Repeated leaves can preserve a merkle root under Bitcoin's odd-leaf padding.
+    if len(set(txids)) != len(txids):
+        raise ValueError("duplicate transaction IDs in txid list")
+    if CBlock.build_merkle_tree_from_txids([lx(txid) for txid in txids])[-1] != header.hashMerkleRoot:
+        raise ValueError("txid list merkle root mismatch")
+    return txids
+
+
+def read_proof(data: bytes, header: CBlockHeader) -> CTransaction:
+    """Read a transaction and the ordered txid list that places it in the header's block."""
+    proof = json.loads(data)
+    if not isinstance(proof, dict) or set(proof) != {"transaction", "txids"}:
+        raise ValueError("proof must be an object with transaction and txids")
+    tx = read_transaction(bytes.fromhex(proof["transaction"]))
+    if b2lx(tx.GetTxid()) not in checked_txids(proof["txids"], header):
+        raise ValueError("proof transaction is not in the block's txid list")
+    return tx
+
+
 def spending_input(transactions: Sequence[CTransaction], txid: bytes, vout: int) -> tuple[CTransaction, int]:
     """Return the one non-coinbase input among these transactions that spends the outpoint."""
     spends = [(tx, index) for tx in transactions if not tx.is_coinbase()

ci/prevouts.py

@@ -15,19 +15,17 @@
 from http.client import HTTPException, IncompleteRead
 import json
 from pathlib import Path
-import re
 import time
 import tempfile
 from typing import TypeVar
 from urllib.request import Request, urlopen
 
-from bitcoin.core import CBlock, CBlockHeader, CTransaction, b2lx, lx
+from bitcoin.core import CBlockHeader, CTransaction, b2lx, lx
 
-from block_evidence import omitted_prevouts, read_transaction
+from block_evidence import HEX64, checked_txids, omitted_prevouts, read_transaction
 
 DEFAULT_APIS = ("https://mempool.space/api", "https://blockstream.info/api")
 PREVOUTS_DIR = Path(".cache/prevouts")
-BLOCK_HASH = re.compile(r"[0-9a-fA-F]{64}")
 PARENT_TXIDS_LIMIT = 2 * 1024 * 1024
 T = TypeVar("T")
 
@@ -145,7 +143,7 @@ def decode_confirmation(data: bytes, txid: str) -> tuple[int, str]:
     if not isinstance(payload, dict) or payload.get("confirmed") is not True:
         raise ValueError(f"parent transaction {txid} is not confirmed")
     height, block_hash = payload.get("block_height"), payload.get("block_hash")
-    if type(height) is not int or height < 0 or not isinstance(block_hash, str) or not BLOCK_HASH.fullmatch(block_hash):
+    if type(height) is not int or height < 0 or not isinstance(block_hash, str) or not HEX64.fullmatch(block_hash):
         raise ValueError(f"malformed confirmation: {txid}")
     return height, block_hash.lower()
 
@@ -165,7 +163,7 @@ def _ascii_text(data: bytes) -> str:
 def decode_block_hash(data: bytes, height: int) -> str:
     """Read a block-height reply as a lowercase block hash."""
     text = _ascii_text(data)
-    if not BLOCK_HASH.fullmatch(text):
+    if not HEX64.fullmatch(text):
         raise ValueError(f"malformed block hash for height {height}")
     return text.lower()
 
@@ -195,17 +193,7 @@ def decode_parent_txids(data: bytes, header: CBlockHeader) -> list[str]:
     """Authenticate a complete, ordered txid list against the parent merkle root."""
     if len(data) > PARENT_TXIDS_LIMIT:
         raise ValueError("oversized parent txid list")
-    txids = json.loads(data)
-    if not isinstance(txids, list) or not txids or any(
-            not isinstance(txid, str) or not BLOCK_HASH.fullmatch(txid) for txid in txids):
-        raise ValueError("parent txid list must be a nonempty array of 64-hex strings")
-    txids = [txid.lower() for txid in txids]
-    # Repeated leaves can preserve a merkle root under Bitcoin's odd-leaf padding.
-    if len(set(txids)) != len(txids):
-        raise ValueError("duplicate transaction IDs in parent evidence")
-    if CBlock.build_merkle_tree_from_txids([lx(txid) for txid in txids])[-1] != header.hashMerkleRoot:
-        raise ValueError("parent transaction merkle root mismatch")
-    return txids
+    return checked_txids(json.loads(data), header)
 
 
 def load_parent_txids(block_hash: str, cache_dir: Path | str = PREVOUTS_DIR, fetch: bool = False,

ci/sanity-check.py

@@ -16,22 +16,23 @@
 from typing import Any
 from urllib.parse import urlparse
 
-from bitcoin.core import CBlock, CBlockHeader, b2lx, lx
+from bitcoin.core import CBlock, CBlockHeader, CTransaction, b2lx, lx
 from bitcoin.core._bignum import vch2bn
 from bitcoin.core.script import CScript, CScriptInvalidError, OP_1NEGATE
 from bitcoin.core.serialize import uint256_from_compact
 
 from block_evidence import (
     MAX_BLOCK_SIGOPS_COST, coinbase_amounts, confirmed_at_or_after, establishes_rule, omitted_prevouts, p2sh_spend_fails,
     spending_input,
-    read_block, reuses_parent_transaction, sigop_cost, verify_witness_commitment,
+    read_block, read_proof, reuses_parent_transaction, sigop_cost, verify_witness_commitment,
 )
 from prevouts import (
     DEFAULT_APIS, PREVOUTS_DIR, load_canonical_hash, load_confirmation, load_parent_txids, load_previous, load_transaction,
 )
 
 DATA_PATH = Path("data/invalid-blocks.jsonl")
 BLOCKS_DIR = Path("blocks")
+PROOFS_DIR = Path("proofs")
 REQUIRED = {"height", "hash", "header", "prev_hash", "nTime", "core_reject_reason", "rule"}
 CONTEXT_FIELDS = {
     "expected_nbits", "parent_mtp", "coinbase_height", "coinbase_scriptsig_hex",
@@ -51,8 +52,8 @@
 # sigops = body plus previous transactions; missing_parent = body plus API
 # evidence for the recorded outpoint; parent_txid_reuse = body plus an
 # authenticated canonical parent txid list; cb_amount = body plus canonical
-# parent and fee prevouts; p2sh = body plus the spent output of the named
-# input. Rule names and reject strings must
+# parent and fee prevouts; p2sh = body or proof file plus the spent output of
+# the named input. Rule names and reject strings must
 # match docs/schema.md.
 RULES = {
     "bad-txns-vout-toolarge": ("bad-txns-vout-toolarge", (), "body"),
@@ -311,12 +312,15 @@ def require_canonical_parent(record: dict[str, Any], prevouts_dir: Path | str, f
 
 
 def check_failure_evidence(record: dict[str, Any], block: CBlock | None, prevouts_dir: Path | str = PREVOUTS_DIR,
-                           fetch_prevouts: bool = False, apis: Sequence[str] = DEFAULT_APIS) -> None:
+                           fetch_prevouts: bool = False, apis: Sequence[str] = DEFAULT_APIS,
+                           proof: CTransaction | None = None) -> None:
     """Require a checked failure; observations cannot substitute for bytes."""
     mode = RULES[record["rule"]][2]
+    if proof is not None and mode != "p2sh":
+        raise ValueError("proof files apply only to the P2SH rule")
     if mode == "local":
         return
-    if block is None:
+    if block is None and proof is None:
         raise ValueError(f"{record['rule']} requires a complete block body")
     if mode == "body" and not establishes_rule(block, record["rule"]):
         raise ValueError(f"committed body does not demonstrate {record['rule']}")
@@ -336,7 +340,8 @@ def check_failure_evidence(record: dict[str, Any], block: CBlock | None, prevout
                   f"fees {amounts['fees']}, excess {amounts['excess']} sat", flush=True)
     if mode == "p2sh":
         txid, vout = record["context"]["failing_prevout"].split(":")
-        tx, index = spending_input(block.vtx, lx(txid), int(vout))
+        transactions = block.vtx if block is not None else [proof]
+        tx, index = spending_input(transactions, lx(txid), int(vout))
         if not p2sh_spend_fails(tx, index, load_transaction(txid, prevouts_dir, fetch_prevouts, apis)):
             raise ValueError("named input does not fail P2SH evaluation")
     if mode == "missing_parent":
@@ -368,11 +373,13 @@ def check_failure_evidence(record: dict[str, Any], block: CBlock | None, prevout
 
 def check_dataset(path: Path | str = DATA_PATH, blocks_dir: Path | str = BLOCKS_DIR,
                   prevouts_dir: Path | str = PREVOUTS_DIR, fetch_prevouts: bool = False,
-                  apis: Sequence[str] = DEFAULT_APIS) -> tuple[list[str], tuple[int, int, int, int]]:
+                  apis: Sequence[str] = DEFAULT_APIS,
+                  proofs_dir: Path | str = PROOFS_DIR) -> tuple[list[str], tuple[int, int, int, int, int]]:
     problems = []
     seen = set()
-    remaining_blocks = {block.name: block for block in Path(blocks_dir).glob("*.bin")}
-    block_count = 0
+    remaining = {"block": {path.name: path for path in Path(blocks_dir).glob("*.bin")},
+                 "proof": {path.name: path for path in Path(proofs_dir).glob("*.json")}}
+    found = {kind: len(paths) for kind, paths in remaining.items()}
     last_key = None
     observation_count = 0
     context_count = 0
@@ -393,8 +400,8 @@ def check_dataset(path: Path | str = DATA_PATH, blocks_dir: Path | str = BLOCKS_
             if record["core_reject_reason"] != RULES[rule][0]:
                 raise ValueError(f"rule {rule} requires core_reject_reason={RULES[rule][0]}")
             block_hash = record["hash"]
-            block = remaining_blocks.pop(f"{height}-{block_hash}.bin", None)
-            block_count += block is not None
+            block = remaining["block"].pop(f"{height}-{block_hash}.bin", None)
+            proof = remaining["proof"].pop(f"{height}-{block_hash}.json", None)
             key = (height, block_hash)
             if last_key is not None and key < last_key:
                 raise ValueError("records must be ordered by height then hash")
@@ -429,12 +436,19 @@ def check_dataset(path: Path | str = DATA_PATH, blocks_dir: Path | str = BLOCKS_
                 if "coinbase_scriptsig_hex" in details:
                     if evidence_block.vtx[0].vin[0].scriptSig.hex() != details["coinbase_scriptsig_hex"]:
                         raise ValueError("coinbase scriptSig does not match context")
-            check_failure_evidence(record, evidence_block, prevouts_dir, fetch_prevouts, apis)
+            proof_transaction = None
+            if proof is not None:
+                if block is not None:
+                    raise ValueError("a record has either a block body or a proof file, not both")
+                proof_transaction = read_proof(proof.read_bytes(), parsed_header)
+            check_failure_evidence(record, evidence_block, prevouts_dir, fetch_prevouts, apis, proof_transaction)
         except (OSError, ValueError) as exc:
             problems.append(f"{where}: {exc}")
-    for block in sorted(remaining_blocks.values()):
-        problems.append(f"{block}: orphan block file; name must match a dataset record")
-    return problems, (len(seen), context_count, observation_count, block_count)
+    for kind, paths in remaining.items():
+        for path in sorted(paths.values()):
+            problems.append(f"{path}: orphan {kind} file; name must match a dataset record")
+    return problems, (len(seen), context_count, observation_count,
+                      found["block"] - len(remaining["block"]), found["proof"] - len(remaining["proof"]))
 
 
 def main() -> int:
@@ -451,7 +465,7 @@ def main() -> int:
         print("\n".join(problems))
         return 1
     print("sanity-check successful")
-    print(f"  {counts[0]} blocks, {counts[1]} contexts, {counts[2]} observations, {counts[3]} block files")
+    print(f"  {counts[0]} blocks, {counts[1]} contexts, {counts[2]} observations, {counts[3]} block files, {counts[4]} proof files")
     return 0