From 3d7758f40dc7ee7ed587afa302fc7b6c6ec4a97c Mon Sep 17 00:00:00 2001 From: lysrain21 Date: Thu, 20 Aug 2026 09:30:10 +0000 Subject: [PATCH] Harden native multihop campaign durability --- scripts/native_multihop_campaign.sh | 109 +++++++++++++--- scripts/run_native_multihop_switching.py | 33 +++-- src/xir_lab/localnet/compose.py | 12 ++ src/xir_lab/native/layerzero_worker.py | 68 ++++++---- .../native/multihop_hyperlane_observer.py | 118 ++++++++++-------- src/xir_lab/native/multihop_preflight.py | 33 ++++- src/xir_lab/native/multihop_runner.py | 16 +-- src/xir_lab/native/multihop_traces.py | 5 +- tests/unit/test_native_layerzero_worker.py | 12 +- ...test_native_multihop_hyperlane_observer.py | 40 ++++++ tests/unit/test_native_multihop_preflight.py | 27 ++++ tests/unit/test_native_multihop_runbook.py | 100 ++++++++++----- tests/unit/test_native_multihop_topology.py | 2 + 13 files changed, 427 insertions(+), 148 deletions(-) diff --git a/scripts/native_multihop_campaign.sh b/scripts/native_multihop_campaign.sh index 24e591d..d7b6ccc 100755 --- a/scripts/native_multihop_campaign.sh +++ b/scripts/native_multihop_campaign.sh @@ -63,6 +63,11 @@ case "$CAMPAIGN_KIND" in exit 2 ;; esac +STOP_AFTER_PHASE=${XIR_MULTIHOP_STOP_AFTER_PHASE:-} +if [[ -n $STOP_AFTER_PHASE && ! ( $CAMPAIGN_KIND == pilot && $STOP_AFTER_PHASE == smoke ) ]]; then + echo "XIR_MULTIHOP_STOP_AFTER_PHASE supports only pilot smoke verification" >&2 + exit 2 +fi TRACE_CONCURRENCY=8 TRACE_SETTLE_SECONDS=0 if [[ $CAMPAIGN_KIND == pilot ]]; then @@ -180,6 +185,25 @@ owned_process_alive() { --identity "$(identity_path_for_pidfile "$pidfile")" >/dev/null 2>&1 } +failed_protocol_writer() { + [[ $MODE == production ]] || return 1 + local name pidfile + for name in layerzero-worker relayer \ + validator-xirlocalchaina validator-xirlocalchainb validator-xirlocalchainc \ + validator-xirlocalchaind validator-xirlocalchaine; do + if [[ $name == layerzero-worker ]]; then + pidfile="$RUNTIME/pids/$name.pid" + else + pidfile="$RUNTIME/hyperlane/agents/pids/$name.pid" + fi + if ! owned_process_alive "$pidfile"; then + printf '%s\n' "$name" + return 0 + fi + done + return 1 +} + complete_pidfile() { local pidfile=$1 rm -f "$pidfile" "$(identity_path_for_pidfile "$pidfile")" @@ -657,20 +681,46 @@ trap 'exit $?' ERR start_lease_supervisor() { rm -f "$LEASE_SUPERVISOR_STOP" - ( - while [[ ! -f $LEASE_SUPERVISOR_STOP ]]; do - date +%s%N >"$RUNTIME/provenance/lease-supervisor-heartbeat-ns" - if [[ $MODE == production && -f $RUNTIME/provenance/lease-supervisor.ready \ - && -L $LEASE && -f $LEASE_TOKEN ]]; then - "$PY" "$REPO/scripts/native_multihop_lease.py" heartbeat \ - --runtime-root "$RUNTIME" --preregistration "$PREREG" \ - --lease "$LEASE" --token "$LEASE_TOKEN" - fi - sleep 1 - done - ) >"$RUNTIME/logs/lease-supervisor.log" 2>&1 & - LEASE_SUPERVISOR_PID=$! - disown "$LEASE_SUPERVISOR_PID" 2>/dev/null || true + local supervisor_pid_path="$RUNTIME/provenance/lease-supervisor.child.pid" + rm -f "$supervisor_pid_path" + nohup bash -c ' + stop_file=$1 + runtime=$2 + mode=$3 + ready=$4 + lease=$5 + token=$6 + prereg=$7 + python=$8 + repo=$9 + pid_path=${10} + ( + while [[ ! -f $stop_file ]]; do + date +%s%N >"$runtime/provenance/lease-supervisor-heartbeat-ns" + if [[ $mode == production && -f $ready && -L $lease && -f $token ]]; then + "$python" "$repo/scripts/native_multihop_lease.py" heartbeat \ + --runtime-root "$runtime" --preregistration "$prereg" \ + --lease "$lease" --token "$token" + fi + sleep 1 + done + ) & + child=$! + printf "%s\\n" "$child" >"$pid_path" + wait "$child" + ' _ "$LEASE_SUPERVISOR_STOP" "$RUNTIME" "$MODE" \ + "$RUNTIME/provenance/lease-supervisor.ready" "$LEASE" "$LEASE_TOKEN" \ + "$PREREG" "$PY" "$REPO" "$supervisor_pid_path" \ + >"$RUNTIME/logs/lease-supervisor.log" 2>&1 /dev/null || true + local pid_wait_deadline=$((SECONDS + 5)) + while [[ ! -s "$supervisor_pid_path" && SECONDS -lt pid_wait_deadline ]]; do sleep 0.05; done + [[ -s "$supervisor_pid_path" ]] || { + echo "lease supervisor did not publish its child PID" >&2 + return 1 + } + LEASE_SUPERVISOR_PID=$(<"$supervisor_pid_path") record_pid_identity "$RUNTIME/provenance/lease-supervisor.pid" \ "$LEASE_SUPERVISOR_PID" } @@ -1140,7 +1190,34 @@ run_phase() { --identity "$(identity_path_for_pidfile "$run/runner.pid")" --signal SIGCONT \ >/dev/null local runner_status=0 + local next_writer_health_check=0 while owned_process_alive "$run/runner.pid"; do + local writer_failure= + if (( SECONDS >= next_writer_health_check )); then + writer_failure=$(failed_protocol_writer || true) + next_writer_health_check=$((SECONDS + 5)) + fi + if [[ -n $writer_failure ]]; then + write_sidecar_failure_stop "$run/submission.stop" \ + "protocol_writer_process_failed:$writer_failure" + record "protocol-writer-failed-during-run:$phase:$writer_failure" + local writer_stop_deadline=$((SECONDS + 120)) + while owned_process_alive "$run/runner.pid" \ + && (( SECONDS < writer_stop_deadline )); do sleep 0.2; done + if owned_process_alive "$run/runner.pid"; then + "$PY" "$REPO/scripts/native_multihop_process_identity.py" signal \ + --identity "$(identity_path_for_pidfile "$run/runner.pid")" --signal SIGTERM \ + >/dev/null || true + fi + wait "$runner_pid" || true + complete_pidfile "$run/runner.pid" + touch "$run/monitor.stop" + wait "$monitor_pid" || true + complete_pidfile "$run/resource-monitor.pid" + quarantine_incomplete_resource_tail "$run" + RESUME_REQUIRED=1 + return 75 + fi if ! owned_process_alive "$run/hyperlane-observer.pid"; then local observer_status=0 wait "$observer_pid" || observer_status=$? @@ -1478,6 +1555,10 @@ production_main() { deploy_and_preflight failpoint preflight run_phase smoke + if [[ $STOP_AFTER_PHASE == smoke ]]; then + record "pilot-smoke-verification-complete" + return 0 + fi run_phase publication_smoke run_phase scale if [[ $CAMPAIGN_KIND == pilot ]]; then diff --git a/scripts/run_native_multihop_switching.py b/scripts/run_native_multihop_switching.py index 80ea6dc..06161a9 100644 --- a/scripts/run_native_multihop_switching.py +++ b/scripts/run_native_multihop_switching.py @@ -10,7 +10,7 @@ import rfc8785 from requests import RequestException -from web3.exceptions import Web3RPCError +from web3.exceptions import TimeExhausted, Web3RPCError from xir_lab.localnet.topology import LocalTopologyError from xir_lab.native.multihop_execution import verify_execution_authority @@ -48,9 +48,7 @@ def main() -> int: parser.add_argument("--root-signer-key-file", type=Path, required=True) parser.add_argument("--state", type=Path, required=True) parser.add_argument("--raw-root", type=Path, required=True) - parser.add_argument( - "--phase", choices=("smoke", "publication_smoke", "scale"), required=True - ) + parser.add_argument("--phase", choices=("smoke", "publication_smoke", "scale"), required=True) parser.add_argument("--submission-stop-file", type=Path) args = parser.parse_args() authority = verify_execution_authority( @@ -81,37 +79,29 @@ def main() -> int: "lease_identity_sha256": authority["lease_identity_sha256"], "lease_identity": authority["lease_identity"], "preflight_sha256": hashlib.sha256(args.preflight.read_bytes()).hexdigest(), - "preflight_semantic_sha256": json.loads( - args.preflight.read_text(encoding="utf-8") - )["semantic_sha256"], + "preflight_semantic_sha256": json.loads(args.preflight.read_text(encoding="utf-8"))[ + "semantic_sha256" + ], "validator_volume_attestation_sha256": preflight_document[ "validator_volume_attestation_sha256" ], "validator_volume_attestation_semantic_sha256": preflight_document[ "validator_volume_attestation_semantic_sha256" ], - "validator_volume_journal_sha256": preflight_document[ - "validator_volume_journal_sha256" - ], + "validator_volume_journal_sha256": preflight_document["validator_volume_journal_sha256"], "validator_volume_journal_semantic_sha256": preflight_document[ "validator_volume_journal_semantic_sha256" ], - "toolchain_preflight_sha256": preflight_document[ - "toolchain_preflight_sha256" - ], + "toolchain_preflight_sha256": preflight_document["toolchain_preflight_sha256"], "toolchain_preflight_semantic_sha256": preflight_document[ "toolchain_preflight_semantic_sha256" ], "config_sha256": hashlib.sha256(args.config.read_bytes()).hexdigest(), "plan_sha256": hashlib.sha256(args.plan.read_bytes()).hexdigest(), "deployment_sha256": hashlib.sha256(args.deployment.read_bytes()).hexdigest(), - "preregistration_sha256": hashlib.sha256( - args.preregistration.read_bytes() - ).hexdigest(), + "preregistration_sha256": hashlib.sha256(args.preregistration.read_bytes()).hexdigest(), } - phase_authority["semantic_sha256"] = hashlib.sha256( - rfc8785.dumps(phase_authority) - ).hexdigest() + phase_authority["semantic_sha256"] = hashlib.sha256(rfc8785.dumps(phase_authority)).hexdigest() serialized_authority = json.dumps(phase_authority, indent=2, sort_keys=True) + "\n" if args.phase_authority_output.exists(): if args.phase_authority_output.read_text(encoding="utf-8") != serialized_authority: @@ -152,6 +142,11 @@ def main() -> int: if "submissions stopped by resource monitor" in str(exc): raise SystemExit(75) from exc raise + except TimeExhausted as exc: + # The signed transaction is durable before receipt polling. Resume must + # re-check the same hash instead of treating a slow QBFT block as a + # terminal experiment failure. + raise SystemExit(75) from exc except (RequestException, Web3RPCError) as exc: if is_transient_rpc_error(exc): raise SystemExit(75) from exc diff --git a/src/xir_lab/localnet/compose.py b/src/xir_lab/localnet/compose.py index f815b4a..912d955 100644 --- a/src/xir_lab/localnet/compose.py +++ b/src/xir_lab/localnet/compose.py @@ -9,6 +9,11 @@ from xir_lab.localnet.topology import LocalIdentityManifest, LocalTopology +# The formal campaign can span more than 100,000 one-second QBFT blocks before +# postprocessing starts. Besu FULL sync retains Bonsai trie logs, but its +# default 512-block load limit still makes older transaction traces unavailable. +BONSAI_HISTORICAL_BLOCK_LIMIT = 250_000 + def _memory_megabytes(value: int) -> str: return f"{value // (1024 * 1024)}m" @@ -68,6 +73,13 @@ def render_compose( "--sync-min-peers=3", "--data-storage-format=BONSAI", ] + if validator_index == 1: + command.extend( + ( + "--bonsai-limit-trie-logs-enabled=false", + f"--bonsai-historical-block-limit={BONSAI_HISTORICAL_BLOCK_LIMIT}", + ) + ) data_mount: str | dict[str, str] = f"{runtime}/{data_path}:/data" if topology.validator_data_storage == "docker-volume": volumes[volume_name] = { diff --git a/src/xir_lab/native/layerzero_worker.py b/src/xir_lab/native/layerzero_worker.py index c6e5244..3ebb529 100644 --- a/src/xir_lab/native/layerzero_worker.py +++ b/src/xir_lab/native/layerzero_worker.py @@ -282,28 +282,52 @@ def intend_action( call_data: bytes, ) -> sqlite3.Row: action_id = "lz_" + hashlib.sha256(f"{guid}:{stage}".encode()).hexdigest()[:24] - self.connection.execute( - """ - INSERT INTO actions( - action_id, guid, stage, destination_chain_id, nonce, target, - calldata_bytes, calldata_hex, calldata_sha256, status, intended_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'intended', ?) - """, - ( - action_id, - guid, - stage, - destination_chain_id, - nonce, - target.lower(), - len(call_data), - "0x" + call_data.hex(), - hashlib.sha256(call_data).hexdigest(), - _now(), - ), - ) - self._record_observation(action_id=action_id, state="intended", details={}) - self.connection.commit() + outer_transaction = self.connection.in_transaction + savepoint = f"intend_action_{action_id}" + if outer_transaction: + self.connection.execute(f"SAVEPOINT {savepoint}") + else: + self.connection.execute("BEGIN IMMEDIATE") + try: + durable_row = self.connection.execute( + "SELECT MAX(nonce) FROM actions WHERE destination_chain_id = ?", + (destination_chain_id,), + ).fetchone() + durable_nonce = None if durable_row is None else durable_row[0] + if durable_nonce is not None: + nonce = max(nonce, int(durable_nonce) + 1) + self.connection.execute( + """ + INSERT INTO actions( + action_id, guid, stage, destination_chain_id, nonce, target, + calldata_bytes, calldata_hex, calldata_sha256, status, intended_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'intended', ?) + """, + ( + action_id, + guid, + stage, + destination_chain_id, + nonce, + target.lower(), + len(call_data), + "0x" + call_data.hex(), + hashlib.sha256(call_data).hexdigest(), + _now(), + ), + ) + self._record_observation(action_id=action_id, state="intended", details={}) + if outer_transaction: + self.connection.execute(f"RELEASE SAVEPOINT {savepoint}") + else: + self.connection.commit() + except BaseException: + if outer_transaction: + self.connection.execute(f"ROLLBACK TO SAVEPOINT {savepoint}") + self.connection.execute(f"RELEASE SAVEPOINT {savepoint}") + else: + self.connection.rollback() + raise return cast( sqlite3.Row, self.connection.execute( diff --git a/src/xir_lab/native/multihop_hyperlane_observer.py b/src/xir_lab/native/multihop_hyperlane_observer.py index 35d0008..a73fe00 100644 --- a/src/xir_lab/native/multihop_hyperlane_observer.py +++ b/src/xir_lab/native/multihop_hyperlane_observer.py @@ -6,6 +6,7 @@ import json import os import time +import urllib.error import urllib.request from pathlib import Path from typing import Any, cast @@ -36,14 +37,19 @@ def _rpc(url: str, method: str, params: list[Any]) -> Any: headers={"content-type": "application/json"}, method="POST", ) - try: - with urllib.request.urlopen(request, timeout=30) as response: - document = json.loads(response.read()) - except (OSError, json.JSONDecodeError) as exc: - raise LocalTopologyError(f"Hyperlane observer RPC failed: {method}") from exc - if not isinstance(document, dict) or document.get("error") is not None: - raise LocalTopologyError(f"Hyperlane observer RPC error: {method}") - return document.get("result") + last_error: BaseException | None = None + for attempt in range(4): + try: + with urllib.request.urlopen(request, timeout=30) as response: + document = json.loads(response.read()) + if not isinstance(document, dict) or document.get("error") is not None: + raise LocalTopologyError(f"Hyperlane observer RPC error: {method}") + return document.get("result") + except (OSError, urllib.error.HTTPError, json.JSONDecodeError, LocalTopologyError) as exc: + last_error = exc + if attempt < 3: + time.sleep(0.25 * (2**attempt)) + raise LocalTopologyError(f"Hyperlane observer RPC failed: {method}") from last_error def _relayer_pid(pid_path: Path) -> int: @@ -64,7 +70,13 @@ def _pending_transactions(rpc_url: str) -> list[dict[str, Any]]: full transaction objects needed by this observer. """ - pending_block = _rpc(rpc_url, "eth_getBlockByNumber", ["pending", True]) + try: + pending_block = _rpc(rpc_url, "eth_getBlockByNumber", ["pending", True]) + except LocalTopologyError: + # Pending-block RPC is an optional submission hint. A busy Besu node + # may reset this expensive full-transaction request; mined-block + # scanning below remains the authoritative, explicitly labeled fallback. + return [] if not isinstance(pending_block, dict) or not isinstance( pending_block.get("transactions"), list ): @@ -202,33 +214,13 @@ def observe_hyperlane_relayer( while True: _relayer_pid(pid_path) for role, rpc_url in role_rpc.items(): - for transaction in _pending_transactions(rpc_url): - transaction_hash = str(transaction.get("hash", "")).lower() - if ( - str(transaction.get("from", "")).lower() == relayer_address - and transaction_hash not in seen_submitted - ): - _append_event( - output_path, - event="submitted_observed", - runtime_root=runtime_root, - relayer_pid_path=pid_path, - chain_role=role, - transaction_hash=transaction_hash, - source="eth_getBlockByNumber:pending", - ) - seen_submitted.add(transaction_hash) - head = int(str(_rpc(rpc_url, "eth_blockNumber", [])), 16) - while next_block[role] <= head: - block_number = next_block[role] - block = _rpc(rpc_url, "eth_getBlockByNumber", [hex(block_number), True]) - if not isinstance(block, dict) or not isinstance(block.get("transactions"), list): - raise LocalTopologyError("Hyperlane observer block is invalid") - for transaction in cast(list[dict[str, Any]], block["transactions"]): - if str(transaction.get("from", "")).lower() != relayer_address: - continue + try: + for transaction in _pending_transactions(rpc_url): transaction_hash = str(transaction.get("hash", "")).lower() - if transaction_hash not in seen_submitted: + if ( + str(transaction.get("from", "")).lower() == relayer_address + and transaction_hash not in seen_submitted + ): _append_event( output_path, event="submitted_observed", @@ -236,23 +228,49 @@ def observe_hyperlane_relayer( relayer_pid_path=pid_path, chain_role=role, transaction_hash=transaction_hash, - block_number=block_number, - source="mined_block_fallback", + source="eth_getBlockByNumber:pending", ) seen_submitted.add(transaction_hash) - if transaction_hash not in seen_mined: - _append_event( - output_path, - event="mined_observed", - runtime_root=runtime_root, - relayer_pid_path=pid_path, - chain_role=role, - transaction_hash=transaction_hash, - block_number=block_number, - source="eth_getBlockByNumber", - ) - seen_mined.add(transaction_hash) - next_block[role] += 1 + head = int(str(_rpc(rpc_url, "eth_blockNumber", [])), 16) + while next_block[role] <= head: + block_number = next_block[role] + block = _rpc(rpc_url, "eth_getBlockByNumber", [hex(block_number), True]) + if not isinstance(block, dict) or not isinstance(block.get("transactions"), list): + raise LocalTopologyError("Hyperlane observer block is invalid") + for transaction in cast(list[dict[str, Any]], block["transactions"]): + if str(transaction.get("from", "")).lower() != relayer_address: + continue + transaction_hash = str(transaction.get("hash", "")).lower() + if transaction_hash not in seen_submitted: + _append_event( + output_path, + event="submitted_observed", + runtime_root=runtime_root, + relayer_pid_path=pid_path, + chain_role=role, + transaction_hash=transaction_hash, + block_number=block_number, + source="mined_block_fallback", + ) + seen_submitted.add(transaction_hash) + if transaction_hash not in seen_mined: + _append_event( + output_path, + event="mined_observed", + runtime_root=runtime_root, + relayer_pid_path=pid_path, + chain_role=role, + transaction_hash=transaction_hash, + block_number=block_number, + source="eth_getBlockByNumber", + ) + seen_mined.add(transaction_hash) + next_block[role] += 1 + except LocalTopologyError: + # A transient RPC reset on one chain must not terminate the + # observer. The next poll retries that chain from its durable + # next_block cursor, preserving mined-block fallback evidence. + continue if stop_file.exists(): if target_blocks is None: if target_blocks_path is not None and target_blocks_path.is_file(): diff --git a/src/xir_lab/native/multihop_preflight.py b/src/xir_lab/native/multihop_preflight.py index aa06133..1461ed9 100644 --- a/src/xir_lab/native/multihop_preflight.py +++ b/src/xir_lab/native/multihop_preflight.py @@ -56,6 +56,36 @@ def _canonical_transaction_hash(value: str) -> str: return "0x" + value.lower().removeprefix("0x") +def _validate_trace_probe(response: Any, *, role: str) -> None: + if not isinstance(response, dict) or response.get("error") is not None: + raise LocalTopologyError(f"Besu TRACE unavailable: {role}") + result = response.get("result") + if not isinstance(result, list) or not result: + raise LocalTopologyError(f"Besu TRACE returned an empty result: {role}") + roots = [ + row + for row in result + if isinstance(row, dict) and row.get("traceAddress") == [] + ] + if len(roots) != 1: + raise LocalTopologyError(f"Besu TRACE lacks one root trace: {role}") + root_result = roots[0].get("result") + if not isinstance(root_result, dict) or "gasUsed" not in root_result: + raise LocalTopologyError(f"Besu TRACE root gas is unavailable: {role}") + gas_used = root_result["gasUsed"] + try: + if isinstance(gas_used, int): + parsed_gas = gas_used + elif isinstance(gas_used, str): + parsed_gas = int(gas_used, 16) if gas_used.startswith("0x") else int(gas_used) + else: + raise TypeError("unsupported gas quantity") + except (TypeError, ValueError) as exc: + raise LocalTopologyError(f"Besu TRACE root gas is invalid: {role}") from exc + if parsed_gas < 0: + raise LocalTopologyError(f"Besu TRACE root gas is invalid: {role}") + + def _load(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) @@ -381,8 +411,7 @@ def run_multihop_preflight( trace = client.provider.make_request( RPCEndpoint("trace_transaction"), [str(sample["transaction_hash"])] ) - if trace.get("error") is not None or not isinstance(trace.get("result"), list): - raise LocalTopologyError(f"Besu TRACE unavailable: {role}") + _validate_trace_probe(trace, role=role) raw_response = client.provider.make_request( RPCEndpoint(BESU_RAW_TRANSACTION_RPC_METHOD), [str(sample["transaction_hash"])], diff --git a/src/xir_lab/native/multihop_runner.py b/src/xir_lab/native/multihop_runner.py index 1235065..69d9853 100644 --- a/src/xir_lab/native/multihop_runner.py +++ b/src/xir_lab/native/multihop_runner.py @@ -544,15 +544,11 @@ def __init__( "validator_volume_attestation_semantic_sha256": self.preflight[ "validator_volume_attestation_semantic_sha256" ], - "validator_volume_journal_sha256": self.preflight[ - "validator_volume_journal_sha256" - ], + "validator_volume_journal_sha256": self.preflight["validator_volume_journal_sha256"], "validator_volume_journal_semantic_sha256": self.preflight[ "validator_volume_journal_semantic_sha256" ], - "toolchain_preflight_sha256": self.preflight[ - "toolchain_preflight_sha256" - ], + "toolchain_preflight_sha256": self.preflight["toolchain_preflight_sha256"], "toolchain_preflight_semantic_sha256": self.preflight[ "toolchain_preflight_semantic_sha256" ], @@ -818,8 +814,7 @@ def _run_batch(self, attempts: list[MultihopAttempt]) -> None: runtime_root=self.runtime_root, preregistration_path=self.preregistration_path, ) - if self.submission_stop_file is not None and self.submission_stop_file.exists(): - raise LocalTopologyError("multihop submissions stopped by resource monitor") + self._raise_if_submissions_stopped() with ThreadPoolExecutor(max_workers=self.concurrency) as pool: futures = [pool.submit(self._run_if_needed, attempt) for attempt in attempts] for future in futures: @@ -1285,6 +1280,7 @@ def _wait_bundle( ), ] while time.monotonic() < deadline: + self._raise_if_submissions_stopped() if all( adapter.functions.verify( receipt.profile_hash, @@ -1311,6 +1307,10 @@ def _wait_bundle( f"timed out waiting for multihop bundle at {destination_role}:{adapter_role}" ) + def _raise_if_submissions_stopped(self) -> None: + if self.submission_stop_file is not None and self.submission_stop_file.exists(): + raise LocalTopologyError("multihop submissions stopped by resource monitor") + @staticmethod def _envelope( record: XIRRecord, diff --git a/src/xir_lab/native/multihop_traces.py b/src/xir_lab/native/multihop_traces.py index 971ece6..265c318 100644 --- a/src/xir_lab/native/multihop_traces.py +++ b/src/xir_lab/native/multihop_traces.py @@ -558,7 +558,8 @@ def capture_multihop_traces( config_document = json.loads(config_path.read_text(encoding="utf-8")) pilot_trace_fallback = ( - isinstance(config_document, dict) + phase == "scale" + and isinstance(config_document, dict) and config_document.get("result_roles", {}).get("scale") == "pilot_diagnostic_only" ) @@ -583,7 +584,7 @@ def fetch(item: tuple[str, str, int]) -> dict[str, Any]: trace_result = _trace_with_retry( role_rpc[role], transaction_hash, - attempts=1 if pilot_trace_fallback else 8, + attempts=8, ) if trace_result is None and pilot_trace_fallback: trace = { diff --git a/tests/unit/test_native_layerzero_worker.py b/tests/unit/test_native_layerzero_worker.py index 5438491..856537d 100644 --- a/tests/unit/test_native_layerzero_worker.py +++ b/tests/unit/test_native_layerzero_worker.py @@ -57,11 +57,21 @@ def test_worker_state_is_resumable_and_intent_precedes_signature(tmp_path: Path) call_data=b"call", ) assert action["calldata_bytes"] == 4 + assert action["nonce"] == 7 + resumed_action = state.intend_action( + guid="0x" + packet.guid.hex(), + stage="commit_verification", + destination_chain_id=3133702, + nonce=7, + target="0x" + "55" * 20, + call_data=b"resumed", + ) + assert resumed_action["nonce"] == 8 state.record_signed(str(action["action_id"]), b"signed", "0xdead") observations = state.connection.execute( "SELECT state FROM observations ORDER BY observation_id" ).fetchall() - assert [row["state"] for row in observations] == ["intended", "signed"] + assert [row["state"] for row in observations] == ["intended", "intended", "signed"] assert state.cursor(49001, 12) == 12 state.advance_cursor(49001, 200) assert state.cursor(49001, 12) == 200 diff --git a/tests/unit/test_native_multihop_hyperlane_observer.py b/tests/unit/test_native_multihop_hyperlane_observer.py index 2d2dd38..8b7a1a5 100644 --- a/tests/unit/test_native_multihop_hyperlane_observer.py +++ b/tests/unit/test_native_multihop_hyperlane_observer.py @@ -11,6 +11,7 @@ from xir_lab.native.multihop_hyperlane_observer import ( SCHEMA, _pending_transactions, + _rpc, load_hyperlane_observer_events, observe_hyperlane_relayer, ) @@ -70,6 +71,45 @@ def _event( } +def test_pending_transactions_treats_exhausted_rpc_disconnect_as_empty( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "xir_lab.native.multihop_hyperlane_observer._rpc", + lambda *_args: (_ for _ in ()).throw(LocalTopologyError("disconnect")), + ) + assert _pending_transactions("http://rpc") == [] + + +def test_rpc_retries_transient_disconnect_before_returning( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = 0 + + def urlopen(_request: object, timeout: int) -> object: + nonlocal calls + calls += 1 + if calls < 3: + raise ConnectionResetError("transient disconnect") + + class Response: + def __enter__(self) -> "Response": + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return b'{"jsonrpc":"2.0","id":1,"result":[]}' + + return Response() + + monkeypatch.setattr("urllib.request.urlopen", urlopen) + monkeypatch.setattr("time.sleep", lambda _seconds: None) + assert _rpc("http://rpc", "eth_getBlockByNumber", ["pending", True]) == [] + assert calls == 3 + + def test_process_capture_requires_stable_observer_and_relayer_identity( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/test_native_multihop_preflight.py b/tests/unit/test_native_multihop_preflight.py index 001d885..d4c7f3a 100644 --- a/tests/unit/test_native_multihop_preflight.py +++ b/tests/unit/test_native_multihop_preflight.py @@ -14,6 +14,7 @@ from xir_lab.native.multihop_preflight import ( _require_review_closure, _semantic_sha256, + _validate_trace_probe, preregistration_review_payload_sha256, verify_multihop_preflight_document, verify_multihop_review_gate, @@ -27,6 +28,32 @@ def _write(path: Path, value: object) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() +def test_trace_probe_requires_nonempty_root_with_gas() -> None: + valid = { + "result": [ + { + "action": {"callType": "call"}, + "result": {"gasUsed": "0x5208"}, + "traceAddress": [], + "type": "call", + } + ] + } + _validate_trace_probe(valid, role="a") + + with pytest.raises(LocalTopologyError, match="empty result"): + _validate_trace_probe({"result": []}, role="a") + with pytest.raises(LocalTopologyError, match="one root trace"): + _validate_trace_probe( + {"result": [{"result": {"gasUsed": "0x1"}, "traceAddress": [0]}]}, + role="a", + ) + with pytest.raises(LocalTopologyError, match="root gas"): + _validate_trace_probe( + {"result": [{"result": {}, "traceAddress": []}]}, role="a" + ) + + def test_review_gate_fails_closed_until_pass_digest_exists(tmp_path: Path) -> None: preregistration = { "status": "preregistered_local_implementation_pending_review", diff --git a/tests/unit/test_native_multihop_runbook.py b/tests/unit/test_native_multihop_runbook.py index 56eee37..0d729ca 100644 --- a/tests/unit/test_native_multihop_runbook.py +++ b/tests/unit/test_native_multihop_runbook.py @@ -66,6 +66,8 @@ def test_runbook_requires_one_fail_closed_versioned_executable() -> None: assert "XIR_MULTIHOP_HOST_LEASE_BASE" in script assert 'HOST_LEASE_BASE="/run/lock/xir-lab-runtime-leases"' in script assert 'HOST_LEASE_BASE="/run/user/' not in script + assert "nohup bash -c '" in script + assert 'disown "$reaper_pid"' in script assert 'export PYTHONPATH="$REPO/src"' in script assert "--target-blocks" in script and "--completion" in script assert "wait_for_durable_ready" in script @@ -76,15 +78,19 @@ def test_runbook_requires_one_fail_closed_versioned_executable() -> None: assert "campaign-signal-resume-required" in script assert "hyperlane_observer_process_failed" in script assert "resource_monitor_process_failed" in script + assert "protocol_writer_process_failed" in script + assert "protocol-writer-failed-during-run" in script + assert "failed_protocol_writer" in script + entrypoint = (ROOT / "scripts/run_native_multihop_switching.py").read_text(encoding="utf-8") + assert "TimeExhausted" in entrypoint + assert "raise SystemExit(75) from exc" in entrypoint assert "SECONDS + 150" in script assert "hyperlane-observer-ready" in script assert "resource-monitor-ready" in script assert "resource-monitor-failed-during-run" in script assert "stage_native_multihop_validator_volumes.py" in script - assert script.index("stage_native_multihop_validator_volumes.py") < script.index( - "up --detach" - ) - assert 'down --remove-orphans' in script + assert script.index("stage_native_multihop_validator_volumes.py") < script.index("up --detach") + assert "down --remove-orphans" in script assert "validator_containers_absent" in script assert "validator_volume_command remove-existing" in script assert "validator_volume_command verify-existing" in script @@ -117,16 +123,20 @@ def test_cleanup_distinguishes_natural_exit_from_live_identity_mismatch() -> Non # Identity verification remains mandatory for every live process, but a # process that exits between kill(0) and pidfd verification is successful # cleanup rather than a live identity mismatch. - assert '>/dev/null 2>&1' in campaign + assert ">/dev/null 2>&1" in campaign assert 'owned_process_alive "$pidfile" || {\n if ! alive "$pid"; then' in campaign assert campaign.count('if ! alive "$pid"; then\n complete_pidfile "$pidfile"') >= 3 - assert '>/dev/null 2>&1' in processes - assert 'if kill -0 "$pid" 2>/dev/null; then\n echo "$label exited; reused PID left untouched"' in processes - assert '>/dev/null 2>&1' in agents + assert ">/dev/null 2>&1" in processes + assert ( + 'if kill -0 "$pid" 2>/dev/null; then\n echo "$label exited; reused PID left untouched"' + in processes + ) + assert ">/dev/null 2>&1" in agents assert ( 'if ! "$python" "$identity_cli" signal' in agents and 'if kill -0 "$pid" 2>/dev/null; then\n' - ' echo "$name PID identity mismatch; refusing signal"' in agents + ' echo "$name PID identity mismatch; refusing signal"' + in agents ) @@ -143,8 +153,8 @@ def test_cleanup_stops_runner_before_resource_monitor_and_rechecks_tail() -> Non 'stop_resource_monitor_pidfile "$run_root"' ) assert "resource_completion_covers_runner_tail" in script - assert 'SELECT COALESCE(MAX(utc_ns),0) FROM events' in script - assert '.last_utc_ns >= $runner_utc' in script + assert "SELECT COALESCE(MAX(utc_ns),0) FROM events" in script + assert ".last_utc_ns >= $runner_utc" in script assert "quarantine_resource_segment" in script @@ -153,8 +163,7 @@ def test_production_publication_namespace_is_derived_from_runtime_identity() -> assert 'RUN_ID=$(basename "$RUNTIME")' in script assert "production runtime basename must be run-NNN" in script assert ( - "local local_public=$WORKSPACE/experiment-results/" - "native-multihop-switching-v1/$RUN_ID" + "local local_public=$WORKSPACE/experiment-results/native-multihop-switching-v1/$RUN_ID" ) in script for terminal_run in ("run-001", "run-002", "run-003"): assert ( @@ -171,8 +180,8 @@ def test_formal_compose_project_matches_the_frozen_topology_identity() -> None: topology = json.loads(MULTIHOP_TOPOLOGY.read_text(encoding="utf-8")) project = topology["project_name"] assert project == "xir-native-multihop-v1" - assert 'COMPOSE_PROJECT=$(jq -er' in script - assert '[[ $COMPOSE_PROJECT == xir-native-multihop-v1 ]]' in script + assert "COMPOSE_PROJECT=$(jq -er" in script + assert "[[ $COMPOSE_PROJECT == xir-native-multihop-v1 ]]" in script compose_commands = re.findall( r"docker compose --project-name ([^ \\\n]+)", script, @@ -193,6 +202,33 @@ def test_formal_compose_project_matches_the_frozen_topology_identity() -> None: ) +def test_smoke_only_stop_is_restricted_to_pilot_verification(tmp_path: Path) -> None: + script = CAMPAIGN.read_text(encoding="utf-8") + assert "XIR_MULTIHOP_STOP_AFTER_PHASE supports only pilot smoke verification" in script + smoke = script.index(" run_phase smoke\n", script.index("production_main()")) + gate = script.index(" if [[ $STOP_AFTER_PHASE == smoke ]]; then\n", smoke) + publication_smoke = script.index(" run_phase publication_smoke\n", smoke) + assert smoke < gate < publication_smoke + assert 'record "pilot-smoke-verification-complete"' in script[gate:publication_smoke] + assert " return 0\n" in script[gate:publication_smoke] + + runtime = tmp_path / "run-902" + result = subprocess.run( + [str(CAMPAIGN), str(ROOT.parent), str(runtime)], + text=True, + capture_output=True, + check=False, + env={ + **os.environ, + "XIR_MULTIHOP_CAMPAIGN_KIND": "formal", + "XIR_MULTIHOP_STOP_AFTER_PHASE": "smoke", + }, + ) + assert result.returncode == 2 + assert "supports only pilot smoke verification" in result.stderr + assert not runtime.exists() + + def test_production_entry_rejects_nonversioned_runtime_before_writes( tmp_path: Path, ) -> None: @@ -232,8 +268,7 @@ def test_volume_bootstrap_direct_entry_requires_review_and_live_lease_before_sta str(runtime / "compose-must-not-be-read.yaml"), "--preregistration", str( - ROOT.parent - / "openspec/changes/measure-multihop-switching-scalability/" + ROOT.parent / "openspec/changes/measure-multihop-switching-scalability/" "artifacts/preregistration-v1.json" ), "--review-gate", @@ -255,7 +290,13 @@ def test_volume_bootstrap_direct_entry_requires_review_and_live_lease_before_sta env={**os.environ, "PYTHONPATH": str(ROOT / "src")}, ) assert result.returncode != 0 - assert "independent review closure has not enabled execution" in result.stderr + assert any( + message in result.stderr + for message in ( + "independent review closure has not enabled execution", + "reviewed implementation source digest drift", + ) + ) assert not runtime.exists() @@ -284,8 +325,7 @@ def test_live_preflight_requires_review_and_lease_before_volume_probe_or_output( str(runtime / "must-not-read-deployment.json"), "--preregistration", str( - ROOT.parent - / "openspec/changes/measure-multihop-switching-scalability/" + ROOT.parent / "openspec/changes/measure-multihop-switching-scalability/" "artifacts/preregistration-v1.json" ), "--review-gate", @@ -307,7 +347,13 @@ def test_live_preflight_requires_review_and_lease_before_volume_probe_or_output( env={**os.environ, "PYTHONPATH": str(ROOT / "src")}, ) assert result.returncode != 0 - assert "independent review closure has not enabled execution" in result.stderr + assert any( + message in result.stderr + for message in ( + "independent review closure has not enabled execution", + "reviewed implementation source digest drift", + ) + ) assert not runtime.exists() @@ -318,13 +364,9 @@ def test_every_runbook_shell_block_and_campaign_are_syntactically_valid() -> Non for block in blocks: subprocess.run(["bash", "-n"], input=block, text=True, check=True) subprocess.run(["bash", "-n", str(CAMPAIGN)], check=True) - direct_python_lines = [ - line for line in runbook.splitlines() if ".venv/bin/python" in line - ] + direct_python_lines = [line for line in runbook.splitlines() if ".venv/bin/python" in line] assert len(direct_python_lines) == 3 - assert all( - line.startswith('PYTHONPATH="$REPO/src" ') for line in direct_python_lines - ) + assert all(line.startswith('PYTHONPATH="$REPO/src" ') for line in direct_python_lines) @pytest.mark.parametrize("script", (PROVISION, WORKER_ROLES)) @@ -523,9 +565,7 @@ def test_layerzero_worker_cannot_omit_runtime_authority_for_five_chain_config( def test_layerzero_worker_rejects_five_chain_config_schema_downgrade_before_state( tmp_path: Path, schema_version: str | None ) -> None: - document: dict[str, object] = { - "chains": [{"chain_id": index} for index in range(5)] - } + document: dict[str, object] = {"chains": [{"chain_id": index} for index in range(5)]} if schema_version is not None: document["schema_version"] = schema_version config = tmp_path / "worker-config.json" diff --git a/tests/unit/test_native_multihop_topology.py b/tests/unit/test_native_multihop_topology.py index 6ed16cd..5fec0d0 100644 --- a/tests/unit/test_native_multihop_topology.py +++ b/tests/unit/test_native_multihop_topology.py @@ -36,6 +36,8 @@ def test_five_chain_topology_renders_twenty_isolated_validators(tmp_path: Path) assert first_sha == second_sha text = first.decode() assert text.count("org.xir.validator-id") == 20 + assert text.count("--bonsai-historical-block-limit=250000") == 5 + assert text.count("--bonsai-limit-trie-logs-enabled=false") == 5 assert "127.0.0.1:58545:8545" in text assert "local-chain-e-v4" in text