feat(bench): route PR benchmarks through txgen

.github/scripts/bench-reth-summary.py

@@ -521,6 +521,7 @@ def main():
     parser.add_argument("--warmup-blocks", default=None, help="Number of warmup blocks")
     parser.add_argument("--wait-time", default=None, help="Wait time interval used between blocks")
     parser.add_argument("--bal-mode", default=None, help="BAL mode (true, feature, baseline)")
+    parser.add_argument("--driver", default=None, help="Benchmark driver used for this run")
     parser.add_argument("--grafana-url", default=None, help="Grafana dashboard URL for this benchmark run")
     args = parser.parse_args()
 
@@ -605,6 +606,7 @@ def main():
 
     summary = {
         "blocks": paired_stats["blocks"],
+        "driver": args.driver,
         "big_blocks": args.big_blocks,
         "warmup_blocks": args.warmup_blocks,
         "wait_time": args.wait_time,

.github/scripts/bench-txgen-build.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#
+# Builds the node binary for the txgen-backed PR benchmark path.
+#
+# Usage: bench-txgen-build.sh <baseline|feature> <source-dir> <commit>
+#
+# This intentionally does not build or install reth-bench. Big-block benchmarks
+# still use the legacy reth-bench path because txgen does not yet replay the
+# reth-bb payload/env-switch/BAL format.
+set -euxo pipefail
+
+MODE="$1"
+SOURCE_DIR="$2"
+COMMIT="$3"
+
+if [ "${BENCH_BIG_BLOCKS:-false}" = "true" ]; then
+  echo "::error::txgen path does not support big-block benchmarks yet; use the reth-bench driver"
+  exit 1
+fi
+
+EXTRA_FEATURES=""
+EXTRA_RUSTFLAGS=""
+if [ "${BENCH_TRACY:-off}" != "off" ]; then
+  EXTRA_FEATURES="tracy,tracy-client/ondemand"
+  EXTRA_RUSTFLAGS=" -C force-frame-pointers=yes"
+fi
+
+build_node_binary() {
+  local features_arg=""
+  local workspace_arg=""
+
+  cd "$SOURCE_DIR"
+  if [ -n "$EXTRA_FEATURES" ]; then
+    features_arg="--features ${EXTRA_FEATURES}"
+    workspace_arg="--workspace"
+  fi
+
+  # shellcheck disable=SC2086
+  RUSTFLAGS="-C target-cpu=native${EXTRA_RUSTFLAGS}" \
+    cargo build --locked --profile profiling --bin reth $workspace_arg $features_arg
+}
+
+case "$MODE" in
+  baseline|main)
+    echo "Building baseline reth (${COMMIT}) from source for txgen benchmark..."
+    build_node_binary
+    ;;
+
+  feature|branch)
+    echo "Building feature reth (${COMMIT}) from source for txgen benchmark..."
+    rustup show active-toolchain || rustup default stable
+    build_node_binary
+    ;;
+
+  *)
+    echo "Usage: $0 <baseline|feature> <source-dir> <commit>"
+    exit 1
+    ;;
+esac

