diff --git a/cmd/p2p/sensor/clickhouse.md b/cmd/p2p/sensor/clickhouse.md new file mode 100644 index 000000000..b113c1628 --- /dev/null +++ b/cmd/p2p/sensor/clickhouse.md @@ -0,0 +1,518 @@ +# Sensor ClickHouse data model + +The tables the `--database=clickhouse` backend writes, and how a devp2p message +becomes rows in them. The writer is `p2p/database/clickhouse.go`; the handlers that +drive it are in `p2p/protocol.go`. + +The DDL itself is not in this repo — it lives in +[sensor-network-tools `clickhouse/schema.sql`](https://github.com/0xPolygon/sensor-network-tools/blob/main/clickhouse/schema.sql) +(canonical) and is mirrored into `polygon-infrastructure`, which applies it to the +ClickHouse VM. + +See also: [Datastore data model](/cmd/p2p/sensor/datastore.md). + +Block numbers collide across networks, so **each network gets its own database** +(`mainnet`, `amoy`) and there is no chain column. The DSN selects it: +`--clickhouse-dsn clickhouse://user:pass@host:9000/mainnet`. + +## Schema + +### The split + +Every kind of data is one of two things. + +**Fact tables** are keyed by a hash, and every column is a pure function of that +hash. Two sensors that see the same block therefore produce byte-identical rows, +so deduplication is a storage optimisation rather than a correctness requirement +and readers never need `FINAL`. + +**Observation tables** are append-only event streams — one row per (thing, +sensor, peer, time). Everything situational lives here and nowhere else: which +sensor, which peer, when, how the block was learned, and the announced total +difficulty. + +The consequence worth internalising: a partially-known block cannot corrupt a +fully-known one. A header carries no transaction count, so instead of writing +`tx_count = 0` onto the block row, body facts live in their own hash-keyed table +that the header path never touches. + +```mermaid +erDiagram + blocks { + UInt64 number PK + String hash PK + String parent_hash FK "-> blocks.hash, the parent block" + DateTime block_time "header consensus timestamp" + LowCardinality signer "precomputed ecrecover" + LowCardinality coinbase + UInt64 difficulty + UInt64 gas_used + UInt64 gas_limit + UInt256 base_fee + String uncle_hash + String state_root + String tx_root + String receipt_root + String logs_bloom "RAW BYTES not hex" + String extra_data "RAW BYTES not hex" + String mix_digest + UInt64 nonce + } + + block_bodies { + String hash PK "keyed by hash alone" + UInt32 tx_count + UInt16 uncle_count + Array uncles + } + + block_txs { + String block_hash PK + UInt32 tx_index PK + String tx_hash FK + Date seen_date "version column; no TTL, kept forever" + } + + transactions { + String hash PK + String from_address + String to_address "empty for contract creation" + UInt256 value + UInt64 gas + UInt256 gas_price + UInt256 gas_fee_cap + UInt256 gas_tip_cap + UInt64 nonce + UInt8 tx_type + UInt64 chain_id + String input_selector "first 4 bytes, no calldata" + UInt32 input_size + UInt16 access_list_size + UInt8 blob_count + UInt8 auth_list_size + Date seen_date "version column, TTL clock" + } + + block_events { + UInt64 block_number PK "denormalised, leads sort key" + String block_hash PK + LowCardinality sensor_id PK + LowCardinality node_id FK "devp2p node id" + LowCardinality source "hash_announce new_block header header_backfill body" + DateTime64 seen_at PK "ms precision" + UInt256 total_difficulty "announced value, 0 otherwise" + } + + tx_events { + String tx_hash PK + LowCardinality sensor_id + LowCardinality node_id FK + LowCardinality source "hash_announce full_tx" + DateTime64 seen_at PK + } + + peers { + LowCardinality sensor_id PK + String node_id PK + String name "client version" + String url "enode URL" + Array caps + DateTime64 seen_at PK + } + + blocks ||--o| block_bodies : "hash (when a body is seen)" + blocks ||--o{ block_txs : "block_hash" + block_txs }o--|| transactions : "tx_hash" + blocks ||--o{ block_events : "number + hash" + transactions ||--o{ tx_events : "hash" + peers ||--o{ block_events : "node_id" + peers ||--o{ tx_events : "node_id" +``` + +`blocks.parent_hash` points at another `blocks.hash`, forming the chain that reorg +and fork analysis walks. It is annotated on the column rather than drawn as a +relationship, because a self-reference renders as an easily-missed loop (or nothing +at all) depending on the mermaid version. + +| Table | Engine | Sort key | Retention | +| ------------------------ | -------------------------------------- | ------------------------------------------------ | --------- | +| `blocks` | `ReplacingMergeTree` (no version) | `(number, hash)` | forever | +| `block_bodies` | `ReplacingMergeTree` | `(hash)` | forever | +| `block_txs` | `ReplacingMergeTree(seen_date)` | `(block_hash, tx_index)` | forever | +| `transactions` | `ReplacingMergeTree(seen_date)` | `(hash)` | 14d | +| `block_events` | `MergeTree` | `(block_number, block_hash, sensor_id, seen_at)` | 14d | +| `tx_events` | `MergeTree` | `(tx_hash, seen_at)` | 14d | +| `peers` | `MergeTree` | `(sensor_id, node_id, seen_at)` | 14d | +| `block_total_difficulty` | `ReplacingMergeTree(total_difficulty)` | `(hash)` | forever | +| `block_events_first` | `AggregatingMergeTree` | `(block_number, block_hash, sensor_id, source)` | 14d | +| `tx_events_first` | `AggregatingMergeTree` | `(tx_hash)` | 14d | +| `block_forks` | `AggregatingMergeTree` | `(number)` | forever | +| `peers_current` | `ReplacingMergeTree(last_seen)` | `(sensor_id, node_id)` | 14d | +| `reorg_detections` | `MergeTree` | `(start_block, depth, detected_at)` | forever | +| `block_latency_metrics` | `MergeTree` | `(scope, hours_analyzed, timestamp)` | forever | + +**Retention is 14 days or forever, never anything in between.** Observations and +anything derived from them expire at 14 days; the content-addressed facts and the +analysis-job tables are kept. So the growth question is only about the `forever` +group — `block_txs` dominates it at roughly 47 GiB/year on mainnet (36.9 B/row on +disk after merge, ~91 transactions per block). + +Things the diagram cannot carry: + +- **`blocks` has no version column.** All rows for a hash are identical, so there + is nothing to order them by. `block_time` is header-derived, which means a hash's + duplicates always land in the same partition and the dedup key can actually + collapse — a dedup key that spans partitions never merges. +- **No encoded block size is stored.** It is only knowable from a full `NewBlock`, + not from a body delivered on its own, so it is not a function of the hash. Storing + it let two sensors write differing rows for one block, and the engine picked + arbitrarily — measured, the `0` won and the real size was discarded. +- **`block_bodies` and `block_txs` are keyed by hash alone, with no `number`.** A + body can arrive before, or without, its header (the sensor requests the two + separately and they race), so the height is not reliably known on that path. + Carrying `number` would mean writing `0` when unknown, reintroducing the exact + partial-row problem the split removes. The height is one join away. +- **`peers` → events is a join on `node_id`, not a foreign key.** It + works only because both sides record the devp2p node id. Get this wrong and the + join silently returns nothing. +- **`logs_bloom` and `extra_data` are raw bytes in a `String` column, not hex.** + Readers that re-run ecrecover depend on it; hex-decoding `extra_data` corrupts it + and silently breaks every signer-derived metric. +- **The post-Shanghai/Cancun header fields are deliberately absent.** `mix_digest` + and `nonce` are all-zero on Bor but stored, because clique's `encodeSigHeader` + includes them unconditionally and ecrecover needs them (as it needs `base_fee`). + `withdrawals_root`, `blob_gas_used`, `excess_blob_gas` and `parent_beacon_root` are + not stored: clique _panics_ if any is non-nil, so they can never take part in the + seal hash, and they are absent from mainnet and amoy headers. Add one back with + `ALTER TABLE ADD COLUMN` if that ever changes. +- **A fact table's partition key must be a function of its dedup key**, because + `ReplacingMergeTree` merges only within a partition. `blocks` satisfies this via + header-derived `block_time`. `transactions` and `block_txs` originally partitioned + on `seen_date`, which is ingest-derived, and so stranded duplicates in separate + partitions permanently: 10.07% of `transactions` rows survived a full + `OPTIMIZE FINAL`, 115k hashes spanning partitions. A transaction has no intrinsic + timestamp — unlike a block, whose header supplies one — so `seen_date` could not + be made key-derived, and both tables now bucket on their own key — + `cityHash64(hash)` for `transactions`, `cityHash64(block_hash)` for `block_txs`, + 16 buckets each. +- **`seen_date` is the version column on both tables.** It is the one column there + that is not a function of the key, so without a version the surviving row's value + was arbitrary. As a version it resolves to the latest sighting, which also means a + re-announced pending transaction's 14 days restart from when it was last seen. +- **Addresses are stored lowercase** — `signer`, `coinbase`, `from_address`, + `to_address`, via `addressHex`, never `common.Address.Hex()`, whose EIP-55 + checksum is a display format. ClickHouse compares case-sensitively and every + validator identity in this pipeline is lowercase, so a checksummed address joins + to nothing and raises nothing. Hashes are lowercase by construction. +- **Fact rows and provenance events are gated separately, per write path.** A + `blocks` row needs `--write-blocks`; an event needs either block-event flag. Mixing + them is the defect that recurred three times — `new_block`/`header`/`body` behind + `--write-block-events` alone, then `full_tx` behind `--write-tx-events` alone, then + `header`/`header_backfill` behind `--write-blocks` because `WriteBlockHeaders` + returned early before reaching the event check. Live headers are requested + regardless of `--write-blocks`, so `header` events were produced and discarded; + `header_backfill` could not even be produced there, parent backfill itself being + gated on `--write-blocks` in `getParentBlock`. +- **`ttl_only_drop_parts` requires a partition no coarser than the TTL.** It + suppresses row-level expiry and drops a part only once every row in it has + expired, so a partition spanning longer than the TTL pins expired rows. Both + `*_first` rollups partitioned monthly against a 14-day TTL: once background + merges combined a month's inserts into one part, a row from the 1st survived + until the 31st's row expired — 44 days, sawtoothing with the calendar. Every + table using the setting now partitions daily; the two that cannot + (`peers_current`, hash-bucketed `transactions`) do not use it. +- **Dedup happens on merge, so duplicates are transient, not absent.** Correct + partitioning makes them converge; it does not stop an unmerged part from holding + two rows for a key. A reader that must not double-count still needs `FINAL` or + `LIMIT 1 BY` — what it no longer needs is to compensate for inflation that never + converges. + +### Derived layer + +Rollups are maintained by materialized views on insert. Every rollup column is a +`SimpleAggregateFunction`, which merges exactly and needs no `-State`/`-Merge` +combinators — readers apply the plain function under a `GROUP BY`. + +The `v_*` views are the intended read surface, so latency arithmetic and fork +detection are defined once rather than in each consumer. + +```mermaid +flowchart LR + subgraph sensor["Written by the sensor"] + BS[(block_events)] + TS[(tx_events)] + PS[(peers)] + BL[(blocks)] + BB[(block_bodies)] + TX[(transactions)] + end + + subgraph jobs["Written by the analysis jobs"] + RD[(reorg_detections)] + BLM[(block_latency_metrics)] + end + + subgraph rollups["Rollups: AggregatingMergeTree fed by MVs"] + BSF[("block_events_first + per block x sensor x source - 14d")] + TSF[("tx_events_first + per tx, fleet-wide - 14d")] + BF[("block_forks + per height - forever")] + PC[("peers_current + per sensor x peer - 14d")] + end + + subgraph views["Read surface"] + VBL[v_block_latency] + VBP[v_block_provenance] + VB[v_blocks] + VF[v_forks] + VSC[v_sensor_coverage] + VP[v_peers] + VTP[v_tx_propagation] + VR[v_reorgs] + end + + BS -->|MV| BSF + TS -->|MV| TSF + BL -->|MV| BF + PS -->|MV| PC + + BSF -->|"propagation sources only"| VBL + BL --> VBL + BSF -->|"every source"| VBP + BSF --> VSC + BF --> VF + PC --> VP + TSF --> VTP + TX --> VTP + BL --> VB + BB --> VB + BTD[(block_total_difficulty)] --> VB + RD --> VR +``` + +`block_latency_metrics` is a report table, one row per job run per scope (`all` vs +`validated`). Its sort key `(scope, hours_analyzed, timestamp)` turns the job's +previous-run lookup into a reverse primary-index seek; on Datastore the same read +had to over-fetch 1000 rows and linearly scan for the matching pair. + +Every `v_*` view distinguishes "not seen" from a zero value with a `has_*` flag +and `NULL`s: `v_blocks.has_body` (so `tx_count`/`uncle_count` are `Nullable`, and +`uncles` is `[]` only because arrays cannot be), +`v_blocks.have_total_difficulty`, `v_block_latency.has_header` (so `block_time`, +`signer` and `latency_ms` are `Nullable`), and `v_tx_propagation.has_tx`. Filter on +the flag rather than testing a column for 0. + +Three traps in the derived layer: + +- **`v_peers` is not optional convenience.** `peers_current` is a genuine upsert + target, so unlike the fact tables its rows are _not_ identical. Joining it raw + fans out over unmerged snapshots — measured at 3x inflation on a freshly loaded + database (74 rows covering 27 distinct peers). `v_peers` collapses them with + `argMax`. +- **Timestamp names carry their scope.** `sensor_first_seen` / `sensor_last_seen` are + one sensor's earliest and latest; plain `first_seen` / `last_seen` are across every + sensor; `first_seen_latency_ms` is how far behind the earliest sensor a given sensor + was. `v_block_latency` exposes the first two side by side, so a per-sensor value is + never mistaken for a fleet-wide one. `tx_events_first` keeps a plain `first_seen` + because its grain is per transaction, which is already across sensors. +- **`v_block_latency.latency_ms` is `NULL` when the header has not been seen.** A + hash announcement is routinely recorded before its header, and the join would + otherwise supply `block_time = epoch` and yield a ~1.8e12 ms "latency" that + destroys any percentile over the column. Filter on `has_header`. Note also that + `latency_ms` is legitimately _negative_ for many Bor blocks: the header timestamp + is the proposer's slot time, which can be ahead of actual propagation. + +## Write paths + +This backend is **pure append**. No `Write*` method reads a row back in order to +modify it; the only read on the path is `HasBlock`, used to decide whether to +backfill a missing parent. + +Every write enqueues onto a per-table `rowBatcher` that flushes on a 1s tick or a +size threshold, retrying a failed batch up to 3 times. `add` is **non-blocking with +drop-on-full**, so a slow or unreachable database can never stall the sensor's hot +path — it drops rows and logs the count instead. + +```mermaid +flowchart LR + subgraph msgs["devp2p / timers"] + M1[NewBlockHashes] + M2[NewBlock] + M3[BlockHeaders] + M4[BlockBodies] + M5["Transactions and + PooledTransactions"] + M6[NewPooledTransactionHashes] + M7(["peer snapshot ticker + --peer-snapshot-interval, 30s"]) + end + + subgraph handlers["p2p/protocol.go"] + H1[handleNewBlockHashes] + H2[handleNewBlock] + H3[handleBlockHeaders] + H4[handleBlockBodies] + H5[processTransactions] + H6[handleNewPooledTransactionHashes] + H7[getParentBlock] + end + + subgraph api["Database interface"] + A1["WriteBlockEvents + carries block heights"] + A1b["WriteBlockHashFirstSeen + NO-OP"] + A2[WriteBlock] + A3[WriteBlockHeaders] + A4[WriteBlockBody] + A5[WriteTransactions] + A6[WriteTransactionEvents] + A7[WritePeers] + A8[HasBlock] + end + + subgraph tables["ClickHouse, via rowBatcher"] + T1[(blocks)] + T2[(block_bodies)] + T3[(block_txs)] + T4[(block_events)] + T5[(transactions)] + T6[(tx_events)] + T7[(peers)] + end + + M1 --> H1 + M2 --> H2 + M3 --> H3 + M4 --> H4 + M5 --> H5 + M6 --> H6 + M7 --> A7 + + H1 --> A1 + H1 --> A1b + H2 --> A2 + H3 --> A3 + H3 --> H7 + H4 --> A4 + H5 --> A5 + H6 --> A6 + H7 --> A8 + + A1 -->|"source=hash_announce"| T4 + A2 --> T1 + A2 -->|"tx_count, uncles"| T2 + A2 -->|"announced value, + never 0"| T9[block_total_difficulty] + A2 --> T3 + A2 -->|"source=new_block + + total_difficulty"| T4 + A2 --> T5 + A3 --> T1 + A3 -->|"source=header or + header_backfill"| T4 + A4 --> T2 + A4 --> T3 + A4 --> T5 + A4 -->|"source=body"| T4 + A5 --> T5 + A5 -->|"source=full_tx"| T6 + A6 -->|"source=hash_announce"| T6 + A7 --> T7 + A8 -.->|"point read, bloom index"| T1 +``` + +### Per-method detail + +| Method | Writes | Notes | +| ------------------------- | ----------------------------------------------------------------------------------------------- | ----------------------------------------------------- | +| `WriteBlock` | `blocks`, `block_bodies`, `block_total_difficulty`, `block_txs`, `block_events`, `transactions` | The only path with the whole block | +| `WriteBlockHeaders` | `blocks`, `block_events` | `isParent` picks `header_backfill`; gated separately | +| `WriteBlockBody` | `block_bodies`, `block_txs`, `transactions`, `block_events` | Event `source = 'body'` | +| `WriteBlockEvents` | `block_events` | Takes `[]BlockAnnouncement`, so heights reach the row | +| `WriteBlockHashFirstSeen` | nothing | Derived instead, see below | +| `WriteTransactions` | `transactions`, `tx_events` | Event under either tx-event flag | +| `WriteTransactionEvents` | `tx_events` | | +| `WritePeers` | `peers` | Own ticker, not the 2s metrics tick | +| `HasBlock` | — | Reads `blocks` by hash, once per new header | + +### Five things worth reading off this + +**`WriteBlockHashFirstSeen` is a no-op.** Earliest first-seen is derived from the +event stream by the `block_events_first` materialized view, so there is nothing to +stamp on the block. The interface method stays because the Datastore backend does +need it. + +Because that rollup is keyed by `source`, it replaces both Datastore column pairs at +once — `TimeFirstSeenHash`/`SensorFirstSeenHash` is `source = 'hash_announce'` and +`TimeFirstSeen`/`SensorFirstSeen` is `source = 'header'`. It expires at 14 days like +its source, so it is a pre-aggregation for query speed rather than a retention play; +`v_block_provenance` presents it. **Any latency read must restrict to the propagation +sources** (`hash_announce`, `new_block`), which `v_block_latency` does: the +sensor-requested sources carry no peer and their timestamps say when it chose to +fetch, not how fast the block arrived. + +**`blocks` is written by two paths, and that is safe.** Both `WriteBlock` and +`WriteBlockHeaders` write a _complete_ header row — every column is a pure function +of the header — so ordering between them cannot matter. The fields a header cannot +carry (`tx_count`, `uncle_count`, `uncles`) go to `block_bodies`, +which the header path never touches. + +This is the defect the schema redesign fixed. Previously the header path wrote +`tx_count = 0` onto the block row, and because the engine kept the row with the +newest version column, a header arriving _after_ the full block — routine during +parent backfill — permanently replaced the real counts with zeros. There is now a +regression test for exactly that ordering +(`TestClickHouseHeaderDoesNotClobberBody`). + +**`total_difficulty` is written twice, to two different kinds of table.** _Which +peer announced which value_ is an observation, so it rides on the `block_events` +row (header and hash-announce events write `0` there). _The value itself_ is a +property of the block, so it also goes to `block_total_difficulty` — hash-keyed and +kept forever, written only by the `NewBlock` path and never as `0`. + +It needs its own table rather than a `blocks` column for the same reason +`block_bodies` does: the header path does not know it and would write `0`, letting a +header clobber a real value. It used to be read out of `block_events_first`, which +worked only while that rollup outlived the raw stream; once retention was +normalised both were 14 days and `v_blocks.total_difficulty` returned `0` for every +older block — the same value that means "no peer ever announced it to us". Now +absence carries that meaning and `v_blocks` exposes it as `Nullable` (the table +column itself is a plain `UInt256`), so there is no sentinel. +`TestClickHouseTotalDifficultySurvivesEventExpiry` covers it. + +**Block heights have to reach `WriteBlockEvents`.** `block_events` leads its sort +key with `block_number`, which is what lets a reader fetch a whole range in one +query instead of one point lookup per hash. `NewBlockHashes` carries the numbers, so +the interface takes `[]database.BlockAnnouncement` rather than bare hashes — +and since `p2p.NewBlockHashesPacket` is defined as a slice of exactly that type, a +decoded packet is handed to the backend with no copy or conversion. + +**Peer identity is the devp2p node id**, not the enode URL, on both event +streams — so events join to `peers` / `peers_current`. The Datastore +backend uses `peer.URLv4()` here, which is why peers and events cannot be joined on +that backend. + +### Write volume + +Batch sizes are per table (`chBlockBatch` and friends): 5,000 blocks, 5,000 bodies, +20,000 block-txs, 50,000 block events, 20,000 transactions, 50,000 tx events, +2,000 peers — and `block_total_difficulty` reuses the 5,000 of bodies, its rows +being at most one per block, the same cardinality. + +`tx_events` is the volume driver, and only its `hash_announce` rows are: every peer +that announces a hash produces one, so the count scales with `--max-peers`. The +`--write-tx-events` / `--write-first-tx-event` pair is what bounds it — whether the +table receives every announcement or only first events. `full_tx` rows are recorded +under either flag, because a delivered body is roughly 2 rows per transaction per +sensor rather than a per-peer stream. Blocks work the same way, via +`recordsBlockEvents` / `recordsTxEvents`. + +Peer snapshots are their own cadence. The 2s tick in `sensor.go` still drives the +Prometheus gauge and the local peer file, but persisting up to `--max-peers` rows +every 2s is a large amount of near-duplicate data to answer one question ("who is +connected now"), so the database write runs on `--peer-snapshot-interval` (30s) and +reads go through `peers_current`. diff --git a/cmd/p2p/sensor/datastore.md b/cmd/p2p/sensor/datastore.md new file mode 100644 index 000000000..8bf535f15 --- /dev/null +++ b/cmd/p2p/sensor/datastore.md @@ -0,0 +1,308 @@ +# Sensor Datastore data model + +The entity kinds the `--database=datastore` backend writes, and how a devp2p message +becomes entities in them. The writer is `p2p/database/datastore.go`; the handlers +that drive it are in `p2p/protocol.go`. Kinds are Datastore's equivalent of tables, +and there is no DDL, so the structs in that file _are_ the schema. + +See also: [ClickHouse data model](/cmd/p2p/sensor/clickhouse.md). + +Unlike the ClickHouse backend, this one is a read-modify-write store: entities are +fetched and mutated in place to keep the earliest-seen timestamps. That shape is what +the ClickHouse schema was designed to stop emulating, so the two are not +column-for-column equivalents. + +## Schema + +### Kinds + +Relationships are Datastore `*datastore.Key` references, not enforced foreign +keys — a key can point at an entity that does not exist yet, and routinely does +(an event is written for a hash before the block itself arrives). + +```mermaid +erDiagram + blocks { + string __key__ PK "NameKey = block hash hex" + Key ParentHash FK "-> blocks (parent block)" + string Number "STRING, indexed - see note" + string GasUsed "indexed" + time Time "indexed" + time TimeFirstSeen "indexed" + time TTL "indexed" + bool IsParent "indexed" + string SensorFirstSeen "indexed" + time TimeFirstSeenHash "indexed" + string SensorFirstSeenHash "indexed" + string TotalDifficulty "noindex" + KeyList Transactions "noindex -> transactions" + KeyList Uncles "noindex -> blocks (uncle headers)" + string UncleHash "noindex" + string Coinbase "noindex" + string Root "noindex" + string TxHash "noindex" + string ReceiptHash "noindex" + bytes Bloom "noindex" + string Difficulty "noindex" + string GasLimit "noindex" + bytes Extra "noindex" + string MixDigest "noindex" + string Nonce "noindex" + string BaseFee "noindex" + } + + transactions { + string __key__ PK "NameKey = tx hash hex" + string From "indexed" + string To "indexed" + time Time "indexed" + time TimeFirstSeen "indexed" + time TTL "indexed" + int16 Type "indexed" + string SensorFirstSeen "indexed" + bytes Data "noindex - can exceed the index size cap" + string Gas "noindex" + string GasPrice "noindex" + string GasFeeCap "noindex" + string GasTipCap "noindex" + string Nonce "noindex" + string Value "noindex" + string V_R_S "noindex - signature" + } + + block_events { + string __key__ PK "IncompleteKey - auto-assigned id" + string SensorId + string PeerId "enode URL" + Key Hash FK "-> blocks" + time Time + time TTL + } + + transaction_events { + string __key__ PK "IncompleteKey - auto-assigned id" + string SensorId + string PeerId "enode URL" + Key Hash FK "-> transactions" + time Time + time TTL + } + + peers { + string __key__ PK "NameKey = devp2p node id" + string Name "client version" + string URL "enode URL" + string LastSeenBy "sensor id" + time TimeLastSeen + time TTL + StringList Caps "noindex" + } + + blocks ||--o{ block_events : "Hash" + transactions ||--o{ transaction_events : "Hash" + blocks }o--o{ transactions : "Transactions key list" +``` + +`ParentHash` and the `Uncles` key list both reference other `blocks` entities. They +are annotated on their columns rather than drawn as relationships, since a +self-reference renders as an easily-missed loop (or nothing) depending on the +mermaid version. + +`block_events` and `transaction_events` are the _same_ Go struct +(`DatastoreEvent`); they are separated only by the kind passed at key-creation +time, and the `Hash` reference points at `blocks` or `transactions` accordingly. + +### What to know before querying it + +- **`Number` is a string.** Range filters on it are therefore lexicographic over + decimal text, so `Number >= "9"` excludes `"10"`. Every consumer that scans block + ranges has to work around this, and `data-analysis/graph.go` carries an explicit + warning about it. This is the single biggest reason the ClickHouse schema uses a + real `UInt64`. +- **`noindex` is not cosmetic.** Datastore caps entities at 200 indexed properties + and indexed byte slices at a maximum size, which is why `Data`, `Bloom` and + `Extra` are excluded. The same cap is why the block-latency job has to drop + `contract_stats` and `seal_time_contract_stats` from its metrics entity before + writing. +- **`TTL` is a plain timestamp field, not an expiry mechanism.** Nothing deletes + these entities automatically; `data-analysis/cleanup.go` queries `TTL <= now` and + issues batched `DeleteMulti` calls. The ClickHouse schema replaces the whole file + with `TTL` clauses that drop partitions. +- **Observation attributes live on the entity.** `TimeFirstSeen`, + `SensorFirstSeen`, `TimeFirstSeenHash`, `SensorFirstSeenHash` and `IsParent` are + stored on the block itself and updated in place, which is what forces the + read-modify-write transactions. +- **Peer identity differs between kinds.** `peers` is keyed by devp2p node id, but + `block_events.PeerId` and `transaction_events.PeerId` hold the _enode URL_ + (`peer.URLv4()`). They are different key spaces, so peers cannot be joined to + events. That is why `NodeList` scans `block_events` ordered by `-Time` instead of + reading `peers`. +- **Uncles are first-class blocks here.** `writeBlock` and `writeBlockBody` write + each uncle header as its own `blocks` entity and link it via the `Uncles` key + list. The ClickHouse backend records only the uncle hashes on `block_bodies`. + +### Rough correspondence to the ClickHouse schema + +Not a migration map — the grain differs on purpose — but useful for orientation. + +| Datastore | ClickHouse | +| ------------------------------------------ | ------------------------------------------------ | +| `blocks` (header fields) | `blocks` | +| `blocks.Transactions` / `Uncles` key lists | `block_txs` / `block_bodies.uncles` | +| `blocks.TotalDifficulty` | `block_events.total_difficulty` | +| `blocks.TimeFirstSeen` / `SensorFirstSeen` | `block_events_first` (derived) | +| `blocks.TimeFirstSeenHash` | `block_events` with `source = 'hash_announce'` | +| `blocks.IsParent` | `block_events` with `source = 'header_backfill'` | +| `block_events` | `block_events` | +| `transactions` | `transactions` (+ `tx_type`, selector, chain id) | +| `transaction_events` | `tx_events` | +| `peers` | `peers` → `peers_current` | +| `TTL` field + cleanup job | `TTL` clauses, whole-partition drops | +| n/a | `block_forks`, `v_*` views | + +## Write paths + +The contrast with the ClickHouse backend is the whole point of this page: almost +every path here is a **read-modify-write** inside a `RunInTransaction` retry loop +(`MaxAttempts = 5`), and four separate paths contend on the _same_ `blocks` entity. +That is what keeps the earliest-seen timestamps correct, and it is the shape the +ClickHouse schema was designed to stop emulating. + +Writes are dispatched through `runAsync`, a semaphore sized by +`--max-db-concurrency` (default 10000). `Close` acquires every slot to guarantee no +write is in flight before closing the client. + +```mermaid +flowchart LR + subgraph msgs["devp2p / timers"] + M1[NewBlockHashes] + M2[NewBlock] + M3[BlockHeaders] + M4[BlockBodies] + M5["Transactions and + PooledTransactions"] + M6[NewPooledTransactionHashes] + M7(["peer ticker, 2s"]) + end + + subgraph api["Database interface"] + A1[WriteBlockEvents] + A1b[WriteBlockHashFirstSeen] + A2[WriteBlock] + A3[WriteBlockHeaders] + A4[WriteBlockBody] + A5[WriteTransactions] + A6[WriteTransactionEvents] + A7[WritePeers] + end + + subgraph rmw["Read-modify-write: RunInTransaction, up to 5 attempts"] + R1["writeBlock + Get then conditional Put"] + R2["writeBlockHeader + Get then conditional Put"] + R3["writeBlockBody + Get then conditional Put"] + R4["writeBlockHashFirstSeen + Get then conditional Put"] + end + + subgraph dedup["Read-then-write"] + D1["writeTransactions + GetMulti then PutMulti + skips earlier TimeFirstSeen"] + end + + subgraph blind["Blind writes"] + B1["writeEvents + PutMulti"] + B2["writeEvent + Put"] + B3["PutMulti"] + end + + subgraph kinds["Datastore kinds"] + K1[(blocks)] + K2[(block_events)] + K3[(transactions)] + K4[(transaction_events)] + K5[(peers)] + end + + M1 --> A1 + M1 --> A1b + M2 --> A2 + M3 --> A3 + M4 --> A4 + M5 --> A5 + M6 --> A6 + M7 --> A7 + + A1 --> B1 + A1b --> R4 + A2 --> B2 + A2 --> R1 + A3 --> R2 + A4 --> R3 + A5 --> D1 + A6 --> B1 + A7 --> B3 + + R1 -.->|"nested, inside the txn"| D1 + R1 -.->|"each uncle, nested"| R2 + R3 -.->|"nested, inside the txn"| D1 + R3 -.->|"each uncle, nested"| R2 + + R1 --> K1 + R2 --> K1 + R3 --> K1 + R4 --> K1 + B1 --> K2 + B1 --> K4 + B2 --> K2 + D1 --> K3 + B3 --> K5 +``` + +### The dotted edges + +They are the part that does not show up in a method list, and they are worth +tracing. `writeBlock` and `writeBlockBody` call `writeTransactions` (itself a +`GetMulti` + `PutMulti`) and `writeBlockHeader` (itself a whole transaction) _from +inside their own transaction closure_. On contention the closure re-runs, so those +nested round trips re-run with it, up to five times. + +### Per-method detail + +| Method | Path | Shape | +| ------------------------- | --------------------------- | ----------------------------------------------------------------------------------------------- | +| `WriteBlock` | `writeEvent` + `writeBlock` | Blind event `Put`, then RMW on the block; conditionally nests transactions and uncle headers | +| `WriteBlockHeaders` | `writeBlockHeader` | RMW per header; skips the write if an equal-or-earlier `TimeFirstSeen` exists | +| `WriteBlockBody` | `writeBlockBody` | RMW; fills `Transactions` / `Uncles` key lists only if still nil | +| `WriteBlockEvents` | `writeEvents` | Batched `PutMulti`, no read. Announced heights are discarded | +| `WriteBlockHashFirstSeen` | `writeBlockHashFirstSeen` | RMW on `blocks` purely to keep the earliest hash-announce time | +| `WriteTransactions` | `writeTransactions` | `GetMulti` to skip txs already stored with an earlier or equal `TimeFirstSeen`, then `PutMulti` | +| `WriteTransactionEvents` | `writeEvents` | Batched `PutMulti`, no read | +| `WritePeers` | `PutMulti` | One entity per connected peer, every 2s | +| `HasBlock` | `Get` | Point lookup by block key | +| `NodeList` | query | Scans `block_events` ordered by `-Time`, collecting distinct `PeerId` | + +Note that `WriteBlockHeaders` and `WriteBlockBody` deliberately write **no** events: +headers and bodies only arrive because the sensor asked for them, so the event is +recorded when the hash announcement comes in instead. + +### Why this differs from ClickHouse + +| | Datastore | ClickHouse | +| ------------------------- | ----------------------------------- | ------------------------------------------ | +| Write shape | read-modify-write per entity | append-only, batched | +| Contention | 4 paths on the same `blocks` entity | none; writers never read | +| "Keep the earliest event" | `tx.Get` then conditional `tx.Put` | derived at read time from the event stream | +| `WriteBlockHashFirstSeen` | its own transaction on `blocks` | no-op | +| Backpressure | semaphore, blocks the caller | fixed buffers, drop-on-full | +| Peer id in events | enode URL (`peer.URLv4()`) | devp2p node id | +| Block ↔ tx link | key list on the block entity | `block_txs` table | +| Uncles | written as full `blocks` entities | hashes on `block_bodies` | +| Expiry | `TTL` field + a manual delete job | `TTL` clauses, whole-partition drops | +| Block numbers | strings, lexicographic ranges | `UInt64` | +| `NodeList` | scan `block_events` by `-Time` | `peers_current` rollup | diff --git a/cmd/p2p/sensor/sensor.go b/cmd/p2p/sensor/sensor.go index e96703ed9..88f417117 100644 --- a/cmd/p2p/sensor/sensor.go +++ b/cmd/p2p/sensor/sensor.go @@ -53,6 +53,7 @@ type ( ShouldWriteTransactionEvents bool ShouldWriteFirstTransactionEvent bool ShouldWritePeers bool + PeerSnapshotInterval time.Duration ShouldBroadcastTx bool ShouldBroadcastTxHashes bool ShouldBroadcastBlocks bool @@ -193,6 +194,12 @@ var SensorCmd = &cobra.Command{ return errors.New("--validator-set-refresh must be greater than zero when --validate-block-signer is enabled with block broadcasting") } } + // Validated here rather than discovered at time.NewTicker, which panics on a + // non-positive interval -- and it would panic after the p2p ports are bound + // and the database is open, i.e. a crash loop rather than a startup error. + if inputSensorParams.PeerSnapshotInterval <= 0 { + return errors.New("--peer-snapshot-interval must be greater than zero") + } return nil }, @@ -318,6 +325,12 @@ var SensorCmd = &cobra.Command{ ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() + // Peer snapshots persist on their own slower cadence: the 2s tick above is + // cheap in-process work, but writing up to --max-peers rows every 2s is not, + // and "who is connected now" does not need that resolution. + peerSnapshotTicker := time.NewTicker(inputSensorParams.PeerSnapshotInterval) + defer peerSnapshotTicker.Stop() + if inputSensorParams.ShouldRunPprof { go handlePprof() } @@ -334,9 +347,10 @@ var SensorCmd = &cobra.Command{ select { case <-ticker.C: peersGauge.Set(float64(server.PeerCount())) - db.WritePeers(ctx, server.Peers(), time.Now()) metrics.Update(conns.HeadBlock().Block, conns.OldestBlock()) writePeers(server.Peers()) + case <-peerSnapshotTicker.C: + db.WritePeers(ctx, server.Peers(), time.Now()) case <-ctx.Done(): log.Info().Msg("Stopping sensor") return nil @@ -540,14 +554,17 @@ will result in less chance of missing data but can significantly increase memory f.BoolVarP(&inputSensorParams.ShouldWriteBlocks, "write-blocks", "B", true, "write blocks to database") f.BoolVar(&inputSensorParams.ShouldWriteBlockEvents, "write-block-events", true, "write block events to database") f.BoolVar(&inputSensorParams.ShouldWriteFirstBlockEvent, "write-first-block-event", false, - "write one block event on first-seen only (requires --write-block-events=false)") + "write one block event on first-seen only; ignored when --write-block-events is set") f.BoolVarP(&inputSensorParams.ShouldWriteTransactions, "write-txs", "t", true, `write transactions to database (this option can significantly increase CPU and memory usage)`) f.BoolVar(&inputSensorParams.ShouldWriteTransactionEvents, "write-tx-events", true, `write transaction events to database (this option can significantly increase CPU and memory usage)`) f.BoolVar(&inputSensorParams.ShouldWriteFirstTransactionEvent, "write-first-tx-event", false, - "write one transaction event on first-seen only (requires --write-tx-events=false)") + "write one transaction event on first-seen only; ignored when --write-tx-events is set") f.BoolVar(&inputSensorParams.ShouldWritePeers, "write-peers", true, "write peers to database") + f.DurationVar(&inputSensorParams.PeerSnapshotInterval, "peer-snapshot-interval", 30*time.Second, + `how often to persist the connected-peer set (requires --write-peers); lower +values multiply write volume by up to --max-peers rows per tick`) f.BoolVar(&inputSensorParams.ShouldBroadcastTx, "broadcast-txs", false, "broadcast full transactions to peers") f.BoolVar(&inputSensorParams.ShouldBroadcastTxHashes, "broadcast-tx-hashes", false, "broadcast transaction hashes to peers") f.BoolVar(&inputSensorParams.ShouldBroadcastBlocks, "broadcast-blocks", false, "broadcast full blocks to peers") diff --git a/cmd/p2p/sensor/usage.md b/cmd/p2p/sensor/usage.md index f2068435a..248e50d0f 100644 --- a/cmd/p2p/sensor/usage.md +++ b/cmd/p2p/sensor/usage.md @@ -2,13 +2,25 @@ Running the sensor will do peer discovery and continue to watch for blocks and transactions from those peers. This is useful for observing the network for forks and reorgs without the need to run the entire full node infrastructure. -The sensor can persist data to various backends including Google Cloud Datastore -or JSON output. If no nodes.json file exists at the specified path, it will be -created automatically. +The sensor can persist data to various backends including ClickHouse, Google Cloud +Datastore, or JSON output. If no nodes.json file exists at the specified path, it +will be created automatically. The bootnodes may change, so refer to the [Polygon Knowledge Layer][bootnodes] if the sensor is not discovering peers. +## Data Model + +The two persistent backends store different shapes, and are documented separately: + +- [ClickHouse data model](/cmd/p2p/sensor/clickhouse.md) — tables and write paths +- [Datastore data model](/cmd/p2p/sensor/datastore.md) — kinds and write paths + +The ClickHouse backend is append-only and batched; the Datastore backend does a +read-modify-write per entity. Select one with `--database`, and for ClickHouse pass +`--clickhouse-dsn`. The ClickHouse DDL lives in `clickhouse/schema.sql` in the +sensor-network-tools repo, not here. + ## JSON-RPC Server The sensor runs a JSON-RPC server on port 8545 (configurable via `--rpc-port`) @@ -16,22 +28,23 @@ that supports a subset of Ethereum JSON-RPC methods using cached data. ### Supported Methods -| Method | Description | -|--------|-------------| -| `eth_chainId` | Returns the chain ID | -| `eth_blockNumber` | Returns the current head block number | -| `eth_gasPrice` | Returns suggested gas price based on recent blocks | -| `eth_getBlockByHash` | Returns block by hash | -| `eth_getBlockByNumber` | Returns block by number (if cached) | -| `eth_getTransactionByHash` | Returns transaction by hash | -| `eth_getTransactionByBlockHashAndIndex` | Returns transaction at index in block | -| `eth_getBlockTransactionCountByHash` | Returns transaction count in block | -| `eth_getUncleCountByBlockHash` | Returns uncle count in block | -| `eth_sendRawTransaction` | Broadcasts signed transaction to peers | +| Method | Description | +| --------------------------------------- | -------------------------------------------------- | +| `eth_chainId` | Returns the chain ID | +| `eth_blockNumber` | Returns the current head block number | +| `eth_gasPrice` | Returns suggested gas price based on recent blocks | +| `eth_getBlockByHash` | Returns block by hash | +| `eth_getBlockByNumber` | Returns block by number (if cached) | +| `eth_getTransactionByHash` | Returns transaction by hash | +| `eth_getTransactionByBlockHashAndIndex` | Returns transaction at index in block | +| `eth_getBlockTransactionCountByHash` | Returns transaction count in block | +| `eth_getUncleCountByBlockHash` | Returns uncle count in block | +| `eth_sendRawTransaction` | Broadcasts signed transaction to peers | ### Limitations Methods requiring state or receipts are not supported: + - `eth_getBalance`, `eth_getCode`, `eth_call`, `eth_estimateGas` - `eth_getTransactionReceipt`, `eth_getLogs` diff --git a/doc/polycli_p2p_sensor.md b/doc/polycli_p2p_sensor.md index 320835bc1..35694fc45 100644 --- a/doc/polycli_p2p_sensor.md +++ b/doc/polycli_p2p_sensor.md @@ -23,13 +23,25 @@ Running the sensor will do peer discovery and continue to watch for blocks and transactions from those peers. This is useful for observing the network for forks and reorgs without the need to run the entire full node infrastructure. -The sensor can persist data to various backends including Google Cloud Datastore -or JSON output. If no nodes.json file exists at the specified path, it will be -created automatically. +The sensor can persist data to various backends including ClickHouse, Google Cloud +Datastore, or JSON output. If no nodes.json file exists at the specified path, it +will be created automatically. The bootnodes may change, so refer to the [Polygon Knowledge Layer][bootnodes] if the sensor is not discovering peers. +## Data Model + +The two persistent backends store different shapes, and are documented separately: + +- [ClickHouse data model](/cmd/p2p/sensor/clickhouse.md) — tables and write paths +- [Datastore data model](/cmd/p2p/sensor/datastore.md) — kinds and write paths + +The ClickHouse backend is append-only and batched; the Datastore backend does a +read-modify-write per entity. Select one with `--database`, and for ClickHouse pass +`--clickhouse-dsn`. The ClickHouse DDL lives in `clickhouse/schema.sql` in the +sensor-network-tools repo, not here. + ## JSON-RPC Server The sensor runs a JSON-RPC server on port 8545 (configurable via `--rpc-port`) @@ -37,22 +49,23 @@ that supports a subset of Ethereum JSON-RPC methods using cached data. ### Supported Methods -| Method | Description | -|--------|-------------| -| `eth_chainId` | Returns the chain ID | -| `eth_blockNumber` | Returns the current head block number | -| `eth_gasPrice` | Returns suggested gas price based on recent blocks | -| `eth_getBlockByHash` | Returns block by hash | -| `eth_getBlockByNumber` | Returns block by number (if cached) | -| `eth_getTransactionByHash` | Returns transaction by hash | -| `eth_getTransactionByBlockHashAndIndex` | Returns transaction at index in block | -| `eth_getBlockTransactionCountByHash` | Returns transaction count in block | -| `eth_getUncleCountByBlockHash` | Returns uncle count in block | -| `eth_sendRawTransaction` | Broadcasts signed transaction to peers | +| Method | Description | +| --------------------------------------- | -------------------------------------------------- | +| `eth_chainId` | Returns the chain ID | +| `eth_blockNumber` | Returns the current head block number | +| `eth_gasPrice` | Returns suggested gas price based on recent blocks | +| `eth_getBlockByHash` | Returns block by hash | +| `eth_getBlockByNumber` | Returns block by number (if cached) | +| `eth_getTransactionByHash` | Returns transaction by hash | +| `eth_getTransactionByBlockHashAndIndex` | Returns transaction at index in block | +| `eth_getBlockTransactionCountByHash` | Returns transaction count in block | +| `eth_getUncleCountByBlockHash` | Returns uncle count in block | +| `eth_sendRawTransaction` | Broadcasts signed transaction to peers | ### Limitations Methods requiring state or receipts are not supported: + - `eth_getBalance`, `eth_getCode`, `eth_call`, `eth_estimateGas` - `eth_getTransactionReceipt`, `eth_getLogs` @@ -143,75 +156,77 @@ polycli p2p sensor amoy-nodes.json \ ## Flags ```bash - --api-port uint port API server will listen on (default 8080) - --blocks-cache-ttl duration time to live for block cache entries (0 for no expiration) (default 10m0s) - -b, --bootnodes string comma separated nodes used for bootstrapping - --broadcast-block-hashes broadcast block hashes to peers - --broadcast-blocks broadcast full blocks to peers - --broadcast-tx-hashes broadcast transaction hashes to peers - --broadcast-txs broadcast full transactions to peers - --broadcast-workers int number of concurrent broadcast workers (default 4) - --cache-only-validated-blocks only cache and serve blocks signed by a known validator (unknown-signer blocks are still recorded to the database); has no effect without --validate-block-signer (default true) - --clickhouse-dsn string ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse) - --database string which database to persist data to, options are: - - datastore (GCP Datastore) - - clickhouse (ClickHouse, see --clickhouse-dsn) - - json (output to stdout) - - none (no persistence) (default "none") - -d, --database-id string datastore database ID - --dial-ratio int ratio of inbound to dialed connections (dial ratio of 2 allows 1/2 of connections to be dialed, setting to 0 defaults to 3) - --discovery-dns string DNS discovery ENR tree URL - --discovery-port int UDP P2P discovery port (default 30303) - --fork-id bytesHex hex encoded fork ID (omit 0x) (default 22D523B2) - --genesis-hash string genesis block hash (default "0xa9c28ce2141b56c474f1dc504bee9b01eb1bd7d1a507580d5519d4437a97de1b") - --heimdall-url string heimdall REST URL for the validator set (used to validate blocks before rebroadcast) (default "https://heimdall-api.polygon.technology") - -h, --help help for sensor - --key string hex-encoded private key (cannot be set with --key-file) - -k, --key-file string private key file (cannot be set with --key) - --known-txs-bloom-hashes uint number of hash functions for known txs bloom filter (default 7) - --known-txs-bloom-size uint bloom filter size in bits for tracking known transactions per peer (default ~40KB per filter, - optimized for ~32K elements with ~1% false positive rate) (default 327680) - --max-blocks int maximum blocks to track across all peers (0 for no limit) (default 1024) - -D, --max-db-concurrency int maximum number of concurrent database operations to perform (increasing this - will result in less chance of missing data but can significantly increase memory usage) (default 10000) - --max-known-blocks int maximum block hashes to track per peer (0 for no limit) (default 1024) - --max-parents int maximum parent block hashes to track per peer (0 for no limit) (default 1024) - -m, --max-peers int maximum number of peers to connect to (default 2000) - --max-queued-txs int maximum transaction announcements to queue per peer (default 4096) - --max-requests int maximum request IDs to track per peer (0 for no limit) (default 2048) - --max-tx-packet-size int target size in bytes for transaction broadcast packets (default 102400) - --max-txs int maximum transactions to cache for serving to peers (0 for no limit) (default 32768) - --nat string NAT port mapping mechanism (any|none|upnp|pmp|pmp:|extip:) (default "any") - -n, --network-id uint filter discovered nodes by this network ID - --no-discovery disable P2P peer discovery - --parents-cache-ttl duration time to live for parent hash cache entries (0 for no expiration) (default 5m0s) - --port int TCP network listening port (default 30303) - --pprof run pprof server - --pprof-port uint port pprof runs on (default 6060) - -p, --project-id string GCP project ID - --prom run Prometheus server (default true) - --prom-port uint port Prometheus runs on (default 2112) - --proxy-rpc proxy unsupported RPC methods to the --rpc endpoint - --proxy-rpc-timeout duration timeout for proxied RPC requests (default 30s) - --requests-cache-ttl duration time to live for requests cache entries (0 for no expiration) (default 5m0s) - --rpc string RPC endpoint used to fetch latest block (default "https://polygon-rpc.com") - --rpc-port uint port for JSON-RPC server to receive transactions (default 8545) - -s, --sensor-id string sensor ID when writing block/tx events - --static-nodes string static nodes file - --trusted-nodes string trusted nodes file - --ttl duration time to live (default 336h0m0s) - --tx-batch-timeout duration timeout for batching transactions before broadcast (default 500ms) - --tx-broadcast-queue-size int capacity of transaction broadcast queue (default 100000) - --txs-cache-ttl duration time to live for transaction cache entries (0 for no expiration) (default 10m0s) - --validate-block-signer only rebroadcast blocks signed by a validator in the heimdall validator set (default true) - --validator-set-refresh duration interval to refresh the validator set from heimdall (default 5m0s) - --write-block-events write block events to database (default true) - -B, --write-blocks write blocks to database (default true) - --write-first-block-event write one block event on first-seen only (requires --write-block-events=false) - --write-first-tx-event write one transaction event on first-seen only (requires --write-tx-events=false) - --write-peers write peers to database (default true) - --write-tx-events write transaction events to database (this option can significantly increase CPU and memory usage) (default true) - -t, --write-txs write transactions to database (this option can significantly increase CPU and memory usage) (default true) + --api-port uint port API server will listen on (default 8080) + --blocks-cache-ttl duration time to live for block cache entries (0 for no expiration) (default 10m0s) + -b, --bootnodes string comma separated nodes used for bootstrapping + --broadcast-block-hashes broadcast block hashes to peers + --broadcast-blocks broadcast full blocks to peers + --broadcast-tx-hashes broadcast transaction hashes to peers + --broadcast-txs broadcast full transactions to peers + --broadcast-workers int number of concurrent broadcast workers (default 4) + --cache-only-validated-blocks only cache and serve blocks signed by a known validator (unknown-signer blocks are still recorded to the database); has no effect without --validate-block-signer (default true) + --clickhouse-dsn string ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse) + --database string which database to persist data to, options are: + - datastore (GCP Datastore) + - clickhouse (ClickHouse, see --clickhouse-dsn) + - json (output to stdout) + - none (no persistence) (default "none") + -d, --database-id string datastore database ID + --dial-ratio int ratio of inbound to dialed connections (dial ratio of 2 allows 1/2 of connections to be dialed, setting to 0 defaults to 3) + --discovery-dns string DNS discovery ENR tree URL + --discovery-port int UDP P2P discovery port (default 30303) + --fork-id bytesHex hex encoded fork ID (omit 0x) (default 22D523B2) + --genesis-hash string genesis block hash (default "0xa9c28ce2141b56c474f1dc504bee9b01eb1bd7d1a507580d5519d4437a97de1b") + --heimdall-url string heimdall REST URL for the validator set (used to validate blocks before rebroadcast) (default "https://heimdall-api.polygon.technology") + -h, --help help for sensor + --key string hex-encoded private key (cannot be set with --key-file) + -k, --key-file string private key file (cannot be set with --key) + --known-txs-bloom-hashes uint number of hash functions for known txs bloom filter (default 7) + --known-txs-bloom-size uint bloom filter size in bits for tracking known transactions per peer (default ~40KB per filter, + optimized for ~32K elements with ~1% false positive rate) (default 327680) + --max-blocks int maximum blocks to track across all peers (0 for no limit) (default 1024) + -D, --max-db-concurrency int maximum number of concurrent database operations to perform (increasing this + will result in less chance of missing data but can significantly increase memory usage) (default 10000) + --max-known-blocks int maximum block hashes to track per peer (0 for no limit) (default 1024) + --max-parents int maximum parent block hashes to track per peer (0 for no limit) (default 1024) + -m, --max-peers int maximum number of peers to connect to (default 2000) + --max-queued-txs int maximum transaction announcements to queue per peer (default 4096) + --max-requests int maximum request IDs to track per peer (0 for no limit) (default 2048) + --max-tx-packet-size int target size in bytes for transaction broadcast packets (default 102400) + --max-txs int maximum transactions to cache for serving to peers (0 for no limit) (default 32768) + --nat string NAT port mapping mechanism (any|none|upnp|pmp|pmp:|extip:) (default "any") + -n, --network-id uint filter discovered nodes by this network ID + --no-discovery disable P2P peer discovery + --parents-cache-ttl duration time to live for parent hash cache entries (0 for no expiration) (default 5m0s) + --peer-snapshot-interval duration how often to persist the connected-peer set (requires --write-peers); lower + values multiply write volume by up to --max-peers rows per tick (default 30s) + --port int TCP network listening port (default 30303) + --pprof run pprof server + --pprof-port uint port pprof runs on (default 6060) + -p, --project-id string GCP project ID + --prom run Prometheus server (default true) + --prom-port uint port Prometheus runs on (default 2112) + --proxy-rpc proxy unsupported RPC methods to the --rpc endpoint + --proxy-rpc-timeout duration timeout for proxied RPC requests (default 30s) + --requests-cache-ttl duration time to live for requests cache entries (0 for no expiration) (default 5m0s) + --rpc string RPC endpoint used to fetch latest block (default "https://polygon-rpc.com") + --rpc-port uint port for JSON-RPC server to receive transactions (default 8545) + -s, --sensor-id string sensor ID when writing block/tx events + --static-nodes string static nodes file + --trusted-nodes string trusted nodes file + --ttl duration time to live (default 336h0m0s) + --tx-batch-timeout duration timeout for batching transactions before broadcast (default 500ms) + --tx-broadcast-queue-size int capacity of transaction broadcast queue (default 100000) + --txs-cache-ttl duration time to live for transaction cache entries (0 for no expiration) (default 10m0s) + --validate-block-signer only rebroadcast blocks signed by a validator in the heimdall validator set (default true) + --validator-set-refresh duration interval to refresh the validator set from heimdall (default 5m0s) + --write-block-events write block events to database (default true) + -B, --write-blocks write blocks to database (default true) + --write-first-block-event write one block event on first-seen only; ignored when --write-block-events is set + --write-first-tx-event write one transaction event on first-seen only; ignored when --write-tx-events is set + --write-peers write peers to database (default true) + --write-tx-events write transaction events to database (this option can significantly increase CPU and memory usage) (default true) + -t, --write-txs write transactions to database (this option can significantly increase CPU and memory usage) (default true) ``` The command also inherits flags from parent commands. diff --git a/p2p/database/clickhouse.go b/p2p/database/clickhouse.go index c6ac21e54..f335f0e31 100644 --- a/p2p/database/clickhouse.go +++ b/p2p/database/clickhouse.go @@ -1,9 +1,12 @@ package database import ( + "bytes" "context" + "encoding/hex" "fmt" "math/big" + "strings" "sync" "sync/atomic" "time" @@ -31,16 +34,36 @@ const ( // transient errors without delaying shutdown. chMaxFlushAttempts = 3 chBlockBatch = 5000 + chBlockBodyBatch = 5000 + chBlockTxBatch = 20000 chBlockEventBatch = 50000 chTxBatch = 20000 chTxEventBatch = 50000 chPeerBatch = 2000 + // How often to restate that the backend is unreachable. + chUnavailableWarnInterval = 1 * time.Minute +) + +// Event sources, so consumers can tell a hash announcement from a delivered +// header or body. +const ( + srcHashAnnounce = "hash_announce" + srcNewBlock = "new_block" + srcHeader = "header" + srcHeaderBackfill = "header_backfill" + srcBody = "body" + srcFullTx = "full_tx" ) // ClickHouse implements the Database interface backed by a ClickHouse cluster. -// The table definitions this writer targets (and the block_first_seen -// materialized view) live in the sensor-network-tools repo -// (clickhouse_schema.sql), not this repo. +// The table definitions live in clickhouse_schema.sql, in the sensor-network-tools +// and polygon-infrastructure repos rather than here. +// +// The schema separates content-addressed facts from observations and this writer +// matches it: every row written to a fact table (blocks, block_bodies, block_txs, +// transactions) is complete and a pure function of its hash, so two sensors emit +// byte-identical rows and no write can partially overwrite another. Everything +// observational goes to the event streams. type ClickHouse struct { conn driver.Conn sensorID string @@ -54,11 +77,26 @@ type ClickHouse struct { shouldWriteFirstTransactionEvent bool shouldWritePeers bool - blocks *rowBatcher[chBlock] - blockEvt *rowBatcher[chEvent] - txs *rowBatcher[chTx] - txEvt *rowBatcher[chEvent] - peers *rowBatcher[chPeer] + blocks *rowBatcher[chBlock] + blockBodies *rowBatcher[chBlockBody] + blockTD *rowBatcher[chBlockTD] + blockTxs *rowBatcher[chBlockTx] + blockEvt *rowBatcher[chBlockEvent] + txs *rowBatcher[chTx] + txEvt *rowBatcher[chTxEvent] + peers *rowBatcher[chPeerSnapshot] + + // discarded approximates the rows dropped because the backend was never + // reachable, so the periodic warning can say roughly how much has been lost. It + // previously counted HasBlock calls instead, which made it read 0 forever in + // exactly the configurations where everything was being dropped. + // + // Approximate in one direction on purpose: the nil-connection check precedes + // each method's flag gates, because with no connection the flags are moot -- so + // a fleet running --write-peers=false still counts the peer rows it would not + // have written. Erring toward over-reporting a dead backend is the right way + // round; the number is a magnitude for an operator, not an accounting figure. + discarded atomic.Uint64 // cancel stops the batcher goroutines; wg tracks them so Close can wait for // their final drain flush before the connection is closed. @@ -101,14 +139,33 @@ func NewClickHouse(ctx context.Context, opts ClickHouseOptions) Database { conn, err := connectClickHouse(ctx, opts.DSN) if err != nil { - log.Error().Err(err).Msg("Could not initialize ClickHouse connection") + // The sensor keeps running so a database outage does not take down a vantage + // point, but that degradation used to be invisible: one error at startup and + // then every write silently discarded, while the sensor peered, tracked the + // head and looked entirely healthy. That is exactly what a ClickHouse auth + // failure did -- two sensors ran for an hour writing nothing. + // + // So keep saying so. A dead backend is now visible in the logs for as long as + // it is dead, not only in the line that scrolled past at boot. + log.Error().Err(err).Msg("Could not initialize ClickHouse connection; ALL WRITES WILL BE DISCARDED") + // Its own cancel, stored on c, so Close stops it. Without this Close waits on + // a goroutine nothing can stop -- the same defect as the batchers inheriting + // the caller's context, and it hangs shutdown rather than losing rows. + wctx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + c.cancel = cancel + c.startUnavailableWarning(wctx) return c } c.conn = conn - // Derive a cancellable context so Close can stop the batchers independently - // of the parent context. - bctx, cancel := context.WithCancel(ctx) + // The batchers must outlive the caller's context, so their own cancellation is + // detached from it -- context.WithCancel(ctx) would inherit it and defeat the + // point. The sensor shuts down on signal-context cancellation and only stops + // serving peers afterwards, so an inherited context makes the batchers drain + // and exit while peers are still writing; those rows land in the buffered + // channel with no reader, are never flushed, and are not counted as dropped, + // so the loss is silent. Close is the only thing that may stop them. + bctx, cancel := context.WithCancel(context.WithoutCancel(ctx)) c.cancel = cancel c.startBatchers(bctx) @@ -166,30 +223,44 @@ func connectClickHouse(ctx context.Context, dsn string) (driver.Conn, error) { // handling lives in newInsertBatcher. func (c *ClickHouse) startBatchers(ctx context.Context) { c.blocks = newInsertBatcher(ctx, c, "blocks", chBlockBatch, - "INSERT INTO blocks (hash, number, parent_hash, block_time, coinbase, signer, difficulty, total_difficulty, gas_used, gas_limit, base_fee, tx_count, uncle_count, uncle_hash, state_root, tx_root, receipt_root, logs_bloom, extra_data, mix_digest, nonce, sensor_id, ingested_at, is_parent)", + "INSERT INTO blocks (number, hash, parent_hash, block_time, signer, coinbase, difficulty, gas_used, gas_limit, base_fee, uncle_hash, state_root, tx_root, receipt_root, logs_bloom, extra_data, mix_digest, nonce)", func(b driver.Batch, r chBlock) error { - return b.Append(r.hash, r.number, r.parentHash, r.blockTime, r.coinbase, r.signer, r.difficulty, r.totalDifficulty, r.gasUsed, r.gasLimit, r.baseFee, r.txCount, r.uncleCount, r.uncleHash, r.stateRoot, r.txRoot, r.receiptRoot, r.logsBloom, r.extraData, r.mixDigest, r.nonce, c.sensorID, r.ingestedAt, r.isParent) + return b.Append(r.number, r.hash, r.parentHash, r.blockTime, r.signer, r.coinbase, r.difficulty, r.gasUsed, r.gasLimit, r.baseFee, r.uncleHash, r.stateRoot, r.txRoot, r.receiptRoot, r.logsBloom, r.extraData, r.mixDigest, r.nonce) + }) + c.blockBodies = newInsertBatcher(ctx, c, "block_bodies", chBlockBodyBatch, + "INSERT INTO block_bodies (hash, tx_count, uncle_count, uncles)", + func(b driver.Batch, r chBlockBody) error { + return b.Append(r.hash, r.txCount, r.uncleCount, r.uncles) + }) + c.blockTD = newInsertBatcher(ctx, c, "block_total_difficulty", chBlockBodyBatch, + "INSERT INTO block_total_difficulty (hash, total_difficulty)", + func(b driver.Batch, r chBlockTD) error { + return b.Append(r.hash, r.totalDifficulty) + }) + c.blockTxs = newInsertBatcher(ctx, c, "block_txs", chBlockTxBatch, + "INSERT INTO block_txs (block_hash, tx_index, tx_hash, seen_date)", + func(b driver.Batch, r chBlockTx) error { + return b.Append(r.blockHash, r.txIndex, r.txHash, r.seenDate) }) - // block_events and transaction_events share the same row shape and column - // order, so both batchers use the same append function. - appendEvent := func(b driver.Batch, r chEvent) error { - return b.Append(r.hash, c.sensorID, r.peerID, r.seenAt) - } c.blockEvt = newInsertBatcher(ctx, c, "block_events", chBlockEventBatch, - "INSERT INTO block_events (block_hash, sensor_id, peer_id, seen_at)", - appendEvent) + "INSERT INTO block_events (block_number, block_hash, sensor_id, node_id, source, seen_at, total_difficulty)", + func(b driver.Batch, r chBlockEvent) error { + return b.Append(r.blockNumber, r.blockHash, c.sensorID, r.nodeID, r.source, r.seenAt, r.totalDifficulty) + }) c.txs = newInsertBatcher(ctx, c, "transactions", chTxBatch, - "INSERT INTO transactions (hash, from_address, to_address, value, gas, gas_price, gas_fee_cap, gas_tip_cap, nonce, tx_type, sensor_id, ingested_at)", + "INSERT INTO transactions (hash, from_address, to_address, value, gas, gas_price, gas_fee_cap, gas_tip_cap, nonce, tx_type, chain_id, input_selector, input_size, access_list_size, blob_count, auth_list_size, seen_date)", func(b driver.Batch, r chTx) error { - return b.Append(r.hash, r.from, r.to, r.value, r.gas, r.gasPrice, r.gasFeeCap, r.gasTipCap, r.nonce, r.txType, c.sensorID, r.ingestedAt) + return b.Append(r.hash, r.from, r.to, r.value, r.gas, r.gasPrice, r.gasFeeCap, r.gasTipCap, r.nonce, r.txType, r.chainID, r.inputSelector, r.inputSize, r.accessListSize, r.blobCount, r.authListSize, r.seenDate) + }) + c.txEvt = newInsertBatcher(ctx, c, "tx_events", chTxEventBatch, + "INSERT INTO tx_events (tx_hash, sensor_id, node_id, source, seen_at)", + func(b driver.Batch, r chTxEvent) error { + return b.Append(r.txHash, c.sensorID, r.nodeID, r.source, r.seenAt) }) - c.txEvt = newInsertBatcher(ctx, c, "transaction_events", chTxEventBatch, - "INSERT INTO transaction_events (tx_hash, sensor_id, peer_id, seen_at)", - appendEvent) c.peers = newInsertBatcher(ctx, c, "peers", chPeerBatch, - "INSERT INTO peers (peer_id, name, url, caps, last_seen_by, time_last_seen)", - func(b driver.Batch, r chPeer) error { - return b.Append(r.peerID, r.name, r.url, r.caps, c.sensorID, r.timeLastSeen) + "INSERT INTO peers (sensor_id, node_id, name, url, caps, seen_at)", + func(b driver.Batch, r chPeerSnapshot) error { + return b.Append(c.sensorID, r.nodeID, r.name, r.url, r.caps, r.seenAt) }) } @@ -236,71 +307,167 @@ func flushBatch[T any](conn driver.Conn, query string, rows []T, appendRow func( // --- row types ------------------------------------------------------------- +// chBlock is a header row. Every field is derived from the header itself, so any +// two sensors that see this hash produce an identical row. Nothing observational +// (which sensor, when, how it was learned) belongs here -- see chBlockEvent. type chBlock struct { + number uint64 + hash string + parentHash string + blockTime time.Time + signer string + coinbase string + difficulty uint64 + gasUsed uint64 + gasLimit uint64 + baseFee *big.Int + uncleHash string + stateRoot string + txRoot string + receiptRoot string + logsBloom []byte + extraData []byte + mixDigest string + nonce uint64 +} + +// chBlockBody holds the facts a header cannot carry. Written only when a body or +// a full block is actually delivered, which is what keeps the header path from +// having to invent a tx_count. +type chBlockBody struct { + hash string + txCount uint32 + uncleCount uint16 + uncles []string +} + +// chBlockTD is the announced total difficulty, which only a NewBlock carries. It +// is a separate table rather than a blocks column because the header path does not +// know it and would have to write 0, letting a header clobber a real value. +type chBlockTD struct { hash string - number uint64 - parentHash string - blockTime time.Time - coinbase string - signer string - difficulty uint64 totalDifficulty *big.Int - gasUsed uint64 - gasLimit uint64 - baseFee uint64 - txCount uint32 - uncleCount uint16 - uncleHash string - stateRoot string - txRoot string - receiptRoot string - logsBloom []byte - extraData []byte - mixDigest string - nonce uint64 - ingestedAt time.Time - isParent bool -} - -type chEvent struct { - hash string - peerID string - seenAt time.Time +} + +type chBlockTx struct { + blockHash string + txIndex uint32 + txHash string + seenDate time.Time +} + +type chBlockEvent struct { + blockNumber uint64 + blockHash string + nodeID string + source string + seenAt time.Time + totalDifficulty *big.Int } type chTx struct { - hash string - from string - to string - value *big.Int - gas uint64 - gasPrice *big.Int - gasFeeCap *big.Int - gasTipCap *big.Int - nonce uint64 - txType uint8 - ingestedAt time.Time -} - -type chPeer struct { - peerID string - name string - url string - caps []string - timeLastSeen time.Time + hash string + from string + to string + value *big.Int + gas uint64 + gasPrice *big.Int + gasFeeCap *big.Int + gasTipCap *big.Int + nonce uint64 + txType uint8 + chainID uint64 + inputSelector string + inputSize uint32 + accessListSize uint16 + blobCount uint8 + authListSize uint8 + seenDate time.Time +} + +type chTxEvent struct { + txHash string + nodeID string + source string + seenAt time.Time +} + +type chPeerSnapshot struct { + nodeID string + name string + url string + caps []string + seenAt time.Time } // --- Database interface ---------------------------------------------------- +// recordsBlockEvents reports whether block_events should be written at all. +// +// The provenance sources -- new_block, header, header_backfill, body -- are about +// one row per block per sensor, measured at 2.8 MB/day across the mainnet fleet, so +// they follow this rather than shouldWriteBlockEvents. That flag exists to bound the +// hash_announce firehose, which is ~52 rows per block per sensor and scales with +// peer count. Gating provenance behind it meant the production config +// (write_block_events=false, write_first_block_event=true) silently recorded no +// total_difficulty, no header timing and no header_backfill marker at all. +func (c *ClickHouse) recordsBlockEvents() bool { + return c.shouldWriteBlockEvents || c.shouldWriteFirstBlockEvent +} + +// recordsTxEvents is the transaction mirror of recordsBlockEvents. full_tx is a +// delivered transaction body -- about 2 rows per transaction per sensor, since the +// sensor's LRU filters repeats, measured at 4.5 GiB over the 14-day TTL for the +// mainnet fleet. hash_announce by contrast is 8+ rows per transaction per sensor and +// climbs with peer count, which is what shouldWriteTransactionEvents exists to bound. +func (c *ClickHouse) recordsTxEvents() bool { + return c.shouldWriteTransactionEvents || c.shouldWriteFirstTransactionEvent +} + func (c *ClickHouse) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time) { if c.conn == nil { + // blocks + block_bodies + block_total_difficulty + the event, then one + // block_txs and one transactions row per transaction. + c.discarded.Add(4 + 2*uint64(len(block.Transactions()))) return } - if c.shouldWriteBlockEvents && peer != nil { - c.blockEvt.add(chEvent{hash: block.Hash().Hex(), peerID: peer.URLv4(), seenAt: tfs}) + // Which peer announced which total difficulty is an observation, so it rides on + // the event. The value itself is a property of the block, so it also goes to its + // own forever-kept table -- the events expire at 14 days, and reading the block's + // total difficulty out of a rollup that expires is what made it read 0 for every + // block older than that. + if c.recordsBlockEvents() && peer != nil { + c.blockEvt.add(chBlockEvent{ + blockNumber: block.NumberU64(), + blockHash: block.Hash().Hex(), + nodeID: peer.ID().String(), + source: srcNewBlock, + seenAt: tfs, + // Copied like the blockTD row below: rows sit in the batcher up to a + // second, and this pointer is raw.TD, also held in the block cache for + // ~10 minutes and handed to BroadcastBlock. Nothing mutates it today; it + // was the one alias left in an otherwise uniform copy discipline. + totalDifficulty: copyBig(td), + }) + } + // Only ever written from here, and never as 0: an absent row is how "no peer + // announced it to us" is expressed, so writing 0 would recreate the sentinel + // this table exists to remove. + // + // Gated on shouldWriteBlocks like every other fact row. Ungated, this was the + // one write that happened with every flag off -- forever-kept rows with no + // blocks row to join to, invisible to v_blocks yet accumulating permanently. + // The value is copied because rows sit in the batcher for up to a second and + // td aliases the caller's big.Int. + if c.shouldWriteBlocks && td != nil && td.Sign() > 0 { + c.blockTD.add(chBlockTD{ + hash: block.Hash().Hex(), + totalDifficulty: copyBig(td), + }) } if c.shouldWriteBlocks { - c.blocks.add(newChBlock(block.Header(), td, tfs, len(block.Transactions()), len(block.Uncles()), false)) + c.blocks.add(newChBlock(block.Header())) + c.writeBlockBody(block.Hash(), block.Transactions(), block.Uncles(), tfs) } if c.shouldWriteTransactions { c.writeTxs(block.Transactions(), tfs) @@ -308,75 +475,182 @@ func (c *ClickHouse) WriteBlock(ctx context.Context, peer *enode.Node, block *ty } func (c *ClickHouse) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) { - if c.conn == nil || !c.shouldWriteBlocks { + // The header row and the header event are separate concerns, gated separately. + // Returning early on !shouldWriteBlocks made header and header_backfill the only + // two provenance sources that also required --write-blocks, so a fleet with it + // off recorded new_block, body, hash_announce and full_tx but silently dropped + // header timing and the backfill marker -- and headers are still requested in + // that configuration, so the events were produced and thrown away. + if c.conn == nil { + c.discarded.Add(uint64(len(headers))) + return + } + if !c.shouldWriteBlocks && !c.recordsBlockEvents() { return } - // A header carries no tx/uncle counts, so they are written as 0; the - // full-block (NewBlock) path writes a separate row with the real counts. - // isParent marks headers fetched as ancestors during backfill. + source := srcHeader + if isParent { + source = srcHeaderBackfill + } for _, h := range headers { - c.blocks.add(newChBlock(h, big.NewInt(0), tfs, 0, 0, isParent)) + // A header row is complete on its own: the fields it cannot carry (tx/uncle + // counts) live in block_bodies, so this path can never overwrite them. + if c.shouldWriteBlocks { + c.blocks.add(newChBlock(h)) + } + if c.recordsBlockEvents() { + c.blockEvt.add(chBlockEvent{ + blockNumber: h.Number.Uint64(), + blockHash: h.Hash().Hex(), + source: source, + seenAt: tfs, + totalDifficulty: big.NewInt(0), + }) + } } } -func (c *ClickHouse) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { - if c.conn == nil || !c.shouldWriteTransactions { +func (c *ClickHouse) WriteBlockBody(ctx context.Context, body *eth.BlockBody, ann BlockAnnouncement, tfs time.Time) { + if c.conn == nil { + c.discarded.Add(1) return } - // The block row is written from the header path; here we only persist the - // transactions carried in the body (no read-modify-write on blocks). + hash := ann.Hash + + // The body arrived: that is true whether or not it decodes, so the event is + // recorded first. Like the header sources it carries no peer -- the sensor + // requested it -- so it is excluded from the propagation rollup, but it makes + // "which sensor got the body, and when" answerable. Recording it first also + // keeps the two decode-failure paths below consistent: previously a failed + // transaction decode skipped the event while a failed uncle decode kept it. + if c.recordsBlockEvents() { + c.blockEvt.add(chBlockEvent{ + blockNumber: ann.Number, + blockHash: hash.Hex(), + source: srcBody, + seenAt: tfs, + totalDifficulty: big.NewInt(0), + }) + } + txs, err := body.Transactions.Items() if err != nil { + // Nothing downstream is derivable: block_bodies needs the count, block_txs + // and transactions need the transactions themselves. log.Error().Err(err).Str("hash", hash.Hex()).Msg("Failed to decode transactions from block body") return } - c.writeTxs(txs, tfs) + + // Each fact is skipped exactly when ITS inputs are undecodable, no wider. + // block_bodies carries uncle facts, so a failed uncle decode skips only that + // row -- writing it with uncle_count = 0 against the NewBlock path's real value + // made the merge survivor arbitrary (the size_bytes defect), but the + // transaction facts are pure functions of the transactions, which decoded + // fine, and the first version of this fix threw those away too. Unreachable + // either way on Bor, which produces no uncles. + uncles, uncleErr := body.Uncles.Items() + if uncleErr != nil { + log.Error().Err(uncleErr).Str("hash", hash.Hex()).Msg("Failed to decode uncles from block body") + } + if c.shouldWriteBlocks { + if uncleErr == nil { + c.writeBlockBody(hash, txs, uncles, tfs) + } else { + c.writeBlockTxs(hash, txs, tfs) + } + } + if c.shouldWriteTransactions { + c.writeTxs(txs, tfs) + } } -func (c *ClickHouse) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if c.conn == nil || peer == nil { +func (c *ClickHouse) WriteBlockEvents(ctx context.Context, peer *enode.Node, anns []BlockAnnouncement, tfs time.Time) { + if c.conn == nil { + c.discarded.Add(uint64(len(anns))) return } - peerID := peer.URLv4() - for _, hash := range hashes { - c.blockEvt.add(chEvent{hash: hash.Hex(), peerID: peerID, seenAt: tfs}) + if peer == nil { + return + } + nodeID := peer.ID().String() + for _, ann := range anns { + c.blockEvt.add(chBlockEvent{ + blockNumber: ann.Number, + blockHash: ann.Hash.Hex(), + nodeID: nodeID, + source: srcHashAnnounce, + seenAt: tfs, + totalDifficulty: big.NewInt(0), + }) } } -// WriteBlockHashFirstSeen is a no-op: ClickHouse derives earliest first-seen at -// query time from the block_events stream (see the block_first_seen -// materialized view), so no per-block stamp is needed. +// WriteBlockHashFirstSeen is a no-op: earliest first-seen is derived from +// block_events by the block_events_first materialized view. func (c *ClickHouse) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time) { } func (c *ClickHouse) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if c.conn == nil || peer == nil { + if c.conn == nil { + c.discarded.Add(uint64(len(hashes))) + return + } + if peer == nil { return } - peerID := peer.URLv4() + nodeID := peer.ID().String() for _, hash := range hashes { - c.txEvt.add(chEvent{hash: hash.Hex(), peerID: peerID, seenAt: tfs}) + c.txEvt.add(chTxEvent{ + txHash: hash.Hex(), + nodeID: nodeID, + source: srcHashAnnounce, + seenAt: tfs, + }) } } func (c *ClickHouse) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { - if c.conn == nil || !c.shouldWriteTransactions { + if c.conn == nil { + c.discarded.Add(uint64(len(txs))) + return + } + // A delivered body is a distinct event from a hash announcement, and worth + // recording under either flag: measured at ~2 rows per transaction per sensor, + // since the sensor's LRU filters repeats, so it is not a per-peer stream. + if c.recordsTxEvents() && peer != nil { + nodeID := peer.ID().String() + for _, tx := range txs { + c.txEvt.add(chTxEvent{ + txHash: tx.Hash().Hex(), + nodeID: nodeID, + source: srcFullTx, + seenAt: tfs, + }) + } + } + if !c.shouldWriteTransactions { return } c.writeTxs(txs, tfs) } func (c *ClickHouse) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time) { - if c.conn == nil || !c.shouldWritePeers { + if c.conn == nil { + c.discarded.Add(uint64(len(peers))) return } + if !c.shouldWritePeers { + return + } + // node_id matches what the event streams record, so peers and events are + // joinable. The enode URL is a column, not the key. for _, peer := range peers { - c.peers.add(chPeer{ - peerID: peer.ID().String(), - name: peer.Fullname(), - url: peer.Node().URLv4(), - caps: peer.Info().Caps, - timeLastSeen: tls, + c.peers.add(chPeerSnapshot{ + nodeID: peer.ID().String(), + name: peer.Fullname(), + url: peer.Node().URLv4(), + caps: peer.Info().Caps, + seenAt: tls, }) } } @@ -384,6 +658,36 @@ func (c *ClickHouse) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time // HasBlock reports whether the block already exists. Called once per new block // (not per event), so an indexed point lookup is cheap. Without a connection it // reports true so the sensor never attempts a backfill it could not persist. +// startUnavailableWarning re-logs while the backend is unreachable, so the failure +// keeps showing up in logs and alerting instead of scrolling past once at startup. +// It does not reconnect: a sensor whose database was down at boot needs a restart, +// and pretending otherwise would hide that. +func (c *ClickHouse) startUnavailableWarning(ctx context.Context) { + c.wg.Add(1) + go func() { + defer c.wg.Done() + ticker := time.NewTicker(chUnavailableWarnInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + log.Error(). + Uint64("rows_discarded", c.discarded.Load()). + Msg("ClickHouse is unavailable; every write is being discarded") + } + } + }() +} + +// HasBlock reports whether the block is already stored, which the sensor uses to +// decide whether to backfill a parent. +// +// With no connection it returns true -- "we have it" -- to suppress backfill rather +// than let every block queue parent requests that can never be satisfied. That is +// the safer degradation, but it does mean a dead backend is silent in this path +// specifically; startUnavailableWarning is what makes it visible. func (c *ClickHouse) HasBlock(ctx context.Context, hash common.Hash) bool { if c.conn == nil { return true @@ -393,12 +697,14 @@ func (c *ClickHouse) HasBlock(ctx context.Context, hash common.Hash) bool { return err == nil && exists == 1 } +// NodeList returns the most recently seen peers' enode URLs, from the narrow +// peers_current rollup rather than grouping over the event firehose. func (c *ClickHouse) NodeList(ctx context.Context, limit int) ([]string, error) { if c.conn == nil { return []string{}, nil } rows, err := c.conn.Query(ctx, - "SELECT peer_id FROM block_events GROUP BY peer_id ORDER BY max(seen_at) DESC LIMIT ?", limit) + "SELECT url FROM peers_current FINAL WHERE url != '' ORDER BY last_seen DESC LIMIT ?", limit) if err != nil { return nil, fmt.Errorf("query node list: %w", err) } @@ -436,50 +742,112 @@ func (c *ClickHouse) ShouldWritePeers() bool { return c.shouldWritePeers } // --- helpers --------------------------------------------------------------- -// newChBlock maps a header (plus data not carried on the header itself) to a -// blocks-table row. -func newChBlock(h *types.Header, td *big.Int, tfs time.Time, txCount, uncleCount int, isParent bool) chBlock { - baseFee := uint64(0) - if h.BaseFee != nil { - baseFee = h.BaseFee.Uint64() +// copyBig returns a defensive copy, or nil for nil. big.Int carries an internal +// slice, so Set is the copy that matters, not the struct assignment. +func copyBig(v *big.Int) *big.Int { + if v == nil { + return nil } - if td == nil { - td = big.NewInt(0) + return new(big.Int).Set(v) +} + +// addressHex renders an address as lowercase 0x hex. +// +// NOT common.Address.Hex(), which applies the EIP-55 checksum and so returns mixed +// case. That is a display format -- it exists so a human can spot a mistyped +// address -- and these come from ecrecover and RLP, never from typing. Stored mixed +// case, an address column silently fails to join: ClickHouse string comparison is +// case-sensitive, and the Polygon staking API, block-latency and data-analysis all +// key validators on lowercase. Measured against the live validator set, 0 of 3 +// real-chain signers matched as stored and all 3 matched lowercased -- a join that +// returns no rows and no error, so "no blocks had a known signer" reads as an +// answer rather than a bug. +// +// Hashes need no equivalent: common.Hash.Hex() is already lowercase, having no +// checksum to apply. +func addressHex(a common.Address) string { + return strings.ToLower(a.Hex()) +} + +// newChBlock maps a header to a blocks-table row. Takes nothing but the header, so +// everything it writes is a pure function of it -- which is what makes duplicate +// rows for a hash byte-identical. +// +// mix_digest and nonce are all-zero on Bor but stored anyway: clique's +// encodeSigHeader includes them unconditionally, so a consumer re-running ecrecover +// needs them, as it needs base_fee. The post-Shanghai/Cancun header fields are +// deliberately not stored -- clique panics if any of them is non-nil, so they can +// never take part in the seal hash. +func newChBlock(h *types.Header) chBlock { + baseFee := new(big.Int) + if h.BaseFee != nil { + baseFee.Set(h.BaseFee) } // Recover the block signer from the header seal so signer-based analytics // don't have to ecrecover on every query. Left empty when it can't be recovered. var signer string if sig, err := util.Ecrecover(h); err == nil { - signer = common.BytesToAddress(sig).Hex() + signer = addressHex(common.BytesToAddress(sig)) } return chBlock{ - hash: h.Hash().Hex(), - number: h.Number.Uint64(), - parentHash: h.ParentHash.Hex(), - blockTime: time.Unix(int64(h.Time), 0).UTC(), - coinbase: h.Coinbase.Hex(), - signer: signer, - difficulty: h.Difficulty.Uint64(), - totalDifficulty: new(big.Int).Set(td), - gasUsed: h.GasUsed, - gasLimit: h.GasLimit, - baseFee: baseFee, - txCount: uint32(txCount), - uncleCount: uint16(uncleCount), - uncleHash: h.UncleHash.Hex(), - stateRoot: h.Root.Hex(), - txRoot: h.TxHash.Hex(), - receiptRoot: h.ReceiptHash.Hex(), - logsBloom: h.Bloom.Bytes(), - extraData: h.Extra, - mixDigest: h.MixDigest.Hex(), - nonce: h.Nonce.Uint64(), - ingestedAt: tfs, - isParent: isParent, + number: h.Number.Uint64(), + hash: h.Hash().Hex(), + parentHash: h.ParentHash.Hex(), + blockTime: time.Unix(int64(h.Time), 0).UTC(), + signer: signer, + coinbase: addressHex(h.Coinbase), + difficulty: h.Difficulty.Uint64(), + gasUsed: h.GasUsed, + gasLimit: h.GasLimit, + baseFee: baseFee, + uncleHash: h.UncleHash.Hex(), + stateRoot: h.Root.Hex(), + txRoot: h.TxHash.Hex(), + receiptRoot: h.ReceiptHash.Hex(), + logsBloom: h.Bloom.Bytes(), + extraData: bytes.Clone(h.Extra), // aliased peer memory; rows outlive the call + mixDigest: h.MixDigest.Hex(), + nonce: h.Nonce.Uint64(), + } +} + +// writeBlockBody records the body facts and the ordered block -> tx mapping. +// +// Deliberately stores no encoded block size: it is only knowable from a full +// NewBlock, not from a body delivered on its own, so it is not a function of the +// hash and two sensors could write differing rows for one block. +func (c *ClickHouse) writeBlockBody(hash common.Hash, txs []*types.Transaction, uncles []*types.Header, tfs time.Time) { + uncleHashes := make([]string, 0, len(uncles)) + for _, u := range uncles { + uncleHashes = append(uncleHashes, u.Hash().Hex()) + } + c.blockBodies.add(chBlockBody{ + hash: hash.Hex(), + txCount: uint32(len(txs)), + uncleCount: uint16(len(uncles)), + uncles: uncleHashes, + }) + c.writeBlockTxs(hash, txs, tfs) +} + +// writeBlockTxs records the ordered block -> transaction mapping alone, for the +// case where the body facts row cannot be written (undecodable uncles) but the +// transactions decoded fine. +func (c *ClickHouse) writeBlockTxs(hash common.Hash, txs []*types.Transaction, tfs time.Time) { + blockHash := hash.Hex() + seenDate := tfs.UTC().Truncate(24 * time.Hour) + for i, tx := range txs { + c.blockTxs.add(chBlockTx{ + blockHash: blockHash, + txIndex: uint32(i), + txHash: tx.Hash().Hex(), + seenDate: seenDate, + }) } } func (c *ClickHouse) writeTxs(txs []*types.Transaction, tfs time.Time) { + seenDate := tfs.UTC().Truncate(24 * time.Hour) for _, tx := range txs { var from, to string chainID := tx.ChainId() @@ -487,23 +855,40 @@ func (c *ClickHouse) writeTxs(txs []*types.Transaction, tfs time.Time) { chainID = c.chainID } if addr, err := types.Sender(types.LatestSignerForChainID(chainID), tx); err == nil { - from = addr.Hex() + from = addressHex(addr) } if tx.To() != nil { - to = tx.To().Hex() + to = addressHex(*tx.To()) + } + // Selector and size but not the calldata: enough for contract-interaction + // analysis at negligible cost. + var selector string + data := tx.Data() + if len(data) >= 4 { + selector = "0x" + hex.EncodeToString(data[:4]) + } + txChainID := uint64(0) + if id := tx.ChainId(); id != nil && id.IsUint64() { + txChainID = id.Uint64() } c.txs.add(chTx{ - hash: tx.Hash().Hex(), - from: from, - to: to, - value: new(big.Int).Set(tx.Value()), - gas: tx.Gas(), - gasPrice: new(big.Int).Set(tx.GasPrice()), - gasFeeCap: new(big.Int).Set(tx.GasFeeCap()), - gasTipCap: new(big.Int).Set(tx.GasTipCap()), - nonce: tx.Nonce(), - txType: tx.Type(), - ingestedAt: tfs, + hash: tx.Hash().Hex(), + from: from, + to: to, + value: new(big.Int).Set(tx.Value()), + gas: tx.Gas(), + gasPrice: new(big.Int).Set(tx.GasPrice()), + gasFeeCap: new(big.Int).Set(tx.GasFeeCap()), + gasTipCap: new(big.Int).Set(tx.GasTipCap()), + nonce: tx.Nonce(), + txType: tx.Type(), + chainID: txChainID, + inputSelector: selector, + inputSize: uint32(len(data)), + accessListSize: uint16(len(tx.AccessList())), + blobCount: uint8(len(tx.BlobHashes())), + authListSize: uint8(len(tx.SetCodeAuthorizations())), + seenDate: seenDate, }) } } diff --git a/p2p/database/clickhouse_test.go b/p2p/database/clickhouse_test.go index a49a02304..83017961e 100644 --- a/p2p/database/clickhouse_test.go +++ b/p2p/database/clickhouse_test.go @@ -3,57 +3,153 @@ package database import ( "context" "math/big" + "net" "os" + "strings" "testing" "time" "github.com/ClickHouse/clickhouse-go/v2" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/clique" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/p2p/enode" ) -// TestClickHouseWrites exercises the ClickHouse backend end-to-end against a -// real server. It is skipped unless POLYCLI_TEST_CLICKHOUSE_DSN is set, e.g. +// Test block heights live in a reserved band far above any real chain head, one +// per test, so a test row is never mistaken for chain data and two tests never +// collide on a height. Spaced by ten, so a test needing a parent or child can +// derive height-1 or height+1 without landing on another test's block -- deriving a +// parent as height-1 from consecutive constants collided, and the resulting failure +// appeared only when the whole suite ran. This matters because these tests write to a real database +// that is usually the local-stack one holding live sensor data: heights 42 and +// 4242 previously produced rows that looked exactly like a transaction included in +// four competing blocks, which is a real reorg signature. +const ( + heightWrites = 900_000_010 + heightHeaderNoClobber = 900_000_020 + heightProductionFlags = 900_000_030 + heightParentCancel = 900_000_040 + heightTotalDiff = 900_000_050 + heightLowercase = 900_000_060 + heightHeaderEvents = 900_000_070 +) + +// cleanupTestHeights removes everything the given test heights produced, in +// dependency order (hash-keyed tables first, via the blocks rows). Registered by +// every test that writes blocks: block_forks has no TTL and v_forks no height +// filter, so without this each run's fresh sealing key added another competing +// hash per height -- six heights wide-N "forks" after N runs, unfilterable and +// permanent. +func cleanupTestHeights(t *testing.T, conn driver.Conn, heights ...uint64) { + t.Helper() + t.Cleanup(func() { + ctx := context.Background() + + // Order matters: every statement below identifies its rows through a table + // that a LATER statement deletes, so reversing any pair silently leaves rows + // behind. Getting this wrong is why two earlier attempts still leaked -- + // deleting tx_events first emptied the subquery that finds the transactions. + // + // transactions is reachable two ways, and both are needed: WriteTransactions + // writes tx_events for it, but writeTxs is ALSO called from WriteBlockBody, + // which writes no tx_events at all -- so a body-only test leaves transactions + // rows that no tx_events row points at. + for _, h := range heights { + for _, q := range []string{ + "ALTER TABLE transactions DELETE WHERE hash IN (SELECT tx_hash FROM block_txs WHERE block_hash IN (SELECT hash FROM blocks WHERE number = ?)) SETTINGS mutations_sync = 1", + "ALTER TABLE tx_events_first DELETE WHERE tx_hash IN (SELECT tx_hash FROM block_txs WHERE block_hash IN (SELECT hash FROM blocks WHERE number = ?)) SETTINGS mutations_sync = 1", + "ALTER TABLE block_txs DELETE WHERE block_hash IN (SELECT hash FROM blocks WHERE number = ?) SETTINGS mutations_sync = 1", + "ALTER TABLE block_bodies DELETE WHERE hash IN (SELECT hash FROM blocks WHERE number = ?) SETTINGS mutations_sync = 1", + "ALTER TABLE block_total_difficulty DELETE WHERE hash IN (SELECT hash FROM blocks WHERE number = ?) SETTINGS mutations_sync = 1", + "ALTER TABLE blocks DELETE WHERE number = ? SETTINGS mutations_sync = 1", + "ALTER TABLE block_events DELETE WHERE block_number = ? SETTINGS mutations_sync = 1", + "ALTER TABLE block_events_first DELETE WHERE block_number = ? SETTINGS mutations_sync = 1", + "ALTER TABLE block_forks DELETE WHERE number = ? SETTINGS mutations_sync = 1", + } { + if err := conn.Exec(ctx, q, h); err != nil { + t.Logf("cleanup height %d: %v", h, err) + } + } + } + + // Whatever the test wrote under its own sensor id, by any path. Last, because + // the height-keyed statements above use tx_events to find nothing but this + // catches what they could not reach. + for _, q := range []string{ + "ALTER TABLE transactions DELETE WHERE hash IN (SELECT tx_hash FROM tx_events WHERE sensor_id LIKE 'test-sensor%') SETTINGS mutations_sync = 1", + "ALTER TABLE tx_events_first DELETE WHERE tx_hash IN (SELECT tx_hash FROM tx_events WHERE sensor_id LIKE 'test-sensor%') SETTINGS mutations_sync = 1", + "ALTER TABLE tx_events DELETE WHERE sensor_id LIKE 'test-sensor%' SETTINGS mutations_sync = 1", + } { + if err := conn.Exec(ctx, q); err != nil { + t.Logf("tx-side cleanup: %v", err) + } + } + }) +} + +// The integration tests in this file are skipped unless +// POLYCLI_TEST_CLICKHOUSE_DSN is set, e.g. // -// POLYCLI_TEST_CLICKHOUSE_DSN=clickhouse://localhost:19000/sensor go test ./p2p/database/ -run TestClickHouseWrites -v +// POLYCLI_TEST_CLICKHOUSE_DSN=clickhouse://localhost:19000/sensor go test ./p2p/database/ -run TestClickHouse -v // -// The target database must already have the schema from -// sensor-network-tools/clickhouse_schema.sql applied. -func TestClickHouseWrites(t *testing.T) { +// The target database must already have clickhouse_schema.sql applied. + +func clickHouseDSN(t *testing.T) string { + t.Helper() dsn := os.Getenv("POLYCLI_TEST_CLICKHOUSE_DSN") if dsn == "" { t.Skip("POLYCLI_TEST_CLICKHOUSE_DSN not set; skipping ClickHouse integration test") } + return dsn +} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - db := NewClickHouse(ctx, ClickHouseOptions{ - DSN: dsn, - SensorID: "test-sensor", - ChainID: 137, - MaxConcurrency: 10, - ShouldWriteBlocks: true, - ShouldWriteBlockEvents: true, - ShouldWriteTransactions: true, - ShouldWriteTransactionEvents: true, - ShouldWritePeers: true, - }) +// signedHeader returns a clique-sealed header plus the address that sealed it, so +// tests can assert the ecrecover-at-ingest path. +func signedHeader(t *testing.T, number int64, blockTime time.Time) (*types.Header, string) { + t.Helper() + priv, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + header := &types.Header{ + Number: big.NewInt(number), + Time: uint64(blockTime.Unix()), + Difficulty: big.NewInt(7), + GasLimit: 30_000_000, + GasUsed: 21_000, + BaseFee: big.NewInt(1_000_000_000), + Extra: make([]byte, crypto.SignatureLength), + } + sig, err := crypto.Sign(clique.SealHash(header).Bytes(), priv) + if err != nil { + t.Fatalf("sign header: %v", err) + } + copy(header.Extra[len(header.Extra)-crypto.SignatureLength:], sig) + // Lowercase, because that is what the writer stores -- see addressHex. + return header, strings.ToLower(crypto.PubkeyToAddress(priv.PublicKey).Hex()) +} - now := time.Now().UTC() +// signedHeaderWithCoinbase is signedHeader for tests that need a specific coinbase. +// The seal covers Coinbase, so it must be set BEFORE signing; assigning it to an +// already-sealed header silently invalidates the signature and ecrecover then +// returns a different address entirely. +func signedHeaderWithCoinbase(t *testing.T, number int64, blockTime time.Time, coinbase common.Address) (*types.Header, string) { + t.Helper() priv, err := crypto.GenerateKey() if err != nil { t.Fatalf("generate key: %v", err) } - wantSigner := crypto.PubkeyToAddress(priv.PublicKey) header := &types.Header{ - Number: big.NewInt(42), - Time: uint64(now.Unix()), + Number: big.NewInt(number), + Time: uint64(blockTime.Unix()), Difficulty: big.NewInt(7), GasLimit: 30_000_000, GasUsed: 21_000, BaseFee: big.NewInt(1_000_000_000), + Coinbase: coinbase, Extra: make([]byte, crypto.SignatureLength), } sig, err := crypto.Sign(clique.SealHash(header).Bytes(), priv) @@ -61,17 +157,60 @@ func TestClickHouseWrites(t *testing.T) { t.Fatalf("sign header: %v", err) } copy(header.Extra[len(header.Extra)-crypto.SignatureLength:], sig) - block := types.NewBlockWithHeader(header) + return header, strings.ToLower(crypto.PubkeyToAddress(priv.PublicKey).Hex()) +} + +func testPeer(t *testing.T) *enode.Node { + t.Helper() + priv, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate peer key: %v", err) + } + return enode.NewV4(&priv.PublicKey, net.IPv4(127, 0, 0, 1), 30303, 30303) +} + +func newTestClickHouse(t *testing.T, dsn string, ctx context.Context) Database { + t.Helper() + return NewClickHouse(ctx, ClickHouseOptions{ + DSN: dsn, + SensorID: "test-sensor", + ChainID: 137, + MaxConcurrency: 10, + ShouldWriteBlocks: true, + ShouldWriteBlockEvents: true, + ShouldWriteTransactions: true, + ShouldWriteTransactionEvents: true, + ShouldWritePeers: true, + }) +} +// TestClickHouseWrites exercises the whole write path against a real server and +// asserts every table the sensor targets receives its row. +func TestClickHouseWrites(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + db := newTestClickHouse(t, dsn, ctx) + + now := time.Now().UTC() + header, wantSigner := signedHeader(t, heightWrites, now) + // The nonce varies per run: with it fixed, each run mapped the SAME tx hash to + // a fresh block hash (the sealing key is fresh, so the header hash changes), + // and block_txs accumulated one-transaction-in-N-competing-blocks -- a real + // reorg signature, manufactured by the test suite. tx := types.NewTx(&types.LegacyTx{ - Nonce: 1, + Nonce: uint64(now.UnixNano()), GasPrice: big.NewInt(2_000_000_000), Gas: 21_000, Value: big.NewInt(1), + Data: []byte{0xa9, 0x05, 0x9c, 0xbb, 0x01, 0x02}, }) + block := types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: []*types.Transaction{tx}}) + peer := testPeer(t) - db.WriteBlock(ctx, nil, block, big.NewInt(100), now) - db.WriteTransactions(ctx, nil, []*types.Transaction{tx}, now) + db.WriteBlock(ctx, peer, block, big.NewInt(100), now) + db.WriteTransactions(ctx, peer, []*types.Transaction{tx}, now) db.WritePeers(ctx, nil, now) // empty peer slice is fine; exercises the path // Close drains the buffered rows synchronously before we verify them. @@ -79,57 +218,571 @@ func TestClickHouseWrites(t *testing.T) { t.Fatalf("close db: %v", cerr) } - conn, err := clickhouse.Open(mustParseDSN(t, dsn)) - if err != nil { - t.Fatalf("open verify conn: %v", err) - } - defer func() { - if err := conn.Close(); err != nil { - t.Errorf("close verify conn: %v", err) - } - }() + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightWrites) + blockHash := block.Hash().Hex() + + // Fact tables. + checkCount(t, conn, "SELECT count() FROM blocks WHERE hash = ?", blockHash) + checkCount(t, conn, "SELECT count() FROM block_bodies WHERE hash = ?", blockHash) + checkCount(t, conn, "SELECT count() FROM block_txs WHERE block_hash = ?", blockHash) + checkCount(t, conn, "SELECT count() FROM transactions WHERE hash = ?", tx.Hash().Hex()) - checkCount(t, conn, "blocks", block.Hash().Hex()) - checkCount(t, conn, "transactions", tx.Hash().Hex()) + // Observation stream, and the rollup the materialized view maintains off it. + checkCount(t, conn, "SELECT count() FROM block_events WHERE block_hash = ?", blockHash) + checkCount(t, conn, "SELECT count() FROM block_events_first WHERE block_hash = ?", blockHash) - // Verify the round-tripped block fields. + // Round-tripped header fields. base_fee is UInt256, so it must be scanned as + // a big.Int rather than the uint64 the old schema used. var ( number uint64 gasUsed uint64 - baseFee uint64 + baseFee big.Int signer string ) row := conn.QueryRow(context.Background(), - "SELECT number, gas_used, base_fee, signer FROM blocks WHERE hash = ? LIMIT 1", block.Hash().Hex()) + "SELECT number, gas_used, base_fee, signer FROM blocks WHERE hash = ? LIMIT 1", blockHash) if err := row.Scan(&number, &gasUsed, &baseFee, &signer); err != nil { t.Fatalf("scan block: %v", err) } - if number != 42 || gasUsed != 21_000 || baseFee != 1_000_000_000 { - t.Fatalf("unexpected block fields: number=%d gas_used=%d base_fee=%d", number, gasUsed, baseFee) + if number != heightWrites || gasUsed != 21_000 || baseFee.Uint64() != 1_000_000_000 { + t.Fatalf("unexpected block fields: number=%d gas_used=%d base_fee=%s", number, gasUsed, baseFee.String()) + } + if signer != wantSigner { + t.Fatalf("signer mismatch: want %s got %s", wantSigner, signer) + } + + // The event must carry the announced total difficulty and the peer's node + // id (not its enode URL), since node id is what joins to the peer tables. + var ( + td big.Int + nodeID string + source string + ) + row = conn.QueryRow(context.Background(), + "SELECT total_difficulty, node_id, source FROM block_events WHERE block_hash = ? AND source = 'new_block' LIMIT 1", blockHash) + if err := row.Scan(&td, &nodeID, &source); err != nil { + t.Fatalf("scan event: %v", err) + } + if td.Uint64() != 100 { + t.Fatalf("total_difficulty: want 100 got %s", td.String()) + } + if nodeID != peer.ID().String() { + t.Fatalf("node_id: want %s got %s", peer.ID().String(), nodeID) + } + + // The block -> tx mapping, which the previous schema could not express. + var mapped string + if err := conn.QueryRow(context.Background(), + "SELECT tx_hash FROM block_txs WHERE block_hash = ? AND tx_index = 0 LIMIT 1", blockHash).Scan(&mapped); err != nil { + t.Fatalf("scan block_txs: %v", err) + } + if mapped != tx.Hash().Hex() { + t.Fatalf("block_txs tx_hash: want %s got %s", tx.Hash().Hex(), mapped) + } + + // Calldata shape is recorded without the calldata. + var ( + selector string + inputSize uint32 + ) + if err := conn.QueryRow(context.Background(), + "SELECT input_selector, input_size FROM transactions WHERE hash = ? LIMIT 1", tx.Hash().Hex()).Scan(&selector, &inputSize); err != nil { + t.Fatalf("scan transaction: %v", err) + } + if selector != "0xa9059cbb" || inputSize != 6 { + t.Fatalf("calldata shape: got selector=%s size=%d", selector, inputSize) + } +} + +// TestClickHouseHeaderDoesNotClobberBody is the regression test for the defect +// that motivated splitting body facts out of the blocks table. +// +// Under the previous schema a header carried no transaction count, so the header +// path wrote tx_count = 0 onto the block row; because the engine kept the row +// with the newest version column, a header arriving after the full block -- +// routine during parent backfill -- permanently replaced the real counts with +// zeros. Body facts now live in their own hash-keyed table that the header path +// never writes, so the ordering cannot matter. +func TestClickHouseHeaderDoesNotClobberBody(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + db := newTestClickHouse(t, dsn, ctx) + + now := time.Now().UTC() + header, _ := signedHeader(t, heightHeaderNoClobber, now) + txs := []*types.Transaction{ + types.NewTx(&types.LegacyTx{Nonce: uint64(now.UnixNano()) + 1, GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(1)}), + types.NewTx(&types.LegacyTx{Nonce: uint64(now.UnixNano()) + 2, GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(2)}), + types.NewTx(&types.LegacyTx{Nonce: uint64(now.UnixNano()) + 3, GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(3)}), + } + block := types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: txs}) + peer := testPeer(t) + + // Full block first, then the same header again a second later -- the ordering + // that used to lose the counts. + db.WriteBlock(ctx, peer, block, big.NewInt(100), now) + db.WriteBlockHeaders(ctx, []*types.Header{block.Header()}, now.Add(time.Second), true) + + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightHeaderNoClobber) + blockHash := block.Hash().Hex() + + var txCount uint32 + if err := conn.QueryRow(context.Background(), + "SELECT max(tx_count) FROM block_bodies WHERE hash = ?", blockHash).Scan(&txCount); err != nil { + t.Fatalf("scan block_bodies: %v", err) + } + if txCount != uint32(len(txs)) { + t.Fatalf("tx_count was clobbered by the header write: want %d got %d", len(txs), txCount) } - if signer != wantSigner.Hex() { - t.Fatalf("signer mismatch: want %s got %s", wantSigner.Hex(), signer) + + // Both events should be recorded, distinguishable by source, and the + // backfilled header must be tagged as such. + var sources []string + rows, err := conn.Query(context.Background(), + "SELECT DISTINCT source FROM block_events WHERE block_hash = ? ORDER BY source", blockHash) + if err != nil { + t.Fatalf("query sources: %v", err) + } + defer func() { + if err := rows.Close(); err != nil { + t.Errorf("close rows: %v", err) + } + }() + for rows.Next() { + var source string + if err := rows.Scan(&source); err != nil { + t.Fatalf("scan source: %v", err) + } + sources = append(sources, source) + } + if len(sources) != 2 || sources[0] != "header_backfill" || sources[1] != "new_block" { + t.Fatalf("want sources [header_backfill new_block], got %v", sources) } } -func mustParseDSN(t *testing.T, dsn string) *clickhouse.Options { +func verifyConn(t *testing.T, dsn string) driver.Conn { t.Helper() opts, err := clickhouse.ParseDSN(dsn) if err != nil { t.Fatalf("parse dsn: %v", err) } - return opts + conn, err := clickhouse.Open(opts) + if err != nil { + t.Fatalf("open verify conn: %v", err) + } + t.Cleanup(func() { + if err := conn.Close(); err != nil { + t.Errorf("close verify conn: %v", err) + } + }) + return conn } -func checkCount(t *testing.T, conn driver.Conn, table, hash string) { +func checkCount(t *testing.T, conn driver.Conn, query, arg string) { t.Helper() var count uint64 - // #nosec G202 -- table is a test-only constant, not user input - if err := conn.QueryRow(context.Background(), - "SELECT count() FROM "+table+" WHERE hash = ?", hash).Scan(&count); err != nil { - t.Fatalf("count %s: %v", table, err) + if err := conn.QueryRow(context.Background(), query, arg).Scan(&count); err != nil { + t.Fatalf("count via %q: %v", query, err) } if count == 0 { - t.Fatalf("expected a row in %s for hash %s, got none", table, hash) + t.Fatalf("expected at least one row from %q for %s, got none", query, arg) + } +} + +// TestClickHouseProductionFlagsRecordProvenance is the regression test for two +// instances of the same defect: provenance events gated on the flag that controls +// the full per-peer announcement stream, rather than on "either event flag is set". +// +// Production runs write_block_events=false / write_first_block_event=true and the +// same pair for transactions. Under that configuration new_block, header, +// header_backfill, body and full_tx were all silently dropped -- the sensors looked +// healthy and the volume-bounded first-event rows kept arriving, so nothing pointed +// at the gap. The block sources were fixed first; full_tx survived one more round +// and is why this test asserts both sides together. +// +// These sources are not volume. Per (block, sensor) the per-peer streams are +// hash_announce at ~52 rows and tx hash_announce at ~8, both scaling with +// --max-peers; the provenance sources are 1-8 rows and full_tx ~2 per transaction. +func TestClickHouseProductionFlagsRecordProvenance(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // The exact terraform default flag set, which is the point of the test. + db := NewClickHouse(ctx, ClickHouseOptions{ + DSN: dsn, + SensorID: "test-sensor-prod-flags", + ChainID: 137, + MaxConcurrency: 10, + ShouldWriteBlocks: true, + ShouldWriteBlockEvents: false, + ShouldWriteFirstBlockEvent: true, + ShouldWriteTransactions: true, + ShouldWriteTransactionEvents: false, + ShouldWriteFirstTransactionEvent: true, + ShouldWritePeers: true, + }) + + now := time.Now().UTC() + header, _ := signedHeader(t, heightProductionFlags, now) + + // The nonce must vary per run. These tables are append-only and the test asserts + // on row presence, so a fixed nonce yields a fixed transaction hash and rows left + // behind by an earlier run satisfy the assertion -- the test then passes even with + // the defect reintroduced, which is how the first draft of it failed to catch the + // very bug it exists for. Block hashes are already unique per run because + // signedHeader seals with a fresh key. + tx := types.NewTx(&types.LegacyTx{ + Nonce: uint64(now.UnixNano()), + GasPrice: big.NewInt(2_000_000_000), + Gas: 21_000, + Value: big.NewInt(1), + }) + block := types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: []*types.Transaction{tx}}) + peer := testPeer(t) + + db.WriteBlock(ctx, peer, block, big.NewInt(555), now) + db.WriteBlockHeaders(ctx, []*types.Header{header}, now, false) + db.WriteTransactions(ctx, peer, []*types.Transaction{tx}, now) + + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightProductionFlags) + blockHash := block.Hash().Hex() + + for _, source := range []string{"new_block", "header"} { + checkCount(t, conn, + "SELECT count() FROM block_events WHERE block_hash = ? AND source = '"+source+"'", blockHash) + } + + // The one that regressed after the block-side fix. + checkCount(t, conn, + "SELECT count() FROM tx_events WHERE tx_hash = ? AND source = 'full_tx'", tx.Hash().Hex()) + + // total_difficulty rides on new_block alone, so losing that source loses the + // column entirely -- assert the value survived, not just the row. + var td big.Int + if err := conn.QueryRow(context.Background(), + "SELECT total_difficulty FROM block_events WHERE block_hash = ? AND source = 'new_block' LIMIT 1", + blockHash).Scan(&td); err != nil { + t.Fatalf("scan total_difficulty: %v", err) + } + if td.Uint64() != 555 { + t.Fatalf("total_difficulty: want 555 got %s", td.String()) + } +} + +// TestClickHouseWritesSurviveParentContextCancel is the regression test for silent +// shutdown data loss. +// +// The sensor shuts down by cancelling its signal context, and only stops serving +// peers afterwards (stopServer/conns.Close are deferred later than db.Close, so +// they run first). While the p2p server winds down, peers keep delivering blocks. +// When the batcher context inherited the caller's, cancellation drained and +// stopped the batchers at the instant of SIGINT, so every row written during that +// window went into the buffered channel with no reader -- never flushed, and not +// counted as dropped because the channel had capacity, so nothing logged it. +// +// Close, and only Close, may stop the batchers. +func TestClickHouseWritesSurviveParentContextCancel(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + + db := newTestClickHouse(t, dsn, ctx) + + // SIGINT arrives. + cancel() + time.Sleep(300 * time.Millisecond) + + // A peer delivers a block while the p2p server is still winding down. + now := time.Now().UTC() + header, _ := signedHeader(t, heightParentCancel, now) + db.WriteBlockHeaders(ctx, []*types.Header{header}, now, false) + + // Only now does the sensor close the database, which must drain. + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightParentCancel) + blockHash := header.Hash().Hex() + checkCount(t, conn, "SELECT count() FROM blocks WHERE hash = ?", blockHash) +} + +// TestClickHouseTotalDifficultySurvivesEventExpiry is the regression test for +// total_difficulty decaying to zero. +// +// total_difficulty reaches the sensor only on a NewBlock announcement, so it used +// to be carried on block_events_first purely to outlive the raw event stream. When +// retention was normalised that rollup became 14 days itself, while blocks stayed +// forever -- so v_blocks returned 0 for every block older than the TTL, which is +// also the documented value for "no peer ever announced it to us". The two cases +// were indistinguishable. +// +// It now lives in its own forever-kept, hash-keyed table, and absence rather than 0 +// means never announced. Deleting the events stands in for the TTL expiring them. +func TestClickHouseTotalDifficultySurvivesEventExpiry(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + db := newTestClickHouse(t, dsn, ctx) + + now := time.Now().UTC() + header, _ := signedHeader(t, heightTotalDiff, now) + block := types.NewBlockWithHeader(header) + wantTD := big.NewInt(987654321) + + db.WriteBlock(ctx, testPeer(t), block, wantTD, now) + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightTotalDiff) + blockHash := block.Hash().Hex() + + // Expire every observation of this block, which is what the 14-day TTL does. + for _, tbl := range []string{"block_events", "block_events_first"} { + if err := conn.Exec(context.Background(), + "ALTER TABLE "+tbl+" DELETE WHERE block_hash = ? SETTINGS mutations_sync = 1", + blockHash); err != nil { + t.Fatalf("expire %s: %v", tbl, err) + } + } + + var ( + have bool + td *big.Int + ) + if err := conn.QueryRow(context.Background(), + "SELECT have_total_difficulty, total_difficulty FROM v_blocks WHERE hash = ? LIMIT 1", + blockHash).Scan(&have, &td); err != nil { + t.Fatalf("scan v_blocks: %v", err) + } + if !have { + t.Fatal("total difficulty was lost with the events it was announced in") + } + if td == nil || td.Cmp(wantTD) != 0 { + t.Fatalf("total_difficulty: want %s got %v", wantTD, td) + } +} + +// TestAddressesAreStoredLowercase guards the address casing convention. +// +// common.Address.Hex() applies the EIP-55 checksum and returns mixed case. Stored +// that way, an address column cannot be joined: ClickHouse comparison is +// case-sensitive and every validator identity in this pipeline -- the Polygon +// staking API, block-latency, data-analysis -- is lowercase. The join returns no +// rows and no error, so the wrong answer looks like a real one. +func TestAddressesAreStoredLowercase(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + db := newTestClickHouse(t, dsn, ctx) + + now := time.Now().UTC() + // A coinbase whose checksummed form is mixed case, set before sealing. + mixedCoinbase := common.HexToAddress("0x25B9fC2ED95BBAa9c030e57C860545a17694F90D") + header, wantSigner := signedHeaderWithCoinbase(t, heightLowercase, now, mixedCoinbase) + // Signed, so types.Sender can recover from_address -- an unsigned transaction + // yields an empty sender and the assertion below would test nothing. + senderKey, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate sender key: %v", err) + } + chainID := big.NewInt(137) + tx, err := types.SignTx( + types.NewTx(&types.LegacyTx{ + Nonce: uint64(now.UnixNano()), + GasPrice: big.NewInt(2_000_000_000), + Gas: 21_000, + To: &mixedCoinbase, + Value: big.NewInt(1), + }), + types.LatestSignerForChainID(chainID), senderKey) + if err != nil { + t.Fatalf("sign tx: %v", err) + } + wantFrom := strings.ToLower(crypto.PubkeyToAddress(senderKey.PublicKey).Hex()) + block := types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: []*types.Transaction{tx}}) + + db.WriteBlock(ctx, testPeer(t), block, big.NewInt(1), now) + db.WriteTransactions(ctx, testPeer(t), []*types.Transaction{tx}, now) + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightLowercase) + + // The signer must round-trip as the lowercase form of the recovered address, + // proving it is lowercased rather than merely absent. + var signer, coinbase string + if err := conn.QueryRow(context.Background(), + "SELECT signer, coinbase FROM blocks WHERE hash = ? LIMIT 1", + block.Hash().Hex()).Scan(&signer, &coinbase); err != nil { + t.Fatalf("scan blocks: %v", err) + } + if signer != wantSigner { + t.Fatalf("signer: want %s got %s", wantSigner, signer) + } + if want := strings.ToLower(mixedCoinbase.Hex()); coinbase != want { + t.Fatalf("coinbase: want %s got %s", want, coinbase) + } + + var from, to string + if err := conn.QueryRow(context.Background(), + "SELECT from_address, to_address FROM transactions WHERE hash = ? LIMIT 1", + tx.Hash().Hex()).Scan(&from, &to); err != nil { + t.Fatalf("scan transactions: %v", err) + } + for name, got := range map[string]string{"from_address": from, "to_address": to} { + if got == "" { + t.Fatalf("%s was empty", name) + } + if got != strings.ToLower(got) { + t.Fatalf("%s is not lowercase: %s", name, got) + } + } + if from != wantFrom { + t.Fatalf("from_address: want %s got %s", wantFrom, from) + } + if want := strings.ToLower(mixedCoinbase.Hex()); to != want { + t.Fatalf("to_address: want %s got %s", want, to) + } +} + +// TestHeaderEventsDoNotRequireWriteBlocks is the third and last instance of the +// provenance-gating defect, after new_block/body and full_tx. +// +// WriteBlockHeaders returned early on !shouldWriteBlocks, before reaching +// recordsBlockEvents, which made header and header_backfill the only two provenance +// sources that also required --write-blocks. A fleet with it off still requests +// headers -- getBlockData has no such gate -- so the events were produced and thrown +// away, silently emptying v_block_provenance's header rows and losing the +// header_backfill marker that replaced Datastore's IsParent. +// +// The header row and the header event are separate concerns and are now gated +// separately. +func TestHeaderEventsDoNotRequireWriteBlocks(t *testing.T) { + dsn := clickHouseDSN(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Block facts off, block events on. + db := NewClickHouse(ctx, ClickHouseOptions{ + DSN: dsn, + SensorID: "test-sensor-header-events", + ChainID: 137, + MaxConcurrency: 10, + ShouldWriteBlocks: false, + ShouldWriteBlockEvents: false, + ShouldWriteFirstBlockEvent: true, + }) + + now := time.Now().UTC() + header, _ := signedHeader(t, heightHeaderEvents, now) + parent, _ := signedHeader(t, heightHeaderEvents-1, now) + + db.WriteBlockHeaders(ctx, []*types.Header{header}, now, false) + db.WriteBlockHeaders(ctx, []*types.Header{parent}, now, true) + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn := verifyConn(t, dsn) + cleanupTestHeights(t, conn, heightHeaderEvents, heightHeaderEvents-1) + + // Both header sources must be recorded on the event flag alone. + for hash, want := range map[string]string{ + header.Hash().Hex(): "header", + parent.Hash().Hex(): "header_backfill", + } { + checkCount(t, conn, + "SELECT count() FROM block_events WHERE block_hash = ? AND source = '"+want+"'", hash) + } + + // And the fact row must still be suppressed -- the two gates are independent, so + // this proves the fix separated them rather than just widening one. + var blocks uint64 + if err := conn.QueryRow(context.Background(), + "SELECT count() FROM blocks WHERE number IN (?, ?)", + uint64(heightHeaderEvents), uint64(heightHeaderEvents-1)).Scan(&blocks); err != nil { + t.Fatalf("scan blocks: %v", err) + } + if blocks != 0 { + t.Fatalf("write_blocks=false should write no blocks rows, got %d", blocks) + } +} + +// TestUnavailableBackendKeepsWarning covers the degraded path, which used to be +// invisible: one error at startup and then every write silently discarded while the +// sensor peered and looked healthy. A ClickHouse auth failure did exactly that for +// an hour across two sensors. +// +// It also guards the shutdown hang found while writing it -- the warning goroutine +// listened on the caller's context while Close only cancels its own, so Close waited +// forever on a goroutine nothing could stop. +func TestUnavailableBackendKeepsWarning(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + db := NewClickHouse(ctx, ClickHouseOptions{ + DSN: "clickhouse://127.0.0.1:1/nope", + SensorID: "warn", + ChainID: 137, + MaxConcurrency: 1, + ShouldWriteBlocks: true, + }) + ch, ok := db.(*ClickHouse) + if !ok { + t.Fatal("not a *ClickHouse") + } + if ch.conn != nil { + t.Skip("unexpectedly connected") + } + // HasBlock on a nil conn must suppress backfill. It is a read, so it must NOT + // count toward discarded -- counting it made the warning report call volume, + // and read 0 forever under --write-blocks=false, where HasBlock is unreachable. + if !db.HasBlock(ctx, [32]byte{1}) { + t.Fatal("HasBlock should return true with no connection, to suppress backfill") + } + if got := ch.discarded.Load(); got != 0 { + t.Fatalf("HasBlock counted toward discarded: %d", got) + } + + // Writes are what the counter measures: a block with two transactions is three + // rows lost. + now := time.Now().UTC() + header, _ := signedHeader(t, heightWrites, now) + tx1 := types.NewTx(&types.LegacyTx{Nonce: uint64(now.UnixNano()), GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(1)}) + tx2 := types.NewTx(&types.LegacyTx{Nonce: uint64(now.UnixNano()) + 1, GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(1)}) + block := types.NewBlockWithHeader(header).WithBody(types.Body{Transactions: []*types.Transaction{tx1, tx2}}) + db.WriteBlock(ctx, testPeer(t), block, big.NewInt(1), now) + // 4 block-level rows (blocks, block_bodies, block_total_difficulty, the event) + // plus block_txs and transactions per transaction. + if got := ch.discarded.Load(); got != 8 { + t.Fatalf("discarded: want 8 (4 block rows + 2 per tx), got %d", got) + } + // The warner goroutine must be registered with the WaitGroup so Close waits. + done := make(chan struct{}) + go func() { _ = db.Close(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Close hung: the warning goroutine is not stopping on cancel") } } diff --git a/p2p/database/database.go b/p2p/database/database.go index 0453e1d5b..2d855d5c8 100644 --- a/p2p/database/database.go +++ b/p2p/database/database.go @@ -12,36 +12,61 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" ) +// BlockAnnouncement is an announced block hash with its height, as carried by eth +// NewBlockHashes. ClickHouse keys observations by block number and needs the +// height; hash-keyed backends ignore it. +// +// p2p.NewBlockHashesPacket is a slice of this type, so a decoded packet reaches +// WriteBlockEvents with no copy or conversion. +type BlockAnnouncement struct { + Hash common.Hash + Number uint64 +} + +// Hashes drops the heights, for backends that only store hashes. +func Hashes(anns []BlockAnnouncement) []common.Hash { + hashes := make([]common.Hash, 0, len(anns)) + for _, ann := range anns { + hashes = append(hashes, ann.Hash) + } + return hashes +} + // Database represents a database solution to write block and transaction data // to. To use another database solution, just implement these methods and // update the sensor to use the new connection. type Database interface { - // WriteBlock will write the both the block and block event to the database - // if ShouldWriteBlocks and ShouldWriteBlockEvents return true, respectively. + // WriteBlock records a block delivered whole (NewBlock). The block row is + // written if ShouldWriteBlocks returns true; the event is written if either + // ShouldWriteBlockEvents or ShouldWriteFirstBlockEvent returns true, because + // a whole-block delivery is one event per block per sensor and is the only + // place a block first seen this way is observed. WriteBlock(context.Context, *enode.Node, *types.Block, *big.Int, time.Time) // WriteBlockHeaders will write the block headers if ShouldWriteBlocks // returns true. WriteBlockHeaders(context.Context, []*types.Header, time.Time, bool) - // WriteBlockEvents appends an inbound block event (peer, hash, time) for - // each hash — one per peer we received the announcement from. The caller - // decides which hashes to pass (every announcement for the full per-peer + // WriteBlockEvents appends an inbound block event (peer, hash, height, time) + // for each announcement — one per peer we received the announcement from. The + // caller decides which announcements to pass (every one for the full per-peer // stream, or just the first-seen ones); the backend only appends. - WriteBlockEvents(context.Context, *enode.Node, []common.Hash, time.Time) + WriteBlockEvents(context.Context, *enode.Node, []BlockAnnouncement, time.Time) - // WriteBlockHashFirstSeen records the earliest sighting of a block hash on + // WriteBlockHashFirstSeen records the earliest event for a block hash on // the block entity itself (Datastore's TimeFirstSeenHash). Backends that // derive first-seen from the event stream (e.g. ClickHouse) treat it as a // no-op. WriteBlockHashFirstSeen(context.Context, *enode.Node, common.Hash, time.Time) // WriteBlockBody writes the transactions carried in the block body (the block - // row itself comes from WriteBlock/WriteBlockHeaders). Backends with a + // row itself comes from WriteBlock/WriteBlockHeaders). The announcement carries + // the height, which the eth block body does not, so backends that key + // observations by block number can record the body's arrival. Backends with a // separate transactions table (e.g. ClickHouse) gate this on // ShouldWriteTransactions; the Datastore backend gates on ShouldWriteBlocks // because it links the transactions and uncles onto the block entity. - WriteBlockBody(context.Context, *eth.BlockBody, common.Hash, time.Time) + WriteBlockBody(context.Context, *eth.BlockBody, BlockAnnouncement, time.Time) // WriteTransactions writes the transaction bodies if ShouldWriteTransactions // returns true. Transaction events are recorded separately via diff --git a/p2p/database/datastore.go b/p2p/database/datastore.go index 79d67a25a..4f0f2e9aa 100644 --- a/p2p/database/datastore.go +++ b/p2p/database/datastore.go @@ -173,13 +173,29 @@ func (d *Datastore) runAsync(fn func()) { }() } +// recordsBlockEvents reports whether a block event should be written for a block +// delivered whole (NewBlock), as opposed to announced by hash. +// +// A NewBlock delivery is one event per block per sensor, so it follows either +// flag rather than shouldWriteBlockEvents alone. That flag exists to bound the +// hash-announcement firehose (~52 events per block per sensor, scaling with peer +// count), which WriteBlockEvents handles. Gating this write behind it meant the +// production config (write-block-events=false, write-first-block-event=true) +// recorded no event at all for blocks whose NewBlock arrived before any +// NewBlockHashes -- common on Bor, where the sensor is often a direct peer of a +// propagator. Later hash announcements do not close the gap: the block is +// already fully cached, so they are skipped as duplicates. +func (d *Datastore) recordsBlockEvents() bool { + return d.shouldWriteBlockEvents || d.shouldWriteFirstBlockEvent +} + // WriteBlock writes the block and the block event to datastore. func (d *Datastore) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time) { if d.client == nil { return } - if d.ShouldWriteBlockEvents() { + if d.recordsBlockEvents() && peer != nil { d.runAsync(func() { d.writeEvent(peer, BlockEventsKind, block.Hash(), BlocksKind, tfs) }) @@ -214,7 +230,8 @@ func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Head // requested. The block events will be written when the hash is received // instead. It will write the uncles and transactions to datastore if they // don't already exist. -func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { +func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, ann BlockAnnouncement, tfs time.Time) { + hash := ann.Hash if d.client == nil || !d.ShouldWriteBlocks() { return } @@ -224,12 +241,14 @@ func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, has }) } -// WriteBlockEvents appends an inbound block event per hash for the given peer. -func (d *Datastore) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if d.client == nil || peer == nil || len(hashes) == 0 { +// WriteBlockEvents appends an inbound block event per announcement for the given +// peer. Announced heights are unused here: Datastore keys events by block key. +func (d *Datastore) WriteBlockEvents(ctx context.Context, peer *enode.Node, anns []BlockAnnouncement, tfs time.Time) { + if d.client == nil || peer == nil || len(anns) == 0 { return } + hashes := Hashes(anns) d.runAsync(func() { d.writeEvents(ctx, peer, BlockEventsKind, hashes, BlocksKind, tfs) }) diff --git a/p2p/database/json.go b/p2p/database/json.go index b8a796d6e..704089756 100644 --- a/p2p/database/json.go +++ b/p2p/database/json.go @@ -78,6 +78,7 @@ type JSONBlockEvent struct { SensorID string `json:"sensor_id"` PeerID string `json:"peer_id"` Hash string `json:"hash"` + Number uint64 `json:"number,omitempty"` Timestamp time.Time `json:"timestamp"` } @@ -218,17 +219,18 @@ func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.H } // WriteBlockEvents writes the block events as JSON. -func (j *JSONDatabase) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if peer == nil || len(hashes) == 0 { +func (j *JSONDatabase) WriteBlockEvents(ctx context.Context, peer *enode.Node, anns []BlockAnnouncement, tfs time.Time) { + if peer == nil || len(anns) == 0 { return } - for _, hash := range hashes { + for _, ann := range anns { event := JSONBlockEvent{ Type: "block_hash", SensorID: j.sensorID, PeerID: peer.URLv4(), - Hash: hash.Hex(), + Hash: ann.Hash.Hex(), + Number: ann.Number, Timestamp: tfs, } @@ -252,7 +254,8 @@ func (j *JSONDatabase) WriteBlockHashFirstSeen(ctx context.Context, peer *enode. } // WriteBlockBody writes the block body as JSON. -func (j *JSONDatabase) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { +func (j *JSONDatabase) WriteBlockBody(ctx context.Context, body *eth.BlockBody, ann BlockAnnouncement, tfs time.Time) { + hash := ann.Hash if !j.ShouldWriteBlocks() { return } diff --git a/p2p/database/nodb.go b/p2p/database/nodb.go index ae9f79459..06a50c52c 100644 --- a/p2p/database/nodb.go +++ b/p2p/database/nodb.go @@ -30,7 +30,7 @@ func (n *nodb) WriteBlockHeaders(ctx context.Context, headers []*types.Header, t } // WriteBlockEvents does nothing. -func (n *nodb) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { +func (n *nodb) WriteBlockEvents(ctx context.Context, peer *enode.Node, anns []BlockAnnouncement, tfs time.Time) { } // WriteBlockHashFirstSeen does nothing. @@ -38,7 +38,7 @@ func (n *nodb) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, ha } // WriteBlockBody does nothing. -func (n *nodb) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { +func (n *nodb) WriteBlockBody(ctx context.Context, body *eth.BlockBody, ann BlockAnnouncement, tfs time.Time) { } // WriteTransactions does nothing. diff --git a/p2p/decode_test.go b/p2p/decode_test.go new file mode 100644 index 000000000..7bca4f8c5 --- /dev/null +++ b/p2p/decode_test.go @@ -0,0 +1,127 @@ +package p2p + +import ( + "crypto/ecdsa" + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/rlp" +) + +// TestDecodeTxSurvivesHostileInput is the regression test for a remote process kill. +// +// decodeTx logged bytes[0] after rlp.DecodeBytes succeeded. 0x80 is a valid RLP +// empty string, so DecodeBytes returns an empty slice with no error and the index +// panicked -- on the peer's own Protocol.Run goroutine, which geth does not recover +// around, from four unauthenticated message paths (TransactionsMsg, +// PooledTransactionsMsg, NewBlockMsg, BlockBodiesMsg). A three-byte message killed +// the sensor. +// +// This is the same failure class as the clique panic util.Ecrecover now converts to +// an error; that fix's comment names the mechanism exactly. +func TestDecodeTxSurvivesHostileInput(t *testing.T) { + c := &conn{} + + for _, tc := range []struct { + name string + raw []byte + }{ + {"empty_rlp_string", []byte{0x80}}, // decodes to []byte{} with no error + {"empty_rlp_list", []byte{0xc0}}, // decodes to a list, not bytes + {"single_zero_byte", []byte{0x00}}, // valid RLP, one zero byte + {"truncated_typed_tx", []byte{0x02}}, // EIP-1559 prefix, nothing after + {"long_string_header_only", []byte{0xb8}}, // claims a length, supplies none + {"nested_empty", []byte{0xc1, 0x80}}, // list containing an empty string + {"garbage", []byte{0xff, 0xff, 0xff, 0xff}}, // not valid RLP at all + } { + t.Run(tc.name, func(t *testing.T) { + // Must return nil rather than panic; a peer controls this input entirely. + if tx := c.decodeTx(tc.raw); tx != nil { + t.Fatalf("unexpectedly decoded a transaction from %x", tc.raw) + } + }) + } +} + +// The list paths must survive the same input, since that is how a peer actually +// delivers it: a Transactions packet is a list of these blobs. +func TestDecodeTxsSurvivesHostileInput(t *testing.T) { + c := &conn{} + raws := []rlp.RawValue{{0x80}, {0xc0}, {0x02}, {0xff}} + if got := c.decodeTxs(raws); len(got) != 0 { + t.Fatalf("expected nothing to decode, got %d transactions", len(got)) + } +} + +// TestBuildBlockBodyRejectsUndecodableTx is the regression test for silent block +// corruption. +// +// buildBlockBody re-encoded whatever survived the lenient decodeTxs, so one garbage +// blob in a peer's BlockBodies response produced a body that was not the body for +// that hash: block_bodies got a tx_count short by the drops, and writeBlockTxs +// indexes by loop position so every later tx_index shifted down. Keyed by a real +// block hash, in ReplacingMergeTree tables with no version column, that row could +// win the merge against an honest sensor's and persist. +func TestBuildBlockBodyRejectsUndecodableTx(t *testing.T) { + c := &conn{} + + good, err := types.SignTx( + types.NewTx(&types.LegacyTx{Nonce: 1, GasPrice: big.NewInt(1), Gas: 21_000, Value: big.NewInt(1)}), + types.LatestSignerForChainID(big.NewInt(137)), mustKey(t)) + if err != nil { + t.Fatalf("sign: %v", err) + } + goodRLP, err := good.MarshalBinary() + if err != nil { + t.Fatalf("marshal: %v", err) + } + encGood, err := rlp.EncodeToBytes(goodRLP) + if err != nil { + t.Fatalf("encode: %v", err) + } + + // A body whose first transaction is fine and whose second is garbage. + body := struct { + Transactions []rlp.RawValue + Uncles []*types.Header + }{ + Transactions: []rlp.RawValue{encGood, {0x80}}, + } + raw, err := rlp.EncodeToBytes(&body) + if err != nil { + t.Fatalf("encode body: %v", err) + } + + if _, buildErr := c.buildBlockBody(raw); buildErr == nil { + t.Fatal("accepted a body with an undecodable transaction; the tx_index mapping would be silently shifted") + } + + // The all-good case must still work, or the guard is useless. + body.Transactions = []rlp.RawValue{encGood} + raw, err = rlp.EncodeToBytes(&body) + if err != nil { + t.Fatalf("encode body: %v", err) + } + built, err := c.buildBlockBody(raw) + if err != nil { + t.Fatalf("rejected a valid body: %v", err) + } + items, err := built.Transactions.Items() + if err != nil { + t.Fatalf("items: %v", err) + } + if len(items) != 1 || items[0].Hash() != good.Hash() { + t.Fatalf("round trip lost the transaction: %d items", len(items)) + } +} + +func mustKey(t *testing.T) *ecdsa.PrivateKey { + t.Helper() + k, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + return k +} diff --git a/p2p/protocol.go b/p2p/protocol.go index 6dcb2e5e9..0c2b22944 100644 --- a/p2p/protocol.go +++ b/p2p/protocol.go @@ -17,6 +17,7 @@ import ( ethp2p "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" "github.com/rs/zerolog" "github.com/rs/zerolog/log" @@ -47,10 +48,11 @@ type conn struct { db database.Database peer *ethp2p.Peer - // requests is used to store the request ID and the block hash. This is used - // when fetching block bodies because the eth protocol block bodies do not - // contain information about the block hash. - requests *ds.LRU[uint64, common.Hash] + // requests is used to store the request ID and the announced block (hash and + // height). This is used when fetching block bodies because the eth protocol + // block bodies carry neither the hash nor the height of the block they belong + // to, and observations are keyed by height. + requests *ds.LRU[uint64, database.BlockAnnouncement] requestNum uint64 // parents tracks hashes of blocks requested as parents to mark them @@ -141,7 +143,7 @@ func NewEthProtocol(version uint, opts EthProtocolOptions) ethp2p.Protocol { logger: log.With().Str("peer", peerURL).Logger(), rw: rw, db: opts.Database, - requests: ds.NewLRU[uint64, common.Hash](opts.RequestsCache), + requests: ds.NewLRU[uint64, database.BlockAnnouncement](opts.RequestsCache), requestNum: 0, parents: ds.NewLRU[common.Hash, struct{}](opts.ParentsCache), peer: p, @@ -445,7 +447,8 @@ func (c *conn) handleBlockRangeUpdate(msg ethp2p.Msg) error { // peer based on what parts of the block we already have. It will return an error // if sending either of the requests failed. The isParent parameter indicates if // this block is being fetched as a parent block. -func (c *conn) getBlockData(hash common.Hash, cache BlockCache, isParent bool) error { +func (c *conn) getBlockData(ann database.BlockAnnouncement, cache BlockCache, isParent bool) error { + hash := ann.Hash // Only request header if we don't have it if cache.Header == nil { headersRequest := &GetBlockHeaders{ @@ -470,7 +473,7 @@ func (c *conn) getBlockData(hash common.Hash, cache BlockCache, isParent bool) e // Only request body if we don't have it if cache.Body == nil { c.requestNum++ - c.requests.Add(c.requestNum, hash) + c.requests.Add(c.requestNum, ann) bodiesRequest := &GetBlockBodies{ RequestId: c.requestNum, @@ -515,7 +518,13 @@ func (c *conn) getParentBlock(ctx context.Context, header *types.Header) error { Str("number", new(big.Int).Sub(header.Number, big.NewInt(1)).String()). Msg("Fetching missing parent block") - return c.getBlockData(header.ParentHash, cache, true) + // The parent sits one below this header, so the height is known without + // fetching it. + parent := database.BlockAnnouncement{ + Hash: header.ParentHash, + Number: header.Number.Uint64() - 1, + } + return c.getBlockData(parent, cache, true) } // eventHashes selects which announced hashes to record as inbound events: every @@ -532,6 +541,18 @@ func eventHashes(all, firstSeen []common.Hash, full, firstOnly bool) []common.Ha } } +// eventAnnouncements is eventHashes for announcements that carry their height. +func eventAnnouncements(all, firstSeen []database.BlockAnnouncement, full, firstOnly bool) []database.BlockAnnouncement { + switch { + case full: + return all + case firstOnly: + return firstSeen + default: + return nil + } +} + func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { var packet NewBlockHashesPacket if err := msg.Decode(&packet); err != nil { @@ -542,15 +563,13 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { c.countMsgReceived(packet.Name(), float64(len(packet))) - // allHashes is every announced hash (the full per-peer event stream); - // uniqueHashes are the first-seen ones (for first-seen events and rebroadcast). - allHashes := make([]common.Hash, 0, len(packet)) - uniqueHashes := make([]common.Hash, 0, len(packet)) - uniqueNumbers := make([]uint64, 0, len(packet)) + // The decoded packet is the full per-peer event stream, passed as-is. + // uniqueAnns collects the first-seen subset for first-seen events and + // rebroadcast. + uniqueAnns := make([]database.BlockAnnouncement, 0, len(packet)) for _, entry := range packet { hash := entry.Hash - allHashes = append(allHashes, hash) // Update latest block info atomically if this block is newer c.latestBlock.Update(func(current latestBlock) (latestBlock, bool) { @@ -582,13 +601,12 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { if !existed { // Write hash first seen time immediately for new blocks. c.db.WriteBlockHashFirstSeen(ctx, c.node, hash, tfs) - uniqueHashes = append(uniqueHashes, hash) - uniqueNumbers = append(uniqueNumbers, entry.Number) + uniqueAnns = append(uniqueAnns, entry) } // Request only the parts we don't have yet (getBlockData inspects the // cache entry and asks for the missing header and/or body). - if err := c.getBlockData(hash, cache, false); err != nil { + if err := c.getBlockData(entry, cache, false); err != nil { return err } } @@ -596,10 +614,10 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { // Record inbound block events: the full per-peer stream (every peer that // announced) or only the first-seen hashes, per the flags. c.db.WriteBlockEvents(ctx, c.node, - eventHashes(allHashes, uniqueHashes, c.db.ShouldWriteBlockEvents(), c.db.ShouldWriteFirstBlockEvent()), tfs) + eventAnnouncements(packet, uniqueAnns, c.db.ShouldWriteBlockEvents(), c.db.ShouldWriteFirstBlockEvent()), tfs) // Only newly-seen hashes are rebroadcast. - if len(uniqueHashes) == 0 { + if len(uniqueAnns) == 0 { return nil } @@ -608,6 +626,12 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { // requests (getBlockData above) are intentionally not gated. When // validation is disabled, rebroadcast the announced hashes immediately. if !c.conns.ValidatesSigners() { + uniqueHashes := make([]common.Hash, 0, len(uniqueAnns)) + uniqueNumbers := make([]uint64, 0, len(uniqueAnns)) + for _, ann := range uniqueAnns { + uniqueHashes = append(uniqueHashes, ann.Hash) + uniqueNumbers = append(uniqueNumbers, ann.Number) + } go c.conns.BroadcastBlockHashes(uniqueHashes, uniqueNumbers) } @@ -822,9 +846,17 @@ func (c *conn) decodeTx(raw []byte) *types.Transaction { return tx } + // bytes[0] only after a length check: 0x80 is a valid RLP empty string, so + // DecodeBytes succeeds with an empty slice and indexing it panicked. That + // panic runs on the peer's own Protocol.Run goroutine, which geth does not + // recover around, so a three-byte message from any peer killed the process. + txType := -1 + if len(bytes) > 0 { + txType = int(bytes[0]) + } c.logger.Warn(). Err(err). - Uint8("type", bytes[0]). + Int("type", txType). Int("size", len(bytes)). Str("hash", crypto.Keccak256Hash(bytes).Hex()). Msg("Failed to decode transaction") @@ -839,9 +871,16 @@ func (c *conn) decodeTx(raw []byte) *types.Transaction { return tx } + // raw is guaranteed non-empty by the guard at the top of this function; the + // length check is repeated so the two log sites read the same way and neither + // depends on a guard twenty lines above it. + prefix := -1 + if len(raw) > 0 { + prefix = int(raw[0]) + } c.logger.Warn(). Err(err). - Uint8("prefix", raw[0]). + Int("prefix", prefix). Int("size", len(raw)). Str("hash", crypto.Keccak256Hash(raw).Hex()). Msg("Failed to decode transaction") @@ -849,6 +888,34 @@ func (c *conn) decodeTx(raw []byte) *types.Transaction { return nil } +// decodeTxsStrict decodes every transaction or fails. +// +// decodeTxs below is lenient on purpose: a Transactions or PooledTransactions packet +// is a batch of INDEPENDENT transactions, so dropping one that will not decode loses +// exactly that transaction. A block body is not a batch -- the block hash commits to +// txRoot, which commits to every transaction in order -- so silently dropping one +// yields a body that is not the body for that hash. Re-encoded and written, it gave +// block_bodies a tx_count short by the dropped transactions and shifted every later +// block_txs.tx_index down, keyed by a hash whose real contents were different. Both +// tables are ReplacingMergeTree, so the corrupt row could win the merge against an +// honest sensor's and persist. +// +// This does not verify the body against the header's txRoot -- the header arrives on +// a separate round trip and may not be held yet -- so a peer can still substitute a +// well-formed body for a hash it does not belong to. Failing closed on undecodable +// input is the part that can be done here. +func (c *conn) decodeTxsStrict(rawTxs []rlp.RawValue) ([]*types.Transaction, error) { + txs := make([]*types.Transaction, 0, len(rawTxs)) + for i, raw := range rawTxs { + tx := c.decodeTx(raw) + if tx == nil { + return nil, fmt.Errorf("transaction %d of %d failed to decode", i, len(rawTxs)) + } + txs = append(txs, tx) + } + return txs, nil +} + // decodeTxs decodes a list of transactions, returning only successfully decoded ones. func (c *conn) decodeTxs(rawTxs []rlp.RawValue) []*types.Transaction { var txs []*types.Transaction @@ -1105,11 +1172,12 @@ func (c *conn) handleBlockBodies(ctx context.Context, msg ethp2p.Msg) error { c.countMsgReceived((*eth.BlockBodiesResponse)(nil).Name(), float64(len(packet.BlockBodiesRLPResponse))) - hash, ok := c.requests.Get(packet.RequestId) + ann, ok := c.requests.Get(packet.RequestId) if !ok { c.logger.Warn().Msg("No block hash found for block body") return nil } + hash := ann.Hash c.requests.Remove(packet.RequestId) // Check if we already have the body in the cache @@ -1123,7 +1191,7 @@ func (c *conn) handleBlockBodies(ctx context.Context, msg ethp2p.Msg) error { return nil } - c.db.WriteBlockBody(ctx, body, hash, tfs) + c.db.WriteBlockBody(ctx, body, ann, tfs) // When cache-only-validated is enabled, only retain the body if the block // already has a cache entry (the announcement marker, or a cached header). @@ -1178,7 +1246,11 @@ func (c *conn) buildBlockBody(raw []byte) (*eth.BlockBody, error) { return nil, fmt.Errorf("failed to decode block body: %w", err) } - txList, err := rlp.EncodeToRawList(c.decodeTxs(decoded.Transactions)) + txs, err := c.decodeTxsStrict(decoded.Transactions) + if err != nil { + return nil, fmt.Errorf("failed to decode block body transactions: %w", err) + } + txList, err := rlp.EncodeToRawList(txs) if err != nil { return nil, fmt.Errorf("failed to encode transactions: %w", err) } @@ -1215,11 +1287,35 @@ func (c *conn) handleNewBlock(ctx context.Context, msg ethp2p.Msg) error { return nil } + // Strict, and then checked against the header. A NewBlock carries its own + // header, so unlike the BlockBodies path the body can be verified rather than + // merely required to parse: DeriveSha over the transactions must equal + // header.TxHash, which the block hash commits to. + // + // Lenient decoding here silently dropped undecodable transactions and kept the + // peer's header, so block_bodies got a tx_count short by the drops and block_txs + // had every later tx_index shifted down -- keyed by a real block hash whose real + // contents differed. Both are ReplacingMergeTree, so that row could beat an + // honest sensor's and persist. + txs, err := c.decodeTxsStrict(raw.Block.Txs) + if err != nil { + c.logger.Warn().Err(err).Msg("Dropping new block with undecodable transactions") + return nil + } block := types.NewBlockWithHeader(raw.Block.Header).WithBody(types.Body{ - Transactions: c.decodeTxs(raw.Block.Txs), + Transactions: txs, Uncles: raw.Block.Uncles, Withdrawals: raw.Block.Withdrawals, }) + if root := types.DeriveSha(types.Transactions(txs), trie.NewStackTrie(nil)); root != raw.Block.Header.TxHash { + c.logger.Warn(). + Str("hash", block.Hash().Hex()). + Str("want_tx_root", raw.Block.Header.TxHash.Hex()). + Str("got_tx_root", root.Hex()). + Int("txs", len(txs)). + Msg("Dropping new block whose body does not match its header") + return nil + } packet := &NewBlockPacket{Block: block, TD: raw.TD} tfs := time.Now() diff --git a/p2p/protocol_events_test.go b/p2p/protocol_events_test.go index ee9ed5691..31e7d21fc 100644 --- a/p2p/protocol_events_test.go +++ b/p2p/protocol_events_test.go @@ -19,11 +19,15 @@ import ( type recordingDB struct { database.Database blockEvents, txEvents int + blockNumbers []uint64 fullBlock, firstBlock, fullTx, firstTx bool } -func (r *recordingDB) WriteBlockEvents(_ context.Context, _ *enode.Node, hashes []common.Hash, _ time.Time) { - r.blockEvents += len(hashes) +func (r *recordingDB) WriteBlockEvents(_ context.Context, _ *enode.Node, anns []database.BlockAnnouncement, _ time.Time) { + r.blockEvents += len(anns) + for _, ann := range anns { + r.blockNumbers = append(r.blockNumbers, ann.Number) + } } func (r *recordingDB) WriteTransactionEvents(_ context.Context, _ *enode.Node, hashes []common.Hash, _ time.Time) { @@ -91,10 +95,34 @@ func TestBlockEventFullVsFirst(t *testing.T) { if rec.blockEvents != tc.want { t.Fatalf("%s: recorded %d block events, want %d", tc.name, rec.blockEvents, tc.want) } + // The announced height must reach the backend: observations are keyed + // by block number, and a zero here would silently key every row to 0. + for _, number := range rec.blockNumbers { + if number != 200 { + t.Fatalf("%s: recorded block number %d, want 200", tc.name, number) + } + } }) } } +func TestEventAnnouncements(t *testing.T) { + all := []database.BlockAnnouncement{{Hash: common.Hash{1}, Number: 1}, {Hash: common.Hash{2}, Number: 2}} + first := []database.BlockAnnouncement{{Hash: common.Hash{1}, Number: 1}} + if got := eventAnnouncements(all, first, true, false); len(got) != 2 { + t.Errorf("full: got %d announcements, want 2", len(got)) + } + if got := eventAnnouncements(all, first, false, true); len(got) != 1 { + t.Errorf("first: got %d announcements, want 1", len(got)) + } + if got := eventAnnouncements(all, first, false, false); got != nil { + t.Errorf("neither: got %v, want nil", got) + } + if got := eventAnnouncements(all, first, true, true); len(got) != 2 { + t.Errorf("full takes precedence: got %d announcements, want 2", len(got)) + } +} + // TestTransactionEventFullVsFirst is the tx mirror of TestBlockEventFullVsFirst. func TestTransactionEventFullVsFirst(t *testing.T) { conns := sharedTestConns(t, false) diff --git a/p2p/protocol_test.go b/p2p/protocol_test.go index 18bcb73d7..b1b7d8641 100644 --- a/p2p/protocol_test.go +++ b/p2p/protocol_test.go @@ -81,7 +81,7 @@ func newTestConn(rw ethp2p.MsgReadWriter, conns *Conns) *conn { logger: zerolog.Nop(), rw: rw, db: database.NoDatabase(), - requests: ds.NewLRU[uint64, common.Hash](ds.LRUOptions{MaxSize: 1024}), + requests: ds.NewLRU[uint64, database.BlockAnnouncement](ds.LRUOptions{MaxSize: 1024}), parents: ds.NewLRU[common.Hash, struct{}](ds.LRUOptions{MaxSize: 1024}), conns: conns, messages: NewPeerMessages(), @@ -177,7 +177,7 @@ func TestHandleBlockBodiesRetainsBodyBeforeHeader(t *testing.T) { // The announcement created a hash-only marker; the header has NOT arrived yet. conns.Blocks().Update(hash, func(bc BlockCache) BlockCache { return bc }) const reqID = 7 - c.requests.Add(reqID, hash) + c.requests.Add(reqID, database.BlockAnnouncement{Hash: hash, Number: 4242}) // Body response arrives before the header. if err := c.handleBlockBodies(ctx, makeBodies(t, reqID)); err != nil { diff --git a/p2p/types.go b/p2p/types.go index af29a4541..817d9af3c 100644 --- a/p2p/types.go +++ b/p2p/types.go @@ -18,6 +18,8 @@ import ( "github.com/ethereum/go-ethereum/p2p/rlpx" "github.com/ethereum/go-ethereum/rlp" "github.com/rs/zerolog" + + "github.com/0xPolygon/polygon-cli/p2p/database" ) type Message interface { @@ -115,10 +117,11 @@ func (*StatusPacket68) Name() string { return "Status" } // NewBlockHashesPacket is the network packet for block hash announcements. // Removed in go-ethereum v1.17.2 but still used by Bor/Polygon nodes. -type NewBlockHashesPacket []struct { - Hash common.Hash - Number uint64 -} +// +// The element type is database.BlockAnnouncement so a decoded packet passes +// straight to Database.WriteBlockEvents. RLP is unaffected: the wire format +// follows field order, not whether the element struct is named. +type NewBlockHashesPacket []database.BlockAnnouncement func (NewBlockHashesPacket) Name() string { return "NewBlockHashes" } diff --git a/util/ecrecover_test.go b/util/ecrecover_test.go new file mode 100644 index 000000000..b82fe2a42 --- /dev/null +++ b/util/ecrecover_test.go @@ -0,0 +1,80 @@ +package util + +import ( + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus/clique" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" +) + +// TestEcrecoverDoesNotPanicOnUntrustedHeader covers every trailing optional field +// clique's encodeSigHeader panics on. Headers arrive from peers, and the sensor's +// write path calls Ecrecover on the peer's own goroutine, which geth does not +// recover around -- so a panic here is a remote process kill, not a bad row. +// +// If a geth bump adds another panicking field, this test keeps passing (the +// recover is generic) but the new field belongs in the table so the coverage +// stays visible. +func TestEcrecoverDoesNotPanicOnUntrustedHeader(t *testing.T) { + hash := common.HexToHash("0xdead") + gas := uint64(1) + + for _, tc := range []struct { + name string + apply func(*types.Header) + }{ + {"withdrawals_hash", func(h *types.Header) { h.WithdrawalsHash = &hash }}, + {"excess_blob_gas", func(h *types.Header) { h.ExcessBlobGas = &gas }}, + {"blob_gas_used", func(h *types.Header) { h.BlobGasUsed = &gas }}, + {"parent_beacon_root", func(h *types.Header) { h.ParentBeaconRoot = &hash }}, + {"slot_number", func(h *types.Header) { h.SlotNumber = &gas }}, // EIP-7843, geth v1.17.4 + } { + t.Run(tc.name, func(t *testing.T) { + h := &types.Header{ + Number: big.NewInt(1), + Difficulty: big.NewInt(1), + Extra: make([]byte, crypto.SignatureLength), + } + tc.apply(h) + + // Must return an error, and must not panic. + signer, err := Ecrecover(h) + if err == nil { + t.Fatalf("expected an error for a header clique rejects, got signer %x", signer) + } + if signer != nil { + t.Fatalf("expected nil signer alongside the error, got %x", signer) + } + }) + } +} + +// A well-formed clique-sealed header must still recover, so the recover() above +// cannot be masking a real regression. +func TestEcrecoverStillRecoversValidHeader(t *testing.T) { + priv, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + h := &types.Header{ + Number: big.NewInt(1), + Difficulty: big.NewInt(1), + Extra: make([]byte, crypto.SignatureLength), + } + sig, err := crypto.Sign(clique.SealHash(h).Bytes(), priv) + if err != nil { + t.Fatalf("sign: %v", err) + } + copy(h.Extra[len(h.Extra)-crypto.SignatureLength:], sig) + + signer, err := Ecrecover(h) + if err != nil { + t.Fatalf("expected recovery to succeed: %v", err) + } + if got, want := common.BytesToAddress(signer), crypto.PubkeyToAddress(priv.PublicKey); got != want { + t.Fatalf("signer mismatch: got %s want %s", got, want) + } +} diff --git a/util/util.go b/util/util.go index c2b912529..d8c71b8dd 100644 --- a/util/util.go +++ b/util/util.go @@ -41,7 +41,24 @@ type ( // Ecrecover recovers the signer address from a block header's seal // (the last SignatureLength bytes of Extra) using the clique seal hash. -func Ecrecover(header *types.Header) ([]byte, error) { +// +// The header may come straight off the p2p wire, so treat it as untrusted. +// clique's encodeSigHeader panics rather than errors on any post-Merge trailing +// field it does not expect -- WithdrawalsHash, ExcessBlobGas, BlobGasUsed, +// ParentBeaconRoot, and as of go-ethereum v1.17.4 SlotNumber. Every one is +// rlp:"optional", so a peer sets them at will, and a caller running on a peer +// goroutine takes the whole process down with it. +// +// The panic is converted to an error rather than pre-checked field by field on +// purpose: that set grows with the geth releases that add one, and a hand-kept +// field list would silently stop covering it at the next dependency bump. +func Ecrecover(header *types.Header) (signer []byte, err error) { + defer func() { + if r := recover(); r != nil { + signer, err = nil, fmt.Errorf("unable to recover signature, clique rejected the header: %v", r) + } + }() + sigStart := len(header.Extra) - crypto.SignatureLength if sigStart < 0 || sigStart > len(header.Extra) { return nil, fmt.Errorf("unable to recover signature") @@ -51,9 +68,8 @@ func Ecrecover(header *types.Header) ([]byte, error) { if err != nil { return nil, err } - signer := crypto.Keccak256(pubkey[1:])[12:] - return signer, nil + return crypto.Keccak256(pubkey[1:])[12:], nil } func EcrecoverTx(tx *types.Transaction) ([]byte, error) {