From 02ee4686ed7479d05bb96c46f02473b405afca2f Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Mon, 6 Jul 2026 12:32:57 +0200 Subject: [PATCH 1/2] perf: reduce hot-path allocation, cloning, and lock contention Implements the eight hot-path optimizations from the "hot-path allocation, cloning, and lock-contention reduction" plan. All changes are perf-only and preserve functional equivalence with Charon v1.7.1. - T01 p2p: add alloc-free `PeerStore::has_connection`; migrate the five emptiness-only `connections_to_peer(..).is_empty()` call sites. - T02 qbft: thread one `Arc` across peers on broadcast instead of an N-1 deep clone; move `msg`/`justification`/`values` out of `pb_msg` on receive; `values_by_hash` consumes its input. - T03 eth2api: `Arc`-wrap the cached validator maps, add a read-lock fast path, and release the write lock across the beacon-node fetch (re-checking on re-acquire). - T04 scheduler: replace the O(N) `to_entries()` status-metric scan with a per-pubkey last-status map, zeroing only the previously-reported series. - T05 scheduler: cache `slots_per_epoch` on the actor at build time instead of a beacon-node round-trip on every duty lookup. - T06 parsigex: acquire the peer-store guard once per broadcast and share the encoded payload via `Arc<[u8]>` instead of a per-target `Vec` copy. - T07 parsigdb: fold the threshold check into `store` under the lock and return only the matched set, dropping the per-store full-accumulator clone. - T08 ssz: pass the buffer by value into `merkleize_impl` (one copy, not two); reuse one `Sha256` in `default_hash_fn`; drop the `put_bitlist` clone. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/monitoringapi/checker.rs | 2 +- crates/consensus/src/qbft/component.rs | 32 ++-- crates/consensus/src/qbft/p2p.rs | 44 +++-- crates/consensus/src/qbft/qbft_run_test.rs | 2 +- crates/core/src/parsigdb/memory.rs | 67 ++++--- .../core/src/parsigdb/memory_internal_test.rs | 82 ++++++++- crates/core/src/scheduler.rs | 166 +++++++++++++++--- crates/dkg/src/bcast/behaviour.rs | 6 +- crates/dkg/src/frostp2p/behaviour.rs | 6 +- crates/eth2api/src/valcache.rs | 126 ++++++++++--- crates/p2p/src/p2p_context.rs | 77 ++++++++ crates/parsigex/src/behaviour.rs | 128 +++++++++++++- crates/parsigex/src/handler.rs | 12 +- crates/ssz/src/hasher.rs | 130 +++++++++++++- 14 files changed, 760 insertions(+), 120 deletions(-) diff --git a/crates/app/src/monitoringapi/checker.rs b/crates/app/src/monitoringapi/checker.rs index 2f90994a..d44500a1 100644 --- a/crates/app/src/monitoringapi/checker.rs +++ b/crates/app/src/monitoringapi/checker.rs @@ -307,7 +307,7 @@ pub fn quorum_peers_connected(p2p_context: &P2PContext) -> bool { let connected = known_peers .iter() .filter(|peer_id| **peer_id != local_peer_id) - .filter(|peer_id| !peer_store.connections_to_peer(peer_id).is_empty()) + .filter(|peer_id| peer_store.has_connection(peer_id)) .count(); connected >= required diff --git a/crates/consensus/src/qbft/component.rs b/crates/consensus/src/qbft/component.rs index f3638186..37c8d8bf 100644 --- a/crates/consensus/src/qbft/component.rs +++ b/crates/consensus/src/qbft/component.rs @@ -186,16 +186,19 @@ pub enum Error { } /// Canonicalizes inbound `Any` values into the hash map used by QBFT messages. -pub(crate) fn values_by_hash(values: &[Any]) -> Result { +/// +/// Consumes `values` and moves each entry into the map instead of cloning, +/// since the caller owns the values and discards the source vector afterwards. +pub(crate) fn values_by_hash(values: Vec) -> Result { let mut out = ValueMap::new(); for value in values { - let decoded = decode_supported_any(value)?; + let decoded = decode_supported_any(&value)?; let hash = match decoded { DecodedValue::UnsignedDataSet(inner) => msg::hash_proto(&inner)?, DecodedValue::PriorityResult(inner) => msg::hash_proto(&inner)?, }; - out.insert(hash, value.clone()); + out.insert(hash, value); } Ok(out) @@ -343,13 +346,14 @@ impl Consensus { /// Validates, wraps, and queues an inbound QBFT consensus message. pub async fn handle( &self, - pb_msg: pbconsensus::QbftConsensusMsg, + mut pb_msg: pbconsensus::QbftConsensusMsg, ct: &CancellationToken, ) -> Result<()> { - let msg = pb_msg.msg.as_ref().ok_or(Error::InvalidConsensusMessage)?; + // Verify the inner message and derive its duty (borrow only). + let inner = pb_msg.msg.as_ref().ok_or(Error::InvalidConsensusMessage)?; - self.verify_msg(msg)?; - let duty = duty_from_msg(msg)?; + self.verify_msg(inner)?; + let duty = duty_from_msg(inner)?; if !(self.duty_gater)(&duty) { return Err(Error::InvalidDuty); @@ -366,8 +370,12 @@ impl Consensus { } } - let values = values_by_hash(&pb_msg.values)?; - let wrapped = msg::Msg::new(msg.clone(), pb_msg.justification.clone(), Arc::new(values))?; + // Move the validated parts out of `pb_msg` (it is dropped after this), + // avoiding a clone of the message, justifications, and values. + let msg = pb_msg.msg.take().ok_or(Error::InvalidConsensusMessage)?; + let justification = std::mem::take(&mut pb_msg.justification); + let values = values_by_hash(std::mem::take(&mut pb_msg.values))?; + let wrapped = msg::Msg::new(msg, justification, Arc::new(values))?; if ct.is_cancelled() { return Err(Error::ReceiveCancelledDuringVerification); @@ -932,7 +940,7 @@ pub(crate) mod tests { #[test] fn values_by_hash_rejects_invalid_type_url() { - let err = values_by_hash(&[Any { + let err = values_by_hash(vec![Any { type_url: "type.googleapis.com/unknown.Type".to_string(), value: vec![], }]) @@ -943,7 +951,7 @@ pub(crate) mod tests { #[test] fn values_by_hash_rejects_malformed_any_value() { - let err = values_by_hash(&[Any { + let err = values_by_hash(vec![Any { type_url: pbcore::UnsignedDataSet::type_url(), value: b"not-protobuf".to_vec(), }]) @@ -955,7 +963,7 @@ pub(crate) mod tests { #[test] fn values_by_hash_hashes_decoded_inner_message() { let any = unsigned_any("a", b"first"); - let values = values_by_hash(std::slice::from_ref(&any)).unwrap(); + let values = values_by_hash(vec![any.clone()]).unwrap(); let decoded = pbcore::UnsignedDataSet::decode(any.value.as_slice()).unwrap(); let hash = msg::hash_proto(&decoded).unwrap(); diff --git a/crates/consensus/src/qbft/p2p.rs b/crates/consensus/src/qbft/p2p.rs index f8175f09..7c5141b0 100644 --- a/crates/consensus/src/qbft/p2p.rs +++ b/crates/consensus/src/qbft/p2p.rs @@ -183,7 +183,13 @@ impl Handle { pub async fn broadcast(&self, msg: pbconsensus::QbftConsensusMsg) -> BroadcastResult { let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); self.cmd_tx - .send(BroadcastCommand { request_id, msg }) + .send(BroadcastCommand { + request_id, + // Wrap once so the fan-out shares a single allocation across all + // peers rather than deep-cloning the (potentially multi-MB) + // payload per target, matching Charon's shared-pointer broadcast. + msg: Arc::new(msg), + }) .map_err(|_| Box::new(Error::BehaviourClosed) as _) } @@ -200,7 +206,7 @@ impl Handle { #[derive(Debug)] struct BroadcastCommand { request_id: u64, - msg: pbconsensus::QbftConsensusMsg, + msg: Arc, } #[doc(hidden)] @@ -208,7 +214,7 @@ struct BroadcastCommand { pub enum ToHandler { Send { request_id: u64, - msg: pbconsensus::QbftConsensusMsg, + msg: Arc, }, } @@ -227,7 +233,7 @@ type ActiveFuture = futures::future::BoxFuture<'static, Option>; pub struct Handler { consensus: Arc, cancellation: CancellationToken, - pending_open: VecDeque<(u64, pbconsensus::QbftConsensusMsg)>, + pending_open: VecDeque<(u64, Arc)>, active_futures: futures::stream::FuturesUnordered, } @@ -272,7 +278,7 @@ impl Handler { &mut self, mut stream: Stream, request_id: u64, - msg: pbconsensus::QbftConsensusMsg, + msg: Arc, ) { stream.ignore_for_keep_alive(); self.active_futures.push( @@ -308,7 +314,7 @@ impl ConnectionHandler for Handler { type FromBehaviour = ToHandler; type InboundOpenInfo = (); type InboundProtocol = ReadyUpgrade; - type OutboundOpenInfo = (u64, pbconsensus::QbftConsensusMsg); + type OutboundOpenInfo = (u64, Arc); type OutboundProtocol = ReadyUpgrade; type ToBehaviour = FromHandler; @@ -455,7 +461,7 @@ where #[derive(Debug)] struct PendingSend { request_id: u64, - msg: pbconsensus::QbftConsensusMsg, + msg: Arc, } /// libp2p behaviour for QBFT consensus messages. @@ -506,12 +512,10 @@ impl Behaviour { /// Returns whether the peer store has any live connection for the peer. fn is_connected(&self, peer_id: &PeerId) -> bool { - !self - .config + self.config .p2p_context .peer_store_lock() - .connections_to_peer(peer_id) - .is_empty() + .has_connection(peer_id) } /// Drains outbound broadcast commands queued through the public handle. @@ -539,7 +543,8 @@ impl Behaviour { peer_id, PendingSend { request_id: command.request_id, - msg: command.msg.clone(), + // Cheap refcount bump; the payload buffer is shared. + msg: Arc::clone(&command.msg), }, ); } @@ -945,6 +950,21 @@ mod tests { assert!(targets.contains(&peer_ids[0])); assert!(targets.contains(&peer_ids[2])); assert!(!targets.contains(&local_peer_id)); + + // The fan-out must share one `Arc` across all targets + // rather than deep-cloning the payload per peer. + let payloads = events + .iter() + .filter_map(|event| match event { + ToSwarm::NotifyHandler { + event: Either::Left(ToHandler::Send { msg, .. }), + .. + } => Some(msg.clone()), + _ => None, + }) + .collect::>(); + assert_eq!(payloads.len(), 2); + assert!(Arc::ptr_eq(&payloads[0], &payloads[1])); Ok(()) } diff --git a/crates/consensus/src/qbft/qbft_run_test.rs b/crates/consensus/src/qbft/qbft_run_test.rs index a7d424aa..8ab088b9 100644 --- a/crates/consensus/src/qbft/qbft_run_test.rs +++ b/crates/consensus/src/qbft/qbft_run_test.rs @@ -422,7 +422,7 @@ async fn replay_sniffed_instance_decides(instance: pbconsensus::SniffedConsensus for sniffed in instance.msgs { let outer = sniffed.msg.expect("sniffed entry has outer message"); let raw = outer.msg.expect("sniffed outer message has inner message"); - let values = component::values_by_hash(&outer.values).expect("sniffed values decode"); + let values = component::values_by_hash(outer.values).expect("sniffed values decode"); let wrapped = msg::Msg::new(raw, outer.justification, Arc::new(values)) .expect("sniffed message wraps"); let wrapped: qbft::Msg = Arc::new(wrapped); diff --git a/crates/core/src/parsigdb/memory.rs b/crates/core/src/parsigdb/memory.rs index 36051332..398b6037 100644 --- a/crates/core/src/parsigdb/memory.rs +++ b/crates/core/src/parsigdb/memory.rs @@ -258,7 +258,7 @@ impl MemDB { let mut output: HashMap> = HashMap::new(); for (pub_key, par_signed) in signed_data.inner().iter() { - let sigs = self + let outcome = self .store( Key { duty: duty.clone(), @@ -268,19 +268,16 @@ impl MemDB { ) .await?; - let Some(sigs) = sigs else { - debug!("Ignoring duplicate partial signature"); - - continue; - }; - - let psigs = get_threshold_matching(&duty.duty_type, &sigs, self.threshold).await?; - - let Some(psigs) = psigs else { - continue; - }; - - output.insert(*pub_key, psigs); + match outcome { + StoreOutcome::Duplicate => { + debug!("Ignoring duplicate partial signature"); + continue; + } + StoreOutcome::Stored => continue, + StoreOutcome::Threshold(psigs) => { + output.insert(*pub_key, psigs); + } + } } if output.is_empty() { @@ -330,7 +327,20 @@ impl MemDB { } } - async fn store(&self, k: Key, value: ParSignedData) -> Result>> { + /// Stores `value` under key `k`, then evaluates the threshold under the + /// same lock against the borrowed accumulator so no full clone of the + /// growing set is produced per store. + /// + /// Returns: + /// - [`StoreOutcome::Duplicate`] if `value` is a duplicate (same + /// `share_idx`, equal data) and nothing was stored. + /// - [`StoreOutcome::Stored`] if stored but the threshold is not yet met. + /// - [`StoreOutcome::Threshold`] carrying exactly the matched set forwarded + /// to threshold subscribers. + /// + /// Returns `Err(ParsigDataMismatch)` on a conflicting duplicate and + /// propagates `MessageRoot` errors from the threshold check. + async fn store(&self, k: Key, value: ParSignedData) -> Result { let mut inner = self.inner.lock().await; // Check if we already have an entry with this ShareIdx @@ -338,8 +348,8 @@ impl MemDB { for s in existing_entries { if s.share_idx == value.share_idx { if s == &value { - // Duplicate, return None to indicate no new data - return Ok(None); + // Duplicate, nothing stored. + return Ok(StoreOutcome::Duplicate); } else { return Err(MemDBError::ParsigDataMismatch { pubkey: k.pub_key, @@ -354,7 +364,7 @@ impl MemDB { .entries .entry(k.clone()) .or_insert_with(Vec::new) - .push(value.clone()); + .push(value); inner .keys_by_duty .entry(k.duty.clone()) @@ -365,13 +375,28 @@ impl MemDB { PARSIG_DB_METRICS.exit_total[&k.pub_key.to_string()].inc(); } - let result = inner.entries.get(&k).cloned().unwrap_or_default(); + // Threshold check runs under the lock against the borrowed slice; only + // the matched subset (if any) is cloned out. No full-accumulator clone. + let sigs = inner.entries.get(&k).map(Vec::as_slice).unwrap_or_default(); - Ok(Some(result)) + match get_threshold_matching(&k.duty.duty_type, sigs, self.threshold)? { + Some(psigs) => Ok(StoreOutcome::Threshold(psigs)), + None => Ok(StoreOutcome::Stored), + } } } -async fn get_threshold_matching( +/// Result of [`MemDB::store`]. +enum StoreOutcome { + /// `value` was a duplicate; nothing stored. + Duplicate, + /// Stored, but the threshold for the key is not yet reached. + Stored, + /// Stored and threshold reached; carries the matched set. + Threshold(Vec), +} + +fn get_threshold_matching( typ: &DutyType, sigs: &[ParSignedData], threshold: u64, diff --git a/crates/core/src/parsigdb/memory_internal_test.rs b/crates/core/src/parsigdb/memory_internal_test.rs index a62aec32..849d0a1e 100644 --- a/crates/core/src/parsigdb/memory_internal_test.rs +++ b/crates/core/src/parsigdb/memory_internal_test.rs @@ -9,7 +9,7 @@ use tokio::{ }; use tokio_util::sync::CancellationToken; -use super::{MemDB, get_threshold_matching, threshold_subscriber}; +use super::{MemDB, MemDBError, get_threshold_matching, threshold_subscriber}; use crate::{ deadline::{DeadlinerTask, NeverExpiringCalculator}, signeddata::{BeaconCommitteeSelection, SignedSyncMessage, VersionedAttestation}, @@ -78,7 +78,6 @@ async fn test_get_threshold_matching(input: Vec, output: Vec) { } let out = get_threshold_matching(&DutyType::SyncMessage, &data, threshold) - .await .expect("threshold matching should succeed"); let expect: Vec<_> = output.iter().map(|idx| data[*idx].clone()).collect(); let expected_out = if expect.is_empty() { @@ -172,3 +171,82 @@ async fn memdb_threshold() { .await .expect("trim task should shut down cleanly"); } + +/// Builds a `MemDB` (with a real never-expiring deadliner) and a shared +/// threshold-subscriber call counter. +fn memdb_with_counter(threshold: u64) -> (Arc, Arc>, CancellationToken) { + let cancel = CancellationToken::new(); + let (deadliner, _drop_rx) = + DeadlinerTask::start(cancel.clone(), "memdb_test", NeverExpiringCalculator); + let db = Arc::new(MemDB::new(cancel.clone(), threshold, deadliner)); + (db, Arc::new(Mutex::new(0usize)), cancel) +} + +#[tokio::test] +async fn store_external_ignores_duplicate() { + const THRESHOLD: u64 = 3; + + let (db, times_called, cancel) = memdb_with_counter(THRESHOLD); + db.subscribe_threshold(threshold_subscriber({ + let times_called = times_called.clone(); + move |_duty, _data| { + let times_called = times_called.clone(); + async move { + *times_called.lock().await += 1; + Ok(()) + } + } + })) + .await; + + let pubkey = random_core_pub_key(); + let attestation = testutil::random_deneb_versioned_attestation(); + let duty = Duty::new_attester_duty(SlotNumber::new(123)); + + let partial = VersionedAttestation::new_partial(attestation.clone(), 1) + .expect("versioned attestation should be valid"); + let mut set = ParSignedDataSet::new(); + set.insert(pubkey, partial); + + // Store the identical partial twice: neither reaches the threshold and the + // duplicate must not error nor fire the subscriber. + db.store_external(&duty, &set).await.expect("first store"); + db.store_external(&duty, &set) + .await + .expect("duplicate store is not an error"); + + assert_eq!(0, *times_called.lock().await); + cancel.cancel(); +} + +#[tokio::test] +async fn store_external_rejects_share_index_mismatch() { + const THRESHOLD: u64 = 3; + + let (db, _times_called, cancel) = memdb_with_counter(THRESHOLD); + + let pubkey = random_core_pub_key(); + let duty = Duty::new_attester_duty(SlotNumber::new(123)); + + // Two different partials that both claim share index 1 for the same key. + let first = + VersionedAttestation::new_partial(testutil::random_deneb_versioned_attestation(), 1) + .expect("versioned attestation should be valid"); + let second = + VersionedAttestation::new_partial(testutil::random_deneb_versioned_attestation(), 1) + .expect("versioned attestation should be valid"); + + let mut set1 = ParSignedDataSet::new(); + set1.insert(pubkey, first); + db.store_external(&duty, &set1).await.expect("first store"); + + let mut set2 = ParSignedDataSet::new(); + set2.insert(pubkey, second); + let err = db + .store_external(&duty, &set2) + .await + .expect_err("conflicting share index must be rejected"); + assert!(matches!(err, MemDBError::ParsigDataMismatch { .. })); + + cancel.cancel(); +} diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index ad6ec3c1..6a53446c 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -1,4 +1,7 @@ -use std::collections::{HashMap, hash_map::Entry}; +use std::{ + collections::{HashMap, hash_map::Entry}, + sync::{LazyLock, Mutex}, +}; use backon::{BackoffBuilder, Retryable}; use tokio::sync; @@ -203,10 +206,16 @@ impl SchedulerBuilder { .await .ok_or(SchedulerError::Terminated)??; + // Fetch the chain constants once at build time (the node is synced at + // this point) and cache `slots_per_epoch` on the actor, mirroring + // Charon's memoized-spec behaviour. + let (_slot_duration, slots_per_epoch) = client.api().fetch_slots_config().await?; + let slot_rx = new_slot_ticker(&client, ct.clone()).await?; let actor = SchedulerActor { client: client.clone(), + slots_per_epoch, // TODO: Figure out what to pass as `pub_keys`. // In Charon, these are not used (dead code) slot_broadcast: self.slot_broadcast, @@ -266,6 +275,12 @@ impl SchedulerHandle { struct SchedulerActor { client: pluto_eth2api::BeaconNodeClient, + /// Cached chain constant: number of slots per epoch. Fetched once at build + /// time. Charon reads this from the memoized beacon-node spec; Pluto's + /// `fetch_slots_config` is not memoized, so we cache it here to avoid a + /// beacon-node round-trip on every duty lookup. + slots_per_epoch: u64, + slot_broadcast: sync::broadcast::Sender, duty_broadcast: sync::broadcast::Sender<(types::Duty, types::DutyDefinitionSet)>, @@ -338,13 +353,11 @@ impl SchedulerActor { return Err(SchedulerError::DeprecatedDutyBuilderProposer); } - // TODO: `client.fetch_slots_config` should be cached. - let (_, slots_per_epoch) = self.client.api().fetch_slots_config().await?; let epoch = duty .slot .inner() - .checked_div(slots_per_epoch) - .expect("non-zero"); + .checked_div(self.slots_per_epoch) + .expect("non-zero: slots_per_epoch is validated non-zero by fetch_slots_config"); if !self.is_epoch_resolved(epoch) { return Err(SchedulerError::EpochNotResolved { epoch, duty }); @@ -662,6 +675,48 @@ struct Validator { v_idx: pluto_eth2api::spec::phase0::ValidatorIndex, } +/// Last-reported `validator_status` label value per validator pubkey. +/// +/// Mirrors Charon's `statusGauge.Reset(pubkey, ...)`: vise's `Family` cannot +/// delete series, so we remember the previously-reported status per pubkey and +/// zero exactly that one series before setting the current status to 1. This +/// avoids the O(N) scan of the whole metric family on every validator. +fn last_validator_status() -> &'static Mutex> { + static MAP: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + &MAP +} + +/// Submits the `validator_status` gauge for `pubkey`, emulating Charon's +/// `statusGauge.Reset(pubkey, ...)` + `Set(1)` in O(1) per validator: it zeroes +/// only the previously-reported series for this pubkey (when the status +/// changed) and sets the current one to 1. +fn submit_validator_status_metric(pubkey: &types::PubKey, status: &str) { + let pubkey_full = pubkey.to_string(); + let pubkey_abbrev = pubkey.abbreviated(); + + { + let mut last = last_validator_status() + .lock() + .expect("validator status map mutex poisoned"); + match last.get(pubkey) { + // Unchanged: the current series is already 1, nothing to reset. + Some(prev) if prev == status => {} + // Status changed: zero the old series, then record the new one. + Some(prev) => { + SCHEDULER_METRICS.validator_status + [&(pubkey_full.clone(), pubkey_abbrev.clone(), prev.clone())] + .set(0); + last.insert(*pubkey, status.to_string()); + } + None => { + last.insert(*pubkey, status.to_string()); + } + } + } + SCHEDULER_METRICS.validator_status[&(pubkey_full, pubkey_abbrev, status.to_string())].set(1); +} + /// Returns the active validators (including their validator index) for the /// epoch. async fn resolve_active_validators( @@ -682,17 +737,10 @@ async fn resolve_active_validators( SCHEDULER_METRICS.validator_balance_gwei[&(pubkey_full.clone(), pubkey_abbrev.clone())] .set(balance); - // Emulate Charon's `statusGauge.Reset`: - // Vise's `Family` cannot delete series, so instead set any previously-reported - // status for this validator to 0 and the current one to 1. + // Emulate Charon's `statusGauge.Reset(pubkey, ...)` in O(1) per + // validator (see `submit_validator_status_metric`). let status = val.status.to_string(); - for ((full, abbrev, prev_status), gauge) in SCHEDULER_METRICS.validator_status.to_entries() - { - if full == pubkey_full && abbrev == pubkey_abbrev && prev_status != status { - gauge.set(0); - } - } - SCHEDULER_METRICS.validator_status[&(pubkey_full, pubkey_abbrev, status)].set(1); + submit_validator_status_metric(&pubkey, &status); // Check for active validators for the given epoch. // The activation epoch needs to be checked in cases where this function is @@ -1088,6 +1136,7 @@ mod tests { fn test_actor(mock: &BeaconMock) -> SchedulerActor { SchedulerActor { client: pluto_eth2api::BeaconNodeClient::new(mock.client().clone()), + slots_per_epoch: 1, slot_broadcast: sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0, duty_broadcast: sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0, resolved_epoch: u64::MAX, @@ -1148,14 +1197,24 @@ mod tests { ct: CancellationToken, } - fn spawn_actor(mock: &BeaconMock) -> TestHarness { + async fn spawn_actor(mock: &BeaconMock) -> TestHarness { let slot_broadcast = sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0; let duty_broadcast = sync::broadcast::channel(CHANNEL_BUFFER_SIZE).0; let slot_sub = slot_broadcast.subscribe(); let duty_sub = duty_broadcast.subscribe(); + let client = pluto_eth2api::BeaconNodeClient::new(mock.client().clone()); + // Cache slots_per_epoch from the mock's spec, mirroring `build`, so + // `get_duty_definition`'s epoch math matches the slots the test drives. + let (_slot_duration, slots_per_epoch) = client + .api() + .fetch_slots_config() + .await + .expect("mock exposes slots config"); + let actor = SchedulerActor { - client: pluto_eth2api::BeaconNodeClient::new(mock.client().clone()), + client, + slots_per_epoch, slot_broadcast, duty_broadcast, resolved_epoch: u64::MAX, @@ -1304,6 +1363,71 @@ mod tests { )); } + #[tokio::test] + async fn get_duty_definition_uses_cached_slots_per_epoch() { + let mock = BeaconMock::builder().build().await.expect("build mock"); + let mut actor = test_actor(&mock); + // Cached at build time, not fetched per lookup. + actor.slots_per_epoch = 32; + + // Slot 64 → epoch 2 (64 / 32). resolved_epoch == u64::MAX, so the + // lookup fails with EpochNotResolved for epoch 2, proving the divisor + // is the cached field rather than any per-lookup network value. + let att = types::Duty::new_attester_duty(types::SlotNumber::new(64)); + assert!(matches!( + actor.get_duty_definition(att).await, + Err(SchedulerError::EpochNotResolved { epoch: 2, .. }) + )); + } + + /// Reads the current value of a `validator_status` gauge series, or 0 if + /// the series has not been created. + fn status_gauge(pubkey: &types::PubKey, status: &str) -> u64 { + SCHEDULER_METRICS + .validator_status + .get(&(pubkey.to_string(), pubkey.abbreviated(), status.to_string())) + .map(|g| g.get()) + .unwrap_or(0) + } + + #[test] + fn submit_validator_status_first_report_sets_current() { + let pk = random_core_pub_key(); + submit_validator_status_metric(&pk, "active_ongoing"); + assert_eq!(status_gauge(&pk, "active_ongoing"), 1); + } + + #[test] + fn submit_validator_status_change_zeros_old_sets_new() { + let pk = random_core_pub_key(); + submit_validator_status_metric(&pk, "active_ongoing"); + submit_validator_status_metric(&pk, "active_exiting"); + assert_eq!(status_gauge(&pk, "active_ongoing"), 0); + assert_eq!(status_gauge(&pk, "active_exiting"), 1); + } + + #[test] + fn submit_validator_status_unchanged_is_idempotent() { + let pk = random_core_pub_key(); + submit_validator_status_metric(&pk, "active_ongoing"); + submit_validator_status_metric(&pk, "active_ongoing"); + assert_eq!(status_gauge(&pk, "active_ongoing"), 1); + } + + #[test] + fn submit_validator_status_pubkeys_are_independent() { + let pk_a = random_core_pub_key(); + let pk_b = random_core_pub_key(); + submit_validator_status_metric(&pk_a, "active_ongoing"); + submit_validator_status_metric(&pk_b, "active_ongoing"); + + // Changing A must not affect B's series. + submit_validator_status_metric(&pk_a, "active_exiting"); + assert_eq!(status_gauge(&pk_a, "active_ongoing"), 0); + assert_eq!(status_gauge(&pk_a, "active_exiting"), 1); + assert_eq!(status_gauge(&pk_b, "active_ongoing"), 1); + } + #[tokio::test] async fn resolve_duties_stores_all_duty_types() { let mock = duties_mock(16).await; @@ -1396,7 +1520,7 @@ mod tests { async fn first_slot_broadcasts_slot_and_triggers_duties(slot_number: u64) { let mock = duties_mock(16).await; mount_head_validators(&mock, validator_set_a_datums()).await; - let mut h = spawn_actor(&mock); + let mut h = spawn_actor(&mock).await; h.slot_tx .send(test_past_slot(slot_number, 16)) @@ -1438,7 +1562,7 @@ mod tests { ) { let mock = duties_mock(16).await; mount_head_validators(&mock, validator_set_a_datums()).await; - let mut h = spawn_actor(&mock); + let mut h = spawn_actor(&mock).await; // Slot is mid-epoch (epoch 0 spans slots 0..=15). With the deterministic // Beacon setup: @@ -1481,7 +1605,7 @@ mod tests { async fn get_duty_success_then_reorg_then_get_duty_fails() { let mock = duties_mock(16).await; mount_head_validators(&mock, validator_set_a_datums()).await; - let mut h = spawn_actor(&mock); + let mut h = spawn_actor(&mock).await; // Drive a slot in epoch 1 and wait for a duty broadcast, which only // happens once `resolve_duties` has completed for the epoch. @@ -1519,7 +1643,7 @@ mod tests { async fn cancellation_during_slot_offset_suppresses_duty_broadcast() { let mock = duties_mock(16).await; mount_head_validators(&mock, validator_set_a_datums()).await; - let mut h = spawn_actor(&mock); + let mut h = spawn_actor(&mock).await; // A mid-epoch slot triggers only the sync-committee contribution duty, // whose broadcast is delayed by 2/3 of the slot duration (~600ms here). diff --git a/crates/dkg/src/bcast/behaviour.rs b/crates/dkg/src/bcast/behaviour.rs index 62066f20..b67b76f3 100644 --- a/crates/dkg/src/bcast/behaviour.rs +++ b/crates/dkg/src/bcast/behaviour.rs @@ -116,11 +116,7 @@ impl Behaviour { } fn is_connected(&self, peer_id: &PeerId) -> bool { - !self - .p2p_context - .peer_store_lock() - .connections_to_peer(peer_id) - .is_empty() + self.p2p_context.peer_store_lock().has_connection(peer_id) } fn new_handler(&self, peer: PeerId) -> Handler { diff --git a/crates/dkg/src/frostp2p/behaviour.rs b/crates/dkg/src/frostp2p/behaviour.rs index 3550890c..54c44ad1 100644 --- a/crates/dkg/src/frostp2p/behaviour.rs +++ b/crates/dkg/src/frostp2p/behaviour.rs @@ -174,11 +174,7 @@ impl FrostP2PBehaviour { } fn is_connected(&self, peer_id: &PeerId) -> bool { - !self - .p2p_context - .peer_store_lock() - .connections_to_peer(peer_id) - .is_empty() + self.p2p_context.peer_store_lock().has_connection(peer_id) } fn drain_commands(&mut self, cx: &mut Context<'_>) { diff --git a/crates/eth2api/src/valcache.rs b/crates/eth2api/src/valcache.rs index 6e086680..0b6b1b93 100644 --- a/crates/eth2api/src/valcache.rs +++ b/crates/eth2api/src/valcache.rs @@ -19,8 +19,11 @@ pub enum ValidatorCacheError { } /// Active validators as [`PubKey`] indexed by their validator index. +/// +/// Internally an `Arc>` so cloning is a refcount bump, not a deep +/// map copy — callers receive cheap clones from [`ValidatorCache`]. #[derive(Debug, Clone, Default, PartialEq)] -pub struct ActiveValidators(HashMap); +pub struct ActiveValidators(Arc>); impl std::ops::Deref for ActiveValidators { type Target = HashMap; @@ -31,8 +34,13 @@ impl std::ops::Deref for ActiveValidators { } /// Complete response of the Beacon node validators endpoint. +/// +/// Internally an `Arc>` so cloning is a refcount bump, not a deep +/// map copy. #[derive(Debug, Clone, Default, PartialEq)] -pub struct CompleteValidators(HashMap); +pub struct CompleteValidators( + Arc>, +); impl std::ops::Deref for CompleteValidators { type Target = HashMap; @@ -47,7 +55,7 @@ impl ActiveValidators { /// Lets consumers outside this crate (e.g. test doubles of /// [`CachedValidatorsProvider`]) construct populated instances. pub fn new(validators: HashMap) -> Self { - Self(validators) + Self(Arc::new(validators)) } /// An [`Iterator`] of active validator indices. @@ -121,10 +129,24 @@ impl ValidatorCache { /// Returns the cached active validators and complete validators response, /// or fetches them if not available populating the cache. pub async fn get_by_head(&self) -> Result<(ActiveValidators, CompleteValidators)> { - let mut inner = self.0.write().await; - - if let (Some(active), Some(complete)) = (&inner.active, &inner.complete) { - return Ok((active.clone(), complete.clone())); + // Fast path: cache already populated. Take only a read lock and return + // cheap `Arc` clones. Mirrors Charon's `activeCached()`/`cached()` + // `RLock` fast path in `app/eth2wrap/valcache.go`. + { + let inner = self.0.read().await; + if let (Some(active), Some(complete)) = (&inner.active, &inner.complete) { + return Ok((active.clone(), complete.clone())); + } + } // read lock released here + + // Slow path: cache miss. Snapshot the client + pubkeys we need, then + // release the lock so the beacon-node round-trip does not block + // concurrent readers. This improves on Charon, which holds the write + // lock across the fetch; observable behaviour is identical (a + // benign double-fetch race is collapsed by the re-check below). + let (eth2_cl, pubkeys) = { + let inner = self.0.read().await; + (inner.eth2_cl.clone(), inner.pubkeys.clone()) }; let request = PostStateValidatorsRequest { @@ -132,13 +154,13 @@ impl ValidatorCache { state_id: "head".into(), }, body: ValidatorRequestBody { - ids: Some(inner.pubkeys.iter().map(format_pubkey).collect()), + ids: Some(pubkeys.iter().map(format_pubkey).collect()), ..Default::default() }, }; - let response = inner - .eth2_cl + // Fetch WITHOUT holding the lock. + let response = eth2_cl .post_state_validators(request) .await .map_err(EthBeaconNodeApiClientError::RequestError) @@ -149,6 +171,15 @@ impl ValidatorCache { let (active_validators, complete_validators) = validators_from_response(response)?; + // Re-acquire the write lock and re-check: another task may have + // populated the cache while we were fetching. If so, prefer the + // existing entry and discard our fetch (idempotent — both fetched + // "head"). + let mut inner = self.0.write().await; + if let (Some(active), Some(complete)) = (&inner.active, &inner.complete) { + return Ok((active.clone(), complete.clone())); + } + inner.active = Some(active_validators.clone()); inner.complete = Some(complete_validators.clone()); @@ -234,8 +265,8 @@ fn validators_from_response( .collect::>>()?; Ok(( - ActiveValidators(active_validators), - CompleteValidators(all_validators), + ActiveValidators(Arc::new(active_validators)), + CompleteValidators(Arc::new(all_validators)), )) } @@ -312,14 +343,14 @@ mod tests { // Check cache is populated. let (actual_active, actual_complete) = cache.get_by_head().await.expect("`get_by_head` succeeds"); - assert_eq!(actual_active.0, expected_active); - assert_eq!(actual_complete.0, expected_complete); + assert_eq!(*actual_active, expected_active); + assert_eq!(*actual_complete, expected_complete); // Check cache is used (no additional request). let (actual_active, actual_complete) = cache.get_by_head().await.expect("`get_by_head` succeeds"); - assert_eq!(actual_active.0, expected_active); - assert_eq!(actual_complete.0, expected_complete); + assert_eq!(*actual_active, expected_active); + assert_eq!(*actual_complete, expected_complete); // Trim cache. cache.trim().await; @@ -327,14 +358,69 @@ mod tests { // Check cache is populated again. let (actual_active, actual_complete) = cache.get_by_head().await.expect("`get_by_head` succeeds"); - assert_eq!(actual_active.0, expected_active); - assert_eq!(actual_complete.0, expected_complete); + assert_eq!(*actual_active, expected_active); + assert_eq!(*actual_complete, expected_complete); // Check cache is used again (no additional request). let (actual_active, actual_complete) = cache.get_by_head().await.expect("`get_by_head` succeeds"); - assert_eq!(actual_active.0, expected_active); - assert_eq!(actual_complete.0, expected_complete); + assert_eq!(*actual_active, expected_active); + assert_eq!(*actual_complete, expected_complete); + } + + #[tokio::test] + async fn get_by_head_concurrent_miss_is_consistent() { + // Concurrent cache misses each return correct, identical data. Because + // the write lock is deliberately released across the beacon-node fetch + // (so a warm cache never blocks readers — see `get_by_head`), a burst + // of *cold* misses may each issue a fetch; the re-check after + // re-acquiring the write lock only guarantees a single stored value, + // not a single request. Once the cache is warm, further reads take the + // read-lock fast path and issue no request (see + // `get_by_head_successful_fetch`). In production `get_by_head` is driven + // by the scheduler's slot tick, so this cold-start burst does not occur. + const CONCURRENCY: u64 = 8; + let pubkeys = (0..3u8).map(test_pubkey).collect::>(); + let datums = vec![ + test_validator_datum(0, &pubkeys[0], ValidatorStatus::ActiveOngoing), + test_validator_datum(1, &pubkeys[1], ValidatorStatus::ActiveOngoing), + test_validator_datum(2, &pubkeys[2], ValidatorStatus::ActiveOngoing), + ]; + + let mock = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/eth/v1/beacon/states/head/validators")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(std::time::Duration::from_millis(50)) + .set_body_json(GetStateValidatorsResponseResponse { + execution_optimistic: false, + finalized: true, + data: datums, + }), + ) + // At least one fetch, at most one per concurrent cold miss. + .expect(1..=CONCURRENCY) + .mount(&mock) + .await; + + let cache = ValidatorCache::new(test_client(&mock), pubkeys.clone()); + + let mut handles = Vec::new(); + for _ in 0..CONCURRENCY { + let cache = cache.clone(); + handles.push(tokio::spawn(async move { cache.get_by_head().await })); + } + + for handle in handles { + let (active, complete) = handle.await.expect("task joins").expect("get_by_head"); + assert_eq!(active.len(), 3); + assert_eq!(complete.len(), 3); + } + + // After warm-up, the read-lock fast path serves without any new request. + let (active, _) = cache.get_by_head().await.expect("warm read"); + assert_eq!(active.len(), 3); } #[tokio::test] diff --git a/crates/p2p/src/p2p_context.rs b/crates/p2p/src/p2p_context.rs index 26e7d6a9..b187a22e 100644 --- a/crates/p2p/src/p2p_context.rs +++ b/crates/p2p/src/p2p_context.rs @@ -140,6 +140,14 @@ impl PeerStore { self.inactive_peers.len() } + /// Returns whether there is any active connection to the given peer. + /// + /// Equivalent to `!connections_to_peer(peer_id).is_empty()` but does not + /// allocate, since it short-circuits on the first match. + pub fn has_connection(&self, peer_id: &PeerId) -> bool { + self.active_peers.iter().any(|p| &p.id == peer_id) + } + /// Returns all active connections to a specific peer. pub fn connections_to_peer(&self, peer_id: &PeerId) -> Vec<&Peer> { self.active_peers @@ -158,3 +166,72 @@ impl PeerStore { self.peer_addresses.get(peer_id) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn peer(id: PeerId, connection_id: usize) -> Peer { + Peer { + id, + connection_id: ConnectionId::new_unchecked(connection_id), + remote_addr: Multiaddr::empty(), + } + } + + #[test] + fn has_connection_empty_store_is_false() { + let store = PeerStore::default(); + assert!(!store.has_connection(&PeerId::random())); + } + + #[test] + fn has_connection_reflects_added_peer() { + let a = PeerId::random(); + let b = PeerId::random(); + let mut store = PeerStore::default(); + store.add_peer(peer(a, 1)); + + assert!(store.has_connection(&a)); + assert!(!store.has_connection(&b)); + } + + #[test] + fn has_connection_true_with_multiple_connections() { + let a = PeerId::random(); + let mut store = PeerStore::default(); + store.add_peer(peer(a, 1)); + store.add_peer(peer(a, 2)); + + assert!(store.has_connection(&a)); + } + + #[test] + fn has_connection_matches_connections_to_peer() { + let a = PeerId::random(); + let b = PeerId::random(); + let c = PeerId::random(); + let mut store = PeerStore::default(); + store.add_peer(peer(a, 1)); + store.add_peer(peer(b, 1)); + + for id in [a, b, c] { + assert_eq!( + store.has_connection(&id), + !store.connections_to_peer(&id).is_empty() + ); + } + } + + #[test] + fn has_connection_false_after_remove() { + let a = PeerId::random(); + let mut store = PeerStore::default(); + let conn = peer(a, 1); + store.add_peer(conn.clone()); + assert!(store.has_connection(&a)); + + store.remove_peer(conn); + assert!(!store.has_connection(&a)); + } +} diff --git a/crates/parsigex/src/behaviour.rs b/crates/parsigex/src/behaviour.rs index 94904533..779df314 100644 --- a/crates/parsigex/src/behaviour.rs +++ b/crates/parsigex/src/behaviour.rs @@ -321,6 +321,9 @@ impl Behaviour { return; } }; + // Share a single refcounted buffer across all broadcast targets instead + // of deep-copying the encoded bytes once per peer. + let message: Arc<[u8]> = Arc::from(message); let peers: Vec<_> = self .config @@ -331,18 +334,19 @@ impl Behaviour { .collect(); let mut pending_peers = HashSet::new(); let mut failure = None; + // Acquire the peer-store read guard once for the whole broadcast rather + // than re-locking per peer. Clone the cheap `Arc`-backed context handle + // so the guard does not keep `self` borrowed while the loop mutably + // borrows other `self` fields (`pending_events`, via + // `emit_broadcast_error`). + let p2p_context = self.config.p2p_context.clone(); + let peer_store = p2p_context.peer_store_lock(); for peer in peers { if peer == self.config.peer_id { continue; } - if self - .config - .p2p_context - .peer_store_lock() - .connections_to_peer(&peer) - .is_empty() - { + if !peer_store.has_connection(&peer) { let error = Failure::Io(std::io::Error::other(format!( "peer {peer} is not connected" ))); @@ -363,6 +367,8 @@ impl Behaviour { }); pending_peers.insert(peer); } + drop(peer_store); + drop(p2p_context); if pending_peers.is_empty() { self.pending_events @@ -775,3 +781,111 @@ mod eth2_verifier_tests { assert!(matches!(err, VerifyError::InvalidShareIndex)); } } + +#[cfg(test)] +mod broadcast_tests { + use libp2p::{Multiaddr, PeerId, swarm::ConnectionId}; + + use pluto_core::types::{Duty, ParSignedDataSet}; + use pluto_p2p::p2p_context::{P2PContext, Peer}; + + use super::*; + + fn trivial_verifier() -> Verifier { + Arc::new(|_duty, _pubkey, _par| Box::pin(async { Ok(()) })) + } + + fn allow_all_gater() -> DutyGaterFn { + Arc::new(|_duty| true) + } + + fn connected_peer(context: &P2PContext, id: PeerId) { + context.peer_store_write_lock().add_peer(Peer { + id, + connection_id: ConnectionId::new_unchecked(1), + remote_addr: Multiaddr::empty(), + }); + } + + /// A broadcast to multiple connected peers must share a single refcounted + /// payload buffer across every `ToHandler::Send` instead of deep-copying + /// the encoded bytes per target. + #[test] + fn broadcast_shares_single_payload_buffer() { + let local = PeerId::random(); + let peer_a = PeerId::random(); + let peer_b = PeerId::random(); + + let context = P2PContext::new([local, peer_a, peer_b]); + connected_peer(&context, peer_a); + connected_peer(&context, peer_b); + + let config = Config::new(local, context, trivial_verifier(), allow_all_gater()); + let (mut behaviour, _handle) = Behaviour::new(config); + + behaviour.handle_command(BroadcastRequest { + request_id: 1, + duty: Duty::new_attester_duty(32.into()), + data_set: ParSignedDataSet::new(), + result_tx: None, + }); + + let payloads: Vec> = behaviour + .pending_events + .iter() + .filter_map(|event| match event { + ToSwarm::NotifyHandler { + event: ToHandler::Send { payload, .. }, + .. + } => Some(payload.clone()), + _ => None, + }) + .collect(); + + assert_eq!(payloads.len(), 2, "one send per connected non-self peer"); + assert!( + Arc::ptr_eq(&payloads[0], &payloads[1]), + "all broadcast targets must share the same payload allocation" + ); + } + + /// A known peer with no active connection still yields a single + /// `BroadcastError` and is excluded from the pending broadcast. + #[test] + fn broadcast_reports_not_connected_peer() { + let local = PeerId::random(); + let peer_a = PeerId::random(); + + let context = P2PContext::new([local, peer_a]); + // peer_a is known but never added to the peer store (not connected). + + let config = Config::new(local, context, trivial_verifier(), allow_all_gater()); + let (mut behaviour, _handle) = Behaviour::new(config); + + behaviour.handle_command(BroadcastRequest { + request_id: 7, + duty: Duty::new_attester_duty(32.into()), + data_set: ParSignedDataSet::new(), + result_tx: None, + }); + + let errors = behaviour + .pending_events + .iter() + .filter(|event| matches!(event, ToSwarm::GenerateEvent(Event::BroadcastError { .. }))) + .count(); + assert_eq!( + errors, 1, + "the unconnected peer produces one BroadcastError" + ); + + // No send events, and the broadcast fails (no connected targets). + assert!(behaviour.pending_events.iter().all(|event| !matches!( + event, + ToSwarm::NotifyHandler { + event: ToHandler::Send { .. }, + .. + } + ))); + } +} diff --git a/crates/parsigex/src/handler.rs b/crates/parsigex/src/handler.rs index 44203f2a..963ecc60 100644 --- a/crates/parsigex/src/handler.rs +++ b/crates/parsigex/src/handler.rs @@ -2,6 +2,7 @@ use std::{ collections::VecDeque, + sync::Arc, task::{Context, Poll}, time::Duration, }; @@ -34,8 +35,8 @@ pub enum ToHandler { Send { /// Request identifier used to correlate broadcast completions. request_id: u64, - /// Encoded protobuf payload. - payload: Vec, + /// Encoded protobuf payload, shared across all broadcast targets. + payload: Arc<[u8]>, }, } @@ -71,8 +72,9 @@ pub struct PendingOpen { /// [`FromHandler::OutboundSuccess`] and [`FromHandler::OutboundError`] /// events back to the originating broadcast. request_id: u64, - /// Encoded protobuf payload to send once the stream is ready. - payload: Vec, + /// Encoded protobuf payload to send once the stream is ready, shared + /// across all broadcast targets. + payload: Arc<[u8]>, } type ActiveFuture = BoxFuture<'static, Option>; @@ -267,7 +269,7 @@ async fn do_recv( Ok((duty, data_set)) } -async fn do_send(mut stream: libp2p::swarm::Stream, payload: Vec) -> Result<(), Failure> { +async fn do_send(mut stream: libp2p::swarm::Stream, payload: Arc<[u8]>) -> Result<(), Failure> { protocol::send_message(&mut stream, &payload) .await .map_err(Failure::Io) diff --git a/crates/ssz/src/hasher.rs b/crates/ssz/src/hasher.rs index c4c8cc71..d0407afd 100644 --- a/crates/ssz/src/hasher.rs +++ b/crates/ssz/src/hasher.rs @@ -138,12 +138,12 @@ impl Hasher { return Err(HasherError::InvalidBufferLength); } let mut result = Vec::with_capacity(src.len() / 2); + let mut hasher = Sha256::new(); for pair in src.chunks(64) { - let mut hasher = Sha256::new(); hasher.update(&pair[..32]); hasher.update(&pair[32..]); - result.extend_from_slice(&hasher.finalize()); + result.extend_from_slice(&hasher.finalize_reset()); } Ok(result) @@ -179,9 +179,12 @@ impl Hasher { 64 - i.leading_zeros() as usize - 1 } - fn merkleize_impl(&mut self, input: &[u8], mut limit: usize) -> Result, HasherError> { + fn merkleize_impl( + &mut self, + mut input: Vec, + mut limit: usize, + ) -> Result, HasherError> { let count = input.len().div_ceil(32); - let mut input = input.to_vec(); if limit == 0 { limit = count; @@ -317,7 +320,12 @@ impl HashWalker for Hasher { let size = parse_bitlist(&mut self.tmp, bb)?; let indx = self.index(); - self.append_bytes32(&self.tmp.clone())?; + // Take the scratch buffer out to satisfy the borrow checker without a + // clone (`append_bytes32` only touches `self.buf`), then restore it so + // its allocation is reused across calls. + let tmp = std::mem::take(&mut self.tmp); + self.append_bytes32(&tmp)?; + self.tmp = tmp; self.merkleize_with_mixin(indx, size, max_size.div_ceil(256))?; Ok(()) } @@ -354,8 +362,8 @@ impl HashWalker for Hasher { self.buf.reserve(32); // Just ensure capacity } - let mut input = self.buf[index..].to_vec(); - input = self.merkleize_impl(&input, 0)?; + let input = self.buf[index..].to_vec(); + let input = self.merkleize_impl(input, 0)?; self.buf.truncate(index); // Truncate without filling self.buf.extend_from_slice(&input); @@ -372,7 +380,7 @@ impl HashWalker for Hasher { let mut input: Vec = self.buf[index..].to_vec(); - input = self.merkleize_impl(&input, limit)?; + input = self.merkleize_impl(input, limit)?; self.buf.truncate(index); self.buf.extend_from_slice(&input); @@ -465,4 +473,110 @@ mod tests { 64 ); } + + #[test] + fn default_hash_fn_zero_pair_matches_zero_hash() { + // Hashing two zero leaves yields the depth-1 zero hash by construction. + assert_eq!( + Hasher::default_hash_fn(&[0u8; 64]).expect("one pair"), + ZERO_HASHES[1].to_vec() + ); + } + + #[test] + fn default_hash_fn_two_pairs_matches_independent_single_pairs() { + // Reusing one `Sha256` via `finalize_reset` must produce the same bytes + // as hashing each 64-byte pair independently. + let mut src = Vec::new(); + src.extend_from_slice(&[1u8; 32]); + src.extend_from_slice(&[2u8; 32]); + src.extend_from_slice(&[3u8; 32]); + src.extend_from_slice(&[4u8; 32]); + + let combined = Hasher::default_hash_fn(&src).expect("two pairs"); + let mut expected = Hasher::default_hash_fn(&src[..64]).expect("pair 0"); + expected.extend_from_slice(&Hasher::default_hash_fn(&src[64..]).expect("pair 1")); + + assert_eq!(combined, expected); + } + + #[test] + fn merkleize_single_chunk_returns_input() { + // limit==1, count==1 path returns the chunk verbatim. + let chunk = [7u8; 32]; + let mut h = Hasher::default(); + h.append_bytes32(&chunk).expect("append"); + h.merkleize(0).expect("merkleize"); + assert_eq!(h.hash().expect("hash"), chunk); + } + + #[test] + fn merkleize_multi_chunk_matches_manual_tree() { + // Three chunks: layer 0 pads to four with the depth-0 zero hash, then + // two hashing layers collapse to a single root. + let c0 = [1u8; 32]; + let c1 = [2u8; 32]; + let c2 = [3u8; 32]; + + let mut h = Hasher::default(); + h.append_bytes32(&c0).expect("c0"); + h.append_bytes32(&c1).expect("c1"); + h.append_bytes32(&c2).expect("c2"); + h.merkleize(0).expect("merkleize"); + let root = h.hash().expect("hash"); + + let mut layer0 = Vec::new(); + layer0.extend_from_slice(&c0); + layer0.extend_from_slice(&c1); + layer0.extend_from_slice(&c2); + layer0.extend_from_slice(&ZERO_HASHES[0]); + let layer1 = Hasher::default_hash_fn(&layer0).expect("layer1"); + let expected = Hasher::default_hash_fn(&layer1).expect("layer2"); + + assert_eq!(root, expected.as_slice()); + } + + #[test] + fn merkleize_with_mixin_matches_manual_tree() { + let c0 = [1u8; 32]; + let c1 = [2u8; 32]; + + let mut h = Hasher::default(); + h.append_bytes32(&c0).expect("c0"); + h.append_bytes32(&c1).expect("c1"); + h.merkleize_with_mixin(0, 2, 4).expect("mixin"); + let root = h.hash().expect("hash"); + + let mut layer0 = Vec::new(); + layer0.extend_from_slice(&c0); + layer0.extend_from_slice(&c1); + let l1 = Hasher::default_hash_fn(&layer0).expect("l1"); + let mut l1_padded = l1.clone(); + l1_padded.extend_from_slice(&ZERO_HASHES[1]); + let tree_root = Hasher::default_hash_fn(&l1_padded).expect("tree root"); + + let mut mixed = tree_root; + let mut tmp = [0u8; 32]; + tmp[..8].copy_from_slice(&2u64.to_le_bytes()); + mixed.extend_from_slice(&tmp); + let expected = Hasher::default_hash_fn(&mixed).expect("mixed"); + + assert_eq!(root, expected.as_slice()); + } + + #[test] + fn put_bitlist_is_deterministic_and_reuses_tmp() { + // Guards the `std::mem::take`/restore in `put_bitlist`: two identical + // calls on fresh hashers must produce the same root, and `self.tmp` + // must be restored (non-take semantics preserved across the call). + let mut a = Hasher::default(); + a.put_bitlist(&[0b0000_1011], 64).expect("put_bitlist a"); + let root_a = a.hash().expect("hash a"); + + let mut b = Hasher::default(); + b.put_bitlist(&[0b0000_1011], 64).expect("put_bitlist b"); + let root_b = b.hash().expect("hash b"); + + assert_eq!(root_a, root_b); + } } From 8ca3df9acdcd9e30bf64ffc033453d0cb9fb45db Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:45:51 +0200 Subject: [PATCH 2/2] fix: address PR review comments - valcache: move immutable `eth2_cl`/`pubkeys` out of the `RwLock` so only the mutable cache entries are guarded (per review). - Drop AI-style comments that merely restate the code; keep/relocate the genuine "why" notes (e.g. Arc rationale onto the type definition). Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/consensus/src/qbft/component.rs | 6 -- crates/consensus/src/qbft/p2p.rs | 6 +- crates/core/src/scheduler.rs | 5 +- crates/eth2api/src/valcache.rs | 90 ++++++++++++-------------- crates/parsigex/src/behaviour.rs | 10 +-- 5 files changed, 50 insertions(+), 67 deletions(-) diff --git a/crates/consensus/src/qbft/component.rs b/crates/consensus/src/qbft/component.rs index 37c8d8bf..1b4a6d97 100644 --- a/crates/consensus/src/qbft/component.rs +++ b/crates/consensus/src/qbft/component.rs @@ -186,9 +186,6 @@ pub enum Error { } /// Canonicalizes inbound `Any` values into the hash map used by QBFT messages. -/// -/// Consumes `values` and moves each entry into the map instead of cloning, -/// since the caller owns the values and discards the source vector afterwards. pub(crate) fn values_by_hash(values: Vec) -> Result { let mut out = ValueMap::new(); @@ -349,7 +346,6 @@ impl Consensus { mut pb_msg: pbconsensus::QbftConsensusMsg, ct: &CancellationToken, ) -> Result<()> { - // Verify the inner message and derive its duty (borrow only). let inner = pb_msg.msg.as_ref().ok_or(Error::InvalidConsensusMessage)?; self.verify_msg(inner)?; @@ -370,8 +366,6 @@ impl Consensus { } } - // Move the validated parts out of `pb_msg` (it is dropped after this), - // avoiding a clone of the message, justifications, and values. let msg = pb_msg.msg.take().ok_or(Error::InvalidConsensusMessage)?; let justification = std::mem::take(&mut pb_msg.justification); let values = values_by_hash(std::mem::take(&mut pb_msg.values))?; diff --git a/crates/consensus/src/qbft/p2p.rs b/crates/consensus/src/qbft/p2p.rs index 7c5141b0..c827e625 100644 --- a/crates/consensus/src/qbft/p2p.rs +++ b/crates/consensus/src/qbft/p2p.rs @@ -185,9 +185,6 @@ impl Handle { self.cmd_tx .send(BroadcastCommand { request_id, - // Wrap once so the fan-out shares a single allocation across all - // peers rather than deep-cloning the (potentially multi-MB) - // payload per target, matching Charon's shared-pointer broadcast. msg: Arc::new(msg), }) .map_err(|_| Box::new(Error::BehaviourClosed) as _) @@ -206,6 +203,8 @@ impl Handle { #[derive(Debug)] struct BroadcastCommand { request_id: u64, + /// Shared so the per-peer fan-out clones a pointer rather than the + /// (potentially multi-MB) payload. msg: Arc, } @@ -543,7 +542,6 @@ impl Behaviour { peer_id, PendingSend { request_id: command.request_id, - // Cheap refcount bump; the payload buffer is shared. msg: Arc::clone(&command.msg), }, ); diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index 6a53446c..a92e2093 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -206,9 +206,8 @@ impl SchedulerBuilder { .await .ok_or(SchedulerError::Terminated)??; - // Fetch the chain constants once at build time (the node is synced at - // this point) and cache `slots_per_epoch` on the actor, mirroring - // Charon's memoized-spec behaviour. + // Cached once here since the node is synced at this point; see the + // `slots_per_epoch` field on `SchedulerActor`. let (_slot_duration, slots_per_epoch) = client.api().fetch_slots_config().await?; let slot_rx = new_slot_ticker(&client, ct.clone()).await?; diff --git a/crates/eth2api/src/valcache.rs b/crates/eth2api/src/valcache.rs index 0b6b1b93..4c74bbb4 100644 --- a/crates/eth2api/src/valcache.rs +++ b/crates/eth2api/src/valcache.rs @@ -98,11 +98,16 @@ impl CachedValidatorsProvider for ValidatorCache { /// A cache for active validators. #[derive(Clone)] -pub struct ValidatorCache(Arc>); +pub struct ValidatorCache(Arc); struct ValidatorCacheInner { eth2_cl: EthBeaconNodeApiClient, pubkeys: Vec, + cached: RwLock, +} + +#[derive(Default)] +struct CachedValidators { active: Option, complete: Option, } @@ -110,57 +115,49 @@ struct ValidatorCacheInner { impl ValidatorCache { /// Creates a new, empty validator cache. pub fn new(eth2_cl: EthBeaconNodeApiClient, pubkeys: Vec) -> Self { - Self(Arc::new(RwLock::new(ValidatorCacheInner { + Self(Arc::new(ValidatorCacheInner { eth2_cl, pubkeys, - active: None, - complete: None, - }))) + cached: RwLock::new(CachedValidators::default()), + })) } /// Clears the cache. This should be called on epoch boundary. pub async fn trim(&self) { - let mut inner = self.0.write().await; + let mut cached = self.0.cached.write().await; - inner.active = None; - inner.complete = None; + cached.active = None; + cached.complete = None; } /// Returns the cached active validators and complete validators response, /// or fetches them if not available populating the cache. pub async fn get_by_head(&self) -> Result<(ActiveValidators, CompleteValidators)> { - // Fast path: cache already populated. Take only a read lock and return - // cheap `Arc` clones. Mirrors Charon's `activeCached()`/`cached()` - // `RLock` fast path in `app/eth2wrap/valcache.go`. + // Warm-cache fast path: a read lock is enough to serve cheap `Arc` + // clones without blocking concurrent readers. { - let inner = self.0.read().await; - if let (Some(active), Some(complete)) = (&inner.active, &inner.complete) { + let cached = self.0.cached.read().await; + if let (Some(active), Some(complete)) = (&cached.active, &cached.complete) { return Ok((active.clone(), complete.clone())); } - } // read lock released here - - // Slow path: cache miss. Snapshot the client + pubkeys we need, then - // release the lock so the beacon-node round-trip does not block - // concurrent readers. This improves on Charon, which holds the write - // lock across the fetch; observable behaviour is identical (a - // benign double-fetch race is collapsed by the re-check below). - let (eth2_cl, pubkeys) = { - let inner = self.0.read().await; - (inner.eth2_cl.clone(), inner.pubkeys.clone()) - }; + } + // Cache miss: fetch without holding any lock so the round-trip does + // not block concurrent readers. A cold-start burst may issue more than + // one fetch; the re-check below keeps a single stored value. let request = PostStateValidatorsRequest { path: PostStateValidatorsRequestPath { state_id: "head".into(), }, body: ValidatorRequestBody { - ids: Some(pubkeys.iter().map(format_pubkey).collect()), + ids: Some(self.0.pubkeys.iter().map(format_pubkey).collect()), ..Default::default() }, }; - // Fetch WITHOUT holding the lock. - let response = eth2_cl + let response = self + .0 + .eth2_cl .post_state_validators(request) .await .map_err(EthBeaconNodeApiClientError::RequestError) @@ -171,17 +168,13 @@ impl ValidatorCache { let (active_validators, complete_validators) = validators_from_response(response)?; - // Re-acquire the write lock and re-check: another task may have - // populated the cache while we were fetching. If so, prefer the - // existing entry and discard our fetch (idempotent — both fetched - // "head"). - let mut inner = self.0.write().await; - if let (Some(active), Some(complete)) = (&inner.active, &inner.complete) { + let mut cached = self.0.cached.write().await; + if let (Some(active), Some(complete)) = (&cached.active, &cached.complete) { return Ok((active.clone(), complete.clone())); } - inner.active = Some(active_validators.clone()); - inner.complete = Some(complete_validators.clone()); + cached.active = Some(active_validators.clone()); + cached.complete = Some(complete_validators.clone()); Ok((active_validators, complete_validators)) } @@ -196,26 +189,29 @@ impl ValidatorCache { &self, slot: u64, ) -> Result<(ActiveValidators, CompleteValidators, bool)> { - let mut inner = self.0.write().await; + // Held across the fetch so concurrent slot refreshes serialize, as + // before. The immutable client/pubkeys are read off the lock. + let mut cached = self.0.cached.write().await; let mut request = PostStateValidatorsRequest { path: PostStateValidatorsRequestPath { state_id: slot.to_string(), }, body: ValidatorRequestBody { - ids: Some(inner.pubkeys.iter().map(format_pubkey).collect()), + ids: Some(self.0.pubkeys.iter().map(format_pubkey).collect()), ..Default::default() }, }; let (response, refreshed_by_slot) = - match inner.eth2_cl.post_state_validators(request.clone()).await { + match self.0.eth2_cl.post_state_validators(request.clone()).await { Ok(PostStateValidatorsResponse::Ok(response)) => (response, true), _ => { // Failed to fetch by slot, fall back to head state request.path.state_id = "head".into(); - let response = inner + let response = self + .0 .eth2_cl .post_state_validators(request) .await @@ -231,8 +227,8 @@ impl ValidatorCache { let (active_validators, complete_validators) = validators_from_response(response)?; - inner.active = Some(active_validators.clone()); - inner.complete = Some(complete_validators.clone()); + cached.active = Some(active_validators.clone()); + cached.complete = Some(complete_validators.clone()); Ok((active_validators, complete_validators, refreshed_by_slot)) } @@ -436,9 +432,9 @@ mod tests { // Verify cache is initially empty { - let inner = cache.0.write().await; - assert!(inner.active.is_none()); - assert!(inner.complete.is_none()); + let cached = cache.0.cached.read().await; + assert!(cached.active.is_none()); + assert!(cached.complete.is_none()); } let result = cache.get_by_head().await; @@ -446,9 +442,9 @@ mod tests { // Verify cache remains empty after failed request { - let inner = cache.0.write().await; - assert!(inner.active.is_none()); - assert!(inner.complete.is_none()); + let cached = cache.0.cached.read().await; + assert!(cached.active.is_none()); + assert!(cached.complete.is_none()); } } diff --git a/crates/parsigex/src/behaviour.rs b/crates/parsigex/src/behaviour.rs index 779df314..8dc6806c 100644 --- a/crates/parsigex/src/behaviour.rs +++ b/crates/parsigex/src/behaviour.rs @@ -321,8 +321,6 @@ impl Behaviour { return; } }; - // Share a single refcounted buffer across all broadcast targets instead - // of deep-copying the encoded bytes once per peer. let message: Arc<[u8]> = Arc::from(message); let peers: Vec<_> = self @@ -334,11 +332,9 @@ impl Behaviour { .collect(); let mut pending_peers = HashSet::new(); let mut failure = None; - // Acquire the peer-store read guard once for the whole broadcast rather - // than re-locking per peer. Clone the cheap `Arc`-backed context handle - // so the guard does not keep `self` borrowed while the loop mutably - // borrows other `self` fields (`pending_events`, via - // `emit_broadcast_error`). + // Clone the cheap `Arc`-backed context so the peer-store guard (held + // once for the whole broadcast) does not keep `self` borrowed while the + // loop mutably borrows other `self` fields via `emit_broadcast_error`. let p2p_context = self.config.p2p_context.clone(); let peer_store = p2p_context.peer_store_lock(); for peer in peers {