.github/scripts/bench-txgen-extract.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Extract raw blocks for txgen send-blocks.
+
+This intentionally uses a numeric `debug_getRawBlock` parameter because some
+public RPC providers reject the hex quantity form that alloy's Debug API emits.
+"""
+
+import argparse
+import json
+import sys
+import time
+import urllib.error
+import urllib.request
+
+
+def rpc_call(url: str, method: str, params: list, retries: int = 12):
+    payload = json.dumps({"jsonrpc": "2.0", "method": method, "params": params, "id": 1}).encode()
+    last_err = None
+    for attempt in range(retries):
+        try:
+            req = urllib.request.Request(
+                url,
+                data=payload,
+                headers={"Content-Type": "application/json", "User-Agent": "reth-bench-txgen"},
+                method="POST",
+            )
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                data = json.loads(resp.read())
+            if "error" in data:
+                raise RuntimeError(data["error"].get("message", data["error"]))
+            return data["result"]
+        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, RuntimeError) as err:
+            last_err = err
+            if attempt + 1 == retries:
+                break
+            time.sleep(min(10.0, 0.5 * (2**attempt)))
+    raise RuntimeError(last_err)
+
+
+def parse_quantity(value: str) -> int:
+    return int(value, 16) if isinstance(value, str) and value.startswith("0x") else int(value)
+
+
+def validate_top_level_rlp(raw: str) -> None:
+    if not isinstance(raw, str) or not raw.startswith("0x"):
+        raise ValueError("raw block is not a hex string")
+    data = bytes.fromhex(raw[2:])
+    if not data:
+        raise ValueError("raw block is empty")
+
+    prefix = data[0]
+    if prefix <= 0x7F:
+        total = 1
+    elif prefix <= 0xB7:
+        total = 1 + prefix - 0x80
+    elif prefix <= 0xBF:
+        len_len = prefix - 0xB7
+        total = 1 + len_len + int.from_bytes(data[1 : 1 + len_len], "big")
+    elif prefix <= 0xF7:
+        total = 1 + prefix - 0xC0
+    else:
+        len_len = prefix - 0xF7
+        total = 1 + len_len + int.from_bytes(data[1 : 1 + len_len], "big")
+
+    if total != len(data):
+        raise ValueError(f"raw block RLP length mismatch: expected {total} bytes, got {len(data)}")
+
+
+def fetch_raw_block(url: str, number: int, retries: int = 12) -> str:
+    last_err = None
+    for attempt in range(retries):
+        try:
+            raw = rpc_call(url, "debug_getRawBlock", [number], retries=1)
+            validate_top_level_rlp(raw)
+            return raw
+        except (RuntimeError, ValueError) as err:
+            last_err = err
+            if attempt + 1 == retries:
+                break
+            time.sleep(min(10.0, 0.5 * (2**attempt)))
+    raise RuntimeError(last_err)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--rpc", required=True)
+    parser.add_argument("--metadata-rpc")
+    parser.add_argument("--from", dest="from_block", type=int, required=True)
+    parser.add_argument("--to", dest="to_block", type=int, required=True)
+    parser.add_argument("-o", "--output", required=True)
+    args = parser.parse_args()
+
+    if args.from_block > args.to_block:
+        parser.error("--from must be <= --to")
+
+    metadata_rpc = args.metadata_rpc or args.rpc
+    total = args.to_block - args.from_block + 1
+    started = time.monotonic()
+    last_log = started
+
+    with open(args.output, "w", encoding="utf-8") as out:
+        for idx, number in enumerate(range(args.from_block, args.to_block + 1), start=1):
+            try:
+                raw = fetch_raw_block(args.rpc, number)
+                block = rpc_call(metadata_rpc, "eth_getBlockByNumber", [hex(number), False])
+            except RuntimeError as err:
+                print(f"failed to fetch block {number}: {err}", file=sys.stderr)
+                return 1
+
+            if not raw or not block:
+                print(f"missing block {number}", file=sys.stderr)
+                return 1
+
+            line = {
+                "raw": raw,
+                "key": block["hash"],
+                "number": parse_quantity(block["number"]),
+                "timestamp": parse_quantity(block["timestamp"]),
+                "gas_used": parse_quantity(block["gasUsed"]),
+                "gas_limit": parse_quantity(block["gasLimit"]),
+                "tx_count": len(block.get("transactions", [])),
+            }
+            out.write(json.dumps(line, separators=(",", ":")) + "\n")
+
+            now = time.monotonic()
+            if idx == total or idx % 100 == 0 or now - last_log >= 5:
+                elapsed = now - started
+                rate = idx / elapsed if elapsed else 0
+                print(f"extracted {idx}/{total} blocks ({idx / total * 100:.1f}%) - {rate:.0f} blocks/s", file=sys.stderr)
+                last_log = now
+
+    print(f"wrote {total} blocks to {args.output}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/bench-txgen-install.sh

@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+#
+# Installs the txgen tools used by the txgen-backed PR benchmark path.
+# Keep this separate from bench-reth-build.sh so scheduled benchmarks can keep
+# using the legacy reth-bench runner until they are migrated explicitly.
+#
+# Required env:
+#   TXGEN_REV   – pinned txgen git revision
+# Optional env:
+#   TXGEN_REPO  – txgen repository URL (default: https://github.com/tempoxyz/txgen)
+set -euxo pipefail
+
+: "${TXGEN_REV:?TXGEN_REV must be set to a pinned txgen revision}"
+
+TXGEN_REPO="${TXGEN_REPO:-https://github.com/tempoxyz/txgen}"
+
+# txgen is private. Prefer the deploy key secret; fall back to token auth for
+# local/manual runs. Use the git CLI so cargo honors the auth configuration.
+if [ -n "${TXGEN_DEPLOY_KEY:-}" ]; then
+  set +x
+  mkdir -p "$HOME/.ssh"
+  printf '%s\n' "$TXGEN_DEPLOY_KEY" > "$HOME/.ssh/txgen_deploy_key"
+  chmod 600 "$HOME/.ssh/txgen_deploy_key"
+  ssh-keyscan github.com >> "$HOME/.ssh/known_hosts" 2>/dev/null
+  export GIT_SSH_COMMAND="ssh -i $HOME/.ssh/txgen_deploy_key -o IdentitiesOnly=yes"
+  set -x
+  TXGEN_REPO="${TXGEN_SSH_REPO:-ssh://[email protected]/tempoxyz/txgen.git}"
+elif [ -n "${TXGEN_TOKEN:-${GH_PROJECT_TOKEN:-${DEREK_PAT:-${DEREK_TOKEN:-}}}}" ]; then
+  AUTH_TOKEN="${TXGEN_TOKEN:-${GH_PROJECT_TOKEN:-${DEREK_PAT:-${DEREK_TOKEN:-}}}}"
+  set +x
+  git config --global url."https://x-access-token:${AUTH_TOKEN}@github.com/".insteadOf "https://github.com/"
+  set -x
+fi
+export CARGO_NET_GIT_FETCH_WITH_CLI=true
+
+cargo install --git "$TXGEN_REPO" --rev "$TXGEN_REV" txgen-ethereum --bin txgen-ethereum --locked
+cargo install --git "$TXGEN_REPO" --rev "$TXGEN_REV" bench-cli --bin bench --locked

.github/scripts/bench-txgen-report-to-reth-csv.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Convert txgen `bench send-blocks` JSON into the legacy reth-bench CSVs.
+
+The PR benchmark rendering pipeline still consumes `combined_latency.csv` and
+`total_gas.csv`. This adapter lets the txgen-backed runner reuse the existing
+summary/charts/slack code while we migrate those consumers to txgen JSON.
+"""
+
+import argparse
+import csv
+import json
+from pathlib import Path
+
+
+def opt_int(value, default=None):
+    if value is None:
+        return default
+    return int(value)
+
+
+def block_latency_us(block: dict) -> tuple[int, int, int]:
+    # txgen currently records server newPayload latency in microseconds but
+    # client-side forkchoiceUpdated latency in milliseconds.
+    new_payload_us = opt_int(block.get("new_payload_server_latency_us"))
+    if new_payload_us is None:
+        new_payload_us = opt_int(block.get("new_payload_ms"), 0) * 1000
+    fcu_us = opt_int(block.get("forkchoice_updated_ms"), 0) * 1000
+    return new_payload_us, fcu_us, new_payload_us + fcu_us
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Convert txgen JSON report to reth-bench CSVs")
+    parser.add_argument("report", help="txgen JSON report path")
+    parser.add_argument("output_dir", help="directory for combined_latency.csv and total_gas.csv")
+    args = parser.parse_args()
+
+    report_path = Path(args.report)
+    output_dir = Path(args.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    with report_path.open() as f:
+        report = json.load(f)
+
+    blocks = report.get("blocks") or []
+    if not blocks:
+        raise SystemExit(f"txgen report {report_path} does not contain any blocks")
+
+    combined_path = output_dir / "combined_latency.csv"
+    with combined_path.open("w", newline="") as f:
+        writer = csv.DictWriter(
+            f,
+            fieldnames=[
+                "block_number",
+                "gas_limit",
+                "transaction_count",
+                "gas_used",
+                "new_payload_latency",
+                "fcu_latency",
+                "total_latency",
+                "persistence_wait",
+                "execution_cache_wait",
+                "sparse_trie_wait",
+            ],
+        )
+        writer.writeheader()
+        for block in blocks:
+            new_payload_us, fcu_us, total_us = block_latency_us(block)
+            writer.writerow(
+                {
+                    "block_number": block["number"],
+                    "gas_limit": block["gas_limit"],
+                    "transaction_count": block["tx_count"],
+                    "gas_used": block["gas_used"],
+                    "new_payload_latency": new_payload_us,
+                    "fcu_latency": fcu_us,
+                    "total_latency": total_us,
+                    "persistence_wait": block.get("persistence_wait_us") or 0,
+                    "execution_cache_wait": block.get("execution_cache_wait_us") or 0,
+                    "sparse_trie_wait": block.get("sparse_trie_wait_us") or 0,
+                }
+            )
+
+    total_gas_path = output_dir / "total_gas.csv"
+    elapsed_us = 0
+    with total_gas_path.open("w", newline="") as f:
+        writer = csv.DictWriter(
+            f,
+            fieldnames=["block_number", "transaction_count", "gas_used", "time"],
+        )
+        writer.writeheader()
+        for block in blocks:
+            _, _, total_us = block_latency_us(block)
+            elapsed_us += total_us
+            writer.writerow(
+                {
+                    "block_number": block["number"],
+                    "transaction_count": block["tx_count"],
+                    "gas_used": block["gas_used"],
+                    "time": elapsed_us,
+                }
+            )
+
+    print(f"Wrote legacy CSVs from {report_path} to {output_dir}")
+
+
+if __name__ == "__main__":
+    main()