diff --git a/tooling/sync/Makefile b/tooling/sync/Makefile index 3b3c419bb99..f5acfa34f97 100644 --- a/tooling/sync/Makefile +++ b/tooling/sync/Makefile @@ -4,7 +4,9 @@ flamegraph-hoodi start-lighthouse start-ethrex backup-db start-mainnet-metrics-d start-sepolia-metrics-docker start-holesky-metrics-docker start-hoodi-metrics-docker \ start-metrics-docker tail-syncing-logs tail-metrics-logs copy_flamegraph import-with-metrics \ multisync-up multisync-down multisync-clean multisync-logs multisync-status \ -multisync-restart multisync-monitor multisync-run multisync-loop multisync-loop-auto +multisync-restart multisync-monitor multisync-run multisync-loop multisync-loop-auto \ +fullsync-bench-bootstrap fullsync-bench-watch fullsync-bench-once fullsync-bench-smoke \ +fullsync-bench-test ETHREX_DIR ?= "../.." @@ -347,3 +349,73 @@ multisync-loop-auto: ## Continuous loop with auto-update: pull latest, build, an --image-tag "$(MULTISYNC_LOCAL_IMAGE)" \ --ethrex-dir "$(ETHREX_DIR)" \ $(if $(MULTISYNC_WATCHED_PHASES),--watched-phases "$(MULTISYNC_WATCHED_PHASES)") + +# ============================================================================= +# FULL-SYNC THROUGHPUT REGRESSION WATCH (issue #7111) +# ============================================================================= +# Keeps a node a fixed distance behind head and, per cycle, measures a fixed window of +# blocks then advances the base by one gap-maintaining step. Observe-only for now: +# thresholds and step detection land once the bootstrap period yields the real noise +# floor. Runs are serial across networks by design — concurrent legs contend and both +# results are junk. + +# Node data, bases and results share one filesystem so `rsync --link-dest` can hardlink +# unchanged SSTs between base generations; see docker-compose.fullsync-bench.yaml. +BENCH_DATA_ROOT ?= /mnt/raid10/fullsync-bench +export BENCH_DATA_ROOT + +FULLSYNC_BENCH_NETWORKS ?= mainnet +FULLSYNC_BENCH_STATE ?= $(BENCH_DATA_ROOT)/state +FULLSYNC_BENCH_RESULTS ?= $(BENCH_DATA_ROOT)/results + +fullsync-bench-bootstrap: ## Snap-sync to the tip and record the first base per network. + python3 fullsync_bench.py \ + --networks $(FULLSYNC_BENCH_NETWORKS) \ + --state-root $(FULLSYNC_BENCH_STATE) \ + --results-dir $(FULLSYNC_BENCH_RESULTS) \ + --bootstrap + +fullsync-bench-watch: ## Continuous full-sync throughput watch (observe-only). + python3 fullsync_bench.py \ + --networks $(FULLSYNC_BENCH_NETWORKS) \ + --state-root $(FULLSYNC_BENCH_STATE) \ + --results-dir $(FULLSYNC_BENCH_RESULTS) + +fullsync-bench-once: ## One measure+advance cycle per network, then exit. + python3 fullsync_bench.py \ + --networks $(FULLSYNC_BENCH_NETWORKS) \ + --state-root $(FULLSYNC_BENCH_STATE) \ + --results-dir $(FULLSYNC_BENCH_RESULTS) \ + --once + +FULLSYNC_SMOKE_NET ?= hoodi +FULLSYNC_SMOKE_ROOT ?= $(BENCH_DATA_ROOT)/smoke +FULLSYNC_SMOKE_MEASURE ?= 2048 +FULLSYNC_SMOKE_GAP ?= 4096 + +fullsync-bench-smoke: ## Rehearse a full cycle now, on a throwaway copy of a real base. + @test -d "$(FULLSYNC_BENCH_STATE)/$(FULLSYNC_SMOKE_NET)/base.0" \ + || { echo "no base for $(FULLSYNC_SMOKE_NET); run fullsync-bench-bootstrap first"; exit 1; } + rm -rf "$(FULLSYNC_SMOKE_ROOT)" + mkdir -p "$(FULLSYNC_SMOKE_ROOT)/state/$(FULLSYNC_SMOKE_NET)" +# Hardlink copy, so rehearsing costs no real disk and cannot touch the live series. + cp -al "$(FULLSYNC_BENCH_STATE)/$(FULLSYNC_SMOKE_NET)/base.0" \ + "$(FULLSYNC_SMOKE_ROOT)/state/$(FULLSYNC_SMOKE_NET)/base.0" +# bench-base.json is rewritten in place, and truncating a hardlink would take the real +# base's metadata with it. Break that one link — with -p, because a base predating +# `created_at` has its age inferred from this file's mtime, and a plain copy resets it to +# now, which reads as a zero-length gap. + cp -p --remove-destination \ + "$(FULLSYNC_BENCH_STATE)/$(FULLSYNC_SMOKE_NET)/base.0/bench-base.json" \ + "$(FULLSYNC_SMOKE_ROOT)/state/$(FULLSYNC_SMOKE_NET)/base.0/bench-base.json" + python3 fullsync_bench.py \ + --networks $(FULLSYNC_SMOKE_NET) \ + --state-root "$(FULLSYNC_SMOKE_ROOT)/state" \ + --results-dir "$(FULLSYNC_SMOKE_ROOT)/results" \ + --measure-blocks $(FULLSYNC_SMOKE_MEASURE) \ + --gap-blocks $(FULLSYNC_SMOKE_GAP) \ + --once + +fullsync-bench-test: ## Unit-test metric extraction and the tip oracle (no node needed). + python3 test_fullsync_metrics.py + python3 test_fullsync_bench.py diff --git a/tooling/sync/README.md b/tooling/sync/README.md index ce738fc5bfb..1cfd03220aa 100644 --- a/tooling/sync/README.md +++ b/tooling/sync/README.md @@ -231,3 +231,192 @@ Notifications include: - Host, branch, and commit info - Per-network status with sync time and blocks processed - Link to the commit on GitHub + +## Full-sync throughput regression watch + +Tracks execution throughput on real mainnet/testnet blocks over time, to catch performance +regressions that land on `main`. Complements the existing coverage rather than duplicating +it: the per-PR CI benchmark runs a synthetic dense-ERC20 import, and multisync validates +*snap* sync completion — neither watches execution throughput drift. Design: issue #7111. + +### How it works + +A node is kept permanently a fixed distance behind the tip. Each cycle both *consumes* a +saved database and *produces* the next one, so the whole thing is a loop that bootstrap +seeds once and never needs re-anchoring: + +``` + ┌────────────────────────────────────────────────┐ + │ │ + ▼ │ + base.0 (a stopped node's datadir, at block B) │ + │ │ + ├── restore ─► MEASURE B → B+M ──► throughput; datadir discarded + │ │ + └── restore ─► ADVANCE B → T−GAP ─► new base.0 ─┘ +``` + +One cycle, per network, once a day: + +1. **Gate.** Read B from the base's `bench-base.json` and estimate the tip T. If + `T − B < M` the whole window has not been produced yet, so skip and try later. +2. **Measure leg.** Restore `base.0` into the datadir — a *copy*; no node ever opens the + base itself. Re-sync the beacon node ahead of B, drop the page cache, start the node, + let it full-sync `B → B+M`, stop it gracefully, and read the per-batch throughput out + of its log. **The resulting database is thrown away**; only the number is kept. +3. **Advance leg.** Restore `base.0` again, back to B. Sync `B → T−GAP`, stop, then + health-check the result by reopening it. Only if it comes back up with state available + does it rotate `base.0→base.1→…` and become the new `base.0`. +4. **Report** the measurement. + +``` + B (7d behind) B+M (4d behind) T (tip) + │ │ │ + ├──── measure: 21,600 ──►│ discarded + ├─ advance: 7,200 ─►│ + new base (6d behind) +``` + +**Why two runs from the same point.** The measurement window `M` must never change or the +history stops being comparable, while the base has to advance at exactly chain rate — one +day per day — or it either catches the tip or falls away forever. Those are two different +distances, so they need two different runs. Promoting the measurement leg's database +instead would advance the base by `M` every day, closing the gap within days and breaking +the mechanism. The measurement is therefore deliberately sterile: it starts from the base, +yields a number, and its state is destroyed. + +The overlap is the payoff. Consecutive three-day windows measured from a base that moved +only one day share **two-thirds of their blocks**, so day-to-day workload variation stays +small and a real regression shows up as a step in the series rather than as noise. + +The advance targets `T − GAP` rather than a fixed block count, so it self-calibrates to +each network's real block production and absorbs missed slots. + +Rough cost per network per day on mainnet: two restores at ~6 min each, ~50 min measuring, +~17 min advancing. + +### Usage + +```bash +make fullsync-bench-bootstrap # first base per network (once) +make fullsync-bench-once # one cycle, then exit +make fullsync-bench-watch # continuous +make fullsync-bench-watch FULLSYNC_BENCH_NETWORKS=mainnet,sepolia,hoodi +make fullsync-bench-test # unit-test metric parsing +``` + +Networks run **serially** on purpose: two legs at once contend for CPU and disk and both +results are junk. An flock enforces this; bootstraps share it, cycles take it exclusively, +and the manual A/B tool (#7112) takes it exclusively too. + +Reporting goes to `SLACK_WEBHOOK_URL_SUCCESS`, or `SLACK_WEBHOOK_URL_FAILED` for a round +where a cycle could not produce a valid measurement — a crashed node, an unclean exit, a +leg that never reached its target. A merely *low* number is not a failure while the watch +is observe-only: there is no baseline to judge it against yet. Both are read from +`tooling/sync/.env` (same format as multisync) or the environment, which wins. + +**Each leg re-checkpoint-syncs the beacon node to the tip first, and this is load-bearing.** +The execution node only does bulk 1024-block batch sync when fork choice hands it a head +far ahead of itself. A beacon node stopped alongside it knows nothing newer than the base, +so the pair crawl forward together importing ~30 blocks at a time — which measures +incremental block import rather than full sync, and emits no throughput metric at all. A +leg whose beacon node cannot reach the tip is reported `cl_not_synced` rather than +recording a number that means something else. + +### Bootstrap + +`make fullsync-bench-bootstrap` snap-syncs each network to the tip, stops it gracefully and +records it as `base.0` plus a `bench-base.json` holding its block number. Snap is used for +this initial fill only — full-syncing from genesis would take weeks; the measurement legs +themselves always run full, since that is the thing being measured. + +The node is then simply left stopped. The gap opens on its own at one day per day as the +chain moves on, so nothing needs anchoring by hand. Cycles skip themselves with a log line +until `head − base ≥ M`; with the default `M` of 3 days that means roughly a three-day wait +after bootstrap before the first real measurement (six to reach the steady-state `GAP`). + +Run **observe-only for 2–3 weeks** after that before wiring up alerting: the real +day-to-day σ is unknown, and thresholds should come from measured data rather than a guess. +Alerting and step detection (trailing median + persistence, not day-vs-day) land later. + +### Rehearsing a cycle + +Waiting `M` days to discover that a log format moved or a webhook is wrong is a poor +feedback loop, so `make fullsync-bench-smoke` runs one complete cycle immediately against +a small window: + +```bash +make fullsync-bench-smoke # hoodi, 2048-block window +make fullsync-bench-smoke FULLSYNC_SMOKE_NET=sepolia +``` + +It exercises the real path end to end — restore, a genuine full-sync leg, graceful stop, +metric extraction from live logs, the result JSON, base rotation and the Slack post — on a +`cp -al` hardlink copy of a real base, so it costs almost no disk and cannot damage the +live series. Stop the watch first; it holds the lock, and the smoke run will refuse. + +Keep the window at or above ~2048 blocks. Throughput is read from the per-batch metric +line, which ethrex emits every 1024 blocks, so a smaller window can finish having logged +no batch at all and be reported `no_batches`. + +`--measure-blocks` and `--gap-blocks` exist for this and nothing else. The runner refuses +to accept them alongside the default results directory, because folding a 2k-block sample +into a series of 21.6k-block ones would drag the trailing median that the comparison rests +on. + +### Storage layout + +Everything lives under `BENCH_DATA_ROOT` (default `/mnt/raid10/fullsync-bench`): + +``` +/data// bind-mounted into the container as /data +/consensus// beacon node database +/state//base.N retained base generations +/results// one JSON + log per leg +``` + +Node data and bases **must stay on one filesystem**. `snapshot()` hardlinks unchanged SST +files between generations with `rsync --link-dest`, which only works within a filesystem; +across two, rsync silently copies instead and every retained generation costs a full +database rather than a delta. The runner asserts this at startup rather than trusting it. + +Bind mounts are used instead of named Docker volumes for the same reason: Docker's +`data-root` is usually on the OS disk, which is both smaller and a different filesystem +from the array holding the bases. + +### Metrics + +Each cycle writes one JSON per leg under `//`: + +| field | why | +|---|---| +| `throughput_ggas_s` (mean/median/stdev/samples) | headline; the **per-batch mean**, not blocks ÷ wall clock | +| `blocks_per_s_mean` | likely the better primary on light testnet blocks | +| `phase_ms_per_mgas` (`validate`/`exec`/`merkle`/`store`) | catches phase-localised regressions invisible end-to-end | +| `state_regen_seconds` | restart cost; sensitive to commit cadence | +| `wall_seconds`, `status`, `reached_block`, `commit`, `host` | validity and attribution | + +Wall clock is deliberately *not* the throughput metric: it includes startup state +regeneration, which is a real signal but a separate one, so it is reported on its own. + +### Operational invariants + +These are lessons from the manual benchmarking that preceded this tool, not preferences: + +- **Graceful stop only** (`docker stop -t 300`). Repeated abrupt stops previously left the + canonical head ahead of any durably-flushed state, after which the node could not + regenerate and needed a full re-sync. +- **Stop condition reads `eth_blockNumber`**, never `eth_syncing.currentBlock` — the latter + goes stale during catch-up and once let a leg run far past its target. +- **Health-gate base rotation**: a new base is promoted only after the node demonstrably + restarts on it with state available; otherwise the previous generation is kept. +- **Never compare across machines.** The same binary has measured 62 s and 81 s on + different CI runners; only same-box comparisons mean anything. +- **Do not `--force-recreate` a `consensus-*` service on its own.** That re-runs its + `setup-jwt-*` dependency, which writes a fresh `jwt.hex`, while the already-running + execution node keeps the secret it read at startup. Engine API calls then fail with + `Auth failed`, the node stops being told where the chain tip is, and the sync quietly + goes nowhere. Restart the matching `ethrex-*` container afterwards. +- **During snap sync `eth_blockNumber` stays at 0** and jumps to the tip only once the + pivot state has landed. A runner sitting at `0 / ` for hours is normal, not a + stall; `docker logs ethrex-` is where the real progress is. diff --git a/tooling/sync/docker-compose.fullsync-bench.yaml b/tooling/sync/docker-compose.fullsync-bench.yaml new file mode 100644 index 00000000000..a2edf5284e3 --- /dev/null +++ b/tooling/sync/docker-compose.fullsync-bench.yaml @@ -0,0 +1,133 @@ +# Overrides multisync's compose for the full-sync throughput bench (issue #7111). +# +# Applied on top of docker-compose.multisync.yaml so the per-network service pairs, JWT +# setup and volumes are defined in exactly one place: +# +# docker compose -p fullsync-bench \ +# -f docker-compose.multisync.yaml -f docker-compose.fullsync-bench.yaml ... +# +# Differences from multisync: +# 1. `--syncmode` defaults to full — this measures execution, not snap-sync completion. +# Bootstrapping the first base overrides it to snap via BENCH_SYNCMODE. +# 2. `mem_limit` — a memory leak must OOM-kill the container, not take down the box. +# 3. the beacon APIs are published, because the runner reads the consensus head to +# compute each cycle's gap-maintaining advance target. +# 4. data lives in bind mounts under BENCH_DATA_ROOT rather than named volumes, so the +# node data and the base snapshots share one filesystem. `rsync --link-dest` only +# hardlinks within a filesystem; split across two it degrades to full copies without +# saying so, and every retained generation costs a whole DB instead of a delta. +# 5. `--purge-db-force` (not multisync's `--purge-db`, which lighthouse ignores when +# not attached to a TTY, so in a container it silently does nothing) is load-bearing: recreating the beacon node +# before each leg re-checkpoint-syncs it to the tip, which is what makes the execution +# node do bulk 1024-block batch sync. Without it the pair crawl forward together ~30 +# blocks at a time and no throughput metric is emitted at all. +services: + # Generate the JWT once and keep it. Multisync mints a fresh one on every start, which is + # fine there because it wipes its volumes between runs — but here the beacon node is + # recreated before every leg, and a new secret would leave the already-running execution + # node holding the old one. Engine auth then fails, fork choice never reaches the node, + # and it sits there syncing nothing while looking healthy. + # Also writes with `>` rather than `tee`, to keep the secret out of the container log. + setup-jwt-hoodi: &bench-jwt + command: > + sh -c 'apk add openssl && + { [ -s /secrets/jwt.hex ] || openssl rand -hex 32 | tr -d "\n" > /secrets/jwt.hex; }' + + setup-jwt-sepolia: + <<: *bench-jwt + + setup-jwt-mainnet: + <<: *bench-jwt + + consensus-hoodi: + volumes: + - secrets-hoodi:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/consensus/hoodi:/root/.lighthouse + ports: + - "5052:5052" + command: > + lighthouse bn + --network hoodi + --http --http-address 0.0.0.0 + --execution-endpoint http://ethrex-hoodi:8551 + --execution-jwt /secrets/jwt.hex + --checkpoint-sync-url https://checkpoint-sync.hoodi.ethpandaops.io + --checkpoint-sync-url-timeout 600 + --purge-db-force + + consensus-sepolia: + volumes: + - secrets-sepolia:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/consensus/sepolia:/root/.lighthouse + ports: + - "5053:5052" + command: > + lighthouse bn + --network sepolia + --http --http-address 0.0.0.0 + --execution-endpoint http://ethrex-sepolia:8551 + --execution-jwt /secrets/jwt.hex + --checkpoint-sync-url https://checkpoint-sync.sepolia.ethpandaops.io + --checkpoint-sync-url-timeout 600 + --purge-db-force + + consensus-mainnet: + volumes: + - secrets-mainnet:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/consensus/mainnet:/root/.lighthouse + ports: + - "5054:5052" + command: > + lighthouse bn + --network mainnet + --http --http-address 0.0.0.0 + --execution-endpoint http://ethrex-mainnet:8551 + --execution-jwt /secrets/jwt.hex + --checkpoint-sync-url https://mainnet-checkpoint-sync.attestant.io + --checkpoint-sync-url-timeout 600 + --purge-db-force + + ethrex-hoodi: &bench-el + mem_limit: 48g + volumes: + - secrets-hoodi:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/data/hoodi:/data + command: > + --http.addr 0.0.0.0 + --http.api eth,net,web3,admin + --network hoodi + --authrpc.addr 0.0.0.0 + --authrpc.jwtsecret /secrets/jwt.hex + --syncmode ${BENCH_SYNCMODE:-full} + --datadir /data + --metrics --metrics.addr 0.0.0.0 --metrics.port 3701 + + ethrex-sepolia: + <<: *bench-el + volumes: + - secrets-sepolia:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/data/sepolia:/data + command: > + --http.addr 0.0.0.0 + --http.api eth,net,web3,admin + --network sepolia + --authrpc.addr 0.0.0.0 + --authrpc.jwtsecret /secrets/jwt.hex + --syncmode ${BENCH_SYNCMODE:-full} + --datadir /data + --metrics --metrics.addr 0.0.0.0 --metrics.port 3701 + + ethrex-mainnet: + <<: *bench-el + volumes: + - secrets-mainnet:/secrets + - ${BENCH_DATA_ROOT:-/mnt/raid10/fullsync-bench}/data/mainnet:/data + command: > + --http.addr 0.0.0.0 + --http.api eth,net,web3,admin + --network mainnet + --authrpc.addr 0.0.0.0 + --authrpc.jwtsecret /secrets/jwt.hex + --syncmode ${BENCH_SYNCMODE:-full} + --datadir /data + --metrics --metrics.addr 0.0.0.0 --metrics.port 3701 diff --git a/tooling/sync/fullsync_bench.py b/tooling/sync/fullsync_bench.py new file mode 100644 index 00000000000..9072901143f --- /dev/null +++ b/tooling/sync/fullsync_bench.py @@ -0,0 +1,771 @@ +#!/usr/bin/env python3 +"""Full-sync throughput regression watch — see issue #7111. + +Keeps a node permanently a fixed distance behind head and, once per cycle per network: + + restore base -> run M blocks (MEASURE, resulting state discarded) + restore base -> advance to head-GAP -> becomes the new base + +The base creeps forward at chain rate, so the node never ages out of relevance and no +anchor or reference commit needs manual upkeep. The measurement window M is fixed and +independent of how fast the base moves; consecutive measurements therefore overlap +heavily, which is what keeps day-over-day workload variation small. + +`run_leg` is deliberately pure — it never decides what happens to the resulting state, +never assumes a particular ref, and never rotates anything. That is what lets the manual +A/B tool (issue #7112) reuse it unchanged. +""" + +import argparse +import fcntl +import json +import os +import shutil +import socket +import subprocess +import sys +import time +from datetime import datetime, timezone + +import requests + +from fullsync_metrics import parse_run, parse_run_file + +HERE = os.path.dirname(os.path.abspath(__file__)) + +# Same `.env` convention as docker_monitor.py, but resolved against this file rather than +# the working directory: the watch runs unattended from wherever a service manager put it, +# and silently losing the Slack webhook to a `cd` is not a good failure. +_ENV_FILE = os.path.join(HERE, ".env") +if os.path.exists(_ENV_FILE): + with open(_ENV_FILE) as _fh: + for _line in _fh: + _line = _line.strip() + if _line and not _line.startswith("#"): + _key, _, _value = _line.partition("=") + os.environ.setdefault(_key.strip(), _value.strip()) + +COMPOSE_FILES = ["docker-compose.multisync.yaml", "docker-compose.fullsync-bench.yaml"] +COMPOSE_PROJECT = "fullsync-bench" + +# Node data, base generations and results all live under here, and must stay on one +# filesystem — see `assert_same_filesystem`. Kept in step with BENCH_DATA_ROOT in the +# compose override, which bind-mounts `/data/` into each container. +DATA_ROOT = os.environ.get("BENCH_DATA_ROOT", "/mnt/raid10/fullsync-bench") +DEFAULT_RESULTS_DIR = os.path.join(DATA_ROOT, "results") + +# One lock for the whole box: two measurement legs at once contend for CPU and disk and +# both results are junk. Cycles take it exclusively, bootstraps share it (see `take_lock`). +# The A/B tool (#7112) must take the same lock exclusively. +LOCK_PATH = "/tmp/ethrex-fullsync-bench.lock" + +# Blocks are ~12s on all three networks, so a nominal day is ~7200 blocks. `measure_blocks` +# is quantised to whole days so every sample spans the same diurnal composition, and must +# stay FIXED once a series starts — changing it breaks comparability of the history. +BLOCKS_PER_DAY = 7200 + +NETWORKS = { + "mainnet": {"rpc_port": 8547, "cl_port": 5054, + "measure_blocks": 3 * BLOCKS_PER_DAY, "gap_blocks": 6 * BLOCKS_PER_DAY}, + "sepolia": {"rpc_port": 8546, "cl_port": 5053, + "measure_blocks": 3 * BLOCKS_PER_DAY, "gap_blocks": 6 * BLOCKS_PER_DAY}, + "hoodi": {"rpc_port": 8545, "cl_port": 5052, + "measure_blocks": 3 * BLOCKS_PER_DAY, "gap_blocks": 6 * BLOCKS_PER_DAY}, +} + +# Retained hardlink generations of each base, for rollback when a cycle ends badly. +# Hardlinks pin SST files that compaction would otherwise free, so this is capped. +KEEP_GENERATIONS = 5 + +STOP_TIMEOUT_SECONDS = 300 # graceful; SIGKILL corrupts the resume state (see #7111) +POLL_SECONDS = 20 + + +def log(msg): + print(f"[{datetime.now(timezone.utc).isoformat(timespec='seconds')}] {msg}", flush=True) + + +def run(cmd, check=True, capture=False): + log(f"$ {' '.join(cmd)}") + return subprocess.run(cmd, check=check, text=True, + stdout=subprocess.PIPE if capture else None) + + +def compose(args, check=True): + cmd = ["docker", "compose", "-p", COMPOSE_PROJECT] + for f in COMPOSE_FILES: + cmd += ["-f", os.path.join(HERE, f)] + return run(cmd + args, check=check) + + +# --------------------------------------------------------------------------- paths + + +def data_dir(net): + """Host path bind-mounted into the node as /data.""" + return os.path.join(DATA_ROOT, "data", net) + + +def base_dir(state_root, net, generation=0): + return os.path.join(state_root, net, f"base.{generation}") + + +def assert_same_filesystem(net, state_root): + """Bases and live data must share a filesystem. + + `snapshot` relies on `rsync --link-dest` hardlinking unchanged SST files, which only + works within one filesystem. Across two, rsync copies instead — no error, just every + retained generation quietly costing a full database. + """ + os.makedirs(data_dir(net), exist_ok=True) + os.makedirs(os.path.join(state_root, net), exist_ok=True) + if os.stat(data_dir(net)).st_dev != os.stat(os.path.join(state_root, net)).st_dev: + raise SystemExit( + f"{net}: data dir {data_dir(net)} and base dir {state_root}/{net} are on " + "different filesystems; --link-dest would silently fall back to full copies" + ) + + +# ------------------------------------------------------------------- node control + + +def rpc_block_number(port, timeout=8): + """Current canonical head. + + Deliberately `eth_blockNumber` and not `eth_syncing.currentBlock`: the latter goes + stale during catch-up, which once let a benchmark leg run far past its stop target. + """ + body = {"jsonrpc": "2.0", "method": "eth_blockNumber", "params": [], "id": 1} + resp = requests.post(f"http://localhost:{port}", json=body, timeout=timeout).json() + return int(resp["result"], 16) + + +def cl_head(net, timeout=8): + """Consensus head, i.e. the tip the node would sync toward.""" + url = f"http://localhost:{NETWORKS[net]['cl_port']}/eth/v2/beacon/blocks/head" + resp = requests.get(url, timeout=timeout).json() + return int(resp["data"]["message"]["body"]["execution_payload"]["block_number"]) + + +SLOT_SECONDS = 12 + +# Slots that actually carry a block. Deliberately below the real figure (~99% on mainnet): +# under-estimating the tip only makes a cycle wait a little longer, while over-estimating +# would aim a leg at a block nobody has produced, which costs its full timeout. +SLOT_FILL_RATIO = 0.97 + + +def cl_is_current(net, timeout=8): + """Whether the beacon node's head can be believed right now.""" + url = f"http://localhost:{NETWORKS[net]['cl_port']}/eth/v1/node/syncing" + data = requests.get(url, timeout=timeout).json()["data"] + return not data["el_offline"] and int(data["sync_distance"]) <= SYNCED_MARGIN_BLOCKS + + +def chain_tip(net, base): + """Best available estimate of the network's current execution head. + + The beacon node cannot be asked for this between cycles. With the execution node + stopped it reports `el_offline` and stops advancing its head — it sat 17 hours behind + and unmoving on all three networks the first time this ran — so `tip - base` would read + zero forever and every cycle would skip itself. Its head is therefore used only when it + is demonstrably current, which in practice means during or just after a leg. + + Otherwise extrapolate from when the base was recorded. Slot timing is fixed and known, + so elapsed wall clock gives a good enough tip for both callers: a "has enough chain + accumulated yet" check, and an advance target a whole GAP below the tip. + """ + try: + if cl_is_current(net): + return cl_head(net) + except Exception: + pass + + meta_path = os.path.join(base, BASE_META) + with open(meta_path) as fh: + meta = json.load(fh) + # Bases written before `created_at` existed fall back to the metadata file's mtime, + # which is when the base was recorded. + created = meta.get("created_at", os.path.getmtime(meta_path)) + elapsed = max(0.0, time.time() - created) + return meta["head"] + int(elapsed / SLOT_SECONDS * SLOT_FILL_RATIO) + + +def start_node(net): + compose(["up", "-d", "--no-deps", f"ethrex-{net}"]) + + +def start_consensus(net): + """Bring up the beacon node (and its JWT setup).""" + compose(["up", "-d", f"consensus-{net}"]) + + +# Checkpoint sync is a download of one finalised state; minutes, not hours. +CL_REFRESH_TIMEOUT_SECONDS = 900 + + +def refresh_consensus(net, min_block, timeout=CL_REFRESH_TIMEOUT_SECONDS): + """Put the beacon node far enough ahead of the base to drive a real leg, and wait. + + This is load-bearing, not hygiene. The execution node only does bulk 1024-block batch + sync when fork choice hands it a head far ahead of where it is. A beacon node that was + stopped alongside it knows nothing newer than the base, so on restart the pair crawl + forward together and the node imports ~30 blocks at a time: that measures incremental + block import, not full sync, and emits no batch throughput metric at all. + + Recreating the container re-runs its `--purge-db-force` checkpoint sync, which lands it + near the tip in a couple of minutes. It also regenerates the JWT, which is precisely why + the execution node must be started *after* this returns. + + The wait is for the beacon node to know a block at or above this leg's target, not for + it to reach the tip. It cannot reach the tip: with no execution node it has nothing to + verify payloads against, so it parks at its checkpoint anchor a couple of epochs back + and from there falls further behind at one slot per slot. Waiting for a small + `sync_distance` would simply time out every time. + """ + compose(["up", "-d", "--force-recreate", f"consensus-{net}"]) + deadline = time.time() + timeout + while time.time() < deadline: + try: + head = cl_head(net) + if head >= min_block: + log(f"{net}: beacon node at block {head}, ahead of the target {min_block}") + return True + log(f"{net}: beacon node at {head}, needs {min_block}") + except Exception: + pass + time.sleep(10) + return False + + +def stop_node(net): + """Graceful stop. Never SIGKILL: an abrupt stop can leave the canonical head ahead of + any durably-flushed state, after which the node cannot regenerate and needs a re-sync.""" + run(["docker", "stop", "-t", str(STOP_TIMEOUT_SECONDS), f"ethrex-{net}"], check=False) + + +def exited_cleanly(net): + out = run(["docker", "inspect", f"ethrex-{net}", "--format", "{{.State.ExitCode}}"], + check=False, capture=True) + return (out.stdout or "").strip() == "0" + + +def drop_page_cache(): + """Cold-start parity between runs.""" + subprocess.run(["sync"], check=False) + try: + with open("/proc/sys/vm/drop_caches", "w") as fh: + fh.write("3") + except OSError as exc: # non-Linux dev box, or not root + log(f"could not drop page cache ({exc}); continuing") + + +# ------------------------------------------------------------------ snapshot / base + + +def restore(net, src): + """Reset the live data dir to `src`.""" + run(["rsync", "-a", "--delete", f"{src}/", f"{data_dir(net)}/"]) + + +def snapshot(net, dest, link_dest=None): + """Hardlink snapshot of the live data dir. + + RocksDB SST files are immutable and uniquely named, so `--link-dest` against the + previous generation makes a new base cost only its delta rather than a full copy. + """ + cmd = ["rsync", "-a", "--delete"] + if link_dest and os.path.isdir(link_dest): + cmd += [f"--link-dest={link_dest}"] + run(cmd + [f"{data_dir(net)}/", f"{dest}/"]) + + +def rotate_generations(state_root, net): + """Shift base.N -> base.N+1, dropping the oldest.""" + oldest = base_dir(state_root, net, KEEP_GENERATIONS - 1) + if os.path.isdir(oldest): + shutil.rmtree(oldest) + for gen in range(KEEP_GENERATIONS - 2, -1, -1): + src = base_dir(state_root, net, gen) + if os.path.isdir(src): + os.rename(src, base_dir(state_root, net, gen + 1)) + + +# ----------------------------------------------------------------------- bootstrap + + +# Snap-syncing mainnet from scratch is an overnight job, and a stall should not hang the +# runner forever. +BOOTSTRAP_TIMEOUT_SECONDS = 48 * 3600 + +# How close to the consensus tip counts as "caught up" — a couple of epochs of slack, so a +# node still importing the last few blocks is not called done early. +SYNCED_MARGIN_BLOCKS = 64 + + +def bootstrap(net, state_root): + """Create the first base for a network. + + Snap-syncs to the tip and snapshots the result. The node is then left stopped, and the + gap the watch needs opens by itself at one day per day as the chain moves on — no + manual anchoring. Cycles can start once head - base >= measure_blocks, which `cycle` + checks and waits for. + """ + cfg = NETWORKS[net] + base = base_dir(state_root, net, 0) + if os.path.isdir(base): + raise SystemExit(f"{net}: base already exists at {base}; remove it to re-bootstrap") + assert_same_filesystem(net, state_root) + + # Snap for the initial fill only: full-syncing from genesis would take weeks. The + # measurement legs themselves always run full — that is the thing being measured. + os.environ["BENCH_SYNCMODE"] = "snap" + start_consensus(net) + start_node(net) + log(f"{net}: snap-syncing to tip; expect hours") + + deadline = time.time() + BOOTSTRAP_TIMEOUT_SECONDS + head = None + try: + while True: + try: + head, tip = rpc_block_number(cfg["rpc_port"]), cl_head(net) + if head >= tip - SYNCED_MARGIN_BLOCKS: + log(f"{net}: caught up at {head} (tip {tip})") + break + log(f"{net}: {head} / {tip} ({tip - head} behind)") + except Exception as exc: + log(f"{net}: waiting for RPC/beacon ({exc})") + if time.time() > deadline: + raise SystemExit(f"{net}: bootstrap timed out after " + f"{BOOTSTRAP_TIMEOUT_SECONDS}s at head {head}") + time.sleep(60) + finally: + os.environ.pop("BENCH_SYNCMODE", None) + stop_node(net) + + if not exited_cleanly(net): + raise SystemExit(f"{net}: node did not exit cleanly; refusing to promote the base") + + os.makedirs(os.path.dirname(base), exist_ok=True) + snapshot(net, base) + write_base_head(base, head) + log(f"{net}: base created at {base} (block {head}); " + f"first cycle once the chain is {cfg['measure_blocks']} blocks past it") + return head + + +# --------------------------------------------------------------------- the primitive + + +def run_leg(net, target_block, log_path, timeout_seconds): + """Sync from the currently-restored state up to `target_block`, then stop. + + Pure: the caller decides what to do with the resulting state. Returns a LegResult + dict; `status` is `ok` only for a run that reached its target and exited cleanly. + """ + cfg = NETWORKS[net] + + # Before the clock starts: the node must be handed a distant head, or this measures + # the wrong thing entirely. Costs a couple of minutes and is excluded from the timing. + if not refresh_consensus(net, target_block): + log(f"{net}: beacon node never learned block {target_block}; refusing to run a leg " + "that would measure incremental import instead of full sync") + return {"network": net, "target_block": target_block, "reached_block": None, + "wall_seconds": 0.0, "status": "cl_not_synced", "log": log_path, + **parse_run([])} + + drop_page_cache() + started = time.time() + start_node(net) + + logger = subprocess.Popen(["docker", "logs", "-f", f"ethrex-{net}"], + stdout=open(log_path, "w"), stderr=subprocess.STDOUT) + status, head = "ok", None + try: + deadline = started + timeout_seconds + while True: + try: + head = rpc_block_number(cfg["rpc_port"]) + except Exception: + pass # RPC not up yet, or a transient blip; keep the last known head + if head is not None and head >= target_block: + break + if time.time() > deadline: + status = "timeout" + break + time.sleep(POLL_SECONDS) + finally: + stop_node(net) + logger.terminate() + + if status == "ok" and not exited_cleanly(net): + # An OOM-kill or crash leaves the DB in a state we must not promote to a base. + status = "unclean_exit" + + metrics = parse_run_file(log_path) + if status == "ok" and not metrics["batches"]: + status = "no_batches" + + return { + "network": net, + "target_block": target_block, + "reached_block": head, + "wall_seconds": round(time.time() - started, 1), + "status": status, + "log": log_path, + **metrics, + } + + +# ------------------------------------------------------------------------- the cycle + + +def cycle(net, state_root, results_dir): + """One measure-and-advance cycle for a single network.""" + cfg = NETWORKS[net] + base = base_dir(state_root, net, 0) + if not os.path.isdir(base): + raise SystemExit(f"no base for {net} at {base}; run with --bootstrap first") + + assert_same_filesystem(net, state_root) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + os.makedirs(os.path.join(results_dir, net), exist_ok=True) + + base_head = read_base_head(net, base, cfg) + + # The whole measurement window has to exist on-chain already. A base less than M + # blocks behind the tip would send the leg after a target nobody has produced yet, + # which can only end in its timeout hours later. + try: + tip = chain_tip(net, base) + except Exception as exc: + log(f"{net}: could not determine the chain tip ({exc}); skipping cycle") + return None + if tip - base_head < cfg["measure_blocks"]: + log(f"{net}: gap is {tip - base_head} blocks, need {cfg['measure_blocks']}; " + "waiting for it to open") + return None + + # --- measurement leg: fixed M blocks, resulting state thrown away ----------- + restore(net, base) + measure_target = base_head + cfg["measure_blocks"] + measure_log = os.path.join(results_dir, net, f"{stamp}.measure.log") + result = run_leg(net, measure_target, measure_log, + timeout_seconds=_leg_timeout(cfg["measure_blocks"])) + result["kind"] = "measure" + result["base_block"] = base_head + result["commit"] = git_commit() + result["host"] = socket.gethostname() + result["timestamp"] = stamp + _write_result(results_dir, net, stamp, result) + + # --- advance leg: gap-maintaining, becomes the next base -------------------- + # Target head-GAP rather than a fixed block count, so the advance self-calibrates to + # the network's real block production and absorbs missed slots. + # Re-read rather than reusing the tip from before the measurement leg: that leg has + # just spent hours running, and the chain moved on while it did. + try: + advance_target = chain_tip(net, base) - cfg["gap_blocks"] + except Exception as exc: + log(f"{net}: could not determine the chain tip ({exc}); skipping advance this cycle") + return result + if advance_target <= base_head: + log(f"{net}: gap has not opened yet (base {base_head} >= target {advance_target}); " + "skipping advance") + return result + + restore(net, base) + advance_log = os.path.join(results_dir, net, f"{stamp}.advance.log") + advanced = run_leg(net, advance_target, advance_log, + timeout_seconds=_leg_timeout(advance_target - base_head)) + + if advanced["status"] != "ok": + log(f"{net}: advance leg ended '{advanced['status']}'; keeping the previous base") + return result + + # --- health-gate the rotation ---------------------------------------------- + # Promote the new state only once the node demonstrably restarts on it with state + # available; otherwise a silently-degraded base would poison every later cycle. + if not base_is_healthy(net, advanced["reached_block"]): + log(f"{net}: new base failed its health check; keeping the previous base") + return result + + rotate_generations(state_root, net) + snapshot(net, base_dir(state_root, net, 0), link_dest=base_dir(state_root, net, 1)) + write_base_head(base_dir(state_root, net, 0), advanced["reached_block"]) + log(f"{net}: base advanced to {advanced['reached_block']}") + return result + + +BASE_META = "bench-base.json" + + +def read_base_head(net, base, cfg): + """Block number the base sits at. + + Recorded alongside the base when it is created, so a cycle does not need an extra + node start/stop just to ask — each start pays a state-regeneration cost we would + otherwise incur twice per cycle for nothing. + """ + meta_path = os.path.join(base, BASE_META) + if os.path.isfile(meta_path): + with open(meta_path) as fh: + return json.load(fh)["head"] + + # Only for a base created outside `bootstrap`, which has no metadata yet. + log(f"{net}: base has no {BASE_META}; reading head once and recording it") + restore(net, base) + start_node(net) + try: + for _ in range(60): + try: + head = rpc_block_number(cfg["rpc_port"]) + write_base_head(base, head) + return head + except Exception: + time.sleep(2) + raise SystemExit(f"{net}: RPC never came up after restore") + finally: + stop_node(net) + + +def write_base_head(base, head): + # `created_at` is what lets the tip be extrapolated while the beacon node is frozen. + with open(os.path.join(base, BASE_META), "w") as fh: + json.dump({"head": head, "created_at": time.time()}, fh) + + +def base_is_healthy(net, expected_head): + """Start briefly and confirm the node reports the expected head with state available.""" + cfg = NETWORKS[net] + start_node(net) + try: + for _ in range(60): + try: + head = rpc_block_number(cfg["rpc_port"]) + except Exception: + time.sleep(2) + continue + if head < expected_head: + return False + body = {"jsonrpc": "2.0", "method": "eth_getBalance", + "params": ["0x0000000000000000000000000000000000000000", hex(head)], "id": 1} + resp = requests.post(f"http://localhost:{cfg['rpc_port']}", json=body, timeout=8).json() + # A node whose head state is missing answers with an error, not a balance. + return "result" in resp + return False + except Exception as exc: + log(f"{net}: health check errored ({exc})") + return False + finally: + stop_node(net) + + +def _leg_timeout(blocks): + """Generous ceiling: legs are expected around 7 blocks/s, so allow well under 1/s.""" + return max(3600, int(blocks / 1.0)) + + +def git_commit(): + try: + return subprocess.check_output(["git", "rev-parse", "--short", "HEAD"], + text=True).strip() + except subprocess.CalledProcessError: + return None + + +def _write_result(results_dir, net, stamp, result): + path = os.path.join(results_dir, net, f"{stamp}.json") + with open(path, "w") as fh: + json.dump(result, fh, indent=2) + log(f"{net}: wrote {path}") + + +# ---------------------------------------------------------------------- reporting + + +def load_history(results_dir, net, limit=30): + folder = os.path.join(results_dir, net) + if not os.path.isdir(folder): + return [] + out = [] + for name in sorted(os.listdir(folder))[-limit:]: + if not name.endswith(".json"): + continue + with open(os.path.join(folder, name)) as fh: + entry = json.load(fh) + if entry.get("kind") == "measure" and entry.get("status") == "ok": + out.append(entry) + return out + + +def summarise(result, history): + """Human-readable line plus the trailing-median delta, when there is enough history. + + Deliberately reports against the trailing median rather than the previous run: a + regression is a step, and a pairwise day-to-day difference carries far more noise. + """ + mean = result["throughput_ggas_s"]["mean"] + if mean is None: + return f"*{result['network']}*: run invalid (`{result['status']}`)" + + line = (f"*{result['network']}*: {mean:.4f} Ggas/s " + f"over {result['batches']} batches (base {result.get('base_block')})") + previous = [h["throughput_ggas_s"]["mean"] for h in history + if h.get("timestamp") != result.get("timestamp")] + if len(previous) >= 3: + ordered = sorted(previous) + median = ordered[len(ordered) // 2] + line += f" — {100 * (mean - median) / median:+.1f}% vs {len(previous)}-run median" + else: + line += " — baseline still building" + return line + + +def post_slack(lines, all_ok=True): + """Report a round of cycles. + + A round where something broke goes to the failure webhook. Note what this does *not* + mean while the watch is observe-only: a throughput number being low is not a failure, + because there is no baseline to judge it against yet. Only a cycle that could not + produce a valid measurement at all counts — a crashed or OOM-killed node, an unclean + exit, a leg that never reached its target. + """ + key = "SLACK_WEBHOOK_URL_FAILED" if not all_ok else "SLACK_WEBHOOK_URL_SUCCESS" + url = os.environ.get(key) + if not url and not all_ok: + # Never drop a failure report just because the failure webhook is unconfigured. + url = os.environ.get("SLACK_WEBHOOK_URL_SUCCESS") + if url: + log(f"{key} unset; sending the failure report to the success webhook") + if not url: + log(f"{key} unset; skipping Slack") + return + # Observe-only by design for now: thresholds and step detection land once the + # bootstrap period has produced enough data to derive them (see #7111). + header = ("Full-sync throughput (observe-only)" if all_ok + else "Full-sync watch: cycle problem") + message = {"blocks": [ + {"type": "header", "text": {"type": "plain_text", "text": header}}, + {"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}}, + ]} + try: + resp = requests.post(url, data=json.dumps(message), + headers={"Content-Type": "application/json"}, timeout=15) + if resp.status_code != 200: + log(f"Slack returned {resp.status_code}") + except Exception as exc: + log(f"Slack post failed: {exc}") + + +# --------------------------------------------------------------------------- main + + +_LOCK_FD = None + + +def take_lock(shared=False): + """Serialise against other runs on this box. + + Measurement legs must never overlap with anything else that loads the machine, so + cycles take the lock exclusively. Bootstraps only take it shared: several networks can + snap-sync at once — nothing is being measured yet — but a cycle still cannot start + while one is in flight. + + Uses flock rather than an O_EXCL lockfile so the kernel drops the lock when the process + dies. A lockfile outlives a SIGKILL or a reboot, which for an unattended watch means + every later run refuses to start until someone removes it by hand. + """ + global _LOCK_FD + _LOCK_FD = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR) + try: + fcntl.flock(_LOCK_FD, (fcntl.LOCK_SH if shared else fcntl.LOCK_EX) | fcntl.LOCK_NB) + except BlockingIOError: + held = "a measurement cycle" if shared else "another run" + raise SystemExit(f"{held} holds {LOCK_PATH}; refusing to run concurrently") + os.truncate(_LOCK_FD, 0) + os.write(_LOCK_FD, str(os.getpid()).encode()) + + +def release_lock(): + global _LOCK_FD + if _LOCK_FD is not None: + fcntl.flock(_LOCK_FD, fcntl.LOCK_UN) + os.close(_LOCK_FD) + _LOCK_FD = None + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--networks", default="mainnet", + help="comma-separated (mainnet,sepolia,hoodi). Run serially.") + parser.add_argument("--state-root", default=os.path.join(DATA_ROOT, "state"), + help="where per-network base generations live") + parser.add_argument("--results-dir", default=DEFAULT_RESULTS_DIR) + parser.add_argument("--bootstrap", action="store_true", + help="snap-sync each network to the tip and record it as the first " + "base, then exit. Run once per network before watching.") + parser.add_argument("--once", action="store_true", + help="run a single cycle per network and exit") + parser.add_argument("--measure-blocks", type=int, + help="override the measurement window. SMOKE TESTS ONLY — the " + "series is only comparable if this stays fixed, so a run " + "using it must also use a throwaway --results-dir.") + parser.add_argument("--gap-blocks", type=int, + help="override the base's distance behind the tip. Smoke tests " + "only, for the same reason.") + args = parser.parse_args() + + if (args.measure_blocks or args.gap_blocks) and args.results_dir == DEFAULT_RESULTS_DIR: + # Mixing a 2k-block sample into a series of 21.6k-block ones would shift the + # trailing median the comparison depends on. + raise SystemExit("--measure-blocks/--gap-blocks change what is being measured; " + "point --results-dir somewhere throwaway so the real series is " + "not polluted") + + nets = [n.strip() for n in args.networks.split(",") if n.strip()] + for net in nets: + if net not in NETWORKS: + raise SystemExit(f"unknown network {net!r}; known: {', '.join(NETWORKS)}") + + # Bootstraps may overlap each other; cycles get the box to themselves. + for net in nets: + if args.measure_blocks: + NETWORKS[net]["measure_blocks"] = args.measure_blocks + if args.gap_blocks: + NETWORKS[net]["gap_blocks"] = args.gap_blocks + + take_lock(shared=args.bootstrap) + try: + if args.bootstrap: + for net in nets: + bootstrap(net, args.state_root) + return 0 + + while True: + lines, all_ok = [], True + for net in nets: # serial on purpose: concurrent legs contend and both are junk + try: + result = cycle(net, args.state_root, args.results_dir) + if result is None: # gap not open yet, or no tip to compare against + continue + if result.get("status") != "ok": + all_ok = False + lines.append(summarise(result, load_history(args.results_dir, net))) + except Exception as exc: # isolate: one network must not stop the others + log(f"{net}: cycle failed: {exc}") + lines.append(f"*{net}*: cycle failed — `{exc}`") + all_ok = False + if lines: + post_slack(lines, all_ok=all_ok) + if args.once: + return 0 + time.sleep(3600) + finally: + release_lock() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tooling/sync/fullsync_metrics.py b/tooling/sync/fullsync_metrics.py new file mode 100644 index 00000000000..0958cf2b8b8 --- /dev/null +++ b/tooling/sync/fullsync_metrics.py @@ -0,0 +1,167 @@ +"""Metric extraction from an ethrex full-sync run log. + +Kept separate from the runner so it can be unit-tested against captured logs without +needing a node, a box or Docker. See issue #7111. + +Throughput is reported as the mean of the per-batch `Gigagas/s` values, deliberately +*not* as blocks divided by wall clock: wall clock includes startup state regeneration +(the node re-executes the uncommitted window on restart), which is a real signal but a +separate one, so it is extracted on its own instead of being folded into throughput. +""" + +import re +from datetime import datetime + +# `... [METRICS] Executed and stored: Range: 1024, Last block num: 25578362, ..., +# Total Gas: 31058623694, Throughput: 0.21086104043277357 Gigagas/s` +# The `(unified pipeline)` variant of the same line is also matched. +BATCH_RE = re.compile( + r"\[METRICS\] Executed and stored.*?" + r"Last block num: (?P\d+).*?" + r"Total Gas: (?P\d+).*?" + r"Throughput: (?P[0-9.]+) Gigagas/s" +) + +# `... Executed and stored 1024 blocks in 78.813 seconds (12.993 blocks/s). ...` +BLOCKS_PER_S_RE = re.compile( + r"Executed and stored \d+ blocks in [0-9.]+ seconds \((?P[0-9.]+) blocks/s\)" +) + +# `... [METRIC] BLOCK 500 0x5f69… | 1.750 Ggas/s | 51.41 ms | 2569 txs | 90 Mgas (98%)` +BLOCK_HEADER_RE = re.compile( + r"\[METRIC\] BLOCK (?P\d+) .*?\| [0-9.]+ Ggas/s \| (?P[0-9.]+) ms " + r"\| \d+ txs \| (?P[0-9.]+) Mgas" +) + +# ` |- exec: 47.41 ms (92%) << BOTTLENECK` / ` `- warmer: 20.33 ms [...]` +PHASE_RE = re.compile(r"[|`]- (?Pvalidate|exec|merkle|store): +(?P[0-9.]+) ms") + +REGEN_START_RE = re.compile(r"Regenerating state from block (?P\d+) to (?P\d+)") +REGEN_END_RE = re.compile(r"Finished regenerating state") + +# Leading tracing timestamp, e.g. `2026-07-28T15:06:49.398765Z INFO ...` +TS_RE = re.compile(r"(?P\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?)Z") + +# Phases whose cost we track per unit of gas. `merkle` is the drain time only: the bulk of +# merkleization runs concurrently with execution and is already reflected in `exec`. +TRACKED_PHASES = ("validate", "exec", "merkle", "store") + + +def _parse_ts(line): + match = TS_RE.search(line) + if not match: + return None + text = match.group("ts") + # Python's fromisoformat rejects more than 6 fractional digits. + if "." in text: + head, frac = text.split(".", 1) + text = f"{head}.{frac[:6]}" + try: + return datetime.fromisoformat(text) + except ValueError: + return None + + +def _mean(values): + return sum(values) / len(values) if values else None + + +def _median(values): + if not values: + return None + ordered = sorted(values) + mid = len(ordered) // 2 + if len(ordered) % 2: + return ordered[mid] + return (ordered[mid - 1] + ordered[mid]) / 2 + + +def _stdev(values): + if len(values) < 2: + return None + mean = _mean(values) + return (sum((v - mean) ** 2 for v in values) / len(values)) ** 0.5 + + +def parse_run(lines): + """Extract the tracked metrics from a run's log lines. + + Returns a dict; every value may be `None` when the log does not contain that signal + (e.g. a run that died before its first batch), so callers must treat the run as + invalid rather than assume zeros. + """ + throughputs = [] + blocks_per_s = [] + last_block = None + total_gas = 0 + + # Per-phase totals, accumulated as (ms, Mgas) so the ratio is gas-normalised. + phase_ms = {name: 0.0 for name in TRACKED_PHASES} + block_mgas = 0.0 + pending_mgas = None + + regen_start = None + regen_end = None + + for line in lines: + batch = BATCH_RE.search(line) + if batch: + throughputs.append(float(batch.group("throughput"))) + last_block = int(batch.group("last_block")) + total_gas += int(batch.group("gas")) + continue + + bps = BLOCKS_PER_S_RE.search(line) + if bps: + blocks_per_s.append(float(bps.group("bps"))) + continue + + header = BLOCK_HEADER_RE.search(line) + if header: + # A new per-block record: the phase lines that follow belong to this block. + pending_mgas = float(header.group("mgas")) + block_mgas += pending_mgas + continue + + phase = PHASE_RE.search(line) + if phase and pending_mgas is not None: + phase_ms[phase.group("phase")] += float(phase.group("ms")) + continue + + if REGEN_START_RE.search(line): + regen_start = _parse_ts(line) + continue + + if REGEN_END_RE.search(line): + regen_end = _parse_ts(line) + + regen_seconds = None + if regen_start and regen_end and regen_end >= regen_start: + regen_seconds = (regen_end - regen_start).total_seconds() + + # ms per Mgas: comparable across runs whose blocks carry different amounts of gas. + phase_ms_per_mgas = None + if block_mgas > 0: + phase_ms_per_mgas = { + name: round(total / block_mgas, 4) for name, total in phase_ms.items() + } + + return { + "batches": len(throughputs), + "throughput_ggas_s": { + "mean": _mean(throughputs), + "median": _median(throughputs), + "stdev": _stdev(throughputs), + "samples": throughputs, + }, + "blocks_per_s_mean": _mean(blocks_per_s), + "last_block": last_block, + "total_gas": total_gas or None, + "phase_ms_per_mgas": phase_ms_per_mgas, + "state_regen_seconds": regen_seconds, + } + + +def parse_run_file(path): + with open(path, "r", errors="replace") as handle: + return parse_run(handle) diff --git a/tooling/sync/test_fullsync_bench.py b/tooling/sync/test_fullsync_bench.py new file mode 100644 index 00000000000..cf1c6ecaba6 --- /dev/null +++ b/tooling/sync/test_fullsync_bench.py @@ -0,0 +1,90 @@ +"""Tests for the runner's chain-tip oracle. + +These pin a failure seen on the box: with the execution node stopped between cycles, the +beacon node reports `el_offline` and stops advancing its head. It sat 17 hours behind and +unmoving on all three networks, so a tip read from it made `tip - base` zero forever and +every cycle skipped itself — the watch would have run for weeks producing nothing. + +Run: python3 tooling/sync/test_fullsync_bench.py +""" + +import json +import os +import sys +import tempfile +import time + +sys.argv = ["test"] # fullsync_bench parses args only under __main__, but be explicit +import fullsync_bench as fb + + +def _base_at(head, age_seconds, with_created_at=True): + """A base directory recorded `age_seconds` ago at block `head`.""" + base = tempfile.mkdtemp() + created = time.time() - age_seconds + meta = {"head": head} + if with_created_at: + meta["created_at"] = created + path = os.path.join(base, fb.BASE_META) + with open(path, "w") as fh: + json.dump(meta, fh) + if not with_created_at: + os.utime(path, (created, created)) + return base + + +def test_tip_advances_while_the_beacon_node_is_frozen(): + # No beacon node is listening on the test port, which is the same situation as one + # that answers but is stuck: the estimate must come from elapsed time either way. + day = 24 * 3600 + base = _base_at(head=1_000_000, age_seconds=3 * day) + tip = fb.chain_tip("mainnet", base) + blocks = tip - 1_000_000 + # 3 days of 12s slots is 21600 slots; at the 0.97 fill ratio, ~20952 blocks. + assert 20_000 < blocks < 21_600, blocks + + +def test_estimate_stays_below_a_full_slot_count(): + # Over-estimating aims a leg at a block that does not exist yet and burns its whole + # timeout, so the estimate must never exceed one block per slot. + base = _base_at(head=500, age_seconds=10 * 24 * 3600) + slots = 10 * 24 * 3600 // fb.SLOT_SECONDS + assert fb.chain_tip("hoodi", base) - 500 < slots + + +def test_gap_opens_far_enough_to_release_a_cycle(): + # The practical question: after M days, has enough chain accumulated to measure? + measure = fb.NETWORKS["mainnet"]["measure_blocks"] + base = _base_at(head=25_691_647, age_seconds=4 * 24 * 3600) + assert fb.chain_tip("mainnet", base) - 25_691_647 >= measure + + +def test_a_fresh_base_releases_nothing(): + measure = fb.NETWORKS["mainnet"]["measure_blocks"] + base = _base_at(head=25_691_647, age_seconds=60) + assert fb.chain_tip("mainnet", base) - 25_691_647 < measure + + +def test_bases_without_created_at_fall_back_to_the_file_mtime(): + # The first bases were written before `created_at` was recorded; they must still work. + base = _base_at(head=42, age_seconds=3 * 24 * 3600, with_created_at=False) + assert "created_at" not in json.load(open(os.path.join(base, fb.BASE_META))) + assert fb.chain_tip("sepolia", base) - 42 > 20_000 + + +def _run_all(): + failures = 0 + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + try: + fn() + print(f"ok {name}") + except AssertionError as exc: + failures += 1 + print(f"FAIL {name}: {exc}") + print("all passed" if not failures else f"{failures} failure(s)") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(_run_all()) diff --git a/tooling/sync/test_fullsync_metrics.py b/tooling/sync/test_fullsync_metrics.py new file mode 100644 index 00000000000..9e5c2b8ebe2 --- /dev/null +++ b/tooling/sync/test_fullsync_metrics.py @@ -0,0 +1,109 @@ +"""Tests for the full-sync metric extraction. + +The fixtures below are real lines captured from ethrex full-sync runs on +`ethrex-mainnet-test-1` during the #7008/#7023 benchmarking, not invented samples — +the parser is only useful if it matches what the node actually emits. + +Run: python3 -m pytest tooling/sync/test_fullsync_metrics.py + or: python3 tooling/sync/test_fullsync_metrics.py +""" + +from fullsync_metrics import parse_run + +BATCH_LINES = [ + "2026-07-28T15:06:49.398765Z INFO [METRICS] Executed and stored: Range: 1024, Last block num: 25578362, Last block gas limit: 60000000, Total transactions: 460710, Total Gas: 31058623694, Throughput: 0.21295251385695252 Gigagas/s", + "2026-07-28T15:09:08.113221Z INFO Executed and stored 1024 blocks in 138.714 seconds (7.382 blocks/s). First block: 25578363 (0x0947…b9cc). Last block: 25579386 (0xe929…f320).", + "2026-07-28T15:09:08.113900Z INFO [METRICS] Executed and stored: Range: 1024, Last block num: 25579386, Last block gas limit: 60000000, Total transactions: 373323, Total Gas: 31381203148, Throughput: 0.21821452280452489 Gigagas/s", + "2026-07-28T15:11:30.552104Z INFO [METRICS] Executed and stored: Range: 1024, Last block num: 25580410, Last block gas limit: 60000000, Total transactions: 401221, Total Gas: 30939154682, Throughput: 0.2454995472715409 Gigagas/s", +] + +# The `(unified pipeline)` wording appeared on the #7008 branch; both must parse. +UNIFIED_LINE = "2026-07-20T20:03:43.225897Z INFO [METRICS] Executed and stored (unified pipeline): Range: 1024, Last block num: 25530262, Last block gas limit: 60000000, Total transactions: 471224, Total Gas: 30939154682, Throughput: 0.3936213097254789 Gigagas/s" + +PER_BLOCK_LINES = [ + "2026-07-23T17:38:20.225580Z INFO [METRIC] BLOCK 500 0x5f6929cf8b4d0fccf50629ed23704c43819811c5ddbe3bfa0aca83f463afaf2e | 1.750 Ggas/s | 51.41 ms | 2569 txs | 90 Mgas (98%)", + "2026-07-23T17:38:20.225584Z INFO |- validate: 3.50 ms ( 7%)", + "2026-07-23T17:38:20.225586Z INFO |- exec: 47.41 ms (92%) << BOTTLENECK", + "2026-07-23T17:38:20.225588Z INFO |- merkle: 0.07 ms ( 0%) [concurrent: 37.38 ms, drain: 0.07 ms, overlap: 100%, queue: 2, start_delay: 0.25 ms]", + "2026-07-23T17:38:20.225591Z INFO |- store: 0.43 ms ( 1%)", + "2026-07-23T17:38:20.225593Z INFO `- warmer: 20.33 ms [finished: 27.08 ms before exec]", +] + +REGEN_LINES = [ + "2026-07-28T15:02:10.100000Z INFO Regenerating state from block 25577200 to 25577338", + "2026-07-28T15:02:52.600000Z INFO Finished regenerating state", +] + + +def test_batch_throughput_mean_and_samples(): + got = parse_run(BATCH_LINES) + assert got["batches"] == 3 + assert got["throughput_ggas_s"]["samples"] == [ + 0.21295251385695252, + 0.21821452280452489, + 0.2454995472715409, + ] + # Mean of the three; the run's headline number. + assert abs(got["throughput_ggas_s"]["mean"] - 0.22555552797767277) < 1e-12 + assert got["last_block"] == 25580410 + assert got["total_gas"] == 31058623694 + 31381203148 + 30939154682 + + +def test_blocks_per_second_is_captured_separately(): + got = parse_run(BATCH_LINES) + assert abs(got["blocks_per_s_mean"] - 7.382) < 1e-9 + + +def test_unified_pipeline_wording_also_parses(): + got = parse_run([UNIFIED_LINE]) + assert got["batches"] == 1 + assert abs(got["throughput_ggas_s"]["mean"] - 0.3936213097254789) < 1e-12 + + +def test_phases_are_normalised_per_mgas(): + got = parse_run(PER_BLOCK_LINES) + phases = got["phase_ms_per_mgas"] + # One block of 90 Mgas: each phase's ms divided by 90. + assert abs(phases["exec"] - 47.41 / 90) < 1e-4 + assert abs(phases["validate"] - 3.50 / 90) < 1e-4 + assert abs(phases["merkle"] - 0.07 / 90) < 1e-4 + assert abs(phases["store"] - 0.43 / 90) < 1e-4 + + +def test_phase_lines_without_a_preceding_block_are_ignored(): + # Guards against a truncated log opening mid-record and skewing the ratio. + got = parse_run(PER_BLOCK_LINES[1:]) + assert got["phase_ms_per_mgas"] is None + + +def test_state_regeneration_time_is_its_own_signal(): + got = parse_run(REGEN_LINES) + assert abs(got["state_regen_seconds"] - 42.5) < 1e-6 + + +def test_run_with_no_batches_yields_none_not_zero(): + # A run that died before its first batch must be reported invalid, never as 0 Ggas/s. + got = parse_run(["2026-07-28T15:00:00.000000Z INFO Starting full sync cycle"]) + assert got["batches"] == 0 + assert got["throughput_ggas_s"]["mean"] is None + assert got["last_block"] is None + assert got["total_gas"] is None + assert got["state_regen_seconds"] is None + + +def _run_all(): + failures = 0 + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + try: + fn() + print(f"ok {name}") + except AssertionError as exc: + failures += 1 + print(f"FAIL {name}: {exc}") + print("all passed" if not failures else f"{failures} failure(s)") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(_run_all())