From 8d41b8376a5d9528df346d92cd1e797c52d290ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:20:20 -0300 Subject: [PATCH 1/2] perf(p2p): verify one KZG batch per block for fetched data columns A range or by-root answer, or a batch of parked columns replayed after their parent imports, carries many columns of the same block. Each column was checked on its own blocking task with its own verify_cell_kzg_proof_batch call, and the chain actor heard nothing until every sidecar of the whole call was done, so data availability waited on the slowest block of the batch. check_and_forward now groups the sidecars by block (the signed header's root for fulu, beacon_block_root for gloas). A group takes one column_check_permits permit, runs every sidecar's rules that do not read the cells, then verifies the proofs of all sidecars still standing in one batch. A batch that fails is re-run sidecar by sidecar, so exactly the sidecars rejected before are rejected. The group's Keep and AwaitParent sidecars go to the chain actor as soon as the group finishes. Verdicts do not change: the proofs are the last rule on the chain path (the lookahead rule after them is the gossip path's alone), so chain_checks is now chain_precheck followed by a KZG step, with gossip's stateful_checks and stateful_checks_gloas keeping their order through the same split halves. On this machine a batch over 9-blob columns costs 3.3x less than separate calls at 8 columns and 4.8x less at 32. --- .../src/beacon/fork_choice.rs | 39 ++ .../src/beacon/gossip/column.rs | 487 +++++++++++++++--- crates/net/p2p/src/beacon/column_checks.rs | 252 +++++++-- docs/beacon_wire.md | 15 + docs/metrics.md | 2 +- 5 files changed, 687 insertions(+), 108 deletions(-) diff --git a/crates/blockchain/state_transition/src/beacon/fork_choice.rs b/crates/blockchain/state_transition/src/beacon/fork_choice.rs index 02c959376..7fce059cc 100644 --- a/crates/blockchain/state_transition/src/beacon/fork_choice.rs +++ b/crates/blockchain/state_transition/src/beacon/fork_choice.rs @@ -4130,6 +4130,45 @@ fn verify_column_cells( kzg::verify_cell_kzg_proof_batch(commitments, &cell_indices, &cells, proofs) } +/// One column's worth of input to [`verify_column_cells_batch`]: the cell +/// position every cell of the column sits at, the cells, and the commitments +/// and proofs they are checked against. +pub struct ColumnCells<'a> { + pub index: u64, + pub column: &'a [fulu::Cell], + pub commitments: &'a [KzgCommitment], + pub proofs: &'a [KzgProof], +} + +/// [`verify_column_cells`] over several columns at once: one +/// `verify_cell_kzg_proof_batch` call whose entries are every cell of every +/// column, in order. +/// +/// `Ok(true)` means every cell of every column verifies. Anything else (a +/// cell that fails, or a malformed entry) does not say which column is at +/// fault, so a caller that must tell them apart verifies the columns one by +/// one afterwards. Each column's three lists must have equal length, as +/// [`verify_data_column_sidecar`] and its gloas counterpart guarantee. +pub fn verify_column_cells_batch(columns: &[ColumnCells<'_>]) -> Result { + let total: usize = columns.iter().map(|column| column.column.len()).sum(); + let mut cell_indices = Vec::with_capacity(total); + let mut cells = Vec::with_capacity(total); + let mut commitments = Vec::with_capacity(total); + let mut proofs = Vec::with_capacity(total); + for column in columns { + cell_indices.extend(std::iter::repeat_n(column.index, column.column.len())); + for cell in column.column { + cells.push( + c_kzg::Cell::from_bytes(&cell[..]) + .map_err(|_| Error::SpecAssert("len(cell) == BYTES_PER_CELL"))?, + ); + } + commitments.extend_from_slice(column.commitments); + proofs.extend_from_slice(column.proofs); + } + kzg::verify_cell_kzg_proof_batch(&commitments, &cell_indices, &cells, &proofs) +} + /// `verify_data_column_sidecar` (gloas `p2p-interface.md`, modified): the /// structural checks a gloas column sidecar must pass before its KZG proofs /// are worth checking, against `kzg_commitments` from the bid of the block the diff --git a/crates/blockchain/state_transition/src/beacon/gossip/column.rs b/crates/blockchain/state_transition/src/beacon/gossip/column.rs index 900d83353..61ffa3f4c 100644 --- a/crates/blockchain/state_transition/src/beacon/gossip/column.rs +++ b/crates/blockchain/state_transition/src/beacon/gossip/column.rs @@ -34,7 +34,7 @@ use crate::beacon::lean_boundary::lean_block_unreachable; use crate::beacon::precheck; use crate::beacon::preset; use crate::beacon::primitives::HashTreeRoot as _; -use crate::beacon::primitives::{Root, Slot, ValidatorIndex}; +use crate::beacon::primitives::{KzgCommitment, Root, Slot, ValidatorIndex}; use crate::beacon::stf; use crate::metrics; @@ -138,43 +138,68 @@ pub enum ChainVerdict { /// Runs off both actors. It costs a KZG batch and a BLS verification per /// sidecar, and those used to run on the chain actor's single thread. pub fn chain_checks(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> ChainVerdict { + settle( + chain_precheck(store, sidecar, now_ms), + &ColumnRef::Fulu(sidecar), + ) +} + +/// [`chain_checks`] up to, and not including, the KZG proofs: every rule that +/// does not read the cells, in the order [`chain_checks`] runs them. +/// +/// The proofs are the last rule on the chain path, so a verdict returned here +/// is final whatever the proofs say, and [`ChainPrecheck::Pending`] means the +/// sidecar is [`ChainVerdict::Keep`] exactly when its proofs verify. +pub fn chain_precheck(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> ChainPrecheck { let config = store.config(); // [REJECT] The sidecar is structurally valid. First, so a peer answering // a fetch with garbage cannot make this pay for a state read or a KZG // batch. if !fork_choice::verify_data_column_sidecar(sidecar, &config) { - return ChainVerdict::Drop(Outcome::Reject(RejectReason::Malformed)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Reject( + RejectReason::Malformed, + ))); } let header = &sidecar.signed_block_header.message; // [IGNORE] Not from a future slot. Also what bounds `advanced_proposer`'s // `process_slots` below: a header naming a slot far ahead would otherwise // make it advance one slot at a time towards that slot. if is_future_slot(&config, header.slot, now_ms) { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::FutureSlot)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::FutureSlot, + ))); } // [IGNORE] From a slot greater than the latest finalized slot. if header.slot <= finalized_start_slot(store) { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::Finalized)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::Finalized, + ))); } // [IGNORE] Already stored. Common during a range sync, whose consecutive // batches ask for overlapping spans of columns, and everything after this // is the expensive part. if store.has_data_column(header.slot, &header.hash_tree_root(), sidecar.index) { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::AlreadyStored)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::AlreadyStored, + ))); } let Ok(Some(parent_state)) = store.get_state(&header.parent_root) else { // Parked rather than dropped: the specification's "MAY be queued" // for a sidecar whose parent has not been seen. The head-state // signature check keeps a forged header from taking a // `PendingDataColumns` row while it waits. - return match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown) { - Outcome::Queue(_) => ChainVerdict::AwaitParent, - outcome => ChainVerdict::Drop(outcome), - }; + return ChainPrecheck::Settled( + match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown) { + Outcome::Queue(_) => ChainVerdict::AwaitParent, + outcome => ChainVerdict::Drop(outcome), + }, + ); }; - match judge_against_parent(store, sidecar, &parent_state, PastLookahead::Advance) { - Outcome::Accept => ChainVerdict::Keep, - outcome => ChainVerdict::Drop(outcome), + match judge_before_kzg(store, sidecar, &parent_state, PastLookahead::Advance) { + Ok(()) => ChainPrecheck::Pending(PendingKzg { + commitments: sidecar.kzg_commitments.iter().cloned().collect(), + }), + Err(outcome) => ChainPrecheck::Settled(ChainVerdict::Drop(outcome)), } } @@ -184,9 +209,18 @@ pub fn chain_checks_for( sidecar: &AnyDataColumnSidecar, now_ms: u64, ) -> ChainVerdict { + settle(chain_precheck_for(store, sidecar, now_ms), &sidecar.into()) +} + +/// [`chain_precheck`] or [`chain_precheck_gloas`], by the sidecar's shape. +pub fn chain_precheck_for( + store: &Store, + sidecar: &AnyDataColumnSidecar, + now_ms: u64, +) -> ChainPrecheck { match sidecar { - AnyDataColumnSidecar::Fulu(sidecar) => chain_checks(store, sidecar, now_ms), - AnyDataColumnSidecar::Gloas(sidecar) => chain_checks_gloas(store, sidecar, now_ms), + AnyDataColumnSidecar::Fulu(sidecar) => chain_precheck(store, sidecar, now_ms), + AnyDataColumnSidecar::Gloas(sidecar) => chain_precheck_gloas(store, sidecar, now_ms), } } @@ -204,11 +238,28 @@ pub fn chain_checks_gloas( sidecar: &gloas::DataColumnSidecar, now_ms: u64, ) -> ChainVerdict { + settle( + chain_precheck_gloas(store, sidecar, now_ms), + &ColumnRef::Gloas(sidecar), + ) +} + +/// [`chain_precheck`] for a gloas sidecar: [`chain_checks_gloas`] up to, and +/// not including, the KZG proofs. The commitments the proofs are checked +/// against come out of the block's bid, so they travel in +/// [`ChainPrecheck::Pending`]. +pub fn chain_precheck_gloas( + store: &Store, + sidecar: &gloas::DataColumnSidecar, + now_ms: u64, +) -> ChainPrecheck { let config = store.config(); // [IGNORE] Not from a future slot. Also what bounds how far ahead a // parked row can name. if is_future_slot(&config, sidecar.slot, now_ms) { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::FutureSlot)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::FutureSlot, + ))); } // [IGNORE] From a slot greater than the latest finalized slot, unless it // is a column of the finalized block itself. Not a gloas rule: gloas has @@ -224,21 +275,146 @@ pub fn chain_checks_gloas( if sidecar.slot <= compute_start_slot_at_epoch(finalized.epoch) && sidecar.beacon_block_root != finalized.root { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::Finalized)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::Finalized, + ))); } // [IGNORE] Already stored, as in [`chain_checks`]. if store.has_data_column(sidecar.slot, &sidecar.beacon_block_root, sidecar.index) { - return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::AlreadyStored)); + return ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Ignore( + IgnoreReason::AlreadyStored, + ))); + } + match gloas_before_kzg(store, sidecar) { + Ok(commitments) => ChainPrecheck::Pending(PendingKzg { commitments }), + Err(Outcome::Queue(QueueReason::BlockUnknown | QueueReason::BlockNotReady)) => { + ChainPrecheck::Settled(ChainVerdict::AwaitParent) + } + Err(outcome) => ChainPrecheck::Settled(ChainVerdict::Drop(outcome)), } - match stateful_checks_gloas(store, sidecar) { - Outcome::Accept => ChainVerdict::Keep, - Outcome::Queue(QueueReason::BlockUnknown | QueueReason::BlockNotReady) => { - ChainVerdict::AwaitParent +} + +/// What [`chain_precheck`] found out without reading a cell. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ChainPrecheck { + /// The verdict, whatever the KZG proofs say: some rule before them + /// already decided. + Settled(ChainVerdict), + /// Every rule before the proofs passed: [`ChainVerdict::Keep`] if the + /// proofs verify, else a drop. Hand it to [`kzg_verdict`], or to + /// [`kzg_verdicts`] along with the other sidecars of the same block. + Pending(PendingKzg), +} + +/// What the KZG proofs of a sidecar that passed [`chain_precheck`] are +/// checked against. +/// +/// Fulu's sidecar carries its commitments, so this is a copy of them (a few +/// dozen bytes each, nothing next to the cells). Gloas's come from the bid of +/// the block the sidecar names, which only the precheck's store read could +/// supply. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingKzg { + commitments: Vec, +} + +/// A sidecar of either fork, borrowed: lets [`chain_checks`] and +/// [`chain_checks_gloas`] share [`settle`] with [`chain_checks_for`] without +/// wrapping (and so cloning) a sidecar into [`AnyDataColumnSidecar`]. +enum ColumnRef<'a> { + Fulu(&'a DataColumnSidecar), + Gloas(&'a gloas::DataColumnSidecar), +} + +impl<'a> From<&'a AnyDataColumnSidecar> for ColumnRef<'a> { + fn from(sidecar: &'a AnyDataColumnSidecar) -> Self { + match sidecar { + AnyDataColumnSidecar::Fulu(sidecar) => Self::Fulu(sidecar), + AnyDataColumnSidecar::Gloas(sidecar) => Self::Gloas(sidecar), + } + } +} + +impl<'a> ColumnRef<'a> { + /// The cells and proofs of the column, as [`fork_choice::ColumnCells`] + /// against `commitments`. + fn cells(&self, commitments: &'a [KzgCommitment]) -> fork_choice::ColumnCells<'a> { + match self { + Self::Fulu(sidecar) => fork_choice::ColumnCells { + index: sidecar.index, + column: &sidecar.column, + commitments, + proofs: &sidecar.kzg_proofs, + }, + Self::Gloas(sidecar) => fork_choice::ColumnCells { + index: sidecar.index, + column: &sidecar.column, + commitments, + proofs: &sidecar.kzg_proofs, + }, } - outcome => ChainVerdict::Drop(outcome), } } +/// A precheck's last step, for the single-sidecar callers. +fn settle(precheck: ChainPrecheck, sidecar: &ColumnRef<'_>) -> ChainVerdict { + match precheck { + ChainPrecheck::Settled(verdict) => verdict, + ChainPrecheck::Pending(pending) => kzg_verdict_ref(sidecar, &pending), + } +} + +/// Verify one pending sidecar's KZG proofs on their own. +pub fn kzg_verdict(sidecar: &AnyDataColumnSidecar, pending: &PendingKzg) -> ChainVerdict { + kzg_verdict_ref(&sidecar.into(), pending) +} + +fn kzg_verdict_ref(sidecar: &ColumnRef<'_>, pending: &PendingKzg) -> ChainVerdict { + let kzg = { + let _timing = metrics::time_data_column_kzg_verify(); + fork_choice::verify_column_cells_batch(&[sidecar.cells(&pending.commitments)]) + }; + // [REJECT] Every cell matches its commitment and proof. + if matches!(kzg, Ok(true)) { + ChainVerdict::Keep + } else { + ChainVerdict::Drop(Outcome::Reject(RejectReason::Kzg)) + } +} + +/// The KZG verdicts of several pending sidecars, one per entry and in order, +/// from one `verify_cell_kzg_proof_batch` call over all their cells. +/// +/// Every entry is [`ChainVerdict::Keep`] when the batch verifies. When it does +/// not, the batch cannot say which sidecar is at fault, so each is verified +/// again on its own: the verdicts are then exactly [`kzg_verdict`]'s, and a +/// forged column costs the rest of its block one extra pass rather than their +/// acceptance. Intended for the sidecars of one block, which is where the +/// cost is shared: the batch pays its fixed work once instead of once per +/// sidecar. +pub fn kzg_verdicts(items: &[(&AnyDataColumnSidecar, &PendingKzg)]) -> Vec { + if let [(sidecar, pending)] = items { + return vec![kzg_verdict(sidecar, pending)]; + } + let refs: Vec> = items.iter().map(|(sidecar, _)| (*sidecar).into()).collect(); + let columns: Vec> = refs + .iter() + .zip(items) + .map(|(sidecar, (_, pending))| sidecar.cells(&pending.commitments)) + .collect(); + let batch = { + let _timing = metrics::time_data_column_kzg_verify(); + fork_choice::verify_column_cells_batch(&columns) + }; + if matches!(batch, Ok(true)) { + return vec![ChainVerdict::Keep; items.len()]; + } + refs.iter() + .zip(items) + .map(|(sidecar, (_, pending))| kzg_verdict_ref(sidecar, pending)) + .collect() +} + /// What [`judge_against_parent`] does when the parent state's proposer /// lookahead has no answer for the sidecar's slot. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -265,17 +441,51 @@ fn judge_against_parent( parent_state: &BeaconState, past_lookahead: PastLookahead, ) -> Outcome { + if let Err(outcome) = judge_before_kzg(store, sidecar, parent_state, past_lookahead) { + return outcome; + } + let header = &sidecar.signed_block_header.message; + // [REJECT] Every cell matches its commitment and proof. + let kzg = { + let _timing = metrics::time_data_column_kzg_verify(); + fork_choice::verify_data_column_sidecar_kzg_proofs(sidecar) + }; + if !matches!(kzg, Ok(true)) { + return Outcome::Reject(RejectReason::Kzg); + } + // [IGNORE] The sidecar's slot is outside the parent state's proposer + // lookahead window (MAY queue). + if past_lookahead == PastLookahead::Queue + && let Err(outcome) = lookahead_covers(parent_state, header.slot) + { + return outcome; + } + Outcome::Accept +} + +/// Every rule [`judge_against_parent`] runs before the KZG proofs, in the same +/// order, so `Ok` means "accepted unless the proofs fail" and an `Err` is the +/// verdict whatever the proofs say. +/// +/// Split out so [`chain_precheck`] can stop here and let the caller verify the +/// proofs of every sidecar of one block in a single batch. The only rule after +/// the proofs is the lookahead one, which is [`PastLookahead::Queue`]'s alone, +/// so on the chain path (`Advance`) the proofs are the last word. +fn judge_before_kzg( + store: &Store, + sidecar: &DataColumnSidecar, + parent_state: &BeaconState, + past_lookahead: PastLookahead, +) -> Result<(), Outcome> { let config = store.config(); let header = &sidecar.signed_block_header.message; // [REJECT] The sidecar is from a higher slot than its parent. if header.slot <= parent_state.slot() { - return Outcome::Reject(RejectReason::NotAfterParent); + return Err(Outcome::Reject(RejectReason::NotAfterParent)); } // [REJECT] Signed by a validator the parent's state knows, over this // header. - if let Err(outcome) = header_signature_is_valid(sidecar, parent_state, &config) { - return outcome; - } + header_signature_is_valid(sidecar, parent_state, &config)?; // [REJECT] Proposed by the proposer the parent's state fixes for the // slot. Read from its lookahead when that covers the slot; past it, only // the chain path pays to find out. @@ -287,38 +497,21 @@ fn judge_against_parent( (None, PastLookahead::Queue) => None, (None, PastLookahead::Advance) => { let Some(expected) = advanced_proposer(parent_state, header.slot, &config) else { - return Outcome::Ignore(IgnoreReason::Internal); + return Err(Outcome::Ignore(IgnoreReason::Internal)); }; Some(expected) } }; if expected.is_some_and(|expected| header.proposer_index != expected) { - return Outcome::Reject(RejectReason::WrongProposer); + return Err(Outcome::Reject(RejectReason::WrongProposer)); } // [REJECT] The finalized checkpoint is an ancestor of the sidecar's block. - if let Err(outcome) = finalized_ancestry(store, header.parent_root).verdict() { - return outcome; - } + finalized_ancestry(store, header.parent_root).verdict()?; // [REJECT] The commitments are the ones the block committed to. if !fork_choice::verify_data_column_sidecar_inclusion_proof(sidecar) { - return Outcome::Reject(RejectReason::InclusionProof); + return Err(Outcome::Reject(RejectReason::InclusionProof)); } - // [REJECT] Every cell matches its commitment and proof. - let kzg = { - let _timing = metrics::time_data_column_kzg_verify(); - fork_choice::verify_data_column_sidecar_kzg_proofs(sidecar) - }; - if !matches!(kzg, Ok(true)) { - return Outcome::Reject(RejectReason::Kzg); - } - // [IGNORE] The sidecar's slot is outside the parent state's proposer - // lookahead window (MAY queue). - if past_lookahead == PastLookahead::Queue - && let Err(outcome) = lookahead_covers(parent_state, header.slot) - { - return outcome; - } - Outcome::Accept + Ok(()) } /// The proposer `parent_state`, advanced to `slot`, names for it: the answer @@ -425,13 +618,33 @@ pub fn cheap_checks_gloas( /// is queued too where the specification rejects it, the same deviation /// fulu's parent rule makes until a bad-block cache exists. pub fn stateful_checks_gloas(store: &Store, sidecar: &gloas::DataColumnSidecar) -> Outcome { - let block = match judgeable_block_gloas(store, &sidecar.beacon_block_root) { - Ok(block) => block, + let kzg_commitments = match gloas_before_kzg(store, sidecar) { + Ok(commitments) => commitments, Err(outcome) => return outcome, }; + // [REJECT] The sidecar's column data passes KZG verification. + let kzg = { + let _timing = metrics::time_data_column_kzg_verify(); + fork_choice::gloas_verify_data_column_sidecar_kzg_proofs(sidecar, &kzg_commitments) + }; + if !matches!(kzg, Ok(true)) { + return Outcome::Reject(RejectReason::Kzg); + } + Outcome::Accept +} + +/// Every rule [`stateful_checks_gloas`] runs before the KZG proofs, in the +/// same order. `Ok` carries the commitments the proofs are checked against +/// (the block's bid's), and means "accepted unless the proofs fail"; an `Err` +/// is the verdict whatever the proofs say. +fn gloas_before_kzg( + store: &Store, + sidecar: &gloas::DataColumnSidecar, +) -> Result, Outcome> { + let block = judgeable_block_gloas(store, &sidecar.beacon_block_root)?; // [REJECT] The sidecar's slot matches the slot of the block. if sidecar.slot != block.slot() { - return Outcome::Reject(RejectReason::SlotMismatch); + return Err(Outcome::Reject(RejectReason::SlotMismatch)); } let kzg_commitments = match &block { SignedBeaconBlock::Gloas(block) => { @@ -450,22 +663,14 @@ pub fn stateful_checks_gloas(store: &Store, sidecar: &gloas::DataColumnSidecar) | SignedBeaconBlock::Capella(_) | SignedBeaconBlock::Deneb(_) | SignedBeaconBlock::Electra(_) - | SignedBeaconBlock::Fulu(_) => return Outcome::Reject(RejectReason::BlockNotGloas), + | SignedBeaconBlock::Fulu(_) => return Err(Outcome::Reject(RejectReason::BlockNotGloas)), SignedBeaconBlock::Lean(_) => lean_block_unreachable("column::stateful_checks_gloas"), }; // [REJECT] The sidecar passes structural validation. if !fork_choice::gloas_verify_data_column_sidecar(sidecar, kzg_commitments) { - return Outcome::Reject(RejectReason::Malformed); - } - // [REJECT] The sidecar's column data passes KZG verification. - let kzg = { - let _timing = metrics::time_data_column_kzg_verify(); - fork_choice::gloas_verify_data_column_sidecar_kzg_proofs(sidecar, kzg_commitments) - }; - if !matches!(kzg, Ok(true)) { - return Outcome::Reject(RejectReason::Kzg); + return Err(Outcome::Reject(RejectReason::Malformed)); } - Outcome::Accept + Ok(kzg_commitments.to_vec()) } /// The block a gloas sidecar is judged against, or the verdict that says why @@ -1372,4 +1577,162 @@ mod tests { .expect("within the list bound"); assert!(!is_parkable_gloas(&config, &custody, &crowded)); } + + // -- KZG batching ------------------------------------------------------ + + /// A blob of `4096` canonical field elements derived from `seed`: the top + /// byte of each 32-byte element is zero, which keeps it below the modulus. + fn blob(seed: u8) -> Vec { + (0..c_kzg::BYTES_PER_BLOB) + .map(|i| { + if i % 32 == 0 { + 0 + } else { + (i as u8).wrapping_mul(31).wrapping_add(seed) + } + }) + .collect() + } + + /// `blobs` blobs' worth of real cells, commitments and proofs, as the + /// fulu sidecars for every column in `columns`, each with its + /// [`PendingKzg`]. Every column carries one cell per blob, as on the wire. + fn real_sidecars(blobs: usize, columns: &[u64]) -> Vec<(AnyDataColumnSidecar, PendingKzg)> { + let mut commitments = Vec::new(); + let mut cells = Vec::new(); + let mut proofs = Vec::new(); + for seed in 0..blobs { + let blob = blob(seed as u8); + commitments.push(crate::beacon::kzg::blob_to_kzg_commitment(&blob).expect("commit")); + let (blob_cells, blob_proofs) = + crate::beacon::kzg::compute_cells_and_kzg_proofs(&blob).expect("cells"); + cells.push(blob_cells); + proofs.push(blob_proofs); + } + columns + .iter() + .map(|&index| { + let column: Vec = cells + .iter() + .map(|blob_cells| { + libssz_types::SszVector::try_from( + blob_cells[index as usize].to_bytes().to_vec(), + ) + .expect("exact cell size") + }) + .collect(); + let column_proofs: Vec = proofs + .iter() + .map(|blob_proofs| blob_proofs[index as usize]) + .collect(); + let mut sidecar = sidecar(5, 0, index); + sidecar.column = column.try_into().expect("within the per-block limit"); + sidecar.kzg_proofs = column_proofs + .try_into() + .expect("within the per-block limit"); + sidecar.kzg_commitments = commitments + .clone() + .try_into() + .expect("within the per-block limit"); + let pending = PendingKzg { + commitments: commitments.clone(), + }; + (AnyDataColumnSidecar::Fulu(sidecar), pending) + }) + .collect() + } + + fn verdicts_of(items: &[(AnyDataColumnSidecar, PendingKzg)]) -> Vec { + let refs: Vec<_> = items.iter().map(|(s, p)| (s, p)).collect(); + kzg_verdicts(&refs) + } + + #[test] + fn a_batch_of_valid_columns_keeps_every_one() { + let items = real_sidecars(2, &[0, 7, 64]); + assert_eq!(verdicts_of(&items), vec![ChainVerdict::Keep; 3]); + } + + /// One forged cell fails the whole batch; the fallback must then give the + /// verdicts the sidecars would have had one at a time: the forged column + /// dropped as a KZG rejection, its neighbours kept. + #[test] + fn one_bad_column_in_a_batch_drops_only_that_column() { + let mut items = real_sidecars(2, &[0, 7, 64]); + let AnyDataColumnSidecar::Fulu(forged) = &mut items[1].0 else { + unreachable!("real_sidecars builds fulu sidecars"); + }; + let mut cell = forged.column[0].to_vec(); + cell[0] ^= 1; + forged.column[0] = libssz_types::SszVector::try_from(cell).expect("exact cell size"); + + let batched = verdicts_of(&items); + assert_eq!( + batched, + vec![ + ChainVerdict::Keep, + ChainVerdict::Drop(Outcome::Reject(RejectReason::Kzg)), + ChainVerdict::Keep, + ] + ); + let one_by_one: Vec<_> = items + .iter() + .map(|(sidecar, pending)| kzg_verdict(sidecar, pending)) + .collect(); + assert_eq!(batched, one_by_one); + } + + /// Proofs swapped between two columns: each proof is a valid proof, for + /// the other column's cell, so only a check of the pairing catches it. + #[test] + fn proofs_swapped_between_columns_are_both_rejected() { + let mut items = real_sidecars(1, &[3, 4]); + let (first, second) = items.split_at_mut(1); + let (AnyDataColumnSidecar::Fulu(a), AnyDataColumnSidecar::Fulu(b)) = + (&mut first[0].0, &mut second[0].0) + else { + unreachable!("real_sidecars builds fulu sidecars"); + }; + std::mem::swap(&mut a.kzg_proofs, &mut b.kzg_proofs); + assert_eq!( + verdicts_of(&items), + vec![ChainVerdict::Drop(Outcome::Reject(RejectReason::Kzg)); 2] + ); + } + + /// Prints the time of one batch over `k` columns against `k` separate + /// calls. Run with `--ignored --nocapture`. + #[test] + #[ignore = "benchmark: prints timings"] + fn bench_kzg_batch_against_separate_calls() { + use std::time::Instant; + const BLOBS: usize = 9; + let all: Vec = (0..32).collect(); + let items = real_sidecars(BLOBS, &all); + for k in [8usize, 32] { + let subset = &items[..k]; + let (mut batched, mut separate) = (Vec::new(), Vec::new()); + for _ in 0..5 { + let start = Instant::now(); + assert!( + verdicts_of(subset) + .iter() + .all(|verdict| *verdict == ChainVerdict::Keep) + ); + batched.push(start.elapsed()); + let start = Instant::now(); + for (sidecar, pending) in subset { + assert_eq!(kzg_verdict(sidecar, pending), ChainVerdict::Keep); + } + separate.push(start.elapsed()); + } + batched.sort(); + separate.sort(); + let (b, s) = (batched[2], separate[2]); + println!( + "k={k} blobs={BLOBS}: batch {b:?}, separate {s:?}, ratio {:.2}", + s.as_secs_f64() / b.as_secs_f64() + ); + } + } } diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..c9764de16 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -13,25 +13,142 @@ //! wait on imported (`BlockChainToP2P::check_data_column_sidecars`): a fulu //! sidecar's parent, a gloas sidecar's own block. //! -//! Each sidecar is checked on a blocking thread, bounded by +//! The sidecars are grouped by the block they belong to, and each group is +//! checked on one blocking thread, bounded by //! [`P2PServer::column_check_permits`]. Unlike a gossip verdict, nothing here -//! has a deadline, so a sidecar with no free permit waits for one rather than +//! has a deadline, so a group with no free permit waits for one rather than //! being dropped. +//! +//! A group takes one permit, not one per sidecar, because the KZG proofs of +//! all its sidecars are verified in one batch: the batch is one unit of work +//! on one thread, so a permit per sidecar would count permits the group does +//! not use at the same time. The pool still bounds the blocking threads this +//! module occupies, at one per block in flight. A group's verdicts are +//! forwarded as soon as it finishes, so the chain actor can count a block's +//! columns toward availability without waiting for the rest of a range batch. + +use std::collections::HashMap; +use std::panic::{AssertUnwindSafe, catch_unwind}; -use ethlambda_state_transition::beacon::gossip::column::{self, ChainVerdict}; +use ethlambda_network_api::P2PToBlockChainRef; +use ethlambda_state_transition::beacon::gossip::column::{ + self, ChainPrecheck, ChainVerdict, PendingKzg, +}; use ethlambda_state_transition::beacon::gossip::{IgnoreReason, Outcome}; +use ethlambda_storage::Store; use ethlambda_types::beacon::containers::DataColumnSidecar; +use ethlambda_types::beacon::primitives::{HashTreeRoot as _, Root}; use ethlambda_types::time::unix_now_ms; use tracing::{error, warn}; use crate::{P2PServer, metrics}; +/// What one block's sidecars came to: the ones to store, the ones to park, and +/// the reason for each one dropped. +#[derive(Default)] +struct GroupVerdicts { + keep: Vec, + awaiting_parent: Vec, + dropped: Vec, +} + +impl GroupVerdicts { + fn record(&mut self, sidecar: DataColumnSidecar, verdict: ChainVerdict) { + match verdict { + ChainVerdict::Keep => self.keep.push(sidecar), + ChainVerdict::AwaitParent => self.awaiting_parent.push(sidecar), + ChainVerdict::Drop(outcome) => self.dropped.push(outcome), + } + } +} + +/// The block a sidecar belongs to: a fulu sidecar's signed header's root, a +/// gloas sidecar's `beacon_block_root`. +fn block_of(sidecar: &DataColumnSidecar) -> Root { + match sidecar { + DataColumnSidecar::Fulu(sidecar) => sidecar.signed_block_header.message.hash_tree_root(), + DataColumnSidecar::Gloas(sidecar) => sidecar.beacon_block_root, + } +} + +/// Group `sidecars` by block, in the order each block first appears. +fn group_by_block(sidecars: Vec) -> Vec> { + let mut positions: HashMap = HashMap::new(); + let mut groups: Vec> = Vec::new(); + for sidecar in sidecars { + let position = *positions.entry(block_of(&sidecar)).or_insert_with(|| { + groups.push(Vec::new()); + groups.len() - 1 + }); + groups[position].push(sidecar); + } + groups +} + +/// Judge every sidecar of one block: each one's checks that do not read the +/// cells, then the KZG proofs of the ones still standing, in one batch. +/// +/// Each sidecar gets the verdict [`column::chain_checks_for`] would give it: +/// the proofs are the last rule on the chain path, so running the other rules +/// first and the proofs together changes nothing but how the proofs are paid +/// for, and a failed batch is re-run sidecar by sidecar. +fn judge_group(store: &Store, sidecars: Vec, now_ms: u64) -> GroupVerdicts { + let mut verdicts = GroupVerdicts::default(); + let mut pending: Vec<(DataColumnSidecar, PendingKzg)> = Vec::new(); + for sidecar in sidecars { + // A panic inside the checks (a `Store` read's `.expect()` on a DB + // error, say) costs that one sidecar, which is what dropping it + // would have done anyway. + let precheck = catch_unwind(AssertUnwindSafe(|| { + column::chain_precheck_for(store, &sidecar, now_ms) + })); + match precheck { + Ok(ChainPrecheck::Settled(verdict)) => verdicts.record(sidecar, verdict), + Ok(ChainPrecheck::Pending(kzg)) => pending.push((sidecar, kzg)), + Err(_) => error!("A data column check panicked; dropping the sidecar"), + } + } + if pending.is_empty() { + return verdicts; + } + let items: Vec<_> = pending + .iter() + .map(|(sidecar, kzg)| (sidecar, kzg)) + .collect(); + let kzg_verdicts = column::kzg_verdicts(&items); + for ((sidecar, _), verdict) in pending.into_iter().zip(kzg_verdicts) { + verdicts.record(sidecar, verdict); + } + verdicts +} + +/// Send the chain actor one group's results: the sidecars to store as one +/// batch, and the ones waiting on a block as another. +fn forward(blockchain: &P2PToBlockChainRef, verdicts: GroupVerdicts) { + for outcome in verdicts.dropped { + count_drop(outcome); + } + if !verdicts.keep.is_empty() { + let _ = blockchain + .new_data_column_sidecars(verdicts.keep) + .inspect_err(|err| warn!(%err, "Failed to forward checked data column sidecars")); + } + if !verdicts.awaiting_parent.is_empty() { + let _ = blockchain + .data_column_sidecars_awaiting_parent(verdicts.awaiting_parent) + .inspect_err( + |err| warn!(%err, "Failed to forward data column sidecars awaiting a parent"), + ); + } +} + /// Run the chain checks on `sidecars`, then send the chain actor the ones that -/// passed as one batch and the ones waiting on a block as another. +/// passed as one batch per block and the ones waiting on a block as another. /// -/// Returns at once: the checks run on a task of their own, so the p2p actor +/// Returns at once: the checks run on tasks of their own, so the p2p actor /// goes on handling swarm events while a range batch's KZG proofs are -/// verified. +/// verified. A block's sidecars are forwarded when that block's group +/// finishes, not when every group of the call has. pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec) { if sidecars.is_empty() { return; @@ -42,52 +159,28 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec checked, - Err(err) => { - error!(%err, "A data column check panicked; dropping the sidecar"); - continue; + let blockchain = blockchain.clone(); + tokio::spawn(async move { + let checked = tokio::task::spawn_blocking(move || { + let _permit = permit; + judge_group(&store, group, unix_now_ms()) + }) + .await; + match checked { + Ok(verdicts) => forward(&blockchain, verdicts), + // Tokio has already caught the panic; what the group + // held is lost, which is what dropping it would have + // done anyway. + Err(err) => error!(%err, "A data column check panicked; dropping the group"), } - }; - match verdict { - ChainVerdict::Keep => keep.push(sidecar), - ChainVerdict::AwaitParent => awaiting_parent.push(sidecar), - ChainVerdict::Drop(outcome) => count_drop(outcome), - } - } - - if !keep.is_empty() { - let _ = blockchain - .new_data_column_sidecars(keep) - .inspect_err(|err| warn!(%err, "Failed to forward checked data column sidecars")); - } - if !awaiting_parent.is_empty() { - let _ = blockchain - .data_column_sidecars_awaiting_parent(awaiting_parent) - .inspect_err( - |err| warn!(%err, "Failed to forward data column sidecars awaiting a parent"), - ); + }); } }); } @@ -257,4 +350,73 @@ mod tests { drop(server); assert_eq!(received.recv().await, None); } + + /// A call mixing blocks forwards each block's sidecars on their own: the + /// chain actor hears about a block as soon as that block's group is done, + /// and a block's sidecars arrive together. + #[tokio::test] + async fn each_block_of_a_mixed_call_is_forwarded_on_its_own() { + let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; + let (sender, mut received) = mpsc::unbounded_channel(); + server.blockchain = Some(Arc::new(RecordingChain(sender))); + + // Two columns of one block (slot 5) and one of another (slot 6), all + // with a parent the store has never seen, interleaved. A third block + // (slot 7) is malformed and goes nowhere. + let a0 = DataColumnSidecar::Fulu(valid_shaped_sidecar(5, 0)); + let b0 = DataColumnSidecar::Fulu(valid_shaped_sidecar(6, 0)); + let a1 = DataColumnSidecar::Fulu(valid_shaped_sidecar(5, 1)); + let mut malformed = valid_shaped_sidecar(7, 0); + malformed.kzg_commitments = Default::default(); + + check_and_forward( + &server, + vec![ + a0.clone(), + b0.clone(), + a1.clone(), + DataColumnSidecar::Fulu(malformed), + ], + ); + + let mut batches = Vec::new(); + for _ in 0..2 { + let Some(Forwarded::AwaitingParent(batch)) = received.recv().await else { + panic!("expected a batch awaiting its parent"); + }; + batches.push(batch); + } + batches.sort_by_key(|batch| batch.len()); + assert_eq!(batches, vec![vec![b0], vec![a0, a1]]); + drop(server); + assert_eq!(received.recv().await, None); + } + + #[test] + fn sidecars_are_grouped_by_block_in_first_seen_order() { + let a0 = DataColumnSidecar::Fulu(valid_shaped_sidecar(5, 0)); + let b0 = DataColumnSidecar::Fulu(valid_shaped_sidecar(6, 0)); + let a1 = DataColumnSidecar::Fulu(valid_shaped_sidecar(5, 1)); + let gloas_of = |byte| { + DataColumnSidecar::Gloas(gloas::DataColumnSidecar { + beacon_block_root: Root::repeat_byte(byte), + ..Default::default() + }) + }; + let (g1, g2, g1_again) = (gloas_of(1), gloas_of(2), gloas_of(1)); + + let groups = group_by_block(vec![ + a0.clone(), + b0.clone(), + a1.clone(), + g1.clone(), + g2.clone(), + g1_again.clone(), + ]); + + assert_eq!( + groups, + vec![vec![a0, a1], vec![b0], vec![g1, g1_again], vec![g2]] + ); + } } diff --git a/docs/beacon_wire.md b/docs/beacon_wire.md index c2138785b..b1fb18f57 100644 --- a/docs/beacon_wire.md +++ b/docs/beacon_wire.md @@ -638,6 +638,21 @@ NUMBER_OF_COLUMNS`) bounds what one request may ask for; an answer over that is truncated, not refused, matching how the block protocols treat a peer built to an older, wider ceiling. +*Checking* what arrives over req/resp, and what gossip left unjudged, is +`beacon/column_checks.rs`. The sidecars of one call are grouped by block (a +fulu sidecar's signed header root, a gloas sidecar's `beacon_block_root`) and +each group is judged on one blocking thread holding one `column_check_permits` +permit. Per sidecar the rules before the KZG proofs run first, in the order +`chain_checks` has always run them (`chain_precheck`); the proofs of every +sidecar still standing then go through one `verify_cell_kzg_proof_batch` call +(`kzg_verdicts`), and the proofs are the last rule on this path, so each +sidecar's verdict is the one it would have had alone. A batch that fails says +nothing about which sidecar is at fault, so each is verified again by itself +and only the ones that fail are rejected. A group is forwarded to the chain +actor when it finishes, not when the whole call has: the sidecars to store as +one batch, the ones waiting on a block as another. Gossip keeps one message, +one verdict, and is not batched. + *Asking* happens on two paths, bulk and per block. The bulk one rides with range sync: every `BeaconBlocksByRange` batch sends a diff --git a/docs/metrics.md b/docs/metrics.md index 1ea07953f..a740958a6 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -586,7 +586,7 @@ spec. |------|------|-------|-------------------------|--------|---------| | `lean_data_columns_stored_total` | Counter | Data column sidecars verified and written to `Table::DataColumns` | On the chain actor storing a sidecar the p2p layer verified | | | | `lean_data_columns_rejected_total` | Counter | Sidecars the p2p layer's chain checks dropped, by reason | On each drop in `beacon::column_checks`, except an already stored sidecar | reason=malformed,future_slot,finalized,not_after_parent,unknown_proposer,bad_signature,wrong_proposer,finalized_not_ancestor,parent_not_ready,inclusion_proof,kzg,internal | | -| `lean_data_column_kzg_verify_seconds` | Histogram | Time spent batch-verifying one sidecar's cells against its own commitments | On each KZG batch verification, in gossip validation or the chain checks | | 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0 | +| `lean_data_column_kzg_verify_seconds` | Histogram | Time spent in one KZG cell-proof batch verification: one sidecar's cells in gossip validation, every still-standing sidecar of one block's group in the chain checks (see [Data column sidecars](beacon_wire.md#data-column-sidecars)), plus one observation per sidecar if that batch fails and each is re-verified alone | On each KZG batch verification, in gossip validation or the chain checks | | 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0 | | `lean_data_column_fetch_failures_total` | Counter | `DataColumnsByRoot` lookups this node gave up on, by reason | On lookup abandonment | reason=no_peers,max_retries | | | `lean_blocks_held_for_columns` | Gauge | Blocks held out of fork choice pending their custody columns | On every hold, release, and finality eviction of the held-block set | | | | `lean_sidecars_awaiting_parent` | Gauge | Sidecars parked until their block's parent has a post-state | On every park, replay, and finality eviction of the parked set | | | From abe310a223bc0ea2b04991eb46f3b761310db606 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:27:43 -0300 Subject: [PATCH 2/2] perf(p2p): verify a block group's shared header signature once Every fulu sidecar of a block carries the same signed header, so the group added in the previous commit repeated one BLS verification per sidecar, now one after another under a single permit. SignatureMemo remembers header_signature_is_valid's answer for the group, keyed on the header root, the signature bytes and (for the head-state check made before parking) the head it read, so a forged signature over a shared header is judged on its own. Only that answer is remembered; every other rule runs per sidecar in its order, so verdicts are unchanged. The memo lives in one group's spawn_blocking closure; gossip and the single-sidecar wrappers use a fresh one per call. --- .../src/beacon/gossip/column.rs | 175 ++++++++++++++++-- crates/net/p2p/src/beacon/column_checks.rs | 6 +- 2 files changed, 165 insertions(+), 16 deletions(-) diff --git a/crates/blockchain/state_transition/src/beacon/gossip/column.rs b/crates/blockchain/state_transition/src/beacon/gossip/column.rs index 61ffa3f4c..74630bd89 100644 --- a/crates/blockchain/state_transition/src/beacon/gossip/column.rs +++ b/crates/blockchain/state_transition/src/beacon/gossip/column.rs @@ -10,6 +10,8 @@ //! actor stores whatever reaches it without checking it //! again, so these are the only checks such a sidecar gets. +use std::collections::HashMap; + use ethlambda_storage::CacheKey; use super::{ @@ -34,7 +36,7 @@ use crate::beacon::lean_boundary::lean_block_unreachable; use crate::beacon::precheck; use crate::beacon::preset; use crate::beacon::primitives::HashTreeRoot as _; -use crate::beacon::primitives::{KzgCommitment, Root, Slot, ValidatorIndex}; +use crate::beacon::primitives::{BlsSignature, KzgCommitment, Root, Slot, ValidatorIndex}; use crate::beacon::stf; use crate::metrics; @@ -85,6 +87,7 @@ pub fn cheap_checks( /// while they run. pub fn stateful_checks(store: &Store, sidecar: &DataColumnSidecar) -> Outcome { let header = &sidecar.signed_block_header.message; + let mut memo = SignatureMemo::default(); let parent_known = store.has_block(&header.parent_root); let parent_state = if parent_known { store.cached_state(CacheKey::BlockState(header.parent_root)) @@ -101,9 +104,15 @@ pub fn stateful_checks(store: &Store, sidecar: &DataColumnSidecar) -> Outcome { } else { QueueReason::ParentUnknown }; - return queue_unless_forged(store, sidecar, reason); + return queue_unless_forged(store, sidecar, reason, &mut memo); }; - judge_against_parent(store, sidecar, &parent_state, PastLookahead::Queue) + judge_against_parent( + store, + sidecar, + &parent_state, + PastLookahead::Queue, + &mut memo, + ) } /// What the chain may do with a sidecar [`chain_checks`] judged. @@ -151,6 +160,18 @@ pub fn chain_checks(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> /// is final whatever the proofs say, and [`ChainPrecheck::Pending`] means the /// sidecar is [`ChainVerdict::Keep`] exactly when its proofs verify. pub fn chain_precheck(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> ChainPrecheck { + chain_precheck_with(store, sidecar, now_ms, &mut SignatureMemo::default()) +} + +/// [`chain_precheck`] sharing `memo` with the other sidecars of the same +/// block, so their identical header signature is verified once. See +/// [`SignatureMemo`] for what is and is not remembered. +pub fn chain_precheck_with( + store: &Store, + sidecar: &DataColumnSidecar, + now_ms: u64, + memo: &mut SignatureMemo, +) -> ChainPrecheck { let config = store.config(); // [REJECT] The sidecar is structurally valid. First, so a peer answering // a fetch with garbage cannot make this pay for a state read or a KZG @@ -189,13 +210,13 @@ pub fn chain_precheck(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) - // signature check keeps a forged header from taking a // `PendingDataColumns` row while it waits. return ChainPrecheck::Settled( - match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown) { + match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown, memo) { Outcome::Queue(_) => ChainVerdict::AwaitParent, outcome => ChainVerdict::Drop(outcome), }, ); }; - match judge_before_kzg(store, sidecar, &parent_state, PastLookahead::Advance) { + match judge_before_kzg(store, sidecar, &parent_state, PastLookahead::Advance, memo) { Ok(()) => ChainPrecheck::Pending(PendingKzg { commitments: sidecar.kzg_commitments.iter().cloned().collect(), }), @@ -217,9 +238,21 @@ pub fn chain_precheck_for( store: &Store, sidecar: &AnyDataColumnSidecar, now_ms: u64, +) -> ChainPrecheck { + chain_precheck_for_with(store, sidecar, now_ms, &mut SignatureMemo::default()) +} + +/// [`chain_precheck_for`] sharing `memo` across the sidecars of one block. +/// Only a fulu sidecar has a header signature to remember; a gloas sidecar +/// carries none, so `memo` goes unused for it. +pub fn chain_precheck_for_with( + store: &Store, + sidecar: &AnyDataColumnSidecar, + now_ms: u64, + memo: &mut SignatureMemo, ) -> ChainPrecheck { match sidecar { - AnyDataColumnSidecar::Fulu(sidecar) => chain_precheck(store, sidecar, now_ms), + AnyDataColumnSidecar::Fulu(sidecar) => chain_precheck_with(store, sidecar, now_ms, memo), AnyDataColumnSidecar::Gloas(sidecar) => chain_precheck_gloas(store, sidecar, now_ms), } } @@ -440,8 +473,9 @@ fn judge_against_parent( sidecar: &DataColumnSidecar, parent_state: &BeaconState, past_lookahead: PastLookahead, + memo: &mut SignatureMemo, ) -> Outcome { - if let Err(outcome) = judge_before_kzg(store, sidecar, parent_state, past_lookahead) { + if let Err(outcome) = judge_before_kzg(store, sidecar, parent_state, past_lookahead, memo) { return outcome; } let header = &sidecar.signed_block_header.message; @@ -476,6 +510,7 @@ fn judge_before_kzg( sidecar: &DataColumnSidecar, parent_state: &BeaconState, past_lookahead: PastLookahead, + memo: &mut SignatureMemo, ) -> Result<(), Outcome> { let config = store.config(); let header = &sidecar.signed_block_header.message; @@ -485,7 +520,7 @@ fn judge_before_kzg( } // [REJECT] Signed by a validator the parent's state knows, over this // header. - header_signature_is_valid(sidecar, parent_state, &config)?; + header_signature_is_valid(sidecar, parent_state, &config, memo, None)?; // [REJECT] Proposed by the proposer the parent's state fixes for the // slot. Read from its lookahead when that covers the slot; past it, only // the chain path pays to find out. @@ -514,6 +549,34 @@ fn judge_before_kzg( Ok(()) } +/// What [`header_signature_is_valid`] answered for the sidecars of one block, +/// so the BLS verification of a header the block's sidecars all carry runs +/// once rather than once per sidecar. +/// +/// Keyed on the header's root, the signature bytes and, for the head-state +/// check [`queue_unless_forged`] makes, the head it read: two sidecars with +/// one header and different signatures are two entries, so a forged signature +/// is judged on its own, and the same header judged against a different state +/// is a different question. Only that function's answer is remembered; every +/// other rule runs for every sidecar in its usual order. Built fresh for each +/// group of one `spawn_blocking` closure and dropped with it, so nothing +/// outlives a block's checks. +#[derive(Default)] +pub struct SignatureMemo { + answers: HashMap<(Root, BlsSignature, Option), Result<(), Outcome>>, + /// How many verifications actually ran: what a test reads to see that the + /// memo is used. + verified: usize, +} + +impl SignatureMemo { + /// The number of header signatures verified, as opposed to answered from + /// the memo. + pub fn verified(&self) -> usize { + self.verified + } +} + /// The proposer `parent_state`, advanced to `slot`, names for it: the answer /// [`precheck::fixed_proposer`] has no window for. `None` when the state /// cannot be advanced or cannot name one. @@ -542,15 +605,18 @@ fn advanced_proposer( /// the *expected* one for the slot. A column queued here is written to the /// chain actor's `PendingDataColumns`, so without this check a forged one /// would sit there rather than being refused up front. -fn queue_unless_forged(store: &Store, sidecar: &DataColumnSidecar, reason: QueueReason) -> Outcome { - let head_state = store - .head() - .ok() - .and_then(|head| store.cached_state(CacheKey::BlockState(head))); +fn queue_unless_forged( + store: &Store, + sidecar: &DataColumnSidecar, + reason: QueueReason, + memo: &mut SignatureMemo, +) -> Outcome { + let head = store.head().ok(); + let head_state = head.and_then(|head| store.cached_state(CacheKey::BlockState(head))); let Some(head_state) = head_state else { return Outcome::Queue(reason); }; - match header_signature_is_valid(sidecar, &head_state, &store.config()) { + match header_signature_is_valid(sidecar, &head_state, &store.config(), memo, head) { Err(Outcome::Reject(RejectReason::BadSignature)) => { Outcome::Reject(RejectReason::BadSignature) } @@ -775,6 +841,30 @@ fn header_signature_is_valid( sidecar: &DataColumnSidecar, state: &BeaconState, config: &Config, + memo: &mut SignatureMemo, + head: Option, +) -> Result<(), Outcome> { + let header = &sidecar.signed_block_header.message; + let key = ( + header.hash_tree_root(), + sidecar.signed_block_header.signature, + head, + ); + if let Some(answer) = memo.answers.get(&key) { + return *answer; + } + let answer = verify_header_signature(sidecar, state, config); + memo.verified += 1; + memo.answers.insert(key, answer); + answer +} + +/// The work behind [`header_signature_is_valid`], run once per distinct +/// answer. +fn verify_header_signature( + sidecar: &DataColumnSidecar, + state: &BeaconState, + config: &Config, ) -> Result<(), Outcome> { let header = &sidecar.signed_block_header.message; let Ok(proposer) = state.validator(header.proposer_index) else { @@ -1314,6 +1404,63 @@ mod tests { ); } + /// The sidecars of one block share a header, so a group's checks verify + /// its signature once; a sidecar carrying a different signature over the + /// same header is its own entry and fails on its own, with the verdict it + /// has when checked alone. + #[test] + fn a_groups_shared_header_signature_is_verified_once() { + let parent = fulu_parent(3); + let parent_root = Root::repeat_byte(0x99); + let store = store_with_parent(&parent, parent_root); + let config = store.config(); + let slot = parent.slot() + 1; + let now = slot_start_ms(&store, slot); + // Signed by the validator it names, which is not the slot's proposer: + // the signature passes and `WrongProposer` follows. + let good = signed_child(&parent, parent_root, slot, 5, 5, &config); + let mut forged = signed_child(&parent, parent_root, slot, 5, 6, &config); + forged.index = 3; + let columns: Vec = (0..3) + .map(|index| DataColumnSidecar { + index, + ..good.clone() + }) + .collect(); + + let mut memo = SignatureMemo::default(); + for sidecar in &columns { + assert_eq!( + chain_precheck_with(&store, sidecar, now, &mut memo), + ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Reject( + RejectReason::WrongProposer + ))) + ); + } + assert_eq!(memo.verified(), 1); + + assert_eq!( + chain_precheck_with(&store, &forged, now, &mut memo), + ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Reject( + RejectReason::BadSignature + ))) + ); + assert_eq!(memo.verified(), 2); + // The forged one's answer is what it is alone, and does not leak onto + // the sidecars that carry the good signature. + assert_eq!( + chain_checks(&store, &forged, now), + ChainVerdict::Drop(Outcome::Reject(RejectReason::BadSignature)) + ); + assert_eq!( + chain_precheck_with(&store, &columns[0], now, &mut memo), + ChainPrecheck::Settled(ChainVerdict::Drop(Outcome::Reject( + RejectReason::WrongProposer + ))) + ); + assert_eq!(memo.verified(), 2); + } + /// Where gossip queues, `chain_checks` drops: the parent already has a /// post-state, so no import will ever let the walk finish, and parking the /// sidecar would only wait for an import that has already happened. diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index c9764de16..c75f4dbc8 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -32,7 +32,7 @@ use std::panic::{AssertUnwindSafe, catch_unwind}; use ethlambda_network_api::P2PToBlockChainRef; use ethlambda_state_transition::beacon::gossip::column::{ - self, ChainPrecheck, ChainVerdict, PendingKzg, + self, ChainPrecheck, ChainVerdict, PendingKzg, SignatureMemo, }; use ethlambda_state_transition::beacon::gossip::{IgnoreReason, Outcome}; use ethlambda_storage::Store; @@ -95,12 +95,14 @@ fn group_by_block(sidecars: Vec) -> Vec, now_ms: u64) -> GroupVerdicts { let mut verdicts = GroupVerdicts::default(); let mut pending: Vec<(DataColumnSidecar, PendingKzg)> = Vec::new(); + // The block's sidecars share a header, so its signature is verified once. + let mut memo = SignatureMemo::default(); for sidecar in sidecars { // A panic inside the checks (a `Store` read's `.expect()` on a DB // error, say) costs that one sidecar, which is what dropping it // would have done anyway. let precheck = catch_unwind(AssertUnwindSafe(|| { - column::chain_precheck_for(store, &sidecar, now_ms) + column::chain_precheck_for_with(store, &sidecar, now_ms, &mut memo) })); match precheck { Ok(ChainPrecheck::Settled(verdict)) => verdicts.record(sidecar, verdict),