Harden native multihop campaign durability

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 &
+  local reaper_pid=$!
+  disown "$reaper_pid" 2>/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

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

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] = {

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(