feat(#869): retry-safe Base RPC failover with chain-id validation

benchmarks/direct-inventory-v1/rpc-failover/check.py

@@ -28,8 +28,12 @@ def read(path: str) -> str:
     if phrase not in tests:
         raise SystemExit(f"shared RPC tests lack {phrase}")
 
+# Run tests as a file so imports resolve without an undeclared PYTHONPATH.
+test_file = ROOT / "scripts" / "test_shared_rpc.py"
+if not test_file.is_file():
+    raise SystemExit("missing required file: scripts/test_shared_rpc.py")
 completed = subprocess.run(
-    [sys.executable, "-m", "unittest", "scripts.test_shared_rpc", "-v"],
+    [sys.executable, str(test_file), "-v"],
     cwd=ROOT,
     text=True,
     stdout=subprocess.PIPE,

scripts/_shared/rpc.py

@@ -1,14 +1,67 @@
-"""JSON-RPC transport shared by local fork rehearsal scripts."""
+"""JSON-RPC transport with retry-safe Base RPC failover and chain-id validation."""
 
 from __future__ import annotations
 
 import json
-from typing import Any
-from urllib.error import URLError
+import time
+from typing import Any, Sequence
+from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
+# Ordered Base mainnet HTTPS endpoints (no credentials in URLs).
+BASE_RPC_ENDPOINTS: Sequence[str] = (
+    "https://mainnet.base.org",
+    "https://base.llamarpc.com",
+    "https://base-rpc.publicnode.com",
+    "https://1rpc.io/base",
+    "https://base.drpc.org",
+)
+
+BASE_CHAIN_ID = 8453
+MAX_RETRIES = 3
+INITIAL_BACKOFF_SECONDS = 1.0
+
+
+class TransportError(RuntimeError):
+    """Retryable transport / HTTP 429 / HTTP 5xx failure."""
+
+    def __init__(self, message: str, *, retryable: bool = True) -> None:
+        super().__init__(message)
+        self.retryable = retryable
+
+
+class RpcError(RuntimeError):
+    """Non-retryable JSON-RPC execution error."""
+
+
+# Public aliases used by tests and callers.
+_TransportError = TransportError
+_RpcError = RpcError
+
+
+def _backoff(attempt: int) -> float:
+    """Deterministic exponential backoff: 1s, 2s, 4s, ..."""
+    return INITIAL_BACKOFF_SECONDS * (2 ** max(attempt - 1, 0))
+
+
+def _retryable_status(code: int) -> bool:
+    return code == 429 or 500 <= code < 600
+
+
+def _redact_endpoint(endpoint: str) -> str:
+    """Never surface credentials if an endpoint URL ever carries them."""
+    if "@" not in endpoint:
+        return endpoint
+    try:
+        scheme, rest = endpoint.split("://", 1)
+        host = rest.split("@", 1)[-1]
+        return f"{scheme}://{host}"
+    except Exception:  # noqa: BLE001
+        return "<redacted-endpoint>"
+
 
 def rpc(url: str, method: str, params: list[Any], request_id: int = 1) -> Any:
+    """Single-endpoint JSON-RPC call (preserved for backward compatibility)."""
     payload = json.dumps(
         {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}
     ).encode("utf-8")
@@ -19,5 +72,188 @@ def rpc(url: str, method: str, params: list[Any], request_id: int = 1) -> Any:
     except URLError as error:
         raise RuntimeError(f"RPC transport failed for {method}: {error}") from error
     if body.get("error"):
-        raise RuntimeError(f"RPC {method} failed: {json.dumps(body['error'], sort_keys=True)}")
+        raise RuntimeError(
+            f"RPC {method} failed: {json.dumps(body['error'], sort_keys=True)}"
+        )
+    return body.get("result")
+
+
+def _rpc_call(endpoint: str, method: str, params: list[Any], request_id: int) -> Any:
+    """Low-level call with transport vs execution error separation."""
+    payload = json.dumps(
+        {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}
+    ).encode("utf-8")
+    request = Request(
+        endpoint, data=payload, headers={"content-type": "application/json"}
+    )
+    try:
+        with urlopen(request, timeout=30) as response:
+            status = getattr(response, "status", None)
+            if status is None and hasattr(response, "getcode"):
+                try:
+                    status = response.getcode()
+                except Exception:  # noqa: BLE001
+                    status = 200
+            if status is None:
+                status = 200
+            body = json.load(response)
+    except HTTPError as error:
+        code = int(getattr(error, "code", 0) or 0)
+        if _retryable_status(code):
+            raise TransportError(
+                f"HTTP {code} from {_redact_endpoint(endpoint)}",
+                retryable=True,
+            ) from error
+        raise TransportError(
+            f"HTTP {code} from {_redact_endpoint(endpoint)}",
+            retryable=False,
+        ) from error
+    except URLError as error:
+        raise TransportError(
+            f"RPC transport failed for {method}: {error}",
+            retryable=True,
+        ) from error
+
+    if _retryable_status(int(status or 0)):
+        raise TransportError(
+            f"HTTP {status} from {_redact_endpoint(endpoint)}",
+            retryable=True,
+        )
+
+    if body.get("error"):
+        raise RpcError(
+            f"RPC {method} failed: {json.dumps(body['error'], sort_keys=True)}"
+        )
     return body.get("result")
+
+
+def _validate_chain(endpoint: str) -> int | None:
+    """Return chain id when endpoint reports Base (8453); otherwise None."""
+    try:
+        result = _rpc_call(endpoint, "eth_chainId", [], 1)
+        if isinstance(result, str):
+            chain_id = int(result, 16) if result.startswith("0x") else int(result)
+        else:
+            chain_id = int(result)
+        if chain_id == BASE_CHAIN_ID:
+            return chain_id
+    except Exception:  # noqa: BLE001 - skip bad endpoints
+        return None
+    return None
+
+
+def _ordered_https_endpoints(
+    preferred: str | None = None,
+    endpoints: Sequence[str] | None = None,
+) -> list[str]:
+    """Build ordered HTTPS endpoint list with optional preferred first."""
+    ordered: list[str] = []
+    pref = (preferred or "").strip()
+    if pref.lower().startswith("https://"):
+        ordered.append(pref)
+    for endpoint in endpoints if endpoints is not None else BASE_RPC_ENDPOINTS:
+        ep = str(endpoint).strip()
+        if ep and ep not in ordered:
+            ordered.append(ep)
+    return ordered
+
+
+def select_working_base_rpc(
+    preferred: str | None = None,
+    endpoints: Sequence[str] | None = None,
+    max_retries: int = MAX_RETRIES,
+) -> str:
+    """Return the first chain-valid HTTPS Base endpoint that accepts eth_chainId.
+
+    Probes preferred first, then the shared ordered list. Does not return a
+    preferred URL that failed validation merely because it is HTTPS.
+    """
+    ordered = _ordered_https_endpoints(preferred, endpoints)
+    if not ordered:
+        raise RuntimeError("RPC failover exhausted: no endpoints configured")
+
+    last_error: Exception | None = None
+    for endpoint in ordered:
+        if not str(endpoint).lower().startswith("https://"):
+            last_error = RuntimeError(
+                f"refusing non-HTTPS endpoint {_redact_endpoint(str(endpoint))}"
+            )
+            continue
+        for attempt in range(1, max_retries + 1):
+            try:
+                chain_id = _validate_chain(endpoint)
+                if chain_id == BASE_CHAIN_ID:
+                    return endpoint
+                last_error = RuntimeError(
+                    f"wrong chain on {_redact_endpoint(endpoint)}"
+                )
+                break
+            except TransportError as error:
+                last_error = error
+                if not error.retryable or attempt >= max_retries:
+                    break
+                time.sleep(_backoff(attempt))
+            except Exception as error:  # noqa: BLE001
+                last_error = error
+                break
+
+    raise RuntimeError(
+        f"RPC failover exhausted for eth_chainId: {last_error}"
+    ) from last_error
+
+
+def rpc_failover(
+    method: str,
+    params: list[Any],
+    request_id: int = 1,
+    endpoints: Sequence[str] | None = None,
+    max_retries: int = MAX_RETRIES,
+    *,
+    preferred: str | None = None,
+) -> Any:
+    """JSON-RPC with ordered HTTPS failover, chain-id gate, and bounded retries.
+
+    Retries only transport failures, HTTP 429, and HTTP 5xx. JSON-RPC execution
+    errors propagate immediately. Endpoint credentials are never logged.
+
+    When ``preferred`` is set it is tried first; the working endpoint for each
+    call is the one that passes chain validation for that attempt.
+    """
+    ordered = _ordered_https_endpoints(preferred, endpoints)
+    if not ordered:
+        raise RuntimeError("RPC failover exhausted: no endpoints configured")
+
+    last_error: Exception | None = None
+    for endpoint in ordered:
+        if not str(endpoint).lower().startswith("https://"):
+            last_error = RuntimeError(
+                f"refusing non-HTTPS endpoint {_redact_endpoint(str(endpoint))}"
+            )
+            continue
+        chain_id = _validate_chain(endpoint)
+        if chain_id != BASE_CHAIN_ID:
+            last_error = RuntimeError(
+                f"wrong chain on {_redact_endpoint(endpoint)}"
+            )
+            continue
+
+        for attempt in range(1, max_retries + 1):
+            try:
+                # After chain validation, perform the requested method.
+                if method == "eth_chainId":
+                    return hex(BASE_CHAIN_ID)
+                return _rpc_call(endpoint, method, params, request_id)
+            except RpcError:
+                raise
+            except TransportError as error:
+                last_error = error
+                if not error.retryable or attempt >= max_retries:
+                    break
+                time.sleep(_backoff(attempt))
+            except Exception as error:  # noqa: BLE001
+                last_error = error
+                break
+
+    raise RuntimeError(
+        f"RPC failover exhausted for {method}: {last_error}"
+    ) from last_error

scripts/bounty_inventory_guard.py

@@ -20,6 +20,13 @@
 from pathlib import Path
 from typing import Any
 
+_SCRIPTS_DIR = Path(__file__).resolve().parent
+if str(_SCRIPTS_DIR) not in sys.path:
+    sys.path.insert(0, str(_SCRIPTS_DIR))
+
+# Inventory automation shares the Base RPC failover transport with readiness checks.
+from _shared.rpc import rpc_failover, select_working_base_rpc
+
 NON_ACTIONABLE_LABELS = frozenset(
     {
         "duplicate",
@@ -179,9 +186,41 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         action="store_true",
         help="Exit with code 2 when count is below threshold",
     )
+    p.add_argument(
+        "--rpc-url",
+        default=os.environ.get("BASE_MAINNET_RPC_URL", ""),
+        help="Preferred Base HTTPS RPC (failover via shared transport; default BASE_MAINNET_RPC_URL)",
+    )
+    p.add_argument(
+        "--probe-rpc",
+        action="store_true",
+        help="Resolve a working Base RPC via shared failover before counting inventory",
+    )
     return p.parse_args(argv)
 
 
+def resolve_inventory_base_rpc(preferred: str | None = None) -> str:
+    """Canonical inventory chain-read endpoint selection (shared failover transport)."""
+    return select_working_base_rpc(preferred=(preferred or "").strip() or None)
+
+
+def probe_inventory_rpc(preferred: str | None = None) -> dict[str, Any]:
+    """Select a working Base endpoint and confirm eth_blockNumber via failover."""
+    endpoint = resolve_inventory_base_rpc(preferred)
+    block = rpc_failover(
+        "eth_blockNumber",
+        [],
+        preferred=endpoint,
+        endpoints=[endpoint],
+        max_retries=2,
+    )
+    return {
+        "base_rpc_endpoint": endpoint,
+        "eth_blockNumber": block,
+        "chain_id": 8453,
+    }
+
+
 def label_names(issue: dict[str, Any]) -> set[str]:
     labels = issue.get("labels") or []
     names: set[str] = set()
@@ -548,6 +587,16 @@ def main(argv: list[str] | None = None) -> int:
     if args.meta_replenishment_target < args.meta_threshold:
         raise SystemExit("meta-replenishment-target must be >= meta-threshold")
 
+    rpc_probe: dict[str, Any] | None = None
+    if args.probe_rpc or (args.rpc_url and not args.fixture):
+        # Live inventory runs share the failover transport; fixtures stay offline.
+        try:
+            rpc_probe = probe_inventory_rpc(args.rpc_url or None)
+        except Exception as error:  # noqa: BLE001 — surface RPC path failures fail-closed
+            if args.probe_rpc:
+                raise SystemExit(f"inventory base RPC probe failed: {error}") from error
+            rpc_probe = {"base_rpc_endpoint": None, "error": str(error)[:500]}
+
     if args.fixture:
         issues = load_fixture(args.fixture)
     else:
@@ -563,6 +612,8 @@ def main(argv: list[str] | None = None) -> int:
         load_claimable_report(args.claimable_report),
     )
     payload = asdict(report)
+    if rpc_probe is not None:
+        payload["base_rpc_probe"] = rpc_probe
     md = report.to_markdown()
 
     print(md)

scripts/check_routed_v3_activation_readiness.py

@@ -9,10 +9,21 @@
 
 import activate_routed_v3_dynamic as dynamic
 import activate_routed_v3_replacements as activation
+from _shared.rpc import select_working_base_rpc
+
+
+def _prefer_failover_base_rpc(rpc_url: str) -> str:
+    """Return the chain-valid HTTPS endpoint that actually passed validation."""
+    preferred = (rpc_url or "").strip()
+    try:
+        return select_working_base_rpc(preferred=preferred or None, max_retries=2)
+    except Exception:
+        return preferred
 
 
 def inspect(rpc_url: str, cast_bin: str) -> dict[str, object]:
     try:
+        rpc_url = _prefer_failover_base_rpc(rpc_url)
         cast = activation.Cast(cast_bin, rpc_url)
         deployment = dynamic.discover_deployment(cast)
         state = activation.policy_state(cast, deployment)