#!/usr/bin/env python3 """Public verification of the repaired-V1 replay check. This is the public version of the validated research replay. The published evidence files contain the complete V1 Transfer history through the migration deadline and the historical contract state used by the research audit. The normal verification run reads those files locally, validates the actual history against the stored contract state, applies the natural reflection-accounting repair, and finds the first real transaction that can no longer execute under the repaired accounting. Published inputs: record/data/v1-counterfactual-transfer-cache.jsonl record/data/v1-counterfactual-transfer-progress.json record/data/v1-counterfactual-block-state.json Local output: record/data/v1-bug-corrected-counterfactual.local.json Run: python3 record/scripts/verify-v1-bug-corrected-counterfactual.py The ordinary public verification run does not reacquire the historical evidence from RPC. It verifies the published evidence used by the research audit. """ from __future__ import annotations import argparse import itertools import json import os import sys import time import urllib.request import urllib.error import threading from collections import defaultdict, deque from pathlib import Path ROOT = Path(__file__).resolve().parents[2] DATA = ROOT / "record" / "data" LOG_CACHE = DATA / "v1-counterfactual-transfer-cache.jsonl" STATE_CACHE = DATA / "v1-counterfactual-block-state.json" STATE_PARTIAL_CACHE = DATA / "v1-counterfactual-block-state-progress.local.jsonl" LOG_PROGRESS = DATA / "v1-counterfactual-transfer-progress.json" OUT = DATA / "v1-bug-corrected-counterfactual.local.json" V1 = "0xb1f4b66104353ec63d8d59d3da42c0b4fb06e7f3" TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" START_BLOCK = 12702978 END_BLOCK = 12770867 T_TOTAL = 10**21 SLOT_EXCLUDED_LEN = 8 SLOT_R_TOTAL = 10 ZERO = "0x" + "0" * 40 def parse_env_file(path): out = {} if not path.exists(): return out for line in path.read_text(errors="ignore").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) out[k.strip()] = v.strip().strip("'").strip('"') return out def rpc_candidates(explicit): """Choose one deterministic RPC for ordinary historical reads. These calls must not round-robin across providers: archive retention and eth_getLogs range limits differ by endpoint. The later historical-state stage has its own adaptive pool which probes archive capability first. """ if explicit: return [explicit] env = dict(parse_env_file(ROOT / ".env")) env.update(os.environ) preferred = [] if env.get("ALCHEMY_ETH_RPC_URL"): preferred.append(env["ALCHEMY_ETH_RPC_URL"]) for key in ("ALCHEMY_API_KEY", "ALCHEMY_KEY", "ALCHEMY_ETH_KEY"): if env.get(key): preferred.append("https://eth-mainnet.g.alchemy.com/v2/" + env[key]) for key in ("ETH_RPC_URL", "MAINNET_RPC_URL"): if env.get(key): preferred.append(env[key]) seen = set() preferred = [u for u in preferred if u and not (u in seen or seen.add(u))] if preferred: return [preferred[0]] # PublicNode has demonstrated archive log availability for this 2021 range. # Keep this stage pinned to one endpoint rather than mixing providers with # different pruning/range policies. return ["https://ethereum-rpc.publicnode.com"] def adaptive_rpc_candidates(explicit): """Return labelled candidates for the adaptive historical-state pool.""" env = dict(parse_env_file(ROOT / ".env")) env.update(os.environ) raw = [] if explicit: raw.append(("explicit", explicit, 6, 2, 10, 80)) for key in ("ETH_RPC_URL", "ALCHEMY_ETH_RPC_URL", "MAINNET_RPC_URL"): if env.get(key): raw.append((key.lower(), env[key], 6, 2, 10, 80)) for key in ("ALCHEMY_API_KEY", "ALCHEMY_KEY", "ALCHEMY_ETH_KEY"): if env.get(key): raw.append(( "alchemy", "https://eth-mainnet.g.alchemy.com/v2/" + env[key], 8, 2, 12, 100, )) extras = env.get("EXTRA_ETH_RPC_URLS", "") for i, url in enumerate((x.strip() for x in extras.split(",") if x.strip()), 1): raw.append((f"extra-{i}", url, 4, 1, 8, 60)) raw += [ ("publicnode", "https://ethereum-rpc.publicnode.com", 4, 1, 8, 60), ("drpc", "https://eth.drpc.org", 4, 1, 8, 60), ("mevblocker", "https://rpc.mevblocker.io", 4, 1, 8, 60), ("1rpc", "https://1rpc.io/eth", 3, 1, 6, 40), ("llamarpc", "https://eth.llamarpc.com", 3, 1, 6, 40), ("flashbots", "https://rpc.flashbots.net", 3, 1, 6, 40), ] seen = set() out = [] for label, url, max_workers, initial_workers, initial_batch, max_batch in raw: if not url or url in seen: continue seen.add(url) out.append({ "label": label, "url": url, "max_workers": max_workers, "initial_workers": initial_workers, "initial_batch": initial_batch, "max_batch": max_batch, }) return out class RPC: def __init__(self, urls): self.urls = urls self.rr = 0 def post(self, payload): last = None body = json.dumps(payload).encode() for attempt in range(max(6, len(self.urls) * 2)): url = self.urls[self.rr % len(self.urls)] self.rr += 1 try: req = urllib.request.Request( url, body, {"content-type": "application/json", "user-agent": "nottingham-audit/1.0"}, ) with urllib.request.urlopen(req, timeout=75) as resp: return json.load(resp) except Exception as exc: last = exc time.sleep(min(6, 0.7 * (attempt + 1))) raise RuntimeError(f"RPC failed: {last}") def one(self, method, params): payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params} data = None for attempt in range(10): data = self.post(payload) if isinstance(data, dict) and "result" in data: return data["result"] error = data.get("error") if isinstance(data, dict) else None if isinstance(error, dict) and error.get("code") == 429: wait = min(20.0, 1.5 * (2 ** attempt)) print( f" {method}: RPC 429; retrying in {wait:.1f}s", file=sys.stderr, flush=True, ) time.sleep(wait) continue raise RuntimeError(data) raise RuntimeError(f"{method} still throttled after retries: {data}") def batch(self, calls, label, size=15, delay=0.35): out = [] total = len(calls) for i in range(0, total, size): chunk = calls[i:i + size] payload = [ {"jsonrpc": "2.0", "id": j, "method": m, "params": p} for j, (m, p) in enumerate(chunk) ] data = None for attempt in range(10): data = self.post(payload) if isinstance(data, list): by_id = {int(x["id"]): x for x in data} errors = [ by_id.get(j) for j in range(len(chunk)) if not by_id.get(j) or "result" not in by_id.get(j) ] if not errors: break throttled = any( isinstance(item, dict) and isinstance(item.get("error"), dict) and item["error"].get("code") == 429 for item in errors ) if throttled: wait = min(20.0, 1.5 * (2 ** attempt)) print( f" {label}: RPC 429 at item {i:,}; " f"retrying same batch in {wait:.1f}s", file=sys.stderr, flush=True, ) time.sleep(wait) continue wait = min(10.0, 1.0 + attempt) print( f" {label}: incomplete batch at item {i:,}; " f"retrying in {wait:.1f}s", file=sys.stderr, flush=True, ) time.sleep(wait) else: raise RuntimeError( f"{label}: still failing after retries at item {i}: " f"{str(data)[:500]}" ) by_id = {int(x["id"]): x for x in data} out.extend(by_id[j]["result"] for j in range(len(chunk))) done = min(i + len(chunk), total) if total: print( f" {label}: {done:,}/{total:,} ({done * 100 / total:5.1f}%)", file=sys.stderr, flush=True, ) if delay: time.sleep(delay) return out def logs(self, start, end, step=1200): found = [] b = start while b <= end: e = min(end, b + step - 1) try: rows = self.one("eth_getLogs", [{ "address": V1, "fromBlock": hex(b), "toBlock": hex(e), "topics": [TRANSFER_TOPIC], }]) except Exception as exc: message = str(exc).lower() range_error = any(phrase in message for phrase in ( "query returned more", "response size", "limited to", "block range", "range limit", "too many blocks", )) if range_error and step > 1: new_step = max(1, step // 2) print( f" log range rejected; reducing step from {step} to {new_step} blocks", file=sys.stderr, flush=True, ) step = new_step continue raise found.extend(rows) print( f" logs {b}-{e}: +{len(rows):,}; total={len(found):,}", file=sys.stderr, flush=True, ) b = e + 1 return found def addr(topic): return "0x" + topic[-40:].lower() def t_amount_candidates(emitted, fee): if not fee: return [emitted] q, _ = divmod(emitted, 8) cands = [] for qq in range(max(0, q - 2), q + 3): for rem in range(10): gross = 10 * qq + rem if gross - gross // 10 - gross // 10 == emitted: cands.append(gross) return sorted(set(cands)) def load_or_fetch_logs(rpc, refresh=False): """Fetch V1 Transfer logs incrementally and resume by completed block range.""" if refresh: if LOG_CACHE.exists(): LOG_CACHE.unlink() if LOG_PROGRESS.exists(): LOG_PROGRESS.unlink() rows = [] if LOG_CACHE.exists(): rows = [ json.loads(x) for x in LOG_CACHE.read_text(errors="ignore").splitlines() if x.strip() ] completed_through = START_BLOCK - 1 if LOG_PROGRESS.exists(): progress = json.loads(LOG_PROGRESS.read_text()) expected_rows = int(progress.get("cached_transfer_rows", len(rows))) if expected_rows != len(rows): raise SystemExit( "Transfer-log cache and progress file disagree " f"({len(rows)} rows vs {expected_rows}). " "Restore the published record/data counterfactual cache files from git." ) completed_through = int(progress.get("completed_through", START_BLOCK - 1)) elif rows: # The older script wrote LOG_CACHE only after the complete fetch # succeeded, so a legacy cache with no progress sidecar is complete. completed_through = END_BLOCK if completed_through >= END_BLOCK: print( f"[1/6] using cached {LOG_CACHE.relative_to(ROOT)} " f"({len(rows):,} transfers)", flush=True, ) rows.sort(key=lambda x: (x["block"], x["tx_index"], x["log_index"])) return rows start = max(START_BLOCK, completed_through + 1) print( f"[1/6] fetching V1 Transfers from block {start} through {END_BLOCK} " f"({len(rows):,} already cached)", flush=True, ) DATA.mkdir(parents=True, exist_ok=True) step = 800 with open(LOG_CACHE, "a") as fh: b = start while b <= END_BLOCK: e = min(END_BLOCK, b + step - 1) try: raw = rpc.one("eth_getLogs", [{ "address": V1, "fromBlock": hex(b), "toBlock": hex(e), "topics": [TRANSFER_TOPIC], }]) except Exception as exc: message = str(exc).lower() range_error = any(phrase in message for phrase in ( "query returned more", "response size", "limited to", "block range", "range limit", "too many blocks", )) if range_error and step > 1: new_step = max(1, step // 2) print( f" log range rejected; reducing step from {step} to {new_step} blocks", file=sys.stderr, flush=True, ) step = new_step continue raise chunk_rows = [] for l in raw: chunk_rows.append({ "block": int(l["blockNumber"], 16), "tx": l["transactionHash"].lower(), "tx_index": int(l["transactionIndex"], 16), "log_index": int(l["logIndex"], 16), "from": addr(l["topics"][1]), "to": addr(l["topics"][2]), "emitted": str(int(l["data"], 16)), }) for item in chunk_rows: fh.write(json.dumps(item) + "\n") fh.flush() rows.extend(chunk_rows) LOG_PROGRESS.write_text(json.dumps({ "completed_through": e, "configured_end_block": END_BLOCK, "cached_transfer_rows": len(rows), }, indent=2) + "\n") print( f" logs {b}-{e}: +{len(chunk_rows):,}; " f"cached={len(rows):,}; {e - START_BLOCK + 1:,}/" f"{END_BLOCK - START_BLOCK + 1:,} blocks", file=sys.stderr, flush=True, ) b = e + 1 time.sleep(0.20) rows.sort(key=lambda x: (x["block"], x["tx_index"], x["log_index"])) return rows class AdaptiveEndpoint: def __init__(self, spec): self.label = spec["label"] self.url = spec["url"] self.max_workers = spec["max_workers"] self.allowed_workers = spec["initial_workers"] self.batch_blocks = spec["initial_batch"] self.max_batch = spec["max_batch"] self.success_streak = 0 self.total_success = 0 self.throttles = 0 self.failures = 0 self.cooldown_until = 0.0 self.disabled = False self.lock = threading.Lock() def snapshot(self): with self.lock: return { "workers": self.allowed_workers, "batch": self.batch_blocks, "success": self.total_success, "throttles": self.throttles, "failures": self.failures, "disabled": self.disabled, } def on_success(self): with self.lock: self.total_success += 1 self.success_streak += 1 self.failures = max(0, self.failures - 1) # Additive increase: cautiously grow request size first, then # concurrency. This finds spare capacity without jumping straight # into another provider throttle. if self.success_streak % 4 == 0 and self.batch_blocks < self.max_batch: step = max(2, self.batch_blocks // 4) self.batch_blocks = min(self.max_batch, self.batch_blocks + step) if self.success_streak % 10 == 0 and self.allowed_workers < self.max_workers: self.allowed_workers += 1 def on_throttle(self): with self.lock: self.throttles += 1 self.success_streak = 0 # Multiplicative decrease. self.batch_blocks = max(2, self.batch_blocks // 2) self.allowed_workers = max(1, self.allowed_workers - 1) penalty = min(30.0, 1.5 * (2 ** min(self.throttles, 4))) self.cooldown_until = max(self.cooldown_until, time.time() + penalty) def on_failure(self): with self.lock: self.failures += 1 self.success_streak = 0 self.batch_blocks = max(2, self.batch_blocks // 2) self.allowed_workers = max(1, self.allowed_workers - 1) self.cooldown_until = max(self.cooldown_until, time.time() + min(10, self.failures)) if self.failures >= 6 and self.total_success == 0: self.disabled = True def post_to_endpoint(endpoint, payload, timeout=45): body = json.dumps(payload).encode() req = urllib.request.Request( endpoint.url, body, {"content-type": "application/json", "user-agent": "nottingham-audit/2.0"}, ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.load(resp) except urllib.error.HTTPError as exc: body_text = "" try: body_text = exc.read().decode("utf-8", errors="ignore") except Exception: pass if exc.code == 429: raise RuntimeError("THROTTLE HTTP 429 " + body_text[:300]) raise RuntimeError(f"HTTP {exc.code}: {body_text[:300]}") except Exception as exc: raise RuntimeError(str(exc)) def classify_rpc_error(data): """Return throttle, archive, or other for a JSON-RPC error response.""" text = str(data).lower() if ( "429" in text or "compute units" in text or "rate limit" in text or "too many requests" in text or "capacity" in text ): return "throttle" if ( "missing trie node" in text or "historical state" in text or "archive" in text or "pruned" in text or "state is not available" in text or "header not found" in text ): return "archive" return "other" def probe_state_endpoint(spec, probe_block): """Verify the endpoint can serve historical V1 storage at our old block.""" ep = AdaptiveEndpoint(spec) payload = [ { "jsonrpc": "2.0", "id": 0, "method": "eth_getStorageAt", "params": [V1, hex(SLOT_R_TOTAL), hex(probe_block)], }, { "jsonrpc": "2.0", "id": 1, "method": "eth_getStorageAt", "params": [V1, hex(SLOT_EXCLUDED_LEN), hex(probe_block)], }, ] last = None for attempt in range(4): try: data = post_to_endpoint(ep, payload, timeout=20) if not isinstance(data, list): last = f"non-batch response: {str(data)[:120]}" else: by_id = {int(x.get("id", -1)): x for x in data} if any(i not in by_id or "result" not in by_id[i] for i in (0, 1)): kind = classify_rpc_error(data) if kind == "archive": return None, f"historical state unavailable: {str(data)[:180]}" last = f"{kind}: {str(data)[:180]}" elif int(by_id[0]["result"], 16) == 0: return None, "historical rTotal returned zero" else: return ep, None except Exception as exc: last = str(exc)[:180] if classify_rpc_error(last) == "archive": return None, last time.sleep(0.5 * (attempt + 1)) return None, last or "probe failed" def state_batch_payload(blocks): calls = [] for b in blocks: calls += [ ("eth_getStorageAt", [V1, hex(SLOT_R_TOTAL), hex(b - 1)]), ("eth_getStorageAt", [V1, hex(SLOT_R_TOTAL), hex(b)]), ("eth_getStorageAt", [V1, hex(SLOT_EXCLUDED_LEN), hex(b)]), ] return [ {"jsonrpc": "2.0", "id": i, "method": method, "params": params} for i, (method, params) in enumerate(calls) ] def decode_state_batch(data, blocks): if not isinstance(data, list): raise RuntimeError(f"non-list batch response: {str(data)[:300]}") by_id = {int(x.get("id", -1)): x for x in data} expected = len(blocks) * 3 errors = [ by_id.get(i) for i in range(expected) if i not in by_id or "result" not in by_id[i] ] if errors: kind = classify_rpc_error(errors) raise RuntimeError(f"{kind.upper()} {str(errors[:3])[:500]}") out = {} for i, b in enumerate(blocks): out[b] = { "r_total_before": str(int(by_id[3 * i]["result"], 16)), "r_total_after": str(int(by_id[3 * i + 1]["result"], 16)), "excluded_len": int(by_id[3 * i + 2]["result"], 16), } return out def load_or_fetch_state(rpc, blocks, refresh=False, explicit_rpc=None): """Adaptive multi-RPC historical state fetcher. Work is distributed across all archive-capable endpoints. Each provider independently tunes both batch size and active worker count: healthy -> additive increase 429/capacity -> multiplicative decrease + cooldown Every successful block is appended to STATE_PARTIAL_CACHE immediately. Existing STATE_CACHE content from older runs is preserved and reused. """ if refresh: if STATE_CACHE.exists(): STATE_CACHE.unlink() if STATE_PARTIAL_CACHE.exists(): STATE_PARTIAL_CACHE.unlink() state = {} if STATE_CACHE.exists(): raw = json.loads(STATE_CACHE.read_text()) state.update({int(k): v for k, v in raw.items()}) if STATE_PARTIAL_CACHE.exists(): for line in STATE_PARTIAL_CACHE.read_text(errors="ignore").splitlines(): if not line.strip(): continue item = json.loads(line) state[int(item["block"])] = { "r_total_before": item["r_total_before"], "r_total_after": item["r_total_after"], "excluded_len": int(item["excluded_len"]), } required = set(blocks) state = {b: v for b, v in state.items() if b in required} missing = [b for b in blocks if b not in state] if not missing: print( f"[2/6] using cached state for all {len(blocks):,} transfer blocks", flush=True, ) return state print( f"[2/6] adaptive multi-RPC state fetch: " f"{len(state):,}/{len(blocks):,} cached; {len(missing):,} remain", flush=True, ) # Probe all candidates concurrently. A public endpoint only joins the pool # if it proves it can read archive storage at the relevant historical block. specs = adaptive_rpc_candidates(explicit_rpc) probe_block = missing[0] probe_results = [None] * len(specs) def do_probe(i, spec): probe_results[i] = probe_state_endpoint(spec, probe_block) probe_threads = [ threading.Thread(target=do_probe, args=(i, spec), daemon=True) for i, spec in enumerate(specs) ] for t in probe_threads: t.start() for t in probe_threads: t.join() endpoints = [] for spec, result in zip(specs, probe_results): ep, error = result if ep: endpoints.append(ep) print( f" + {ep.label}: archive OK, start workers=" f"{ep.allowed_workers}, batch={ep.batch_blocks}", file=sys.stderr, flush=True, ) else: print( f" - {spec['label']}: skipped ({error})", file=sys.stderr, flush=True, ) if not endpoints: raise RuntimeError( "No configured RPC endpoint passed the historical-state probe. " "Set ALCHEMY_API_KEY, ETH_RPC_URL, or EXTRA_ETH_RPC_URLS." ) DATA.mkdir(parents=True, exist_ok=True) work = deque(missing) work_lock = threading.Lock() state_lock = threading.Lock() file_lock = threading.Lock() print_lock = threading.Lock() stop = threading.Event() target_total = len(blocks) initial_done = len(state) start_time = time.time() last_report = {"done": initial_done, "at": start_time} progress_fh = open(STATE_PARTIAL_CACHE, "a", buffering=1) def total_done(): with state_lock: return len(state) def pop_work(n): with work_lock: out = [] while work and len(out) < n: out.append(work.popleft()) return out def requeue(items): with work_lock: for b in reversed(items): work.appendleft(b) def maybe_report(force=False): now = time.time() done = total_done() with print_lock: if not force and done - last_report["done"] < 500 and now - last_report["at"] < 8: return elapsed = max(0.001, now - start_time) rate = max(0, done - initial_done) / elapsed tuning = [] for ep in endpoints: snap = ep.snapshot() status = "off" if snap["disabled"] else f"w{snap['workers']}/b{snap['batch']}" if snap["throttles"]: status += f"/429x{snap['throttles']}" tuning.append(f"{ep.label}:{status}") print( f" state: {done:,}/{target_total:,} " f"({done * 100 / target_total:5.1f}%) | " f"{rate:,.1f} blocks/s | " + " ".join(tuning), file=sys.stderr, flush=True, ) last_report["done"] = done last_report["at"] = now def worker(ep, worker_index): consecutive_other_errors = 0 while not stop.is_set(): if total_done() >= target_total: return snap = ep.snapshot() if snap["disabled"]: return if worker_index >= snap["workers"]: time.sleep(0.15) continue with ep.lock: cooldown = ep.cooldown_until batch_blocks = ep.batch_blocks now = time.time() if cooldown > now: time.sleep(min(0.5, cooldown - now)) continue chunk = pop_work(batch_blocks) if not chunk: # Other workers may still own the final chunks. if total_done() >= target_total: return time.sleep(0.10) continue payload = state_batch_payload(chunk) try: data = post_to_endpoint(ep, payload, timeout=50) decoded = decode_state_batch(data, chunk) newly_committed = [] with state_lock: for b in chunk: if b in state: continue state[b] = decoded[b] newly_committed.append((b, decoded[b])) if newly_committed: with file_lock: for b, item in newly_committed: progress_fh.write(json.dumps({ "block": b, **item, }) + "\n") progress_fh.flush() consecutive_other_errors = 0 ep.on_success() maybe_report() except Exception as exc: message = str(exc) kind = classify_rpc_error(message) requeue(chunk) if kind == "throttle": ep.on_throttle() elif kind == "archive": with ep.lock: ep.disabled = True with print_lock: print( f" {ep.label}: disabled after archive-state failure: " f"{message[:180]}", file=sys.stderr, flush=True, ) return else: consecutive_other_errors += 1 ep.on_failure() if consecutive_other_errors >= 6: with ep.lock: ep.disabled = True with print_lock: print( f" {ep.label}: disabled after repeated errors: " f"{message[:180]}", file=sys.stderr, flush=True, ) return threads = [] for ep in endpoints: for worker_index in range(ep.max_workers): t = threading.Thread( target=worker, args=(ep, worker_index), daemon=True, name=f"state-{ep.label}-{worker_index}", ) threads.append(t) t.start() try: while total_done() < target_total: alive = any(t.is_alive() for t in threads) enabled = any(not ep.snapshot()["disabled"] for ep in endpoints) if not alive or not enabled: raise RuntimeError( f"Adaptive RPC pool stopped with " f"{target_total - total_done():,} blocks still missing." ) maybe_report() time.sleep(0.5) except KeyboardInterrupt: stop.set() maybe_report(force=True) print( " interrupted; completed state reads are checkpointed and will resume", file=sys.stderr, flush=True, ) raise finally: stop.set() for t in threads: t.join(timeout=2) progress_fh.close() maybe_report(force=True) # Consolidate the append-only progress cache into the canonical JSON only # after every required block is present. complete = {b: state[b] for b in blocks} STATE_CACHE.write_text(json.dumps( {str(k): v for k, v in sorted(complete.items())}, indent=1, ) + "\n") return complete def solve_block(logs, start_r, end_r, excluded_fee): items = [] for l in logs: fee = l["from"] != ZERO and l["from"] not in excluded_fee and l["to"] not in excluded_fee cands = t_amount_candidates(int(l["emitted"]), fee) if not cands: return None, {"status": "no_gross_candidate", "log": l} items.append((l, fee, cands)) ambiguous = [i for i, x in enumerate(items) if len(x[2]) > 1] def replay(low): r_total = start_r chosen = [] for i, (l, fee, cands) in enumerate(items): rate = r_total // T_TOTAL gross = cands[0] if i in low else cands[-1] chosen.append((l, fee, gross, rate)) if fee: r_total -= (gross // 10) * rate return r_total, chosen hi_end, _ = replay(set()) unit = start_r // T_TOTAL m = round((end_r - hi_end) / unit) if unit else 0 if not (0 <= m <= len(ambiguous)): return None, { "status": "r_total_incompatible", "ambiguous": len(ambiguous), "required_low": m, "actual_end": str(end_r), "modeled_high_end": str(hi_end), } max_combos = 30000 for n, combo in enumerate(itertools.combinations(ambiguous, m)): if n >= max_combos: break r_end, chosen = replay(set(combo)) if r_end == end_r: return chosen, {"status": "exact", "ambiguous": len(ambiguous), "tested": n + 1} # Same bounded fallback used by the exact-rate builder. Candidate gross # amounts differ by only raw token units. Accept only inside that proven # rounding envelope and mark the block as bounded, never exact. combo = set(ambiguous[:m]) r_end, chosen = replay(combo) bound = len(ambiguous) * unit if abs(r_end - end_r) <= bound: return chosen, { "status": "bounded", "ambiguous": len(ambiguous), "residual": str(r_end - end_r), "bound": str(bound), } return None, { "status": "no_solution", "ambiguous": len(ambiguous), "actual_end": str(end_r), "modeled_end": str(r_end), } def main(): ap = argparse.ArgumentParser() ap.add_argument("--rpc") args = ap.parse_args() rpc = RPC(rpc_candidates(args.rpc)) logs = load_or_fetch_logs(rpc, False) by_block = defaultdict(list) for l in logs: by_block[l["block"]].append(l) blocks = sorted(by_block) deploy_logs = by_block.get(START_BLOCK, []) mint = [l for l in deploy_logs if l["from"] == ZERO] if len(mint) != 1: raise SystemExit(f"Expected one deployment mint Transfer, found {len(mint)}") if len(deploy_logs) != 1: raise SystemExit( "Deployment block contains more than the mint; initialize this script " "with an explicit constructor-block replay before using its result." ) deployer = mint[0]["to"] print(f" deployer={deployer}") state = load_or_fetch_state(rpc, blocks, False, args.rpc) print("[3/6] validating full-history assumptions against actual _rTotal") if any(int(state[b]["excluded_len"]) != 0 for b in blocks): bad = [b for b in blocks if int(state[b]["excluded_len"]) != 0][:10] raise SystemExit( f"Reflection-excluded accounts exist in transfer blocks {bad}; " "simple rTotal/tTotal replay is not valid." ) excluded_fee = {V1, deployer} solved = {} statuses = {} failures = [] previous_end = int(state[START_BLOCK]["r_total_after"]) process_blocks = [b for b in blocks if b > START_BLOCK] for n, b in enumerate(process_blocks, 1): start_r = int(state[b]["r_total_before"]) end_r = int(state[b]["r_total_after"]) if previous_end is not None and start_r != previous_end: failures.append({ "block": b, "status": "unseen_r_total_change_between_transfer_blocks", "previous_transfer_block_end": str(previous_end), "this_block_start": str(start_r), }) break chosen, meta = solve_block(by_block[b], start_r, end_r, excluded_fee) statuses[b] = meta if chosen is None: failures.append({"block": b, **meta}) break solved[b] = chosen previous_end = end_r if n % 500 == 0 or n == len(process_blocks): print( f" validate blocks: {n:,}/{len(process_blocks):,} " f"({n * 100 / len(process_blocks):5.1f}%)", file=sys.stderr, ) exact_blocks = sum(1 for x in statuses.values() if x.get("status") == "exact") bounded_blocks = sum(1 for x in statuses.values() if x.get("status") == "bounded") if failures: result = { "status": "actual_history_validation_failed", "deployer": deployer, "transfer_logs": len(logs), "transfer_blocks": len(blocks), "exact_blocks_before_failure": exact_blocks, "bounded_blocks_before_failure": bounded_blocks, "failure": failures[0], "meaning": ( "The assumptions needed for the counterfactual replay do not yet " "reproduce the actual V1 chain history. Do not interpret a repaired " "counterfactual until this validation failure is resolved." ), } OUT.write_text(json.dumps(result, indent=2) + "\n") print(json.dumps(result, indent=2)) raise SystemExit(2) print(f" actual history validated: exact={exact_blocks:,}, bounded={bounded_blocks:,}") print("[4/6] replaying the same transfers through repaired reflection accounting") initial_r = int(state[START_BLOCK]["r_total_after"]) balances = defaultdict(int) balances[deployer] = initial_r corrected_r_total = initial_r first_impossible = None applied = 0 for bn, b in enumerate(process_blocks, 1): actual_start = int(state[b]["r_total_before"]) if corrected_r_total != actual_start: first_impossible = { "block": b, "reason": "corrected_r_total_diverged_from_actual_rate_path", "corrected_r_total": str(corrected_r_total), "actual_r_total": str(actual_start), } break for l, fee, gross, rate in solved[b]: sender = l["from"] recipient = l["to"] if sender == ZERO: continue r_amount = gross * rate sender_before = balances[sender] if sender_before < r_amount: deficit = r_amount - sender_before first_impossible = { "block": b, "tx": l["tx"], "tx_index": l["tx_index"], "log_index": l["log_index"], "sender": sender, "recipient": recipient, "fee_bearing": fee, "gross_token_raw": str(gross), "rate": str(rate), "corrected_sender_r_owned": str(sender_before), "required_r_amount": str(r_amount), "deficit_r_units": str(deficit), "deficit_as_fraction_of_required": float(deficit / r_amount) if r_amount else None, "reason": "sender_cannot_make_actual_transfer_in_repaired_ledger", } break balances[sender] -= r_amount if fee: t_fee = gross // 10 t_team = gross // 10 r_fee = t_fee * rate r_team = t_team * rate r_transfer = r_amount - r_fee - r_team balances[recipient] += r_transfer balances[V1] += r_team corrected_r_total -= r_fee else: balances[recipient] += r_amount applied += 1 if first_impossible: break if bn % 500 == 0 or bn == len(process_blocks): print( f" corrected replay: {bn:,}/{len(process_blocks):,} blocks " f"({bn * 100 / len(process_blocks):5.1f}%)", file=sys.stderr, ) print("[5/6] interpreting counterfactual") if first_impossible: status = "same_transaction_history_becomes_impossible" meaning = ( "The natural bug repair cannot reproduce the real transaction history " "without changing at least one holder action. A complete bug-free ledger " "therefore requires an additional counterfactual policy: what amount that " "holder would have transferred, and how every later actor would have behaved." ) print(" FIRST IMPOSSIBLE TRANSFER") print(json.dumps(first_impossible, indent=2)) else: status = "same_transaction_history_remained_executable" meaning = ( "The repaired ledger can execute the same nominal transfer history through " "the deadline. A separate entitlement rule is still required to decide " "whether counterfactual balances, actual deposited units, or another measure " "should control V2 allocation." ) print(" repaired ledger executed the full actual transfer history") print("[6/6] writing audit") result = { "status": status, "code_level_repair": "rTransferAmount = rAmount - rFee - rTeam", "deployment_block": START_BLOCK, "deadline_last_block": END_BLOCK, "deployer": deployer, "fee_exempt_assumption": [V1, deployer], "actual_history_validation": { "transfer_logs": len(logs), "transfer_blocks": len(blocks), "exact_blocks": exact_blocks, "bounded_blocks": bounded_blocks, "reflection_excluded_accounts_in_transfer_blocks": 0, "validated_through_deadline": True, }, "corrected_replay": { "transfers_applied_before_divergence": applied, "first_impossible": first_impossible, }, "meaning": meaning, } OUT.write_text(json.dumps(result, indent=2) + "\n") print(f" wrote {OUT.relative_to(ROOT)}") if __name__ == "__main__": main()