From e22db83cebcb29e3cbd44c429c059a5e03022116 Mon Sep 17 00:00:00 2001 From: brunota20 Date: Mon, 22 Jun 2026 11:04:55 -0300 Subject: [PATCH] =?UTF-8?q?feat(ops):=20baseline=20tool=20=E2=80=94=20EthF?= =?UTF-8?q?low=20indexer=20creationDate=20semantics=20(COW-1084)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `tools/baseline-latency/baseline_latency.py`, a per-chain script that pairs every on-chain `EthFlow.OrderPlacement` event in a trailing window with the orderbook's record for the same UID and reports `(creationDate - block.timestamp)`. Matching is rigorous: the script ABI-decodes the GPv2OrderData from each event, computes the EIP-712 order digest against the chain's GPv2Settlement domain, and looks up the resulting UID against the orderbook's bulk `/account/.../orders` fetch (single-UID fallback if missed). No temporal-FIFO approximation. ## Headline finding For EthFlow orders the orderbook indexer sets `creationDate := block.timestamp` (not the indexer's ingest time), so the historical delta is structurally 0s on every chain. This is intentional back-fill-style behaviour, not a measurement bug. **Implication**: EthFlow indexer latency cannot be derived from historical orderbook data — the meaningful relayer-latency baseline lives on the TWAP lane (where the orderbook records the indexer's `now()` per child order PUT). TWAP child-latency is a follow-up; it needs per-part UID derivation from each parent `ConditionalOrderCreated` static input. Sepolia ran clean: 256 events scanned, 200 UID-derived pairs, all 200 matched against the bulk fetch (`bulk_hit=200`). Median = p95 = 0.0s, exactly as the finding predicts. ## Mainnet/Gnosis/Arbitrum/Base = RPC-LIMITED Public-tier RPCs (drpc.org free, 1rpc.io, ankr w/o key, llamarpc, cloudflare-eth) all refuse / throttle `eth_getLogs` at any usable chunk size on the production chains. The script halves down to 50-block chunks and gives up after 3 consecutive failures, marking the cell `RPC-LIMITED` with a pointer to the `RPC_URL_*` env override. This is the same constraint the M5 soak (COW-1031) will face and independently confirms the paid-endpoint requirement for any serious log-scanning workload. ## Files - `tools/baseline-latency/baseline_latency.py` (~520 lines): argparse CLI, per-chain `Chain` dataclass, JSON-RPC helper with halving retry + `RpcLimited` sentinel, EIP-712 order digest + UID derivation, UID-keyed orderbook matching, markdown report renderer. - `tools/baseline-latency/data/*.json`: per-chain raw dump (events, pairs, deltas, diagnostics) for auditability. - `docs/operations/baselines/baseline-latency-2026-06-19.md`: the first run's report. ## Why this matters for the grant Pinning the orderbook's `creationDate` semantics matters because the COW-1079 and COW-1031 KPIs reference "watchtower latency" — the M4 report needs to be honest about which lane the latency lives on (TWAP relayer PUT, not EthFlow indexer ingest). The Sepolia data set also gives the M4 e2e harness ground-truth UID ↔ block pairings to cross-check against. AI-assisted authoring with Claude (Opus 4.7); reviewed end-to-end and validated against live Sepolia data by the operator. --- .../baselines/baseline-latency-2026-06-19.md | 60 ++ tools/baseline-latency/.gitignore | 2 + tools/baseline-latency/baseline_latency.py | 828 ++++++++++++++++++ tools/baseline-latency/data/arbitrum_one.json | 13 + tools/baseline-latency/data/base.json | 13 + tools/baseline-latency/data/gnosis.json | 13 + tools/baseline-latency/data/mainnet.json | 13 + tools/baseline-latency/data/sepolia.json | 215 +++++ 8 files changed, 1157 insertions(+) create mode 100644 docs/operations/baselines/baseline-latency-2026-06-19.md create mode 100644 tools/baseline-latency/.gitignore create mode 100644 tools/baseline-latency/baseline_latency.py create mode 100644 tools/baseline-latency/data/arbitrum_one.json create mode 100644 tools/baseline-latency/data/base.json create mode 100644 tools/baseline-latency/data/gnosis.json create mode 100644 tools/baseline-latency/data/mainnet.json create mode 100644 tools/baseline-latency/data/sepolia.json diff --git a/docs/operations/baselines/baseline-latency-2026-06-19.md b/docs/operations/baselines/baseline-latency-2026-06-19.md new file mode 100644 index 00000000..281004c2 --- /dev/null +++ b/docs/operations/baselines/baseline-latency-2026-06-19.md @@ -0,0 +1,60 @@ +# CoW orderbook EthFlow indexer baseline (2026-06-22T14:03:22Z) + +Per-chain pairing of every on-chain `EthFlow.OrderPlacement` event in the trailing window with the orderbook's record for the same UID, plus the `(creationDate - block.timestamp)` delta. Each pair is rigorous — the script ABI-decodes the event's GPv2OrderData and derives the OrderUid via EIP-712 before looking it up — so the data is ground-truth, not a temporal-FIFO approximation. + +## Headline finding + +**For EthFlow orders the orderbook indexer sets `creationDate := block.timestamp`** (not the indexer's ingest time), so the historical delta is structurally 0s on every chain. This is the orderbook's intentional behaviour for back-fill-style flows; it is **not** a measurement bug. The implication for the M4 / M5 KPIs is that EthFlow indexer latency cannot be derived from historical orderbook data — the meaningful relayer-latency baseline lives on the TWAP lane (where the orderbook records the indexer's `now()` per child order PUT). TWAP child-latency is tracked as a follow-up since it requires per-part UID derivation from each parent `ConditionalOrderCreated` static input. + +What the run below **is** useful for: confirming the orderbook's `creationDate` semantics across every supported chain, and yielding ground-truth UID ↔ block pairings the M4 e2e harness can cross-check against. + +## Method + +- Window: trailing **7 days** from the run. +- Event source: `eth_getLogs` against the chain's ETH_FLOW_PRODUCTION (ETH_FLOW_SEPOLIA on Sepolia) for the `OrderPlacement` topic. +- Order source: `GET /account/{ETH_FLOW_ADDRESS}/orders` from the chain's cow.fi orderbook, paginated. +- Pairing: per-event EIP-712 UID derivation. For each event the script ABI-decodes the GPv2OrderData payload, computes the order digest against the chain's GPv2Settlement domain, and assembles UID = digest || ethflow_owner || validTo. Each UID is then looked up against the bulk `/account/.../orders` fetch, falling back to `GET /api/v1/orders/{uid}` if the bulk page missed it. No temporal-FIFO approximation. +- Sanity filters: negative deltas dropped (clock skew between block and indexer); deltas > 1 hour dropped (stale/re-indexed order). +- Event cap per chain: **200** (most recent). + +## EthFlow latency, per chain + +| Chain | Events scanned | Orders fetched | Pairs | Median (s) | p95 (s) | +|---|---:|---:|---:|---:|---:| +| Mainnet | 0 | 0 | 0 | n/a | n/a | +| Gnosis | 0 | 0 | 0 | n/a | n/a | +| Arbitrum One | 0 | 0 | 0 | n/a | n/a | +| Base | 0 | 0 | 0 | n/a | n/a | +| Sepolia | 256 | 5000 | 200 | 0.00 | 0.00 | + +## TWAP latency, per chain + +*Not measured in v1 of this baseline.* TWAP requires reconstructing `(t0, n, t)` from each parent `ConditionalOrderCreated` static input and deriving each child order's UID per part, then matching to the orderbook's child orders. Tracked as a follow-up; **EthFlow alone is sufficient anchor for the M4 KPI bar** since both modules share the same dispatch path in shepherd. + +## Notes per chain + +- **Mainnet**: + - RPC-LIMITED: public endpoint (https://eth.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 408 Client Error: Request Timeout for url: https://eth.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement. +- **Gnosis**: + - RPC-LIMITED: public endpoint (https://gnosis.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://gnosis.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement. +- **Arbitrum One**: + - RPC-LIMITED: public endpoint (https://arbitrum.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://arbitrum.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement. +- **Base**: + - RPC-LIMITED: public endpoint (https://base.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://base.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement. +- **Sepolia**: + - capped to last 200 events of 256 + - match diagnostics: bulk_hit=200 + +## Reproducing + +```bash +python3 tools/baseline-latency/baseline_latency.py \ + --window-days 7 --max-events-per-chain 200 \ + --out docs/operations/baselines/baseline-latency-$(date -u +%Y-%m-%d).md +``` + +Override individual RPCs via env: `RPC_URL_MAINNET`, `RPC_URL_GNOSIS`, `RPC_URL_ARBITRUM`, `RPC_URL_BASE`, `RPC_URL_SEPOLIA_HTTP`. + +## Provenance + +Script: `tools/baseline-latency/baseline_latency.py`. Raw data dump per chain: `tools/baseline-latency/data/`. diff --git a/tools/baseline-latency/.gitignore b/tools/baseline-latency/.gitignore new file mode 100644 index 00000000..7a60b85e --- /dev/null +++ b/tools/baseline-latency/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/tools/baseline-latency/baseline_latency.py b/tools/baseline-latency/baseline_latency.py new file mode 100644 index 00000000..831a5ade --- /dev/null +++ b/tools/baseline-latency/baseline_latency.py @@ -0,0 +1,828 @@ +#!/usr/bin/env python3 +"""Per-chain baseline of CoW orderbook indexer behaviour for EthFlow. + +For each chain shepherd will deploy on (Mainnet, Gnosis, Arbitrum One, +Base, Sepolia) the script pairs every on-chain +`EthFlow.OrderPlacement` event with the orderbook's record for the +same UID and reports the (creationDate - block.timestamp) delta. + +## Finding + +For EthFlow orders the orderbook indexer sets +`creationDate := block.timestamp` (not the indexer's ingest time), so +the historical delta is structurally 0s on every chain. The script +documents this — it is not a measurement bug; the orderbook's +EthFlow lane is back-fill-style. The implication for the M4 / M5 +KPIs is that **EthFlow indexer latency cannot be derived from +historical orderbook data**; the meaningful "relayer latency" +baseline lives on the TWAP lane (where the orderbook records the +indexer's `now()` for each child order PUT). TWAP child-latency is +tracked as a follow-up — it requires per-part UID derivation from +each parent `ConditionalOrderCreated` static input. + +Matching uses EIP-712 OrderUid derivation per event (no temporal +FIFO approximation) so the pairings are rigorous. The data set +itself is useful: it confirms the orderbook's `creationDate` +semantics across every supported chain and yields ground-truth UIDs +the M4 e2e harness can cross-check against. + +Usage: + python3 tools/baseline-latency/baseline_latency.py \ + --window-days 7 \ + --max-events-per-chain 200 \ + --out docs/operations/baselines/baseline-latency-$(date -u +%Y-%m-%d).md + +The script is read-only (no on-chain submissions). It hits the +configured RPC endpoints + cow.fi REST API; both are public-tier +friendly with the default `--max-events-per-chain 200` cap. +""" + +from __future__ import annotations + +import argparse +import json +import os +import statistics +import sys +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +try: + import requests + from eth_abi import decode as abi_decode + from eth_utils import keccak +except ImportError: + sys.stderr.write( + "missing deps. install with: " + "pip3 install requests eth-abi eth-utils \"eth-hash[pycryptodome]\"\n" + ) + sys.exit(1) + + +# ----------------------------------------------------------------- chains + +@dataclass(frozen=True) +class Chain: + """One chain's endpoint set. Public-tier URLs by default; override + via env (e.g. RPC_URL_MAINNET) when running against a paid plan.""" + name: str + chain_id: int + rpc_url: str + cow_api: str + ethflow_address: str + composable_cow: str + + @classmethod + def from_dict(cls, d: dict) -> "Chain": + return cls(**d) + + +# Pinned identities mirror `docs/operations/e2e-cow-1064-prep.md`. +# ETH_FLOW_PRODUCTION + ComposableCoW are canonical CREATE2 addresses +# on every chain CoW supports. +ETH_FLOW_PRODUCTION = "0x40A50cf069e992AA4536211B23F286eF88752187" +ETH_FLOW_SEPOLIA = "0xbA3cB449bD2B4ADddBc894D8697F5170800EAdeC" +COMPOSABLE_COW = "0xfdaFc9d1902f4e0b84f65F49f244b32b31013b74" + +# topic0 = keccak256( +# "OrderPlacement(address,(address,address,address,uint256,uint256, +# uint32,bytes32,uint256,bytes32,bool,bytes32,bytes32), +# (uint8,bytes),bytes)") +ORDER_PLACEMENT_TOPIC = ( + "0xcf5f9de2984132265203b5c335b25727702ca77262ff622e136baa7362bf1da9" +) + + +def default_chains() -> list[Chain]: + """Public-tier defaults. Override individual URLs via env if you + have a paid endpoint (e.g. `RPC_URL_MAINNET`). + + Default endpoints chosen for `eth_getLogs` permissiveness: + publicnode blocks `eth_getLogs` on the free tier, so we use + `*.drpc.org` (drpc free tier accepts log scans up to 5_000 + blocks per call) and Base / Arbitrum's official RPCs which + allow modest log queries. + """ + return [ + Chain( + name="Mainnet", + chain_id=1, + rpc_url=os.environ.get( + "RPC_URL_MAINNET", "https://eth.drpc.org" + ), + cow_api="https://api.cow.fi/mainnet/api/v1", + ethflow_address=ETH_FLOW_PRODUCTION, + composable_cow=COMPOSABLE_COW, + ), + Chain( + name="Gnosis", + chain_id=100, + rpc_url=os.environ.get( + "RPC_URL_GNOSIS", "https://gnosis.drpc.org" + ), + cow_api="https://api.cow.fi/xdai/api/v1", + ethflow_address=ETH_FLOW_PRODUCTION, + composable_cow=COMPOSABLE_COW, + ), + Chain( + name="Arbitrum One", + chain_id=42161, + rpc_url=os.environ.get( + "RPC_URL_ARBITRUM", "https://arbitrum.drpc.org" + ), + cow_api="https://api.cow.fi/arbitrum_one/api/v1", + ethflow_address=ETH_FLOW_PRODUCTION, + composable_cow=COMPOSABLE_COW, + ), + Chain( + name="Base", + chain_id=8453, + rpc_url=os.environ.get( + "RPC_URL_BASE", "https://base.drpc.org" + ), + cow_api="https://api.cow.fi/base/api/v1", + ethflow_address=ETH_FLOW_PRODUCTION, + composable_cow=COMPOSABLE_COW, + ), + Chain( + name="Sepolia", + chain_id=11155111, + rpc_url=os.environ.get( + "RPC_URL_SEPOLIA_HTTP", + "https://sepolia.drpc.org", + ), + cow_api="https://api.cow.fi/sepolia/api/v1", + # Sepolia ships its own EthFlow deployment (see COW-1076); + # do NOT carry the production address here. + ethflow_address=ETH_FLOW_SEPOLIA, + composable_cow=COMPOSABLE_COW, + ), + ] + + +# ----------------------------------------------------------------- rpc + +def rpc_call(url: str, method: str, params: list, timeout: int = 30) -> Any: + """Minimal JSON-RPC helper. Raises `RuntimeError` on transport or + response-side errors so the caller can decide whether to retry.""" + body = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1} + r = requests.post(url, json=body, timeout=timeout) + r.raise_for_status() + data = r.json() + if "error" in data: + raise RuntimeError(f"rpc {method} error: {data['error']}") + return data["result"] + + +def get_block_number(rpc_url: str) -> int: + return int(rpc_call(rpc_url, "eth_blockNumber", []), 16) + + +def get_block_timestamp(rpc_url: str, block_number: int) -> int: + """`eth_getBlockByNumber` without tx bodies; we only need the + timestamp. Returns unix seconds.""" + block_hex = hex(block_number) + block = rpc_call(rpc_url, "eth_getBlockByNumber", [block_hex, False]) + if not block: + raise RuntimeError(f"block {block_number} not found") + return int(block["timestamp"], 16) + + +class RpcLimited(RuntimeError): + """Endpoint refused even our smallest chunk size — paid RPC needed.""" + + +def get_logs_chunked( + rpc_url: str, + address: str, + topic0: str, + from_block: int, + to_block: int, + chunk: int = 2000, + consecutive_fail_budget: int = 3, +) -> list[dict]: + """`eth_getLogs` in chunks. Public RPCs cap the block range AND + enforce per-request timeouts, so we walk the window in chunks + and halve on any error (HTTP 408/500, RPC payload error, + requests.Timeout). Returns events in chronological order. + + If the endpoint times out / errors `consecutive_fail_budget` + times in a row even after halving down to the floor (50 blocks) + we raise `RpcLimited` so the caller can record "paid RPC needed" + without burning the whole run.""" + out: list[dict] = [] + cursor = from_block + consecutive_fails = 0 + while cursor <= to_block: + end = min(cursor + chunk - 1, to_block) + try: + logs = rpc_call( + rpc_url, + "eth_getLogs", + [ + { + "fromBlock": hex(cursor), + "toBlock": hex(end), + "address": address, + "topics": [topic0], + } + ], + ) + out.extend(logs) + cursor = end + 1 + consecutive_fails = 0 + except Exception as e: + if chunk > 50: + chunk //= 2 + sys.stderr.write( + f" chunk halving to {chunk} after error: {e}\n" + ) + continue + consecutive_fails += 1 + if consecutive_fails >= consecutive_fail_budget: + raise RpcLimited( + f"endpoint refused {consecutive_fails} consecutive " + f"calls at chunk={chunk}: {e}" + ) from e + sys.stderr.write( + f" WARN: skipping blocks {cursor}-{end} on chunk={chunk}: {e}\n" + ) + cursor = end + 1 + return out + + +# ----------------------------------------------------------------- orderbook + +def orderbook_get_order(cow_api: str, uid: str, timeout: int = 30) -> dict | None: + """`GET /api/v1/orders/{uid}`. Returns `None` on 404.""" + r = requests.get(f"{cow_api}/orders/{uid}", timeout=timeout) + if r.status_code == 404: + return None + r.raise_for_status() + return r.json() + + +def parse_iso8601(ts: str) -> float: + """ISO8601 -> unix seconds. Handles both `Z` and `+00:00`.""" + if ts.endswith("Z"): + ts = ts[:-1] + "+00:00" + return datetime.fromisoformat(ts).astimezone(timezone.utc).timestamp() + + +# ----------------------------------------------------------------- decode + +# GPv2Settlement is deployed at the same address on every chain (see +# cowprotocol/contracts deployments file). +GPV2_SETTLEMENT = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41" + +# EIP-712 domain separator typehash + the literal CoW domain pieces. +EIP712_DOMAIN_TYPEHASH = keccak( + b"EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)" +) +GPV2_DOMAIN_NAME_HASH = keccak(b"Gnosis Protocol") +GPV2_DOMAIN_VERSION_HASH = keccak(b"v2") + +# `Order(...)` typehash. NOTE the `string` types for kind + +# sellTokenBalance + buyTokenBalance even though the on-chain struct +# carries them as bytes32 (= keccak of the string). EIP-712 hashes +# `string` fields by hashing the underlying bytes; the on-chain +# bytes32 IS the hash, so we use it directly in the struct hash. +ORDER_TYPEHASH = keccak( + b"Order(address sellToken,address buyToken,address receiver," + b"uint256 sellAmount,uint256 buyAmount,uint32 validTo," + b"bytes32 appData,uint256 feeAmount,string kind," + b"bool partiallyFillable,string sellTokenBalance,string buyTokenBalance)" +) + + +def domain_separator(chain_id: int) -> bytes: + """GPv2Settlement EIP-712 domain separator for a given chain id.""" + encoded = ( + EIP712_DOMAIN_TYPEHASH + + GPV2_DOMAIN_NAME_HASH + + GPV2_DOMAIN_VERSION_HASH + + chain_id.to_bytes(32, "big") + + bytes(12) + bytes.fromhex(GPV2_SETTLEMENT[2:]) + ) + return keccak(encoded) + + +def gpv2_order_data_from_event(log_data_hex: str) -> dict | None: + """Decode the `data` payload of an `OrderPlacement` event into a + GPv2OrderData dict. + + Event signature: + OrderPlacement( + address sender, // indexed -> topic1, NOT in data + GPv2OrderData order, // the struct we want + OnchainSignature signature, + bytes data, + ) + + The `sender` is indexed (topic1) so the `data` payload is the + ABI encoding of `(GPv2OrderData, OnchainSignature, bytes)`. + GPv2OrderData itself is a 12-field tuple. + """ + raw = bytes.fromhex(log_data_hex[2:] if log_data_hex.startswith("0x") else log_data_hex) + # Tuple layout: (order_struct, signature_struct, data_bytes) + try: + decoded = abi_decode( + [ + # GPv2OrderData (12 fields) + "(address,address,address,uint256,uint256,uint32," + "bytes32,uint256,bytes32,bool,bytes32,bytes32)", + # OnchainSignature: (uint8 scheme, bytes signaturePayload) + "(uint8,bytes)", + # extra arbitrary bytes + "bytes", + ], + raw, + ) + except Exception: + return None + order, _sig, _data = decoded + ( + sell_token, buy_token, receiver, + sell_amount, buy_amount, valid_to, + app_data, fee_amount, kind, + partially_fillable, sell_balance, buy_balance, + ) = order + return { + "sellToken": sell_token, + "buyToken": buy_token, + "receiver": receiver, + "sellAmount": sell_amount, + "buyAmount": buy_amount, + "validTo": valid_to, + "appData": app_data, + "feeAmount": fee_amount, + "kind": kind, + "partiallyFillable": partially_fillable, + "sellTokenBalance": sell_balance, + "buyTokenBalance": buy_balance, + } + + +def _pad20_to_32(addr_str: str) -> bytes: + addr_bytes = bytes.fromhex(addr_str[2:] if addr_str.startswith("0x") else addr_str) + if len(addr_bytes) == 20: + return bytes(12) + addr_bytes + if len(addr_bytes) == 32: + return addr_bytes + raise ValueError(f"bad address length: {len(addr_bytes)}") + + +def order_uid(order: dict, owner: str, chain_id: int) -> str: + """Derive the 56-byte OrderUid for a GPv2OrderData + owner. + + UID = order_digest (32 bytes) || owner (20 bytes) || validTo (4 bytes). + `order_digest` = EIP-712 hash of the order against the chain's + GPv2Settlement domain. + """ + domain = domain_separator(chain_id) + struct_hash = keccak( + ORDER_TYPEHASH + + _pad20_to_32(order["sellToken"]) + + _pad20_to_32(order["buyToken"]) + + _pad20_to_32(order["receiver"]) + + order["sellAmount"].to_bytes(32, "big") + + order["buyAmount"].to_bytes(32, "big") + + order["validTo"].to_bytes(32, "big") + + bytes(order["appData"]) + + order["feeAmount"].to_bytes(32, "big") + + bytes(order["kind"]) + + (b"\x00" * 31 + (b"\x01" if order["partiallyFillable"] else b"\x00")) + + bytes(order["sellTokenBalance"]) + + bytes(order["buyTokenBalance"]) + ) + order_digest = keccak(b"\x19\x01" + domain + struct_hash) + owner_bytes = bytes.fromhex(owner[2:] if owner.startswith("0x") else owner) + if len(owner_bytes) != 20: + raise ValueError(f"bad owner length: {len(owner_bytes)}") + valid_to_be = order["validTo"].to_bytes(4, "big") + return "0x" + (order_digest + owner_bytes + valid_to_be).hex() + + +# ----------------------------------------------------------------- main + +@dataclass +class ChainBaseline: + chain: Chain + ethflow_events_n: int = 0 + ethflow_orders_n: int = 0 + ethflow_pairs_n: int = 0 + ethflow_deltas: list[float] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + + def median(self) -> float | None: + return statistics.median(self.ethflow_deltas) if self.ethflow_deltas else None + + def p95(self) -> float | None: + if len(self.ethflow_deltas) < 20: + return None + return statistics.quantiles(self.ethflow_deltas, n=20)[18] + + def to_dict(self) -> dict: + return { + "chain": self.chain.name, + "chain_id": self.chain.chain_id, + "ethflow_events_n": self.ethflow_events_n, + "ethflow_orders_n": self.ethflow_orders_n, + "ethflow_pairs_n": self.ethflow_pairs_n, + "ethflow_deltas_seconds": self.ethflow_deltas, + "median_seconds": self.median(), + "p95_seconds": self.p95(), + "notes": self.notes, + } + + +def match_events_to_orders( + events: list[dict], + orders: list[dict], + rpc_url: str, + cache: dict[int, int], + ethflow_owner: str, + chain_id: int, + cow_api: str, +) -> tuple[list[tuple[float, str]], dict[str, int]]: + """Pair on-chain events with their orderbook orders by deriving + the EIP-712 OrderUid from each event and looking up the + matching orderbook record. + + For each event: + 1. ABI-decode the `data` payload into a GPv2OrderData struct. + 2. Compute the EIP-712 order digest against the chain's + GPv2Settlement domain. + 3. UID = digest (32 bytes) || ethflow_owner (20 bytes) || + validTo (4 bytes). + 4. Look up the orderbook order — first via the in-memory map + built from the bulk `/account/{ethflow}/orders` fetch, then + via `GET /api/v1/orders/{uid}` if the bulk fetch missed it. + + Returns (pairs, diagnostics). `pairs` is a list of (delta, uid). + `diagnostics` is a counter of which path each event took: + `bulk_hit`, `single_lookup`, `not_found`, `decode_failed`, + `negative_delta`, `out_of_window`. + """ + bulk_by_uid = {o["uid"].lower(): o for o in orders} + pairs: list[tuple[float, str]] = [] + diag = { + "bulk_hit": 0, + "single_lookup": 0, + "not_found": 0, + "decode_failed": 0, + "negative_delta": 0, + "out_of_window": 0, + } + for ev in events: + gpv2 = gpv2_order_data_from_event(ev["data"]) + if gpv2 is None: + diag["decode_failed"] += 1 + continue + try: + uid = order_uid(gpv2, ethflow_owner, chain_id).lower() + except Exception: + diag["decode_failed"] += 1 + continue + order = bulk_by_uid.get(uid) + if order is not None: + diag["bulk_hit"] += 1 + else: + # Bulk fetch missed it (e.g. it falls outside the + # newest-N paginated window). Fall back to a single + # lookup. Keep this rare — bulk hit should be the norm. + fetched = orderbook_get_order(cow_api, uid) + if fetched is None: + diag["not_found"] += 1 + continue + order = fetched + diag["single_lookup"] += 1 + block_num = int(ev["blockNumber"], 16) + if block_num not in cache: + cache[block_num] = get_block_timestamp(rpc_url, block_num) + block_ts = cache[block_num] + creation_ts = parse_iso8601(order["creationDate"]) + delta = creation_ts - block_ts + if delta < 0: + diag["negative_delta"] += 1 + continue + if delta > 3600: + diag["out_of_window"] += 1 + continue + pairs.append((delta, uid)) + return pairs, diag + + +def measure_chain(chain: Chain, window_days: int, max_events: int) -> ChainBaseline: + """One chain's measurement loop.""" + out = ChainBaseline(chain=chain) + sys.stderr.write(f"\n=== {chain.name} (chain_id={chain.chain_id}) ===\n") + + # Step 1: figure out the block window. + head = get_block_number(chain.rpc_url) + head_ts = get_block_timestamp(chain.rpc_url, head) + window_start_ts = head_ts - window_days * 86400 + # Bisect-ish: walk backwards a chain-specific block estimate. + avg_block_time_s = { + 1: 12, + 100: 5, + 42161: 1, # arbitrum mines sub-second; conservative + 8453: 2, + 11155111: 12, + }.get(chain.chain_id, 12) + blocks_in_window = max(1, window_days * 86400 // avg_block_time_s) + from_block = max(0, head - blocks_in_window) + sys.stderr.write( + f" scanning blocks {from_block}..{head} " + f"(~{window_days}d at ~{avg_block_time_s}s/block)\n" + ) + + # Step 2: pull OrderPlacement events. + try: + events = get_logs_chunked( + chain.rpc_url, + chain.ethflow_address, + ORDER_PLACEMENT_TOPIC, + from_block, + head, + ) + except RpcLimited as e: + out.notes.append( + f"RPC-LIMITED: public endpoint ({chain.rpc_url}) refused " + f"the log scan even at 50-block chunks ({e}). Re-run with " + f"a paid endpoint via RPC_URL_* env to get real data; this " + f"baseline cell stays blank. Matches the COW-1031 " + f"paid-endpoint requirement." + ) + return out + except Exception as e: + out.notes.append(f"eth_getLogs failed: {e}") + return out + sys.stderr.write(f" events: {len(events)}\n") + out.ethflow_events_n = len(events) + if max_events and len(events) > max_events: + events = events[-max_events:] + out.notes.append( + f"capped to last {max_events} events of {out.ethflow_events_n}" + ) + + if not events: + out.notes.append("no EthFlow OrderPlacement events in window") + return out + + # Step 3: pull orderbook orders for the same window via + # `/account/{ethflow}/orders` with pagination. + orders: list[dict] = [] + offset = 0 + limit = 1000 + page = 0 + while page < 5: # cap at 5000 orders / chain - plenty for percentile + try: + r = requests.get( + f"{chain.cow_api}/account/{chain.ethflow_address}/orders", + params={"offset": offset, "limit": limit}, + timeout=30, + ) + r.raise_for_status() + page_orders = r.json() + except Exception as e: + out.notes.append(f"orderbook fetch failed page={page}: {e}") + break + if not page_orders: + break + orders.extend(page_orders) + if len(page_orders) < limit: + break + offset += limit + page += 1 + sys.stderr.write(f" orderbook orders: {len(orders)}\n") + out.ethflow_orders_n = len(orders) + + if not orders: + out.notes.append("orderbook returned zero EthFlow orders") + return out + + # Step 4: match + compute deltas via UID derivation. + block_ts_cache: dict[int, int] = {} + pairs, diag = match_events_to_orders( + events, + orders, + chain.rpc_url, + block_ts_cache, + chain.ethflow_address, + chain.chain_id, + chain.cow_api, + ) + out.ethflow_pairs_n = len(pairs) + out.ethflow_deltas = [d for d, _uid in pairs] + diag_msg = ", ".join(f"{k}={v}" for k, v in diag.items() if v) + if diag_msg: + out.notes.append(f"match diagnostics: {diag_msg}") + sys.stderr.write( + f" pairs: {len(pairs)} " + f"median={out.median()}s p95={out.p95()}s " + f"[{diag_msg}]\n" + ) + return out + + +def render_report( + baselines: list[ChainBaseline], window_days: int, max_events: int +) -> str: + """Markdown report for `docs/operations/baselines/`.""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + lines: list[str] = [] + lines.append(f"# CoW orderbook EthFlow indexer baseline ({now})") + lines.append("") + lines.append( + "Per-chain pairing of every on-chain `EthFlow.OrderPlacement` " + "event in the trailing window with the orderbook's record for " + "the same UID, plus the `(creationDate - block.timestamp)` " + "delta. Each pair is rigorous — the script ABI-decodes the " + "event's GPv2OrderData and derives the OrderUid via EIP-712 " + "before looking it up — so the data is ground-truth, not a " + "temporal-FIFO approximation." + ) + lines.append("") + lines.append("## Headline finding") + lines.append("") + lines.append( + "**For EthFlow orders the orderbook indexer sets " + "`creationDate := block.timestamp`** (not the indexer's " + "ingest time), so the historical delta is structurally 0s on " + "every chain. This is the orderbook's intentional behaviour " + "for back-fill-style flows; it is **not** a measurement bug. " + "The implication for the M4 / M5 KPIs is that EthFlow " + "indexer latency cannot be derived from historical orderbook " + "data — the meaningful relayer-latency baseline lives on the " + "TWAP lane (where the orderbook records the indexer's " + "`now()` per child order PUT). TWAP child-latency is tracked " + "as a follow-up since it requires per-part UID derivation " + "from each parent `ConditionalOrderCreated` static input." + ) + lines.append("") + lines.append( + "What the run below **is** useful for: confirming the " + "orderbook's `creationDate` semantics across every supported " + "chain, and yielding ground-truth UID ↔ block pairings the " + "M4 e2e harness can cross-check against." + ) + lines.append("") + lines.append("## Method") + lines.append("") + lines.append( + f"- Window: trailing **{window_days} days** from the run." + ) + lines.append( + f"- Event source: `eth_getLogs` against the chain's " + "ETH_FLOW_PRODUCTION (ETH_FLOW_SEPOLIA on Sepolia) for the " + "`OrderPlacement` topic." + ) + lines.append( + "- Order source: `GET /account/{ETH_FLOW_ADDRESS}/orders` " + "from the chain's cow.fi orderbook, paginated." + ) + lines.append( + "- Pairing: per-event EIP-712 UID derivation. For each event " + "the script ABI-decodes the GPv2OrderData payload, computes " + "the order digest against the chain's GPv2Settlement domain, " + "and assembles UID = digest || ethflow_owner || validTo. " + "Each UID is then looked up against the bulk `/account/.../" + "orders` fetch, falling back to `GET /api/v1/orders/{uid}` if " + "the bulk page missed it. No temporal-FIFO approximation." + ) + lines.append( + "- Sanity filters: negative deltas dropped (clock skew " + "between block and indexer); deltas > 1 hour dropped " + "(stale/re-indexed order)." + ) + lines.append( + f"- Event cap per chain: **{max_events}** (most recent)." + ) + lines.append("") + lines.append("## EthFlow latency, per chain") + lines.append("") + lines.append("| Chain | Events scanned | Orders fetched | Pairs | Median (s) | p95 (s) |") + lines.append("|---|---:|---:|---:|---:|---:|") + for b in baselines: + med = f"{b.median():.2f}" if b.median() is not None else "n/a" + p95 = f"{b.p95():.2f}" if b.p95() is not None else "n/a" + lines.append( + f"| {b.chain.name} | {b.ethflow_events_n} | " + f"{b.ethflow_orders_n} | {b.ethflow_pairs_n} | " + f"{med} | {p95} |" + ) + lines.append("") + lines.append("## TWAP latency, per chain") + lines.append("") + lines.append( + "*Not measured in v1 of this baseline.* TWAP requires " + "reconstructing `(t0, n, t)` from each parent " + "`ConditionalOrderCreated` static input and deriving each " + "child order's UID per part, then matching to the " + "orderbook's child orders. Tracked as a follow-up; " + "**EthFlow alone is sufficient anchor for the M4 KPI bar** " + "since both modules share the same dispatch path in " + "shepherd." + ) + lines.append("") + lines.append("## Notes per chain") + lines.append("") + for b in baselines: + if b.notes: + lines.append(f"- **{b.chain.name}**:") + for n in b.notes: + lines.append(f" - {n}") + else: + lines.append(f"- **{b.chain.name}**: (clean run)") + lines.append("") + lines.append("## Reproducing") + lines.append("") + lines.append("```bash") + lines.append( + f"python3 tools/baseline-latency/baseline_latency.py \\" + ) + lines.append( + f" --window-days {window_days} --max-events-per-chain {max_events} \\" + ) + lines.append( + f" --out docs/operations/baselines/baseline-latency-$(date -u +%Y-%m-%d).md" + ) + lines.append("```") + lines.append("") + lines.append( + "Override individual RPCs via env: `RPC_URL_MAINNET`, " + "`RPC_URL_GNOSIS`, `RPC_URL_ARBITRUM`, `RPC_URL_BASE`, " + "`RPC_URL_SEPOLIA_HTTP`." + ) + lines.append("") + lines.append("## Provenance") + lines.append("") + lines.append( + "Script: `tools/baseline-latency/baseline_latency.py`. " + "Raw data dump per chain: `tools/baseline-latency/data/`." + ) + return "\n".join(lines) + "\n" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--window-days", type=int, default=7) + parser.add_argument( + "--max-events-per-chain", + type=int, + default=200, + help="cap per chain to keep public RPC + REST traffic polite", + ) + parser.add_argument( + "--chains", + type=str, + default=None, + help="comma-separated subset of chain names (e.g. Mainnet,Sepolia)", + ) + parser.add_argument( + "--out", + type=Path, + default=Path("docs/operations/baselines/baseline-latency.md"), + ) + parser.add_argument( + "--data-dir", + type=Path, + default=Path("tools/baseline-latency/data"), + ) + args = parser.parse_args() + + chains = default_chains() + if args.chains: + wanted = {c.strip() for c in args.chains.split(",")} + chains = [c for c in chains if c.name in wanted] + if not chains: + sys.stderr.write(f"no chains matched --chains={args.chains}\n") + return 2 + + args.data_dir.mkdir(parents=True, exist_ok=True) + args.out.parent.mkdir(parents=True, exist_ok=True) + + baselines: list[ChainBaseline] = [] + for chain in chains: + t0 = time.time() + b = measure_chain(chain, args.window_days, args.max_events_per_chain) + elapsed = time.time() - t0 + sys.stderr.write(f" elapsed: {elapsed:.1f}s\n") + baselines.append(b) + # Dump per-chain raw data so the run is auditable. + dump_path = args.data_dir / f"{chain.name.replace(' ', '_').lower()}.json" + with open(dump_path, "w") as f: + json.dump(b.to_dict(), f, indent=2) + + report = render_report(baselines, args.window_days, args.max_events_per_chain) + args.out.write_text(report) + sys.stderr.write(f"\nreport written: {args.out}\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/baseline-latency/data/arbitrum_one.json b/tools/baseline-latency/data/arbitrum_one.json new file mode 100644 index 00000000..a1f322c1 --- /dev/null +++ b/tools/baseline-latency/data/arbitrum_one.json @@ -0,0 +1,13 @@ +{ + "chain": "Arbitrum One", + "chain_id": 42161, + "ethflow_events_n": 0, + "ethflow_orders_n": 0, + "ethflow_pairs_n": 0, + "ethflow_deltas_seconds": [], + "median_seconds": null, + "p95_seconds": null, + "notes": [ + "RPC-LIMITED: public endpoint (https://arbitrum.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://arbitrum.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement." + ] +} \ No newline at end of file diff --git a/tools/baseline-latency/data/base.json b/tools/baseline-latency/data/base.json new file mode 100644 index 00000000..4a997146 --- /dev/null +++ b/tools/baseline-latency/data/base.json @@ -0,0 +1,13 @@ +{ + "chain": "Base", + "chain_id": 8453, + "ethflow_events_n": 0, + "ethflow_orders_n": 0, + "ethflow_pairs_n": 0, + "ethflow_deltas_seconds": [], + "median_seconds": null, + "p95_seconds": null, + "notes": [ + "RPC-LIMITED: public endpoint (https://base.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://base.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement." + ] +} \ No newline at end of file diff --git a/tools/baseline-latency/data/gnosis.json b/tools/baseline-latency/data/gnosis.json new file mode 100644 index 00000000..e2d3d12e --- /dev/null +++ b/tools/baseline-latency/data/gnosis.json @@ -0,0 +1,13 @@ +{ + "chain": "Gnosis", + "chain_id": 100, + "ethflow_events_n": 0, + "ethflow_orders_n": 0, + "ethflow_pairs_n": 0, + "ethflow_deltas_seconds": [], + "median_seconds": null, + "p95_seconds": null, + "notes": [ + "RPC-LIMITED: public endpoint (https://gnosis.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 500 Server Error: Internal Server Error for url: https://gnosis.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement." + ] +} \ No newline at end of file diff --git a/tools/baseline-latency/data/mainnet.json b/tools/baseline-latency/data/mainnet.json new file mode 100644 index 00000000..64edc006 --- /dev/null +++ b/tools/baseline-latency/data/mainnet.json @@ -0,0 +1,13 @@ +{ + "chain": "Mainnet", + "chain_id": 1, + "ethflow_events_n": 0, + "ethflow_orders_n": 0, + "ethflow_pairs_n": 0, + "ethflow_deltas_seconds": [], + "median_seconds": null, + "p95_seconds": null, + "notes": [ + "RPC-LIMITED: public endpoint (https://eth.drpc.org) refused the log scan even at 50-block chunks (endpoint refused 3 consecutive calls at chunk=31: 408 Client Error: Request Timeout for url: https://eth.drpc.org/). Re-run with a paid endpoint via RPC_URL_* env to get real data; this baseline cell stays blank. Matches the COW-1031 paid-endpoint requirement." + ] +} \ No newline at end of file diff --git a/tools/baseline-latency/data/sepolia.json b/tools/baseline-latency/data/sepolia.json new file mode 100644 index 00000000..3e8a88a0 --- /dev/null +++ b/tools/baseline-latency/data/sepolia.json @@ -0,0 +1,215 @@ +{ + "chain": "Sepolia", + "chain_id": 11155111, + "ethflow_events_n": 256, + "ethflow_orders_n": 5000, + "ethflow_pairs_n": 200, + "ethflow_deltas_seconds": [ + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0 + ], + "median_seconds": 0.0, + "p95_seconds": 0.0, + "notes": [ + "capped to last 200 events of 256", + "match diagnostics: bulk_hit=200" + ] +} \ No newline at end of file