From 69a23e283dbb8961bccfe6018be8c29916dd3aad Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Thu, 9 Jul 2026 18:58:41 +0100 Subject: [PATCH 1/2] avoid loopback event --- Cargo.lock | 20 +- Cargo.toml | 6 +- crates/beacon_state/tile/src/tile/block.rs | 2 + crates/bin/src/main.rs | 3 + crates/common/src/gossip.rs | 102 +++++++ crates/common/src/identity.rs | 4 + crates/common/src/lib.rs | 4 +- crates/common/src/rpc_rate_limit.rs | 4 + crates/common/src/spine/messages.rs | 10 + crates/common/src/spine/stream_protocol.rs | 4 +- crates/common/src/ticker.rs | 4 + crates/control/src/tile.rs | 315 ++++++++++++++------- crates/e2e/src/stack.rs | 7 + crates/e2e/src/utils.rs | 1 + crates/gossip/src/control.rs | 9 +- crates/gossip/src/{tile.rs => handler.rs} | 77 ++++- crates/gossip/src/lib.rs | 4 +- crates/gossip/src/mcache.rs | 44 ++- crates/gossip/src/message.rs | 11 +- crates/network/src/p2p/mod.rs | 2 +- crates/network/src/p2p/quic/peer.rs | 43 ++- crates/network/src/socket/linux.rs | 2 + crates/network/src/socket/mod.rs | 3 + crates/peer/src/counters.rs | 62 ++++ crates/peer/src/database.rs | 8 +- crates/peer/src/lib.rs | 1 + crates/peer/src/manager.rs | 197 ++++++++++--- crates/peer/src/manager/rpc.rs | 21 -- crates/storage/src/store.rs | 13 + crates/storage/src/store/io.rs | 8 +- crates/storage/src/tile.rs | 34 ++- crates/surfer/src/app.rs | 40 ++- crates/surfer/src/discovery.rs | 3 +- crates/surfer/src/render/counters_pane.rs | 32 ++- crates/surfer/src/schema.rs | 32 +++ crates/surfer/src/sources/counters.rs | 31 ++ docs/spine-message-flow.md | 12 +- 37 files changed, 930 insertions(+), 245 deletions(-) rename crates/gossip/src/{tile.rs => handler.rs} (75%) create mode 100644 crates/peer/src/counters.rs diff --git a/Cargo.lock b/Cargo.lock index 273fe55a..c95ec7c7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1487,7 +1487,7 @@ dependencies = [ [[package]] name = "flux" version = "0.0.38" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "bitcode", "core_affinity", @@ -1508,11 +1508,13 @@ dependencies = [ [[package]] name = "flux-communication" version = "0.0.27" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "directories", "flux-timing", "flux-utils", + "libc", + "mio", "rand 0.9.4", "shared_memory", "thiserror 1.0.69", @@ -1522,7 +1524,7 @@ dependencies = [ [[package]] name = "flux-profiler" version = "0.0.1" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "bytesize", "clap", @@ -1541,7 +1543,7 @@ dependencies = [ [[package]] name = "flux-profiler-macros" version = "0.0.1" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "quote", "syn", @@ -1550,7 +1552,7 @@ dependencies = [ [[package]] name = "flux-timing" version = "0.0.16" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "bitcode", "chrono", @@ -1567,7 +1569,7 @@ dependencies = [ [[package]] name = "flux-utils" version = "0.0.23" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "bytes", "core_affinity", @@ -4844,7 +4846,7 @@ dependencies = [ [[package]] name = "spine-derive" version = "0.0.10" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "proc-macro2", "quote", @@ -5336,12 +5338,12 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "type-hash" version = "0.0.1" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" [[package]] name = "type-hash-derive" version = "0.0.4" -source = "git+https://github.com/gattaca-com/flux?rev=ceddf228cdcda60bd134ddb49cc0a8375a09b853#ceddf228cdcda60bd134ddb49cc0a8375a09b853" +source = "git+https://github.com/gattaca-com/flux?rev=2d1b4691a93842dd0cc602cf668ad8b9f71f3904#2d1b4691a93842dd0cc602cf668ad8b9f71f3904" dependencies = [ "proc-macro-crate", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 48c33b11..90e310db 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,9 +73,9 @@ silver_network = {path = "crates/network" } silver_peer = {path = "crates/peer" } silver_storage = { path = "crates/storage" } silver_engine = { path = "crates/engine"} -flux = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853"} -flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853", features = ["bytes"]} -flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853"} +flux = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904"} +flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904", features = ["bytes"]} +flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904"} # External backtrace = "0.3" diff --git a/crates/beacon_state/tile/src/tile/block.rs b/crates/beacon_state/tile/src/tile/block.rs index ffe73450..a9435a38 100644 --- a/crates/beacon_state/tile/src/tile/block.rs +++ b/crates/beacon_state/tile/src/tile/block.rs @@ -60,6 +60,8 @@ impl BeaconStateTile { ?source, head_slot = self.head_state_slot(), block_slot, + wall_slot = self.ticker.current_slot(), + time_into_slot = ?self.ticker.slot_time_elapsed(), "applied block: {:?}", f ); diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index e3f0f3f9..fbaddc66 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -68,6 +68,8 @@ fn main() -> Result<(), Box> { incoming_rpc_producer.cache_ref().random_access("ds_persist_incoming_rpc", true)?; let incoming_rpc_consumer_eng = incoming_rpc_producer.cache_ref().random_access("eng_incoming_rpc", true)?; + let incoming_rpc_consumer_ctl = + incoming_rpc_producer.cache_ref().random_access("ctl_incoming_rpc", true)?; let incoming_engine_resp_producer = TCache::producer( "incoming_engine_resp", config.engine_config().incoming_engine_resp_tcache_size, @@ -185,6 +187,7 @@ fn main() -> Result<(), Box> { gossip_handler, outgoing_rpc_producer.clone(), spec.clone(), + incoming_rpc_consumer_ctl, ); // A finalized checkpoint state is mandatory (no genesis or runtime sync): diff --git a/crates/common/src/gossip.rs b/crates/common/src/gossip.rs index f2c92395..940426c5 100644 --- a/crates/common/src/gossip.rs +++ b/crates/common/src/gossip.rs @@ -73,12 +73,98 @@ impl From for String { } } +pub const ATTESTATION_SUBNETS: usize = 64; +pub const SYNC_COMMITTEE_SUBNETS: usize = 4; + +const ATTESTATION_BASE: usize = 2; +const VOLUNTARY_EXIT_SLOT: usize = ATTESTATION_BASE + ATTESTATION_SUBNETS; +const SYNC_CONTRIB_SLOT: usize = VOLUNTARY_EXIT_SLOT + 3; +const SYNC_COMMITTEE_BASE: usize = SYNC_CONTRIB_SLOT + 1; +const LC_FINALITY_SLOT: usize = SYNC_COMMITTEE_BASE + SYNC_COMMITTEE_SUBNETS; +const DATA_COLUMN_BASE: usize = LC_FINALITY_SLOT + 3; +const EXECUTION_BID_SLOT: usize = DATA_COLUMN_BASE + crate::NUMBER_OF_CUSTODY_GROUPS as usize; + +/// Dense per-topic slot space for counters/telemetry: subnet topics get one +/// slot per subnet, everything else one slot. The layout is fixed so +/// observers label by the same arithmetic (`gossip_topic_for_counter_slot`). +pub const GOSSIP_TOPIC_COUNTER_SLOTS: usize = EXECUTION_BID_SLOT + 5; + +/// Inverse of `GossipTopic::counter_slot`, for observer-side labelling. +pub fn gossip_topic_for_counter_slot(slot: usize) -> Option { + Some(match slot { + 0 => GossipTopic::BeaconBlock, + 1 => GossipTopic::BeaconAggregateAndProof, + s if s < VOLUNTARY_EXIT_SLOT => { + GossipTopic::BeaconAttestation((s - ATTESTATION_BASE) as u64) + } + s if s == VOLUNTARY_EXIT_SLOT => GossipTopic::VoluntaryExit, + s if s == VOLUNTARY_EXIT_SLOT + 1 => GossipTopic::ProposerSlashing, + s if s == VOLUNTARY_EXIT_SLOT + 2 => GossipTopic::AttesterSlashing, + s if s == SYNC_CONTRIB_SLOT => GossipTopic::SyncCommitteeContributionAndProof, + s if s < LC_FINALITY_SLOT => GossipTopic::SyncCommittee((s - SYNC_COMMITTEE_BASE) as u64), + s if s == LC_FINALITY_SLOT => GossipTopic::LightClientFinalityUpdate, + s if s == LC_FINALITY_SLOT + 1 => GossipTopic::LightClientOptimisticUpdate, + s if s == LC_FINALITY_SLOT + 2 => GossipTopic::BlsToExecutionChange, + s if s < EXECUTION_BID_SLOT => { + GossipTopic::DataColumnSidecar((s - DATA_COLUMN_BASE) as u64) + } + s if s == EXECUTION_BID_SLOT => GossipTopic::ExecutionPayloadBid, + s if s == EXECUTION_BID_SLOT + 1 => GossipTopic::ExecutionPayload, + s if s == EXECUTION_BID_SLOT + 2 => GossipTopic::PayloadAttestationMessage, + s if s == EXECUTION_BID_SLOT + 3 => GossipTopic::ProposerPreferences, + s if s == EXECUTION_BID_SLOT + 4 => GossipTopic::InclusionList, + _ => return None, + }) +} + impl GossipTopic { /// Format the full wire topic `/eth2/{fork_digest_hex}/{name}/ssz_snappy`. pub fn to_wire(&self, fork_digest_hex: &str) -> String { format!("/eth2/{fork_digest_hex}/{self}/ssz_snappy") } + /// Slot in the dense counter space; out-of-range subnet ids clamp to + /// their kind's last slot. + pub fn counter_slot(self) -> usize { + fn subnet(base: usize, id: u64, count: usize) -> usize { + base + (id as usize).min(count - 1) + } + match self { + Self::BeaconBlock => 0, + Self::BeaconAggregateAndProof => 1, + Self::BeaconAttestation(id) => subnet(ATTESTATION_BASE, id, ATTESTATION_SUBNETS), + Self::VoluntaryExit => VOLUNTARY_EXIT_SLOT, + Self::ProposerSlashing => VOLUNTARY_EXIT_SLOT + 1, + Self::AttesterSlashing => VOLUNTARY_EXIT_SLOT + 2, + Self::SyncCommitteeContributionAndProof => SYNC_CONTRIB_SLOT, + Self::SyncCommittee(id) => subnet(SYNC_COMMITTEE_BASE, id, SYNC_COMMITTEE_SUBNETS), + Self::LightClientFinalityUpdate => LC_FINALITY_SLOT, + Self::LightClientOptimisticUpdate => LC_FINALITY_SLOT + 1, + Self::BlsToExecutionChange => LC_FINALITY_SLOT + 2, + Self::DataColumnSidecar(id) => { + subnet(DATA_COLUMN_BASE, id, crate::NUMBER_OF_CUSTODY_GROUPS as usize) + } + Self::ExecutionPayloadBid => EXECUTION_BID_SLOT, + Self::ExecutionPayload => EXECUTION_BID_SLOT + 1, + Self::PayloadAttestationMessage => EXECUTION_BID_SLOT + 2, + Self::ProposerPreferences => EXECUTION_BID_SLOT + 3, + Self::InclusionList => EXECUTION_BID_SLOT + 4, + } + } + + pub fn p3_scored(&self) -> bool { + matches!( + self, + Self::BeaconBlock | + Self::BeaconAggregateAndProof | + Self::BeaconAttestation(_) | + Self::SyncCommitteeContributionAndProof | + Self::SyncCommittee(_) | + Self::ExecutionPayload | + Self::PayloadAttestationMessage + ) + } + /// Parse the full wire topic `/eth2/{fork_digest_hex}/{name}/ssz_snappy`. /// Verifies the envelope and that the fork digest matches /// `fork_digest_hex`. @@ -245,4 +331,20 @@ mod tests { "fork digest mismatch must fail" ); } + + #[test] + fn counter_slots_round_trip() { + for slot in 0..GOSSIP_TOPIC_COUNTER_SLOTS { + let topic = gossip_topic_for_counter_slot(slot) + .unwrap_or_else(|| panic!("slot {slot} has no topic")); + assert_eq!(topic.counter_slot(), slot, "{topic:?}"); + } + assert!(gossip_topic_for_counter_slot(GOSSIP_TOPIC_COUNTER_SLOTS).is_none()); + + // Out-of-range subnets clamp to their kind's last slot. + assert_eq!( + GossipTopic::BeaconAttestation(9999).counter_slot(), + GossipTopic::BeaconAttestation(ATTESTATION_SUBNETS as u64 - 1).counter_slot(), + ); + } } diff --git a/crates/common/src/identity.rs b/crates/common/src/identity.rs index 0ce9d156..01398ce2 100644 --- a/crates/common/src/identity.rs +++ b/crates/common/src/identity.rs @@ -208,6 +208,10 @@ impl Identify { (self, keypair).into() } + pub fn user_agent(&self) -> &str { + str::from_utf8(&self.user_agent[..self.user_agent_len]).unwrap_or("unknown") + } + /// Walk the four eth2-shaped address fields, yielding one `Eth2Addr` /// per populated slot. fn address_iter(&self) -> impl Iterator + '_ { diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 7370d47d..5bf03a44 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -3,8 +3,8 @@ extern crate self as silver_common; pub use crate::{ error::Error, gossip::{ - GossipTopic, MESSAGE_ID_LEN, MessageId, MessageIdHasher, msg_id_invalid_snappy, - msg_id_valid_snappy, + GOSSIP_TOPIC_COUNTER_SLOTS, GossipTopic, MESSAGE_ID_LEN, MessageId, MessageIdHasher, + gossip_topic_for_counter_slot, msg_id_invalid_snappy, msg_id_valid_snappy, }, id::{Keypair, PeerId, decode_protobuf_pubkey, encode_secp256k1_protobuf}, identity::{ diff --git a/crates/common/src/rpc_rate_limit.rs b/crates/common/src/rpc_rate_limit.rs index a24c04fa..f6837094 100644 --- a/crates/common/src/rpc_rate_limit.rs +++ b/crates/common/src/rpc_rate_limit.rs @@ -202,6 +202,10 @@ mod tests { ); assert_eq!( outbound.peek_outbound(protocol, 128, now + Duration::from_secs(30)), + RpcRateLimit::TooSoon + ); + assert_eq!( + outbound.peek_outbound(protocol, 128, now + Duration::from_secs(60)), RpcRateLimit::Allowed ); } diff --git a/crates/common/src/spine/messages.rs b/crates/common/src/spine/messages.rs index 440983f6..2acdd8c1 100644 --- a/crates/common/src/spine/messages.rs +++ b/crates/common/src/spine/messages.rs @@ -379,6 +379,16 @@ pub enum PeerEvent { p2p_peer: usize, iwant: TCacheRead, }, + /// A data column sidecar validated from a non-gossip source (RPC + /// by-root / EL blobs). Control re-publishes it on its subnet: the + /// gossip handler wraps the SSZ (a ref into `incoming_rpc`) as + /// protobuf and PM fans it out to the topic mesh, excluding + /// `originator` (the peer that served it to us). + PublishDataColumn { + originator: P2pStreamId, + topic: GossipTopic, + ssz: TCacheRead, + }, /// Emitted in order to trigger sending of a gossip message. /// Peer manager will generate select peers to send to. SendGossip { diff --git a/crates/common/src/spine/stream_protocol.rs b/crates/common/src/spine/stream_protocol.rs index ae8d7d03..8e8dacc5 100644 --- a/crates/common/src/spine/stream_protocol.rs +++ b/crates/common/src/spine/stream_protocol.rs @@ -108,10 +108,10 @@ impl StreamProtocol { Self::BeaconBlocksByRoot | Self::ExecutionPayloadEnvelopesByRange | Self::ExecutionPayloadEnvelopesByRoot => { - Some(RpcQuota::n_every(MAX_BLOCK_RATE_LIMIT_TOKENS, 30)) // 30s appears to be the default period of Prysm. Lighthouse is 10s. + Some(RpcQuota::n_every(MAX_BLOCK_RATE_LIMIT_TOKENS, 60)) // TODO lighthouse permits per 10s } Self::DataColumnSidecarsByRange | Self::DataColumnSidecarsByRoot => { - Some(RpcQuota::n_every(MAX_SIDECAR_RATE_LIMIT_TOKENS, 15)) + Some(RpcQuota::n_every(MAX_SIDECAR_RATE_LIMIT_TOKENS, 120)) // TODO, lighthouse permits per 30s } } } diff --git a/crates/common/src/ticker.rs b/crates/common/src/ticker.rs index 7c2e5bc0..7a1ce26f 100644 --- a/crates/common/src/ticker.rs +++ b/crates/common/src/ticker.rs @@ -101,6 +101,10 @@ impl SlotTicker { self.anchor_genesis_ms + self.anchor.elapsed().as_millis() as u64 } + pub fn slot_time_elapsed(&self) -> Duration { + Duration::from_millis(self.since_genesis_ms() % self.slot_ms) + } + pub fn current_slot(&self) -> u64 { self.millis_since_genesis() / self.slot_ms } diff --git a/crates/control/src/tile.rs b/crates/control/src/tile.rs index 4400e43d..5c7491e7 100644 --- a/crates/control/src/tile.rs +++ b/crates/control/src/tile.rs @@ -27,6 +27,9 @@ pub struct Controller { /// (peer-pick, caps, send) and owns column sync. sync_engine: SyncEngine, rpc_producer: TMultiProducer, + /// Reads `incoming_rpc` sidecar payloads referenced by + /// `PeerEvent::PublishDataColumn`. + rpc_ssz_consumer: TRandomAccess, last_tick: Instant, last_ping: Instant, last_status: Instant, @@ -49,6 +52,7 @@ impl Controller { gossip_handler: GossipHandler, rpc_producer: TMultiProducer, spec: Arc, + rpc_ssz_consumer: TRandomAccess, ) -> Self { let sync_engine = SyncEngine::new( peer_manager.syncing_config(), @@ -61,6 +65,7 @@ impl Controller { gossip_handler, sync_engine, rpc_producer, + rpc_ssz_consumer, last_tick: Instant::now(), last_ping: Instant::now(), last_status: Instant::now(), @@ -118,12 +123,11 @@ impl Controller { impl Tile for Controller { fn loop_body(&mut self, adapter: &mut SpineAdapter) { let now = Instant::now(); + self.rpc_ssz_consumer.free(); // Local status must land before the sync drive below: issuance is // capped against the imported head, and a one-loop-stale watermark - // stalls lock-step tests (and costs a loop of latency live). This - // block also cannot sit after `handle_peer_control` is created — the - // closure holds `self.gossip_handler` for the rest of the body. + // stalls lock-step tests (and costs a loop of latency live). let mut latest_status_event = None; adapter.consume(|beacon_event: BeaconStateEvent, _producers| match beacon_event { BeaconStateEvent::Status { ssz, latest_block_slot, wall_slot, .. } => { @@ -149,99 +153,79 @@ impl Tile for Controller { }); let fork_digest_changed = self.handle_latest_status(now, latest_status_event); + if fork_digest_changed { + tracing::info!("fork digest changed; re-announcing gossip subscriptions"); + self.peer_manager + .fan_out_subscriptions(&mut |evt| handle_peer_control(&mut self.gossip_handler, + &mut self.rpc_producer, evt, &mut adapter.producers)); + } - let mut handle_peer_control = |pc: PeerControl, producers: &mut SilverSpineProducers| { - self.gossip_handler.handle_peer_control(pc, &mut |event| { - if let GossipHandlerEvent::SendGossip(gossip_msg_out) = event { - producers.p2p_send.produce(&P2pSend::Gossip(gossip_msg_out).into()); - } - }); - - match pc { - PeerControl::P2pSend(send) => { - producers.p2p_send.produce(&send.into()); - } - PeerControl::P2pBlockByRootRequest { app_id, peer, block_root } => { - match allocate_blocks_by_root(&mut self.rpc_producer, &block_root) { - Ok(read) => { - producers.p2p_send.produce( - &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { - application_id: app_id, - peer, - request: silver_common::RpcRequest::BlockByRoot(read), - })) - .into(), - ); - } - Err(e) => { - tracing::error!(?e, "failed to allocate blocks by root request"); - } - } - } - PeerControl::P2pEnvelopeByRootRequest { app_id, peer, block_root } => { - match allocate_blocks_by_root(&mut self.rpc_producer, &block_root) { - Ok(read) => { - producers.p2p_send.produce( - &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { - application_id: app_id, - peer, - request: - silver_common::RpcRequest::ExecutionPayloadEnvelopesByRoot( - read, - ), - })) - .into(), + adapter.consume(|event: PeerEvent, producers| { + if let PeerEvent::PublishDataColumn { originator, topic, ssz } = event { + let read = self.rpc_ssz_consumer.acquire(ssz); + match read.buffer() { + Ok((bytes, _)) => { + if let Some((msg_hash, protobuf)) = + self.gossip_handler.publish(topic, bytes) + { + self.peer_manager.handle_event( + PeerEvent::SendGossip { + originator_stream_id: originator, + topic, + msg_hash, + recv_ts: Nanos::now(), + protobuf, + }, + now, + &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + producers, + ) + }, ); } - Err(e) => { - tracing::error!(?e, "failed to allocate envelopes by root request"); - } } + Err(e) => tracing::warn!(?e, ?topic, "publish column ssz read failed"), } - PeerControl::P2pDataColumnsRequest { app_id, peer, block_root, columns } => { - tracing::debug!(peer, columns, "emit data columns P2pSend"); - match allocate_data_columns_by_root( - &mut self.rpc_producer, - &block_root, - columns, - ) { - Ok(tcache) => { - producers.p2p_send.produce( - &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { - application_id: app_id, - peer, - request: silver_common::RpcRequest::DataColumnsByRoot(tcache), - })) - .into(), - ); - } - Err(e) => { - tracing::error!(?e, "failed to allocate data columns request"); - } - }; - } - other => { - producers.peer_control.produce(&other.into()); - } + return; } - }; - - if fork_digest_changed { - tracing::info!("fork digest changed; re-announcing gossip subscriptions"); - self.peer_manager - .fan_out_subscriptions(&mut |evt| handle_peer_control(evt, &mut adapter.producers)); - } - adapter.consume(|event: PeerEvent, producers| { self.sync_engine.peer_event(event, now, self.peer_manager.our_fork_digest()); - self.peer_manager - .handle_event(event, now, &mut |evt| handle_peer_control(evt, producers)); + + if let PeerEvent::SendGossip { + originator_stream_id: _, + topic, + msg_hash, + recv_ts: _, + protobuf, + } = &event + { + self.gossip_handler.mcache_insert(*msg_hash, *topic, *protobuf); + } + + self.peer_manager.handle_event(event, now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + producers, + ) + }); }); adapter.consume(|rpc: RpcInbound, producers| { self.sync_engine.rpc_event(now, &rpc, self.peer_manager.our_fork_digest()); - self.peer_manager - .on_rpc_inbound(rpc, now, &mut |evt| handle_peer_control(evt, producers)); + self.peer_manager.on_rpc_inbound(rpc, now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + producers, + ) + }); }); // Target selection: the engine is authoritative. Produce its target @@ -264,10 +248,22 @@ impl Tile for Controller { // run on the periodic 300s heartbeat. if self.sync_engine.take_just_synced() { self.last_status = now; - self.peer_manager - .fan_out_status(now, &mut |evt| handle_peer_control(evt, &mut adapter.producers)); - self.peer_manager - .fan_out_subscriptions(&mut |evt| handle_peer_control(evt, &mut adapter.producers)); + self.peer_manager.fan_out_status(now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); + self.peer_manager.fan_out_subscriptions(&mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); } // Sync issuance: the engine (Syncing or Following) decides the ranges + @@ -279,19 +275,36 @@ impl Tile for Controller { if self.last_drain.elapsed() > OUTBOUND_DRAIN_INTERVAL { self.last_drain = now; self.peer_manager.drain_pending_outbound(now, &mut |evt| { - handle_peer_control(evt, &mut adapter.producers) + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) }); } if self.last_tick.elapsed() > Duration::from_millis(700) { self.last_tick = now; - self.peer_manager - .tick(now, &mut |evt| handle_peer_control(evt, &mut adapter.producers)); + self.peer_manager.tick(now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); if self.auto_ping && self.last_ping.elapsed() > Duration::from_secs(17) { self.last_ping = now; - self.peer_manager - .fan_out_ping(now, &mut |evt| handle_peer_control(evt, &mut adapter.producers)); + self.peer_manager.fan_out_ping(now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); } } @@ -300,6 +313,8 @@ impl Tile for Controller { for peer in self.peer_manager.live_peers_with_status() { if let Some(enr) = peer.enr.as_ref() { handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, PeerControl::PersistPeer { enr: *enr }, &mut adapter.producers, ); @@ -315,19 +330,115 @@ impl Tile for Controller { self.sync_engine.fell_behind() && self.last_status.elapsed() > Duration::from_secs(1); if fell_behind || self.last_status.elapsed() > Duration::from_secs(30) { self.last_status = now; - self.peer_manager - .fan_out_status(now, &mut |evt| handle_peer_control(evt, &mut adapter.producers)); + self.peer_manager.fan_out_status(now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); } - if self.gossip_handler.spin(&mut |event| match event { - GossipHandlerEvent::PeerEvent(peer_event) => adapter.produce(peer_event), - GossipHandlerEvent::NewGossip(new_gossip_msg) => adapter.produce(new_gossip_msg), - GossipHandlerEvent::SendGossip(gossip_msg_out) => { - adapter.produce(P2pSend::Gossip(gossip_msg_out)) - } - }) { + if self.gossip_handler.spin() { adapter.mark_work(); } + while let Some(event) = self.gossip_handler.pop_event() { + match event { + GossipHandlerEvent::PeerEvent(peer_event) => { + self.sync_engine.peer_event( + peer_event, + now, + self.peer_manager.our_fork_digest(), + ); + self.peer_manager.handle_event(peer_event, now, &mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); + } + GossipHandlerEvent::NewGossip(new_gossip_msg) => adapter.produce(new_gossip_msg), + GossipHandlerEvent::SendGossip(gossip_msg_out) => { + adapter.produce(P2pSend::Gossip(gossip_msg_out)) + } + } + } + } +} + +fn handle_peer_control( + gossip_handler: &mut GossipHandler, + rpc_producer: &mut TMultiProducer, + pc: PeerControl, + producers: &mut SilverSpineProducers, +) { + gossip_handler.handle_peer_control(pc); + + match pc { + PeerControl::P2pSend(send) => { + producers.p2p_send.produce(&send.into()); + } + PeerControl::P2pBlockByRootRequest { app_id, peer, block_root } => { + match allocate_blocks_by_root(rpc_producer, &block_root) { + Ok(read) => { + producers.p2p_send.produce( + &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { + application_id: app_id, + peer, + request: silver_common::RpcRequest::BlockByRoot(read), + })) + .into(), + ); + } + Err(e) => { + tracing::error!(?e, "failed to allocate blocks by root request"); + } + } + } + PeerControl::P2pDataColumnsRequest { app_id, peer, block_root, columns } => { + tracing::debug!(peer, columns, "emit data columns P2pSend"); + match allocate_data_columns_by_root(rpc_producer, &block_root, columns) { + Ok(tcache) => { + producers.p2p_send.produce( + &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { + application_id: app_id, + peer, + request: silver_common::RpcRequest::DataColumnsByRoot(tcache), + })) + .into(), + ); + } + Err(e) => { + tracing::error!(?e, "failed to allocate data columns request"); + } + }; + } + PeerControl::P2pEnvelopeByRootRequest { app_id, peer, block_root } => { + match allocate_blocks_by_root(rpc_producer, &block_root) { + Ok(read) => { + producers.p2p_send.produce( + &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { + application_id: app_id, + peer, + request: + silver_common::RpcRequest::ExecutionPayloadEnvelopesByRoot( + read, + ), + })) + .into(), + ); + } + Err(e) => { + tracing::error!(?e, "failed to allocate envelopes by root request"); + } + } + } + other => { + producers.peer_control.produce(&other.into()); + } } } diff --git a/crates/e2e/src/stack.rs b/crates/e2e/src/stack.rs index 1f21427a..f1dfe40d 100644 --- a/crates/e2e/src/stack.rs +++ b/crates/e2e/src/stack.rs @@ -212,6 +212,8 @@ impl PublisherStack { // (multipart-RPC) read via `rpc_in_ra`. Regular consumer also // attached for keep-alive plus future controller use. let rpc_in_producer = TCache::producer("e2e_stack", TCACHE_SIZE); + let rpc_in_ctl = + rpc_in_producer.cache_ref().random_access("ctl_e2e", true).expect("ctl rpc ra"); let rpc_in_consumer = rpc_in_producer.cache_ref().consumer("e2e").ok(); let rpc_in_ra = rpc_in_producer.cache_ref().random_access("e2e", true).expect("rpc_in random_access"); @@ -272,6 +274,7 @@ impl PublisherStack { .unwrap(), TCache::multi_producer("dummy_rpc_out", 32), // dummpy rpc out Arc::new(SpecConfig::mainnet()), + rpc_in_ctl, ); // Spine + per-tile adapters. @@ -382,6 +385,10 @@ impl EchoStack { compression, TCache::multi_producer("dummy_rpc_out", 32), // dummpy rpc out Arc::new(SpecConfig::mainnet()), + TCache::producer("ctl_rpc_in_dummy", 32) + .cache_ref() + .random_access("ctl_e2e", true) + .expect("ctl rpc ra"), ); let mut spine = SilverSpine::new_with_base_dir(base_dir, Some(path_suffix)); diff --git a/crates/e2e/src/utils.rs b/crates/e2e/src/utils.rs index f5b69292..461a8a45 100644 --- a/crates/e2e/src/utils.rs +++ b/crates/e2e/src/utils.rs @@ -184,6 +184,7 @@ impl PmBsHarness { gossip_handler, TCache::multi_producer("rpc_out_dummy", 32), Arc::new(SpecConfig::mainnet()), + rpc_p.cache_ref().random_access("ctl_test", true).expect("ctl rpc ra"), ); let mut ctl_a = SpineAdapter::connect_tile(&ctl, &mut spine); diff --git a/crates/gossip/src/control.rs b/crates/gossip/src/control.rs index ee7bbf34..927d5656 100644 --- a/crates/gossip/src/control.rs +++ b/crates/gossip/src/control.rs @@ -106,7 +106,14 @@ pub(super) fn handle_iwants<'a>( tcache, })); } else { - tracing::warn!(?stream_id, "WANT message not in cache"); + match mcache.history(&hash) { + Some(ts) => { + tracing::info!(?stream_id, ?hash, elapsed=?ts.elapsed(), "IWANT for message > 8.4s old"); + } + None => { + tracing::warn!(?stream_id, ?hash, "WANT message not in cache"); + } + } } } } diff --git a/crates/gossip/src/tile.rs b/crates/gossip/src/handler.rs similarity index 75% rename from crates/gossip/src/tile.rs rename to crates/gossip/src/handler.rs index 731efb26..0dd16787 100644 --- a/crates/gossip/src/tile.rs +++ b/crates/gossip/src/handler.rs @@ -1,9 +1,9 @@ -use std::time::Instant; +use std::{collections::VecDeque, time::Instant}; use buffa::MessageView; use silver_common::{ - Error, GossipMsgOut, MessageId, P2pStreamId, PeerControl, PeerEvent, TCacheProducer, TConsumer, - TProducer, ssz_view::StatusView, + Error, GossipMsgOut, GossipTopic, MessageId, P2pStreamId, PeerControl, PeerEvent, + TCacheProducer, TCacheRead, TConsumer, TProducer, msg_id_valid_snappy, ssz_view::StatusView, }; use crate::{ @@ -15,7 +15,7 @@ use crate::{ dedup::DedupCache, generated::RPCView, mcache::MessageCache, - message::handle_incoming, + message::{copy_compressed_to_protobuf_output, handle_incoming}, }; /// Reads all incoming gossip protobuf messages (sequential consumer): @@ -39,6 +39,12 @@ pub struct GossipHandler { // scratch buffer for iwant meessage ids. iwant_buffer: Vec, + + // snappy block-compression for locally published messages. + snap_encoder: snap::raw::Encoder, + snap_scratch: Vec, + + events: VecDeque, } impl GossipHandler { @@ -60,6 +66,9 @@ impl GossipHandler { mcache_publish: protobuf_gossip_publish, mcache, iwant_buffer: Vec::with_capacity(256), + snap_encoder: snap::raw::Encoder::new(), + snap_scratch: Vec::new(), + events: VecDeque::with_capacity(64), }) } @@ -85,11 +94,59 @@ impl GossipHandler { } } + /// Publish a locally-obtained, already-validated message on `topic`: + /// compress, wrap as protobuf into the outgoing tcache, register with + /// dedup + mcache (so gossip copies dedupe and IWANTs can be served). + /// Returns `None` when the message was already seen via gossip, or + /// pre-Status (no fork digest yet). + pub fn publish(&mut self, topic: GossipTopic, ssz: &[u8]) -> Option<(MessageId, TCacheRead)> { + if self.fork_digest_hex.is_empty() { + return None; + } + let wire = topic.to_wire(&self.fork_digest_hex); + self.snap_scratch.resize(snap::raw::max_compress_len(ssz.len()), 0); + let n = match self.snap_encoder.compress(ssz, &mut self.snap_scratch) { + Ok(n) => n, + Err(e) => { + tracing::error!(?e, ?topic, "publish snappy compress failed"); + return None; + } + }; + let msg_id = msg_id_valid_snappy(&wire, ssz); + let fast_hash = self.dedup_cache.contains_fast(&self.snap_scratch[..n])?; + if !self.dedup_cache.insert(fast_hash, msg_id) { + return None; + } + let read = copy_compressed_to_protobuf_output( + &mut self.mcache_publish, + &self.snap_scratch[..n], + &wire, + ) + .inspect_err(|e| tracing::error!(?e, ?topic, "publish protobuf write failed")) + .ok()?; + self.mcache.insert(msg_id, topic, read); + Some((msg_id, read)) + } + pub fn set_fork_digest(&mut self, status_ssz: &[u8; 92]) { self.fork_digest_hex = hex::encode(StatusView::fork_digest(status_ssz)); } - pub fn handle_peer_control( + pub fn handle_peer_control(&mut self, peer_control: PeerControl) { + let mut events = std::mem::take(&mut self.events); + self.handle_peer_control_inner(peer_control, &mut |e| events.push_back(e)); + self.events = events; + } + + pub fn mcache_insert(&mut self, id: MessageId, topic: GossipTopic, protobuf: TCacheRead) { + self.mcache.insert(id, topic, protobuf); + } + + pub fn pop_event(&mut self) -> Option { + self.events.pop_front() + } + + fn handle_peer_control_inner( &mut self, peer_control: PeerControl, emit: &mut impl FnMut(GossipHandlerEvent), @@ -151,7 +208,14 @@ impl GossipHandler { } } - pub fn spin(&mut self, emit: &mut impl FnMut(GossipHandlerEvent)) -> bool { + pub fn spin(&mut self) -> bool { + let mut events = std::mem::take(&mut self.events); + let did_work = self.spin_inner(&mut |e| events.push_back(e)); + self.events = events; + did_work + } + + fn spin_inner(&mut self, emit: &mut impl FnMut(GossipHandlerEvent)) -> bool { let mut did_work = false; let now = Instant::now(); self.dedup_cache.maybe_rotate(now); @@ -220,7 +284,6 @@ impl GossipHandler { &mut self.dedup_cache, &mut self.incoming_gossip_publish, &mut self.mcache_publish, - &mut self.mcache, emit, ) { tracing::error!( diff --git a/crates/gossip/src/lib.rs b/crates/gossip/src/lib.rs index 8fb42400..ce2a6333 100644 --- a/crates/gossip/src/lib.rs +++ b/crates/gossip/src/lib.rs @@ -5,16 +5,16 @@ mod dedup; #[rustfmt::skip] #[allow(clippy::all)] mod generated; +mod handler; mod mcache; mod message; -mod tile; pub use control::{ copy_grafts_to_protobuf_output, copy_prunes_to_protobuf_output, copy_subscribes_to_protobuf_output, copy_unsubscribes_to_protobuf_output, }; +pub use handler::GossipHandler; use silver_common::{GossipMsgOut, NewGossipMsg, PeerEvent}; -pub use tile::GossipHandler; /// Events emitted by the GossipHandler. #[allow(clippy::large_enum_variant)] diff --git a/crates/gossip/src/mcache.rs b/crates/gossip/src/mcache.rs index fb057cef..8ddf3128 100644 --- a/crates/gossip/src/mcache.rs +++ b/crates/gossip/src/mcache.rs @@ -5,7 +5,9 @@ use std::{ time::{Duration, Instant}, }; -use silver_common::{GossipTopic, MessageId, MessageIdHasher, TCacheRead, TRandomAccess, TRead}; +use silver_common::{ + GossipTopic, MessageId, MessageIdHasher, TCacheRead, TRandomAccess, TRead, Wheel, +}; /// Another rotating bucket cache. Each bucket optionally maps a message id to /// TCacheRead. This cache maintains a tail of the TCache containing the cached @@ -31,6 +33,7 @@ impl Default for Bucket { pub(crate) struct MessageCache { buckets: Box<[Bucket]>, + history: Wheel, // last 64 * 700ms cached messages = ~45seconds cache_consumer: TRandomAccess, current_bucket: usize, last_rotation: Instant, @@ -41,6 +44,7 @@ impl MessageCache { pub(crate) fn new(cache_consumer: TRandomAccess) -> Self { Self { buckets: vec![Bucket::default(); BUCKETS].into_boxed_slice(), + history: Wheel::new(ROTATION_INTERVAL), cache_consumer, current_bucket: 0, last_rotation: Instant::now(), @@ -56,6 +60,7 @@ impl MessageCache { bucket.ihaves.entry(topic).and_modify(|v| v.push(id)).or_insert_with(|| vec![id]); bucket.tcache_min_seq = bucket.tcache_min_seq.min(acquired.seq()); bucket.messages.insert(id, acquired); + self.history.insert(id, Instant::now()); } pub(crate) fn has(&self, id: &MessageId) -> bool { @@ -66,6 +71,10 @@ impl MessageCache { self.buckets.iter().find_map(|b| b.messages.get(id)).map(|a| &a.read).copied() } + pub(crate) fn history(&self, id: &MessageId) -> Option<&Instant> { + self.history.get(id) + } + pub(crate) fn get_ihaves( &self, topic: &GossipTopic, @@ -98,6 +107,7 @@ impl MessageCache { self.cache_consumer.free(); self.last_rotation = now; + self.history.maybe_rotate(now, &mut |_, _| true); } } } @@ -115,8 +125,8 @@ struct IHaveIterator<'a> { impl<'a> IHaveIterator<'a> { fn new(topic: GossipTopic, mcache: &'a MessageCache) -> Self { let mut len = 0; - for i in 0..3 { - let idx = (mcache.current_bucket + i) % BUCKETS; + for i in 0..IHAVE_BUCKETS { + let idx = (mcache.current_bucket + BUCKETS - i) % BUCKETS; len += mcache.buckets[idx].ihaves.get(&topic).map(|v| v.len()).unwrap_or_default(); } let iter = mcache.buckets[mcache.current_bucket].ihaves.get(&topic).map(|v| v.iter()); @@ -146,7 +156,7 @@ impl<'a> Iterator for IHaveIterator<'a> { None if self.buckets_left == 0 => return None, None => { self.buckets_left -= 1; - self.bucket = (self.bucket + 1) % BUCKETS; + self.bucket = (self.bucket + BUCKETS - 1) % BUCKETS; self.iter = self.mcache.buckets[self.bucket].ihaves.get(&self.topic).map(|v| v.iter()); } @@ -190,6 +200,32 @@ mod tests { assert!(matches!(mcache.get(&id), Some(_))); } + fn force_rotate(mcache: &mut MessageCache) { + mcache.last_rotation -= ROTATION_INTERVAL * 2; + mcache.maybe_rotate(Instant::now()); + } + + #[test] + fn ihaves_cover_three_most_recent_buckets() { + let (mut mcache, mut producer) = mk_mcache(); + + let ids: Vec<_> = (1u8..=4).map(|b| MessageId { id: [b; 20] }).collect(); + for (i, id) in ids.iter().enumerate() { + let tc = mk_tcache_read(&mut producer); + mcache.insert(*id, GossipTopic::BeaconBlock, tc); + if i < ids.len() - 1 { + force_rotate(&mut mcache); + } + } + assert_eq!(mcache.current_bucket, 3); + + let iter = mcache.get_ihaves(&GossipTopic::BeaconBlock); + assert_eq!(iter.len(), 3); + let mut ihaves: Vec<_> = iter.copied().collect(); + ihaves.sort_by_key(|m| m.id); + assert_eq!(ihaves, ids[1..]); + } + #[test] fn serve_iwant_unknown_for_missing_id() { let (mcache, _producer) = mk_mcache(); diff --git a/crates/gossip/src/message.rs b/crates/gossip/src/message.rs index e8585e06..83775137 100644 --- a/crates/gossip/src/message.rs +++ b/crates/gossip/src/message.rs @@ -10,10 +10,7 @@ use silver_common::{ TCacheRead, TProducer, TReservation, msg_id_invalid_snappy, msg_id_valid_snappy, }; -use crate::{ - GossipHandlerEvent, control::copy_idontwants_to_protobuf_output, dedup::DedupCache, - mcache::MessageCache, -}; +use crate::{GossipHandlerEvent, control::copy_idontwants_to_protobuf_output, dedup::DedupCache}; #[allow(clippy::too_many_arguments)] pub(super) fn handle_incoming( @@ -25,7 +22,6 @@ pub(super) fn handle_incoming( dedup_cache: &mut DedupCache, incoming_gossip_publish: &mut TProducer, mcache_publish: &mut TProducer, - mcache: &mut MessageCache, emit: &mut impl FnMut(GossipHandlerEvent), ) -> Result<(), Error> { // Fast duplicate check. @@ -84,9 +80,6 @@ pub(super) fn handle_incoming( } })?; - // Add message to message cache. - mcache.insert(msg_id, topic, mcache_read); - // Flush the reservation matching the gossip message available downstream. reservation.flush()?; @@ -128,7 +121,7 @@ fn decompress_to_reservation( Ok(msg_id) } -fn copy_compressed_to_protobuf_output( +pub(crate) fn copy_compressed_to_protobuf_output( producer: &mut TProducer, snappy_data: &[u8], topic: &str, diff --git a/crates/network/src/p2p/mod.rs b/crates/network/src/p2p/mod.rs index 93a50c3e..b52a7b44 100644 --- a/crates/network/src/p2p/mod.rs +++ b/crates/network/src/p2p/mod.rs @@ -267,7 +267,7 @@ impl P2p { pub fn enqueue_rpc_out(&mut self, msg: RpcOutbound, context: &mut Context) -> SendResult { match self.peers.get_mut(&ConnectionHandle(msg.peer_id())) { Some(peer) => { - tracing::debug!(protocol=?msg.protocol(), "enqueue outbound rpc request"); + tracing::debug!(protocol=?msg.protocol(), peer=msg.peer_id(), "enqueue outbound rpc request"); let acquired_msg = AcquiredRpcOutbound::from((msg, &mut context.rpc_consumer)); peer.send_rpc(acquired_msg) } diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index 8a2558d0..e3029014 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -730,11 +730,20 @@ impl OutBuffer { seq & (self.len - 1) } - /// Returns `true` if adding the new message overwrote an old message. + /// Returns `true` if adding the new message dropped the oldest queued + /// message. fn add_msg(&mut self, msg: T) -> bool { - let old_msg = self.msgs[self.pos(self.head)].replace(msg); + let dropped = self.head - self.tail == self.msgs.len(); + if dropped { + // Full: pos(head) == pos(tail), so the overwrite below replaces + // the oldest message. Advance tail with it — otherwise head/tail + // desync and is_empty() reports non-empty while pop() yields + // None, leaving the stream flagged needs_spin forever. + self.tail += 1; + } + self.msgs[self.pos(self.head)].replace(msg); self.head += 1; - old_msg.is_some() + dropped } /// Called when the current read is complete. @@ -744,7 +753,10 @@ impl OutBuffer { self.tail += 1; Some(msg) } - None => None, + None => { + debug_assert!(self.head == self.tail, "OutBuffer head/tail desync"); + None + } } } @@ -768,6 +780,29 @@ mod tests { const TCACHE_BYTES: usize = 64 * 1024; + #[test] + fn out_buffer_overflow_keeps_ring_consistent() { + let mut buf = OutBuffer::new(4); + for i in 0..4usize { + assert!(!buf.add_msg(i)); + } + + // Two overwrites drop the two oldest messages. + assert!(buf.add_msg(4)); + assert!(buf.add_msg(5)); + assert_eq!(buf.len(), 4); + + let drained: Vec<_> = std::iter::from_fn(|| buf.pop()).collect(); + assert_eq!(drained, vec![2, 3, 4, 5]); + assert!(buf.is_empty()); + assert!(buf.pop().is_none()); + + // Buffer must remain usable after an overflow episode. + assert!(!buf.add_msg(6)); + assert_eq!(buf.pop(), Some(6)); + assert!(buf.is_empty()); + } + /// Per-peer test plumbing. Owns the four tcaches plus the `Context` /// passed into `Peer::spin`. struct PeerHarness { diff --git a/crates/network/src/socket/linux.rs b/crates/network/src/socket/linux.rs index adb5dfa4..cf311fc3 100644 --- a/crates/network/src/socket/linux.rs +++ b/crates/network/src/socket/linux.rs @@ -249,6 +249,8 @@ impl TxBatch { return false; } tracing::error!(error=?err, "sendmmsg"); + // Try to avoid any issue with specifc packets + self.clear(); return false; } self.send_idx += ret as usize; diff --git a/crates/network/src/socket/mod.rs b/crates/network/src/socket/mod.rs index a9824d68..09a25c9a 100644 --- a/crates/network/src/socket/mod.rs +++ b/crates/network/src/socket/mod.rs @@ -111,6 +111,9 @@ impl Socket { F: FnMut(&mut Vec) -> Option, { let buf_idx = self.tx_batch.entries.len(); + if buf_idx >= self.tx_batch.bufs.len() { + return false; + } self.tx_batch.bufs[buf_idx].clear(); let Some(tx) = f(&mut self.tx_batch.bufs[buf_idx]) else { diff --git a/crates/peer/src/counters.rs b/crates/peer/src/counters.rs new file mode 100644 index 00000000..2f9ec069 --- /dev/null +++ b/crates/peer/src/counters.rs @@ -0,0 +1,62 @@ +use std::sync::atomic::{AtomicPtr, AtomicU64, Ordering}; + +use silver_common::{GOSSIP_TOPIC_COUNTER_SLOTS, GossipTopic, metrics::map_counters}; + +/// Subnet-resolved gossip traffic counters, shmem-backed like +/// `declare_counters!` output but indexed arithmetically: +/// `counter_slot(topic) * 4` is sent, `+ 1` is recv, `+ 2` is the mesh-size +/// gauge, `+ 3` is the subscribed-peers gauge. +pub struct GossipTopicCounters; + +pub const PER_TOPIC: usize = 4; + +const COUNT: usize = GOSSIP_TOPIC_COUNTER_SLOTS * PER_TOPIC; +const BYTES: usize = COUNT * size_of::(); + +impl GossipTopicCounters { + pub fn sent(topic: GossipTopic) { + Self::slot(topic.counter_slot() * PER_TOPIC).fetch_add(1, Ordering::Relaxed); + } + + pub fn recv(topic: GossipTopic) { + Self::slot(topic.counter_slot() * PER_TOPIC + 1).fetch_add(1, Ordering::Relaxed); + } + + pub fn mesh(topic: GossipTopic, size: usize) { + Self::slot(topic.counter_slot() * PER_TOPIC + 2).store(size as u64, Ordering::Relaxed); + } + + /// `counts` indexed by `GossipTopic::counter_slot`. + pub fn subscribed(counts: &[u16; GOSSIP_TOPIC_COUNTER_SLOTS]) { + for (slot, count) in counts.iter().enumerate() { + Self::slot(slot * PER_TOPIC + 3).store(*count as u64, Ordering::Relaxed); + } + } + + fn shmem() -> &'static AtomicPtr { + static SHMEM: AtomicPtr = AtomicPtr::new(std::ptr::null_mut()); + &SHMEM + } + + fn slot(i: usize) -> &'static AtomicU64 { + let mut p = Self::shmem().load(Ordering::Relaxed); + if p.is_null() { + p = Self::lazy_init(); + } + unsafe { &*p.add(i) } + } + + #[cold] + #[inline(never)] + fn lazy_init() -> *mut AtomicU64 { + map_counters( + flux::utils::directories::local_share_dir().as_ref(), + "silver", + "gossip_topics", + BYTES, + Self::shmem(), + ) + .unwrap_or_else(|e| panic!("GossipTopicCounters init failed: {e}")); + Self::shmem().load(Ordering::Relaxed) + } +} diff --git a/crates/peer/src/database.rs b/crates/peer/src/database.rs index a8aee8e4..f1fa2223 100644 --- a/crates/peer/src/database.rs +++ b/crates/peer/src/database.rs @@ -98,8 +98,8 @@ impl PeerDatabase { } } - pub fn peer_disconnected(&mut self, p2p_id: usize) { - self.by_p2p_id.remove(&p2p_id); + pub fn peer_disconnected(&mut self, p2p_id: usize) -> Option<&PeerRecord> { + self.by_p2p_id.remove(&p2p_id).and_then(|idx| self.peers.get(idx)) } pub fn p2p_status(&mut self, p2p_id: usize, status: PeerStatus, earliest_slot: Option) { @@ -188,6 +188,10 @@ impl PeerDatabase { .filter_map(|(_, idx)| self.peers.get(*idx)) .filter(|record| record.status.is_some()) } + + pub fn by_p2p_id(&self, p2p: usize) -> Option<&PeerRecord> { + self.by_p2p_id.get(&p2p).and_then(|idx| self.peers.get(*idx)) + } } fn status_bytes(s: &PeerStatus) -> &[u8] { diff --git a/crates/peer/src/lib.rs b/crates/peer/src/lib.rs index 96873791..80fbde95 100644 --- a/crates/peer/src/lib.rs +++ b/crates/peer/src/lib.rs @@ -1,3 +1,4 @@ +mod counters; mod database; mod manager; mod scoring; diff --git a/crates/peer/src/manager.rs b/crates/peer/src/manager.rs index 2af0eb55..85042344 100644 --- a/crates/peer/src/manager.rs +++ b/crates/peer/src/manager.rs @@ -520,10 +520,10 @@ impl PeerManager { self.on_unsubscribe(p2p_peer, topic, now, emit); } PeerEvent::P2pGossipTopicGraft { p2p_peer, topic } => { - self.on_remote_graft(p2p_peer, topic, now); + self.on_remote_graft(p2p_peer, topic, now, emit); } PeerEvent::P2pGossipTopicPrune { p2p_peer, topic } => { - self.on_remote_prune(p2p_peer, topic, now); + self.on_remote_prune(p2p_peer, topic, now, emit); } PeerEvent::P2pGossipHave { p2p_peer, topic: _, hash, already_seen } => { self.on_ihave(p2p_peer, hash, already_seen, now); @@ -558,6 +558,9 @@ impl PeerManager { PeerEvent::OutboundIHave { topic, msg_count: _, protobuf } => { self.on_outbound_ihave(topic, protobuf, emit); } + // Consumed by the control tile (gossip republish); PM sees only + // the SendGossip it turns into. + PeerEvent::PublishDataColumn { .. } => {} PeerEvent::OutboundIWant { p2p_peer, iwant } => { self.on_outbound_iwant(p2p_peer, iwant, emit); } @@ -582,9 +585,8 @@ impl PeerManager { tracing::trace!(p2p_peer, "Got peer metadata"); self.database.p2p_metadata(p2p_peer, metadata_ssz) } - PeerEvent::P2pPeerGoodbye { p2p_peer: _, status: _ } => { - // TODO send p2p disconnect - // TODO scoring based on status? + PeerEvent::P2pPeerGoodbye { p2p_peer, status } => { + self.on_p2p_peer_goodbye(p2p_peer, now, status, emit); } PeerEvent::P2pPeerIdentity { p2p_peer, identify } => { tracing::trace!(p2p_peer, ?identify, "Got peer identify"); @@ -628,6 +630,28 @@ impl PeerManager { // ── Cold path: tick ───────────────────────────────────────────────── pub fn tick(&mut self, now: Instant, emit: &mut impl FnMut(PeerControl)) { + // Mesh-size gauges refreshed here rather than at each mutation site — + // mesh entries persist (empty vecs stay), so this is exact. + for (topic, mesh_peers) in &self.mesh { + crate::counters::GossipTopicCounters::mesh(*topic, mesh_peers.len()); + } + // Peers announce SUBSCRIBE for all their subnets; only count + // subscribers on topics we participate in ourselves. + let mut ours = [false; silver_common::GOSSIP_TOPIC_COUNTER_SLOTS]; + for topic in &self.our_topics { + ours[topic.counter_slot()] = true; + } + let mut subs = [0u16; silver_common::GOSSIP_TOPIC_COUNTER_SLOTS]; + for peer in self.peers.values() { + for topic in &peer.topics { + let slot = topic.counter_slot(); + if ours[slot] { + subs[slot] = subs[slot].saturating_add(1); + } + } + } + crate::counters::GossipTopicCounters::subscribed(&subs); + // 1) Heartbeat rollover: reset per-heartbeat counters, sweep broken promises. if now.saturating_duration_since(self.last_heartbeat) >= self.params.heartbeat_interval { self.heartbeat(now); @@ -735,11 +759,13 @@ impl PeerManager { } } - fn on_disconnected(&mut self, conn: usize, now: Instant, emit: &mut impl FnMut(PeerControl)) { - tracing::debug!("removing disconnected peer: {conn}"); - let Some(mut state) = self.peers.remove(&conn) else { - return; - }; + fn on_disconnected( + &mut self, + conn: usize, + now: Instant, + emit: &mut impl FnMut(PeerControl), + ) -> Option<&PeerRecord> { + let mut state = self.peers.remove(&conn)?; // De-index IP colocation. if let Some(v) = self.ip_colocations.get_mut(&state.ip_prefix) { @@ -777,8 +803,8 @@ impl PeerManager { archived_at: now, }); - self.database.peer_disconnected(conn); self.on_range_peer_disconnected(conn, emit); + self.database.peer_disconnected(conn) } // ── Gossip event handlers ─────────────────────────────────────────── @@ -837,34 +863,70 @@ impl PeerManager { } } - /// Peer grafted us — they consider us in their mesh. From our side this - /// just means subsequent deliveries from them will count as mesh - /// deliveries (P3 tracking), which needs a timestamp. - fn on_remote_graft(&mut self, conn: usize, topic: GossipTopic, now: Instant) { - if let Some(peer) = self.peers.get_mut(&conn) { - let t = peer.topic_stats.entry(topic).or_default(); - t.meshed_since = Some(now); - t.mesh_active = false; // activates after grace window + fn on_remote_graft( + &mut self, + conn: usize, + topic: GossipTopic, + now: Instant, + emit: &mut impl FnMut(PeerControl), + ) { + let Some(peer) = self.peers.get(&conn) else { + if let Some(record) = self.database.by_p2p_id(conn) && + let Some(id) = record.peer_id + { + emit(PeerControl::P2pDisconnect { p2p: id, p2p_connection: conn }) + } + return; + }; + let peer_id = peer.peer_id; + let score = peer.cached_score; + let mesh_size = self.mesh.get(&topic).map(|m| m.len()).unwrap_or(0); + let accept = self.our_topics.contains(&topic) && + mesh_size < self.params.d_high as usize && + score >= self.params.gossip_threshold && + !self.is_backed_off(conn, topic, now); + if accept { + self.do_graft(conn, peer_id, topic, now, emit); + tracing::info!(p2p_peer = conn, ?topic, mesh_size, "PM peer GRAFTed us: accepted"); + } else { + self.do_prune(conn, peer_id, topic, now, emit); + tracing::info!(p2p_peer = conn, ?topic, mesh_size, "PM peer GRAFTed us: refused"); } - tracing::info!(p2p_peer = conn, ?topic, "PM peer GRAFTed us"); } - fn on_remote_prune(&mut self, conn: usize, topic: GossipTopic, now: Instant) { - if let Some(peer) = self.peers.get_mut(&conn) { - if let Some(t) = peer.topic_stats.get_mut(&topic) { - // Carry any active deficit into the failure penalty. - if t.mesh_active && - t.mesh_deliveries < self.params.mesh_message_deliveries_threshold - { - t.mesh_failure_penalty += - self.params.mesh_message_deliveries_threshold - t.mesh_deliveries; - } - t.meshed_since = None; - t.mesh_active = false; + fn on_remote_prune( + &mut self, + conn: usize, + topic: GossipTopic, + now: Instant, + emit: &mut impl FnMut(PeerControl), + ) { + let mut mesh_size = 0; + if let Some(mesh_peers) = self.mesh.get_mut(&topic) && + let Some(idx) = mesh_peers.iter().position(|c| *c == conn) + { + mesh_peers.swap_remove(idx); + mesh_size = mesh_peers.len(); + } + let Some(peer) = self.peers.get_mut(&conn) else { + if let Some(record) = self.database.by_p2p_id(conn) && + let Some(id) = record.peer_id + { + emit(PeerControl::P2pDisconnect { p2p: id, p2p_connection: conn }) } - peer.backoffs.insert(topic, now + self.params.prune_backoff); + return; + }; + if let Some(t) = peer.topic_stats.get_mut(&topic) { + // Carry any active deficit into the failure penalty. + if t.mesh_active && t.mesh_deliveries < self.params.mesh_message_deliveries_threshold { + t.mesh_failure_penalty += + self.params.mesh_message_deliveries_threshold - t.mesh_deliveries; + } + t.meshed_since = None; + t.mesh_active = false; } - tracing::debug!(p2p_peer = conn, ?topic, "PM peer PRUNEd us"); + peer.backoffs.insert(topic, now + self.params.prune_backoff); + tracing::info!(p2p_peer = conn, ?topic, mesh_size, "PM peer PRUNEd us"); } fn on_ihave(&mut self, conn: usize, hash: MessageId, already_seen: bool, now: Instant) { @@ -949,6 +1011,8 @@ impl PeerManager { idontwant: TCacheRead, emit: &mut impl FnMut(PeerControl), ) { + crate::counters::GossipTopicCounters::recv(topic); + // Any peer who promised this id is released — they did their job; // we just got another copy from someone else first. self.promises.remove(&msg_hash); @@ -1063,6 +1127,7 @@ impl PeerManager { // dontwant continue; } + crate::counters::GossipTopicCounters::sent(topic); emit(PeerControl::P2pSend(P2pSend::Gossip(GossipMsgOut { peer_id: *peer, tcache }))); } } @@ -1161,7 +1226,12 @@ impl PeerManager { // `target_peers` up to `max_priority_peers` for under-meshed validator // subnets. let connected = self.peers.len() + self.dialing.len(); - let priority = enr_matches_subnets(&enr, self.required_attnets, self.required_syncnets); + let priority = enr_matches_subnets( + &enr, + self.required_attnets, + self.required_syncnets, + self.custody_columns, + ); let dial = connected < self.params.target_peers || (priority && connected < self.params.max_priority_peers); if !dial { @@ -1267,6 +1337,28 @@ impl PeerManager { self.database.p2p_status(p2p_peer, status_ssz, earliest_slot); } + pub(crate) fn on_p2p_peer_goodbye( + &mut self, + p2p_peer: usize, + now: Instant, + code: u64, + emit: &mut impl FnMut(PeerControl), + ) { + if let Some(peer_record) = self.on_disconnected(p2p_peer, now, emit) && + let Some(peer_id) = peer_record.peer_id + { + let user_agent = peer_record.identify.as_ref().map(|i| i.user_agent()); + tracing::info!( + p2p_peer, + code = Self::goodbye_reason(code), + ?user_agent, + "received goodbye" + ); + emit(PeerControl::P2pDisconnect { p2p: peer_id, p2p_connection: p2p_peer }) + } + self.peers.remove(&p2p_peer); + } + /// Translate RPC misbehaviour severity into a P5 application-score /// delta. Calibrated so a single `Fatal` report drops the peer below the /// default `graylist_threshold = -80`, triggering eviction on the next @@ -1326,6 +1418,7 @@ impl PeerManager { t.meshed_since = Some(now); t.mesh_active = false; } + tracing::info!(?topic, conn, "GRAFT peer"); emit(PeerControl::P2pGossipGraft { p2p: peer_id, p2p_connection: conn, topic }); } @@ -1377,16 +1470,18 @@ impl PeerManager { }); !waiters.is_empty() }); - for (conn, count) in penalties { - self.add_behaviour_penalty(conn, count as f64, "broken gossip promises"); + for (conn, _count) in penalties { + // TODO seem to be over eagerly banning people here + self.add_behaviour_penalty(conn, 1.0, "broken gossip promises"); } } fn activate_p3_where_due(&mut self, now: Instant) { let activation = self.params.mesh_message_deliveries_activation_s; for peer in self.peers.values_mut() { - for t in peer.topic_stats.values_mut() { - if !t.mesh_active && + for (topic, t) in peer.topic_stats.iter_mut() { + if topic.p3_scored() && + !t.mesh_active && let Some(since) = t.meshed_since && now.saturating_duration_since(since).as_secs_f64() >= activation { @@ -1581,6 +1676,20 @@ impl PeerManager { } }); } + + /// Human label for a Goodbye reason code (per eth2 spec). + fn goodbye_reason(code: u64) -> &'static str { + match code { + 1 => "ClientShutdown", + 2 => "IrrelevantNetwork", + 3 => "Error", + 128 => "Banned", + 129 => "BannedIP", + 250 => "ScoreTooLow", + 251 => "Fault", + _ => "Unknown", + } + } } /// Session-level blacklist. Inserted into by beacon-state's `BlockRejected` @@ -1649,7 +1758,12 @@ fn build_subnet_masks(our_topics: &[GossipTopic]) -> ([u8; 8], u8) { /// True iff the ENR advertises subscription to at least one attnet/syncnet /// we also subscribe to. Both bitfields are SSZ Bitvectors so a bytewise /// AND is sufficient — any non-zero result means at least one shared bit. -fn enr_matches_subnets(enr: &Enr, attnets_mask: [u8; 8], syncnets_mask: u8) -> bool { +fn enr_matches_subnets( + enr: &Enr, + attnets_mask: [u8; 8], + syncnets_mask: u8, + custody_columns: u128, +) -> bool { if let Some(enr_attnets) = enr.attnets() { for i in 0..8 { if enr_attnets[i] & attnets_mask[i] != 0 { @@ -1662,6 +1776,9 @@ fn enr_matches_subnets(enr: &Enr, attnets_mask: [u8; 8], syncnets_mask: u8) -> b { return true; } + if let Some(cgc) = enr.cgc() { + return enr.node_id().custody_groups(cgc as u8) & custody_columns != 0 + } false } diff --git a/crates/peer/src/manager/rpc.rs b/crates/peer/src/manager/rpc.rs index bcae3c63..23b1963d 100644 --- a/crates/peer/src/manager/rpc.rs +++ b/crates/peer/src/manager/rpc.rs @@ -128,20 +128,6 @@ impl RootReqKind { } } -/// Human label for a Goodbye reason code (per eth2 spec). -fn goodbye_reason(code: u64) -> &'static str { - match code { - 1 => "ClientShutdown", - 2 => "IrrelevantNetwork", - 3 => "Error", - 128 => "Banned", - 129 => "BannedIP", - 250 => "ScoreTooLow", - 251 => "Fault", - _ => "Unknown", - } -} - /// Does this `response` terminate an outbound RPC stream we initiated? /// Single-chunk protocols (Status/Ping/MetaData/Goodbye) have no `Complete` /// sentinel — the one response chunk is the terminator. Multi-chunk @@ -482,14 +468,7 @@ impl PeerManager { } } RpcRequest::Goodbye(goodbye) => { - let code = u64::from_le_bytes(goodbye); crate::PeerCounters::GoodbyeReceived.inc(); - tracing::info!( - peer = stream_id.peer(), - code, - reason = goodbye_reason(code), - "received goodbye" - ); self.handle_event( PeerEvent::P2pPeerGoodbye { p2p_peer: stream_id.peer(), diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index a375a83d..ae49e58e 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -602,6 +602,7 @@ impl Store { // TODO should not return 'Complete' should return rate limit error if self.query_queue.len() >= MAX_INFLIGHT_QUERIES { + tracing::warn!(?stream_id, "queries at capacity"); self.query_queue.push_back(PendingQuery { stream_id, units: VecDeque::new() }); return; } @@ -629,6 +630,9 @@ impl Store { .chunks_exact(8) .map(|chunk| u64::from_le_bytes(chunk.try_into().unwrap())) .collect(); + + tracing::info!(?stream_id, start, count, "storage query"); + // `(slot, column)` order per fulu p2p-interface: outer slot, // inner column. Serve the canonical block's column from the // unfinalized store, else the finalized flat store. @@ -655,6 +659,9 @@ impl Store { let request_columns = DataColumnsByRootIdentifierView::columns(buf) .chunks_exact(8) .map(|chunk| u64::from_le_bytes(chunk.try_into().unwrap())); + + tracing::info!(?stream_id, ?root, "storage query"); + // Serve a specific block's columns regardless of // canonicity: unfinalized (by block_root) first, else // the finalized flat store. @@ -679,6 +686,9 @@ impl Store { let count = BeaconBlocksByRangeRequestView::count(&req_bytes).min(MAX_REQUEST_BLOCKS); let end = start.saturating_add(count); + + tracing::info!(?stream_id, start, count, "storage query"); + // Per slot: the unfinalized canonical block if present, else the // finalized flat store (`serve_file` skips an absent slot). let canonical = self.canonical_chain_in_range(start, end); @@ -699,6 +709,9 @@ impl Store { if let Ok((buf, _)) = read.buffer() { if BeaconBlocksByRootRequestView::check_size(buf) { let count = BeaconBlocksByRootRequestView::count(buf); + + tracing::info!(?stream_id, count, "storage query"); + for i in 0..count { let root = BeaconBlocksByRootRequestView::root(buf, i); // Serve any block we hold by root regardless of diff --git a/crates/storage/src/store/io.rs b/crates/storage/src/store/io.rs index 50b300f6..844e015d 100644 --- a/crates/storage/src/store/io.rs +++ b/crates/storage/src/store/io.rs @@ -326,7 +326,13 @@ impl Store { self.query_queue.push_back(query); } ServeResult::Missing => { - self.query_queue.push_back(query); + let error = "resource unavailable".as_bytes(); + let mut msg = [0u8; 256]; + msg[..error.len()].copy_from_slice(error); + let response = RpcResponse::Error { error: 3, msg, len: error.len() }; + emit(IoEvent::P2pSend(P2pSend::Rpc(RpcOutbound::Response( + RpcResponseOutbound { stream_id: query.stream_id, response }, + )))); } ServeResult::ProducerFull => { // Tcache full — un-consume and retry this request first diff --git a/crates/storage/src/tile.rs b/crates/storage/src/tile.rs index a6e8032f..fbe446a8 100644 --- a/crates/storage/src/tile.rs +++ b/crates/storage/src/tile.rs @@ -339,7 +339,7 @@ impl StorageTile { gloas_root } - #[inline] + #[timed] fn data_columns( &mut self, stream_id: P2pStreamId, @@ -621,6 +621,16 @@ impl StorageTile { } } + if stream_id.protocol() != StreamProtocol::GossipSub && self.store.is_synced() { + // Validated from a non-gossip source: the mesh never saw us + // deliver it. Hand it to control for re-publish on its subnet. + peer(PeerEvent::PublishDataColumn { + originator: stream_id, + topic: GossipTopic::DataColumnSidecar(column_index), + ssz: sidecar.read, + }); + } + let validated = self.validated_columns.entry(block_root).or_default(); *validated |= column_bitmask; let validated = *validated; @@ -631,6 +641,7 @@ impl StorageTile { tracing::info!( block = hex::encode(block_root), slot, + is_gossip = matches!(stream_id.protocol(), StreamProtocol::GossipSub), "DataColumnsAvailable: custody set complete" ); emit(DataColumnsAvailable { block_root, slot }) @@ -673,14 +684,15 @@ impl StorageTile { stream_id: P2pStreamId, is_gloas: bool, producers: &mut SilverSpineProducers, - ) { + ) -> bool { let emit_da = &mut |msg: DataColumnsAvailable| { producers.data_columns.produce(&msg.into()); }; let Some((block_root, columns)) = self.data_columns(stream_id, t_read, is_gloas, emit_da) else { - return; + // Validated + return true; }; // Validation failed - score down the peer and retransmit @@ -692,6 +704,7 @@ impl StorageTile { .into(), ); producers.peer_events.produce(&self.column_request(block_root, columns).into()); + false } } @@ -718,9 +731,20 @@ impl Tile for StorageTile { tracing::debug!(_custody_group, "data column sidecar over gossip"); let t_read = self.gossip_consumer.acquire(gossip.ssz); - // Gossip is `is_synced`-gated, so head ≈ wall selects the layout. + let is_gloas = self.spec.is_gloas_at_slot(self.store.head_slot() + 1); - self.handle_data_column_sidecar(t_read, gossip.stream_id, is_gloas, producers); + if self.handle_data_column_sidecar(t_read, gossip.stream_id, is_gloas, producers) { + producers.peer_events.produce( + &PeerEvent::SendGossip { + originator_stream_id: gossip.stream_id, + topic: gossip.topic, + msg_hash: gossip.msg_hash, + recv_ts: gossip.recv_ts, + protobuf: gossip.protobuf, + } + .into(), + ); + } } _ => {} }); diff --git a/crates/surfer/src/app.rs b/crates/surfer/src/app.rs index 92434394..cac649d6 100644 --- a/crates/surfer/src/app.rs +++ b/crates/surfer/src/app.rs @@ -116,9 +116,10 @@ impl App { let mut idx = 0; for (i, set) in self.counters.iter().enumerate() { if i == sel_set { - return idx + 1 + sel_slot; + let visible_before = (0..sel_slot).filter(|&s| set.slot_visible(s)).count(); + return idx + 1 + visible_before; } - idx += 1 + set.slot_count(); + idx += 1 + set.visible_slots(); } idx } @@ -146,7 +147,9 @@ impl App { } } } - self.counters.sort_by(|a, b| a.name.cmp(&b.name)); + self.counters.sort_by(|a, b| { + crate::schema::sort_key(&a.name).cmp(&crate::schema::sort_key(&b.name)) + }); if let Some(n) = sel_name { if let Some(idx) = self.counters.iter().position(|c| c.name == n) { self.counters_selection.0 = idx; @@ -277,18 +280,27 @@ impl App { } let (mut set, mut slot) = self.counters_selection; let n_sets = self.counters.len(); - let cur_len = self.counters[set].slot_count() as i32; - let new_slot = slot as i32 + dir; - if new_slot < 0 { - set = (set + n_sets - 1) % n_sets; - slot = self.counters[set].slot_count().saturating_sub(1); - } else if new_slot >= cur_len { - set = (set + 1) % n_sets; - slot = 0; - } else { - slot = new_slot as usize; + let total: usize = self.counters.iter().map(|s| s.slot_count()).sum(); + // Step one slot at a time (wrapping across sets), skipping hidden + // zero-valued slots; bounded by the total slot count. + for _ in 0..total { + if dir > 0 { + slot += 1; + if slot >= self.counters[set].slot_count() { + set = (set + 1) % n_sets; + slot = 0; + } + } else if slot == 0 { + set = (set + n_sets - 1) % n_sets; + slot = self.counters[set].slot_count().saturating_sub(1); + } else { + slot -= 1; + } + if self.counters[set].slot_visible(slot) { + self.counters_selection = (set, slot); + return; + } } - self.counters_selection = (set, slot); } fn move_timing_selection(&mut self, dir: i32) { diff --git a/crates/surfer/src/discovery.rs b/crates/surfer/src/discovery.rs index 4c12f7a9..c4bdbd25 100644 --- a/crates/surfer/src/discovery.rs +++ b/crates/surfer/src/discovery.rs @@ -66,7 +66,8 @@ pub fn discover(base_dir: &std::path::Path, app_name: &str) -> io::Result 0 && app.counters_table_state.offset() >= flat_idx { + *app.counters_table_state.offset_mut() = flat_idx - 1; + } f.render_stateful_widget(table, area, &mut app.counters_table_state); } @@ -122,7 +132,7 @@ fn draw_chart(f: &mut Frame, area: Rect, app: &mut App) { return; }; let label = set.slot_names.get(slot_idx).map(String::as_str).unwrap_or("?"); - let Some(hist) = set.history.get(slot_idx) else { + let Some(hist) = set.value_history.get(slot_idx) else { f.render_widget(Block::default().borders(Borders::ALL).title(" history "), area); return; }; @@ -142,12 +152,18 @@ fn draw_chart(f: &mut Frame, area: Rect, app: &mut App) { let data: Vec<(f64, f64)> = hist.iter().enumerate().map(|(i, &v)| (i as f64, v as f64)).collect(); - let y_max = data.iter().map(|(_, y)| *y).fold(0.0f64, f64::max).max(1.0); + // Value plot: auto-range to the data's span (padded) rather than + // anchoring at 0, so the trajectory is visible at any magnitude. + let data_lo = data.iter().map(|(_, y)| *y).fold(f64::INFINITY, f64::min); + let data_hi = data.iter().map(|(_, y)| *y).fold(f64::NEG_INFINITY, f64::max); + let pad = ((data_hi - data_lo) * 0.05).max(1.0); + let y_min = (data_lo - pad).max(0.0); + let y_max = data_hi + pad; let x_max = n.saturating_sub(1).max(1) as f64; let datasets = vec![ Dataset::default() - .name(format!("Δ {secs}s")) + .name("value") .marker(symbols::Marker::Braille) .style(Style::default().fg(Color::Cyan)) .graph_type(GraphType::Line) @@ -160,8 +176,8 @@ fn draw_chart(f: &mut Frame, area: Rect, app: &mut App) { Line::from("now"), ]; let y_labels = vec![ - Line::from("0"), - Line::from(fmt_u64((y_max / 2.0).round() as u64)), + Line::from(fmt_u64(y_min.round() as u64)), + Line::from(fmt_u64(((y_min + y_max) / 2.0).round() as u64)), Line::from(fmt_u64(y_max.round() as u64)), ]; @@ -175,7 +191,7 @@ fn draw_chart(f: &mut Frame, area: Rect, app: &mut App) { ) .y_axis( Axis::default() - .bounds([0.0, y_max * 1.1]) + .bounds([y_min, y_max]) .labels(y_labels) .style(Style::default().fg(Color::DarkGray)), ); diff --git a/crates/surfer/src/schema.rs b/crates/surfer/src/schema.rs index 5ecbaa4e..5541b61c 100644 --- a/crates/surfer/src/schema.rs +++ b/crates/surfer/src/schema.rs @@ -15,6 +15,19 @@ pub fn lookup(file_name: &str) -> Option<&'static [&'static str]> { } } +/// Counters-pane ordering: bulky per-topic groups sink below the compact +/// process-wide ones. +pub fn sort_key(file_name: &str) -> (u8, &str) { + let rank = if file_name == "gossip_topics" { 1 } else { 0 }; + (rank, file_name) +} + +/// Groups whose zero-valued slots are hidden in the counters pane — +/// dense pre-allocated layouts where only touched slots are interesting. +pub fn hide_zero(file_name: &str) -> bool { + file_name == "gossip_topics" +} + /// `lookup` with a positional fallback. Returns `(names, registered)` /// — `registered = false` means surfer is displaying positional /// labels because no schema was wired up for this file. @@ -22,6 +35,25 @@ pub fn names_for(file_name: &str, slot_count: usize) -> (Vec, bool) { if let Some(arr) = lookup(file_name) { return (arr.iter().map(|s| s.to_string()).collect(), true); } + // Gossip topic counters: arithmetic layout shared with + // `GossipTopic::counter_slot` — (topic slot) x (sent, recv, mesh, subs). + if file_name == "gossip_topics" { + let names = (0..slot_count) + .map(|i| { + let kind = match i % 4 { + 0 => "sent", + 1 => "recv", + 2 => "mesh", + _ => "subs", + }; + match silver_common::gossip_topic_for_counter_slot(i / 4) { + Some(topic) => format!("{topic}_{kind}"), + None => format!("slot_{i}"), + } + }) + .collect(); + return (names, true); + } // TCache files have a fixed semantic layout decoded from slot count. if file_name.starts_with("tcache-") && slot_count >= 2 { let mut names = vec!["capacity".to_string(), "head_seq".to_string()]; diff --git a/crates/surfer/src/sources/counters.rs b/crates/surfer/src/sources/counters.rs index a672e2cd..b1733583 100644 --- a/crates/surfer/src/sources/counters.rs +++ b/crates/surfer/src/sources/counters.rs @@ -34,6 +34,9 @@ pub struct CounterSet { /// are positional fallbacks because no schema was wired up /// (`false`). pub schema_registered: bool, + /// Zero-valued slots are hidden (dense pre-allocated layouts where + /// only touched slots carry signal). + hide_zero: bool, /// Companion read-only mmap into `tcache-names-{name}` when the /// CounterSet is a tcache. `n_consumers × 32` bytes of zero-padded /// UTF-8. `None` for non-tcache counters and for tcaches whose @@ -51,6 +54,9 @@ pub struct CounterSet { bucket_start: Vec, /// Per-slot ring of completed-bucket deltas (newest at back). pub history: Vec>, + /// Per-slot ring of absolute values sampled at each bucket close — + /// drives the drill-in value chart. + pub value_history: Vec>, /// `false` until the first `sample()` call. The first sample /// primes `previous` and `bucket_start` so initial deltas start /// at 0 rather than a wraparound (matters for slots initialised @@ -73,6 +79,7 @@ impl CounterSet { let map_bytes = slot_count * 8; let base = mmap_readonly(&file.path, map_bytes)?; let (slot_names, schema_registered) = crate::schema::names_for(&file.name, slot_count); + let hide_zero = crate::schema::hide_zero(&file.name); // For tcache counters, try to open the companion names file // `tcache-names-{tcache_name}`. Failure is non-fatal — surfer @@ -98,6 +105,7 @@ impl CounterSet { name: file.name.clone(), slot_names, schema_registered, + hide_zero, consumer_names_base, consumer_names_bytes, base, @@ -107,6 +115,9 @@ impl CounterSet { previous: vec![0; slot_count], bucket_start: vec![0; slot_count], history: (0..slot_count).map(|_| VecDeque::with_capacity(BUCKET_HISTORY_LEN)).collect(), + value_history: (0..slot_count) + .map(|_| VecDeque::with_capacity(BUCKET_HISTORY_LEN)) + .collect(), primed: false, }) } @@ -155,6 +166,19 @@ impl CounterSet { self.slot_count } + /// Traffic counters are monotonic so rows appear and stay; gauge + /// slots (mesh size) can drop back to zero and re-hide. + pub fn slot_visible(&self, i: usize) -> bool { + !self.hide_zero || self.current.get(i).copied().unwrap_or(0) != 0 + } + + pub fn visible_slots(&self) -> usize { + if !self.hide_zero { + return self.slot_count; + } + self.current.iter().filter(|&&v| v != 0).count() + } + /// Close the current bucket: for each slot, compute /// `current - bucket_start` and push to the per-slot history ring /// (drop oldest when full). Then snapshot `current` into @@ -178,6 +202,13 @@ impl CounterSet { h.pop_front(); } h.push_back(delta); + + let v = &mut self.value_history[i]; + if v.len() == BUCKET_HISTORY_LEN { + v.pop_front(); + } + v.push_back(self.current[i]); + self.bucket_start[i] = self.current[i]; } } diff --git a/docs/spine-message-flow.md b/docs/spine-message-flow.md index fa65046d..b93060af 100644 --- a/docs/spine-message-flow.md +++ b/docs/spine-message-flow.md @@ -34,7 +34,6 @@ flowchart LR NET -->|peer_events : PeerEvent| CTL ST -->|peer_events : PeerEvent| CTL BS -->|peer_events : PeerEvent| CTL - CTL -->|"peer_events : PeerEvent (gossip)"| CTL CTL -->|peer_control : PeerControl| NET CTL -->|peer_control : PeerControl| ST @@ -66,11 +65,10 @@ flowchart LR Solid arrows are spine queues (`queue : MessageType`), one per consumer since queues are SPMC. The dashed edge is the exception: `incoming_gossip` is the *tcache* carrying raw protobuf Network → Control (no spine queue on that hop — the gossip handler's queued -output is `new_gossip`). The Control self-edge is the gossip handler's `PeerEvent`s -(gossipsub scoring/misbehaviour): produced onto the queue and consumed by Control's own -`PeerEvent` pass next loop. Gossip's former `peer_control` and `beacon_events` -consumption is now in-tile — Control forwards `PeerControl` to the handler directly and -sets its fork digest from the `Status` it already consumes. `engine_health` is omitted +output is `new_gossip`). The gossip handler's other traffic is in-tile, not on the +spine: its `PeerEvent`s (gossipsub scoring/misbehaviour) go straight to the +`PeerManager`, `PeerControl` is forwarded to the handler directly, and its fork digest +is set from the `Status` Control already consumes. `engine_health` is omitted from the diagram: Engine produces it but no tile currently consumes it. Both Storage↔Engine edges carry only the `GetBlobs` variants (EL-mempool blob fetch); the queues are broadcast, so Storage sees every `EngineResp` and ignores the rest. @@ -82,7 +80,7 @@ queues are broadcast, so Storage sees every `EngineResp` and ignores the rest. | `new_gossip` | `NewGossipMsg` | Control _(gossip)_ | BeaconState, Storage | refs → `incoming_gossip`, `ssz_gossip` | | `p2p_send` | `P2pSend` | Control, Storage | Network | refs → `outgoing_gossip` / `outgoing_rpc` | | `rpc_inbound` | `RpcInbound` | Network | Control, BeaconState, Storage | ref → `incoming_rpc` | -| `peer_events` | `PeerEvent` | Network, Control _(gossip)_, Storage, BeaconState | Control | inline | +| `peer_events` | `PeerEvent` | Network, Storage, BeaconState | Control | inline | | `peer_control` | `PeerControl` | Control | Network, Storage | inline | | `beacon_events` | `BeaconStateEvent` | BeaconState | Control, Storage | inline | | `data_columns` | `DataColumnsAvailable` | Storage | BeaconState | inline | From 066f8619a1acaaaaefbbd39b59ff864d4835cd76 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Mon, 13 Jul 2026 18:45:13 +0100 Subject: [PATCH 2/2] merge --- crates/common/src/gossip.rs | 4 +--- crates/common/src/ticker.rs | 2 +- crates/control/src/tile.rs | 23 ++++++++++++--------- crates/storage/src/tile.rs | 40 ++++++++++++++++++++++--------------- kurtosis/net.yaml | 25 +++++++++++++++++++++++ 5 files changed, 65 insertions(+), 29 deletions(-) diff --git a/crates/common/src/gossip.rs b/crates/common/src/gossip.rs index 940426c5..049023f5 100644 --- a/crates/common/src/gossip.rs +++ b/crates/common/src/gossip.rs @@ -87,7 +87,7 @@ const EXECUTION_BID_SLOT: usize = DATA_COLUMN_BASE + crate::NUMBER_OF_CUSTODY_GR /// Dense per-topic slot space for counters/telemetry: subnet topics get one /// slot per subnet, everything else one slot. The layout is fixed so /// observers label by the same arithmetic (`gossip_topic_for_counter_slot`). -pub const GOSSIP_TOPIC_COUNTER_SLOTS: usize = EXECUTION_BID_SLOT + 5; +pub const GOSSIP_TOPIC_COUNTER_SLOTS: usize = EXECUTION_BID_SLOT + 4; /// Inverse of `GossipTopic::counter_slot`, for observer-side labelling. pub fn gossip_topic_for_counter_slot(slot: usize) -> Option { @@ -112,7 +112,6 @@ pub fn gossip_topic_for_counter_slot(slot: usize) -> Option { s if s == EXECUTION_BID_SLOT + 1 => GossipTopic::ExecutionPayload, s if s == EXECUTION_BID_SLOT + 2 => GossipTopic::PayloadAttestationMessage, s if s == EXECUTION_BID_SLOT + 3 => GossipTopic::ProposerPreferences, - s if s == EXECUTION_BID_SLOT + 4 => GossipTopic::InclusionList, _ => return None, }) } @@ -148,7 +147,6 @@ impl GossipTopic { Self::ExecutionPayload => EXECUTION_BID_SLOT + 1, Self::PayloadAttestationMessage => EXECUTION_BID_SLOT + 2, Self::ProposerPreferences => EXECUTION_BID_SLOT + 3, - Self::InclusionList => EXECUTION_BID_SLOT + 4, } } diff --git a/crates/common/src/ticker.rs b/crates/common/src/ticker.rs index 7a1ce26f..0b7cbfb7 100644 --- a/crates/common/src/ticker.rs +++ b/crates/common/src/ticker.rs @@ -102,7 +102,7 @@ impl SlotTicker { } pub fn slot_time_elapsed(&self) -> Duration { - Duration::from_millis(self.since_genesis_ms() % self.slot_ms) + Duration::from_millis(self.millis_since_genesis() % self.slot_ms) } pub fn current_slot(&self) -> u64 { diff --git a/crates/control/src/tile.rs b/crates/control/src/tile.rs index 5c7491e7..91686b9f 100644 --- a/crates/control/src/tile.rs +++ b/crates/control/src/tile.rs @@ -7,8 +7,9 @@ use std::{ use flux::{spine::SpineAdapter, tile::Tile}; use silver_chain_spec::SpecConfig; use silver_common::{ - BeaconStateEvent, P2pSend, PeerControl, PeerEvent, RpcInbound, RpcOutbound, RpcRequestOutbound, - SilverSpine, SilverSpineProducers, TCacheProducer, TCacheRead, TMultiProducer, msg_is_backfill, + BeaconStateEvent, Nanos, P2pSend, PeerControl, PeerEvent, RpcInbound, RpcOutbound, + RpcRequestOutbound, SilverSpine, SilverSpineProducers, TCacheProducer, TCacheRead, + TMultiProducer, TRandomAccess, msg_is_backfill, ssz_view::{BLOCKS_BY_RANGE_REQ_SIZE, METADATA_SIZE, STATUS_V2_SIZE, StatusView}, }; use silver_gossip::{GossipHandler, GossipHandlerEvent}; @@ -155,9 +156,14 @@ impl Tile for Controller { let fork_digest_changed = self.handle_latest_status(now, latest_status_event); if fork_digest_changed { tracing::info!("fork digest changed; re-announcing gossip subscriptions"); - self.peer_manager - .fan_out_subscriptions(&mut |evt| handle_peer_control(&mut self.gossip_handler, - &mut self.rpc_producer, evt, &mut adapter.producers)); + self.peer_manager.fan_out_subscriptions(&mut |evt| { + handle_peer_control( + &mut self.gossip_handler, + &mut self.rpc_producer, + evt, + &mut adapter.producers, + ) + }); } adapter.consume(|event: PeerEvent, producers| { @@ -423,10 +429,9 @@ fn handle_peer_control( &P2pSend::Rpc(RpcOutbound::Request(RpcRequestOutbound { application_id: app_id, peer, - request: - silver_common::RpcRequest::ExecutionPayloadEnvelopesByRoot( - read, - ), + request: silver_common::RpcRequest::ExecutionPayloadEnvelopesByRoot( + read, + ), })) .into(), ); diff --git a/crates/storage/src/tile.rs b/crates/storage/src/tile.rs index fbe446a8..19e1db3c 100644 --- a/crates/storage/src/tile.rs +++ b/crates/storage/src/tile.rs @@ -11,10 +11,10 @@ use flux::{spine::SpineAdapter, tile::Tile}; use flux_profiler::timed; use silver_beacon_state_data::{B256, BeaconStateReader, SLOTS_PER_EPOCH, SpecConfig}; use silver_common::{ - BASE_REQUEST_ID, BeaconStateEvent, DataColumnsAvailable, EngineReq, EngineResp, NewGossipMsg, - P2pSend, P2pStreamId, PeerControl, PeerEvent, ReplayBlock, RpcInbound, RpcSeverity, - SilverSpine, SilverSpineProducers, StreamProtocol, SyncUpdate, SyncingStrategy, TCacheProducer, - TMultiProducer, TProducer, TRandomAccess, TRead, Wheel, msg_is_backfill, + BASE_REQUEST_ID, BeaconStateEvent, DataColumnsAvailable, EngineReq, EngineResp, GossipTopic, + NewGossipMsg, P2pSend, P2pStreamId, PeerControl, PeerEvent, ReplayBlock, RpcInbound, + RpcSeverity, SilverSpine, SilverSpineProducers, StreamProtocol, SyncUpdate, SyncingStrategy, + TCacheProducer, TMultiProducer, TProducer, TRandomAccess, TRead, Wheel, msg_is_backfill, msg_is_column_backfill, msg_is_live_column_request, msg_is_post_gloas, ssz_view::{ DataColumnSidecarFuluView, DataColumnSidecarGloasView, NUMBER_OF_COLUMNS, @@ -621,16 +621,6 @@ impl StorageTile { } } - if stream_id.protocol() != StreamProtocol::GossipSub && self.store.is_synced() { - // Validated from a non-gossip source: the mesh never saw us - // deliver it. Hand it to control for re-publish on its subnet. - peer(PeerEvent::PublishDataColumn { - originator: stream_id, - topic: GossipTopic::DataColumnSidecar(column_index), - ssz: sidecar.read, - }); - } - let validated = self.validated_columns.entry(block_root).or_default(); *validated |= column_bitmask; let validated = *validated; @@ -641,7 +631,6 @@ impl StorageTile { tracing::info!( block = hex::encode(block_root), slot, - is_gossip = matches!(stream_id.protocol(), StreamProtocol::GossipSub), "DataColumnsAvailable: custody set complete" ); emit(DataColumnsAvailable { block_root, slot }) @@ -795,7 +784,26 @@ impl Tile for StorageTile { tracing::debug!("data column sidecar over rpc"); let t_read = self.rpc_consumer.acquire(ssz); let is_gloas = msg_is_post_gloas(rsp.application_id); - self.handle_data_column_sidecar(t_read, rsp.stream_id, is_gloas, producers); + + let Ok((buffer, _)) = t_read.buffer() else { + return; + }; + let column_index = if is_gloas { + DataColumnSidecarGloasView::index(buffer) + } else { + DataColumnSidecarFuluView::index(buffer) + }; + + if self.handle_data_column_sidecar(t_read, rsp.stream_id, is_gloas, producers) { + if rsp.stream_id.protocol() != StreamProtocol::GossipSub && self.store.is_synced() { + // Publish to gossip + producers.peer_events.produce(&PeerEvent::PublishDataColumn { + originator: rsp.stream_id, + topic: GossipTopic::DataColumnSidecar(column_index), + ssz, + }.into()); + } + } } silver_common::RpcResponse::Error { error, msg, len } if msg_is_column_backfill(rsp.application_id)=> { let err_msg = String::from_utf8_lossy(&msg[..len]).to_string(); diff --git a/kurtosis/net.yaml b/kurtosis/net.yaml index 1f570167..a6e97fcc 100644 --- a/kurtosis/net.yaml +++ b/kurtosis/net.yaml @@ -19,26 +19,47 @@ participants: # Three lighthouse nodes on distinct ELs -> guaranteed QUIC peers + EL # diversity. These are silver's reliable counterparties. + # + # Scoring visibility: `global_log_level: debug` covers lighthouse's peer + # manager (score-driven disconnects, bans, goodbyes-with-reason). The + # gossipsub-internal penalties (graft-in-backoff, IWANT flood, invalid + # message rejects) log under the gossipsub tracing target -- enabled via + # RUST_LOG below. Per-peer live scores (incl. silver's) need no flag: + # GET /lighthouse/peers on each node's HTTP API. - el_type: geth cl_type: lighthouse cl_extra_params: - --target-peers=64 + cl_extra_env_vars: + RUST_LOG: "gossipsub=debug,libp2p_gossipsub=debug" count: 1 - el_type: reth cl_type: lighthouse cl_extra_params: - --target-peers=64 + cl_extra_env_vars: + RUST_LOG: "gossipsub=debug,libp2p_gossipsub=debug" count: 1 - el_type: nethermind cl_type: lighthouse cl_extra_params: - --target-peers=64 + cl_extra_env_vars: + RUST_LOG: "gossipsub=debug,libp2p_gossipsub=debug" count: 1 # One non-lighthouse CL for protocol-diversity in disconnect/scoring/goodbye # behaviour. Confirm it advertises QUIC; if not, silver will skip it (which # is itself a useful observation in your new logs). + # + # Prysm at debug logs per-message validation results ("could not + # validate...") -- the fastest check that silver's (re)published columns are + # accepted -- and its own scorer's "disconnecting bad peer". go-libp2p does + # NOT log gossipsub penalties; per-peer scores come from the debug RPC + # endpoints enabled below. - el_type: geth cl_type: prysm + cl_extra_params: + - --enable-debug-rpc-endpoints count: 1 network_params: @@ -72,6 +93,10 @@ global_log_level: debug additional_services: - dora - spamoor + # prometheus scrapes every CL's metrics endpoint: lighthouse exports + # gossipsub score distributions + ban/disconnect counters -- the trend view + # that log lines can't give. + - prometheus_grafana # Blob spammer config. NOTE: service name (`spamoor`), scenario (`blobs`), and # keys (`throughput`/`sidecars`) are verified against ethereum-package @6.1.0 +