From 03a2f5555bc874578b0aa0585ae92e3ffdaad8b5 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Sat, 22 Aug 2026 14:12:27 +0100 Subject: [PATCH 1/4] remove allocation on outbound gossip --- crates/bin/Cargo.toml | 1 + crates/bin/src/main.rs | 2 +- crates/common/src/lib.rs | 21 +- crates/common/src/spine.rs | 4 +- crates/common/src/spine/tcache.rs | 42 +++- crates/common/src/spine/tcache/consumer.rs | 81 +++++++- crates/network/src/lib.rs | 4 + crates/network/src/p2p/quic/mod.rs | 3 + crates/network/src/p2p/quic/peer.rs | 160 ++++++++++++++- crates/network/src/p2p/quic/stream.rs | 10 + crates/network/src/p2p/streams/gossip_in.rs | 8 + crates/network/src/p2p/streams/gossip_out.rs | 190 ++++++++++++++++-- crates/network/src/p2p/streams/mod.rs | 7 +- crates/network/src/p2p/streams/negotiate.rs | 8 + .../network/src/p2p/streams/rpc/request_in.rs | 8 + .../src/p2p/streams/rpc/response_in.rs | 8 + crates/network/src/p2p/streams/state.rs | 17 +- crates/network/src/tile.rs | 7 + crates/storage/src/store/backfill.rs | 2 +- crates/surfer/src/app.rs | 5 + crates/surfer/src/main.rs | 1 + crates/surfer/src/render/peers_pane.rs | 9 +- scripts/start_silver_with_ethrex.sh | 2 +- 23 files changed, 549 insertions(+), 51 deletions(-) diff --git a/crates/bin/Cargo.toml b/crates/bin/Cargo.toml index e7380f65..b989a4d0 100644 --- a/crates/bin/Cargo.toml +++ b/crates/bin/Cargo.toml @@ -31,6 +31,7 @@ tracing.workspace = true workspace = true [features] +default = ["thread_park"] alloc-profile = ["silver_common/alloc-profile"] # Hardware-counter dimension for `#[timed]`: per-call counters via rdpmc on # perf-{fn} queues (dev; needs `sudo sysctl kernel.perf_event_paranoid=2`). diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 6e28d3e1..1ecbc7c8 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -140,7 +140,7 @@ fn main() -> Result<(), Box> { gossip_producer: incoming_gossip_producer, gossip_consumer: outgoing_gossip_producer .cache_ref() - .random_access("p2p_outgoing_gossip", true)?, + .strict_random_access("p2p_outgoing_gossip", true)?, rpc_producer: incoming_rpc_producer, rpc_consumer: outgoing_rpc_producer.cache_ref().random_access("p2p_outgoing_rpc", true)?, identify: Some(ProtoIdentify::from((&config.identify()?, &keypair))), diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 0325e67c..946b9d36 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -15,16 +15,17 @@ pub use crate::{ }, request::{DataKind, Origin, RequestId, Scope, SyncRequest}, spine::{ - ALL_PROTOCOLS, AcquiredRead as TRead, AgentString, BeaconStateEvent, BlockSource, - ColumnSource, Consumer as TConsumer, DataColumnsEvent, ELSyncStatus, EngineFcuReq, - EngineFcuResp, EngineGetBlobsReq, EngineGetBlobsResp, EngineGetPayloadBodiesByHashReq, - EngineGetPayloadBodiesByRangeReq, EngineGetPayloadBodiesResp, EngineGetPayloadReq, - EngineGetPayloadResp, EngineHealthEvent, EngineNewPayloadEnvelopeReq, EngineNewPayloadReq, - EngineNewPayloadResp, EnginePreparePayloadReq, EngineReq, EngineResp, Error as TCacheError, - GossipMsgIn, GossipMsgOut, IpBytes, MAX_BLOBS_PER_BLOCK, MAX_PAYLOAD_BODIES_PER_REQ, - MULTISTREAM_V1, MultiProducer as TMultiProducer, NewGossipMsg, P2pConnectionStats, P2pSend, - P2pStreamId, PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, PeerStats, - PeerStatus, PeerTopicScores, Producer as TProducer, REJECT_RESPONSE, RPC_PROTOCOLS, + ALL_PROTOCOLS, AcquiredRead as TRead, AcquiredWithOffset, AgentString, BeaconStateEvent, + BlockSource, ColumnSource, Consumer as TConsumer, DataColumnsEvent, ELSyncStatus, + EngineFcuReq, EngineFcuResp, EngineGetBlobsReq, EngineGetBlobsResp, + EngineGetPayloadBodiesByHashReq, EngineGetPayloadBodiesByRangeReq, + EngineGetPayloadBodiesResp, EngineGetPayloadReq, EngineGetPayloadResp, EngineHealthEvent, + EngineNewPayloadEnvelopeReq, EngineNewPayloadReq, EngineNewPayloadResp, + EnginePreparePayloadReq, EngineReq, EngineResp, Error as TCacheError, GossipMsgIn, + GossipMsgOut, IpBytes, MAX_BLOBS_PER_BLOCK, MAX_PAYLOAD_BODIES_PER_REQ, MULTISTREAM_V1, + MultiProducer as TMultiProducer, NewGossipMsg, P2pConnectionStats, P2pSend, P2pStreamId, + PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, PeerStats, PeerStatus, + PeerTopicScores, Producer as TProducer, REJECT_RESPONSE, RPC_PROTOCOLS, RandomAccessConsumer as TRandomAccess, ReplayBlock, Reservation as TReservation, RpcInbound, RpcOutbound, RpcRequest, RpcRequestInbound, RpcRequestOutbound, RpcResponse, RpcResponseInbound, RpcResponseOutbound, RpcSeverity, SilverSpine, SilverSpineProducers, diff --git a/crates/common/src/spine.rs b/crates/common/src/spine.rs index 95062e0e..f86677a3 100644 --- a/crates/common/src/spine.rs +++ b/crates/common/src/spine.rs @@ -18,8 +18,8 @@ pub use stream_protocol::{ ALL_PROTOCOLS, MULTISTREAM_V1, REJECT_RESPONSE, RPC_PROTOCOLS, StreamProtocol, }; pub use tcache::{ - AcquiredRead, Consumer, Error, MultiProducer, Producer, RandomAccessConsumer, Reservation, - TCache, TCacheProducer, TCacheRead, TCacheRef, + AcquiredRead, AcquiredWithOffset, Consumer, Error, MultiProducer, Producer, + RandomAccessConsumer, Reservation, TCache, TCacheProducer, TCacheRead, TCacheRef, }; mod messages; diff --git a/crates/common/src/spine/tcache.rs b/crates/common/src/spine/tcache.rs index 5b201f30..1f1819a2 100644 --- a/crates/common/src/spine/tcache.rs +++ b/crates/common/src/spine/tcache.rs @@ -8,7 +8,7 @@ use std::{ sync::atomic::{AtomicU64, Ordering}, }; -pub use consumer::{AcquiredRead, Consumer, RandomAccessConsumer, TCacheRead}; +pub use consumer::{AcquiredRead, AcquiredWithOffset, Consumer, RandomAccessConsumer, TCacheRead}; use flux::{Timer, timing::Nanos, tracing}; pub use producer::{MultiProducer, Producer, Reservation, TCacheProducer}; use thiserror::Error; @@ -203,6 +203,26 @@ impl TCache { &self, name: &'static str, auto_free: bool, + ) -> Result { + self.ra_consumer(name, auto_free, false) + } + + /// Strict random access consumer cannot be force reset + /// on idle and therefore can block producer - only use + /// for high throughput consumers. + pub fn strict_random_access( + &self, + name: &'static str, + auto_free: bool, + ) -> Result { + self.ra_consumer(name, auto_free, true) + } + + fn ra_consumer( + &self, + name: &'static str, + auto_free: bool, + strict: bool, ) -> Result { let seq = self.head().seq.load(Ordering::Acquire); @@ -225,16 +245,23 @@ impl TCache { self.record_tail(index, seq); self.record_consumer_name(index, name); + let active = if strict { + Buckets::strict(32 * 1024, self.len as u64, seq) + } else { + Buckets::new(32 * 1024, self.len as u64, seq) + }; + Ok(RandomAccessConsumer { cache: TCacheRef { cache: addr_of!(*self) as *const c_void }, index, name, - active: Buckets::new(32 * 1024, self.len as u64, seq), + active, auto_free, timer: self.create_consumer_timer(name), last_read: Nanos::now(), last_head: seq, lag_threshold: lag_threshold(self.len), + strict, }) } @@ -324,6 +351,17 @@ impl TCache { Ok(slot.reserve_ns) } + fn check_seq(&self, seq: u64) -> bool { + let idx = self.index(seq); + let slot: &Slot = self.slot_at(idx); + if slot.magic != MAGIC { + return false; + } + + let slot_seq = slot.seq.load(Ordering::Acquire); + slot_seq == seq + } + fn reserve_len(&self, seq: u64, requested_len: usize) -> usize { let mut data_len = requested_len + size_of::(); diff --git a/crates/common/src/spine/tcache/consumer.rs b/crates/common/src/spine/tcache/consumer.rs index b4449001..ae8a0f5c 100644 --- a/crates/common/src/spine/tcache/consumer.rs +++ b/crates/common/src/spine/tcache/consumer.rs @@ -132,6 +132,7 @@ pub struct RandomAccessConsumer { pub(super) last_read: Nanos, pub(super) last_head: u64, pub(super) lag_threshold: u64, + pub(super) strict: bool, } impl RandomAccessConsumer { @@ -148,13 +149,33 @@ impl RandomAccessConsumer { AcquiredRead { consumer: self as *const Self, read } } + pub fn acquire_strict(&mut self, read: TCacheRead) -> Option { + let now = Nanos::now(); + self.last_read = now; + + self.active + .acquire(read.seq) + .then(|| { + if let Some(timer) = &mut self.timer { + if let Ok(reserve_ns) = read.cache_ts() { + timer.emit_latency_from_nanos(reserve_ns, now); + } + } + AcquiredRead { consumer: self as *const Self, read } + }) + .and_then(|ar| { + // check slot seq. + self.cache.check_seq(read.seq).then(|| ar) + }) + } + /// Should be called periodically to publish the tail offset so it is /// visible to the Producer. pub fn free(&mut self) { let mut tail = self.active.tail_seq; if tail != u64::MAX { let cache_head = self.cache.head(); - if self.last_read.elapsed() > IDLE_INTERVAL_NS { + if !self.strict && self.last_read.elapsed() > IDLE_INTERVAL_NS { // check lagging let head = cache_head.seq.load(Ordering::Relaxed); if head.saturating_sub(tail) > self.lag_threshold { @@ -206,7 +227,7 @@ impl Drop for RandomAccessConsumer { /// for the lifetime of every `AcquiredRead` it hands out — guaranteed by /// drop-order discipline (see NetworkTile field ordering) - order containers /// of reads before consumer. -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct AcquiredRead { consumer: *const RandomAccessConsumer, pub read: TCacheRead, @@ -225,6 +246,13 @@ impl AcquiredRead { } consumer.cache.read(self.read.seq).map(|(data, _, ts)| (data, ts)) } + + pub fn with_offset(&self, offset: usize) -> Option { + let consumer = unsafe { &mut *(self.consumer as *mut RandomAccessConsumer) }; + consumer + .acquire_strict(self.read) + .and_then(|read| Some(AcquiredWithOffset { read, offset })) + } } impl Deref for AcquiredRead { @@ -252,6 +280,30 @@ impl Drop for AcquiredRead { } } +impl Clone for AcquiredRead { + fn clone(&self) -> Self { + unsafe { + let consumer = &mut *(self.consumer as *mut RandomAccessConsumer); + consumer.active.acquire(self.read.seq); + } + Self { consumer: self.consumer, read: self.read } + } +} + +pub struct AcquiredWithOffset { + read: AcquiredRead, + offset: usize, +} + +impl AsRef<[u8]> for AcquiredWithOffset { + fn as_ref(&self) -> &[u8] { + match self.read.buffer() { + Ok((buffer, _)) => &buffer[self.offset..], + Err(_) => &[], + } + } +} + pub(super) struct Buckets { buckets: Box<[u16]>, tail_seq: u64, @@ -260,16 +312,26 @@ pub(super) struct Buckets { bucket_shift: u64, bucket_mask: u64, // max difference between head and tail, before 'forced' eviction + // for 'strict' consumers this is set to cache length so that it never + // triggers lag_threshold: u64, // Out-of-order acquire lookback: the tail never advances within this // many seqs of the newest acquire's bucket, so late acquires up to - // this far behind still land at or above the tail. 10% of capacity, + // this far behind still land at or above the tail. 20% of capacity, // rounded up to a bucket. guard: u64, } impl Buckets { pub(super) fn new(bucket_size: u64, cache_capacity: u64, seq: u64) -> Self { + Self::create(bucket_size, cache_capacity, seq, false) + } + + pub(super) fn strict(bucket_size: u64, cache_capacity: u64, seq: u64) -> Self { + Self::create(bucket_size, cache_capacity, seq, true) + } + + pub(super) fn create(bucket_size: u64, cache_capacity: u64, seq: u64, strict: bool) -> Self { assert!(bucket_size.is_power_of_two()); let mut number_of_buckets = cache_capacity / bucket_size; if !cache_capacity.is_multiple_of(bucket_size) || !number_of_buckets.is_power_of_two() { @@ -282,12 +344,16 @@ impl Buckets { bucket_size, bucket_shift: bucket_size.trailing_zeros() as u64, bucket_mask: !(bucket_size - 1), - lag_threshold: lag_threshold(cache_capacity as u32), - guard: (cache_capacity / 10).next_multiple_of(bucket_size).max(bucket_size), + lag_threshold: if strict { + cache_capacity + } else { + lag_threshold(cache_capacity as u32) + }, + guard: (cache_capacity / 5).next_multiple_of(bucket_size).max(bucket_size), } } - fn acquire(&mut self, seq: u64) { + fn acquire(&mut self, seq: u64) -> bool { // Out-of-order acquire beyond the guard window: the producer may // already be reclaiming this slot, and bucket_index aliases behind // the tail onto in-window buckets — counting it would corrupt a @@ -295,7 +361,7 @@ impl Buckets { // the read surfaces as StaleSeq at buffer() time. if self.tail_seq != u64::MAX && seq < self.tail_seq { tracing::warn!(seq, tail = self.tail_seq, head = self.head_seq, "acquire below tail"); - return; + return false; } let bucket_idx = self.bucket_index(seq); @@ -325,6 +391,7 @@ impl Buckets { } self.tail_seq += self.bucket_size; } + true } fn release(&mut self, seq: u64, name: &str) { diff --git a/crates/network/src/lib.rs b/crates/network/src/lib.rs index 37bca5fd..b0e7a506 100644 --- a/crates/network/src/lib.rs +++ b/crates/network/src/lib.rs @@ -33,6 +33,10 @@ silver_common::declare_counters! { // A peer's read-timeout gave up on our response (their reset carried // the response-timeout code): direct we-are-slow signal. RemoteResponseTimeout, + // Stale gossip skipped + GossipMsgSkipped, + // Gossip stream stalled (read or write) — connection closed. + GossipStallDisconnect, } } diff --git a/crates/network/src/p2p/quic/mod.rs b/crates/network/src/p2p/quic/mod.rs index 05cf5ccb..18486aef 100644 --- a/crates/network/src/p2p/quic/mod.rs +++ b/crates/network/src/p2p/quic/mod.rs @@ -67,4 +67,7 @@ pub enum SendResult { StreamGone, MessageDropped, UnknownPeer, + /// Connection is closing/draining: nothing sent on it can be delivered, + /// and opening a stream would misreport as credit exhaustion. + ConnectionClosing, } diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index 708fa810..3e33d61f 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -140,6 +140,9 @@ impl Peer { } pub(crate) fn send_gossip(&mut self, msg: TRead) -> SendResult { + if self.connection.is_closed() { + return SendResult::ConnectionClosing; + } self.dirty = true; if let Some(stream) = match &self.outbound_gossip { Some(id) => self.streams.get_mut(id), @@ -158,6 +161,9 @@ impl Peer { } pub(crate) fn send_rpc(&mut self, msg: AcquiredRpcOutbound) -> SendResult { + if self.connection.is_closed() { + return SendResult::ConnectionClosing; + } tracing::debug!(id=?self.id, protocol=?msg.protocol(), "outbound rpc"); self.dirty = true; @@ -193,6 +199,9 @@ impl Peer { } pub(crate) fn send_identify(&mut self) -> SendResult { + if self.connection.is_closed() { + return SendResult::ConnectionClosing; + } self.dirty = true; match self.open_stream(StreamProtocol::Identity) { Some(_) => SendResult::Ok, @@ -212,6 +221,29 @@ impl Peer { self.dirty = true; self.pending_shutdown = None; self.connection.close(now, VarInt::from_u32(0), Bytes::new()); + self.clear_streams(); + } + + /// A stalled gossip stream costs the connection, not just the stream: + /// reopening on the same connection spends one of rust-libp2p's five + /// per-connection substream attempts, after which the remote disables + /// gossipsub on it silently. A redial gives both sides a fresh budget. + fn disconnect_on_stall(&mut self, now: Instant) { + crate::NetworkCounters::GossipStallDisconnect.inc(); + tracing::warn!(id = ?self.id, "gossip stall: closing connection"); + self.shutdown(now); + } + + /// Drop every stream state — and the queued messages (and tcache + /// acquires) in their outbound buffers — as soon as the connection can + /// no longer deliver, rather than at the drained reap up to 3×PTO later. + /// No per-stream `StreamClosed` events: the PM tears the peer down on + /// disconnect, and a close event for an outgoing RPC would read as the + /// peer abandoning a response. + fn clear_streams(&mut self) { + self.streams.clear(); + self.inbound_gossip = None; + self.outbound_gossip = None; } /// Stream-leak diagnostics: one line per over-populated connection @@ -395,6 +427,7 @@ impl Peer { zombie, "connection lost" ); + self.clear_streams(); } quinn_proto::Event::Stream(stream_event) => { self.handle_stream_event(stream_event, now, context, on_event); @@ -411,7 +444,7 @@ impl Peer { // Drive only streams flagged for non-event work; quinn-I/O parks are // re-driven by Readable/Writable via `handle_stream_event`. - let to_remove = spin_streams( + let (to_remove, stalled) = spin_streams( now, &mut self.connection, context, @@ -425,12 +458,15 @@ impl Peer { for id in to_remove { self.end_stream(id, now); } + if stalled { + self.disconnect_on_stall(now); + } // Read-response timeouts only fire inside a spin; sweep everything // when the earliest deadline lapses. if self.next_deadline.is_some_and(|d| now >= d) { self.next_deadline = None; - let to_remove = spin_streams( + let (to_remove, stalled) = spin_streams( now, &mut self.connection, context, @@ -444,6 +480,9 @@ impl Peer { for id in to_remove { self.end_stream(id, now); } + if stalled { + self.disconnect_on_stall(now); + } } } @@ -463,6 +502,10 @@ impl Peer { }; let result = stream.spin(&mut self.connection, context, now, &mut self.inbound_rpc_limits, on_event); + if let SpinResult::Stalled = result { + self.disconnect_on_stall(now); + return; + } if let SpinResult::End = result { self.end_stream(id, now); return; @@ -628,17 +671,23 @@ fn spin_streams( inbound_gossip: &mut Option, all: bool, on_event: &mut E, -) -> ArrayVec +) -> (ArrayVec, bool) where E: FnMut(crate::NetEvent), { let mut to_remove = ArrayVec::new(); + let mut stalled = false; for (id, stream) in streams { if !all && !stream.needs_spin { continue; } let result = stream.spin(connection, context, now, inbound_rpc_limits, on_event); + if let SpinResult::Stalled = result { + stalled = true; + to_remove.push(*id); + continue; + } if let SpinResult::End = result { to_remove.push(*id); continue; @@ -658,7 +707,7 @@ where } } } - to_remove + (to_remove, stalled) } /// Bucket a `ConnectionLost` reason into the `NetworkCounters` disconnect @@ -729,6 +778,8 @@ struct Stream { enum SpinResult { Ok, End, + /// Gossip stream stalled: the owner closes the whole connection. + Stalled, Protocol(StreamProtocol), } @@ -836,7 +887,11 @@ impl Stream { // TODO error info. on_event(NetEvent::StreamClosed { stream: self.p2p_id }); - SpinResult::End + if matches!(e, StreamError::GossipReadStall | StreamError::GossipWriteStall) { + SpinResult::Stalled + } else { + SpinResult::End + } } } } @@ -1398,6 +1453,101 @@ mod tests { assert!(!pair.client_peer.streams.contains_key(&stream)); } + /// Closing drops stream state (and the queued acquires in its buffers) + /// immediately on both ends — locally at `shutdown`, remotely on + /// `ConnectionLost` — instead of holding it until the drained reap, and + /// a send into a closing connection is refused rather than misreported + /// as stream-credit exhaustion. + #[test] + fn closing_connection_refuses_sends_and_drops_streams() { + let mut pair = PeerPair::new(); + let mut client_h = PeerHarness::new(); + let mut server_h = PeerHarness::new(); + let now = Instant::now(); + + let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); + let stream_id = P2pStreamId::new( + pair.client_peer.id.connection, + sid.into(), + StreamProtocol::GossipSub, + false, + ); + client_h.send_gossip(stream_id, b"ping", &mut pair.client_peer); + wait_for(&mut pair, &mut client_h, &mut server_h, 200, |_, s| !s.received.is_empty()); + assert!(!pair.server_peer.streams.is_empty(), "server holds the inbound gossip stream"); + + pair.client_peer.shutdown(now); + assert!(pair.client_peer.streams.is_empty(), "local close drops streams at once"); + assert!(pair.client_peer.outbound_gossip.is_none()); + let mut res = client_h.gossip_out_producer.reserve(4, true).unwrap(); + res.write_all(b"late").unwrap(); + let late = client_h.context.gossip_consumer.acquire(res.read()); + assert!(matches!(pair.client_peer.send_gossip(late), SendResult::ConnectionClosing)); + assert!(pair.client_peer.streams.is_empty(), "refused send opens nothing"); + + let mut noop_c = |_: NetEvent| {}; + let mut noop_s = |_: NetEvent| {}; + for _ in 0..200 { + pair.step(now, &mut client_h, &mut server_h, &mut noop_c, &mut noop_s); + if pair.server_peer.connection.is_closed() { + break; + } + } + assert!(pair.server_peer.connection.is_closed(), "CONNECTION_CLOSE must reach the server"); + assert!(pair.server_peer.streams.is_empty(), "connection loss drops the server's streams"); + } + + /// A gossip stall (here: the server's inbound frame parked mid-body past + /// the window) closes the connection rather than just the stream. + #[test] + fn gossip_stall_closes_connection() { + use crate::p2p::streams::{ + gossip_in::{GOSSIP_BODY_STALL_TIMEOUT, GossipReadState}, + gossip_out::GossipWriteState, + }; + + let mut pair = PeerPair::new(); + let mut client_h = PeerHarness::new(); + let mut server_h = PeerHarness::new(); + + let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); + let stream_id = P2pStreamId::new( + pair.client_peer.id.connection, + sid.into(), + StreamProtocol::GossipSub, + false, + ); + client_h.send_gossip(stream_id, b"ping", &mut pair.client_peer); + wait_for(&mut pair, &mut client_h, &mut server_h, 200, |_, s| !s.received.is_empty()); + + let t0 = Instant::now(); + let stalled_since = t0 - GOSSIP_BODY_STALL_TIMEOUT - Duration::from_secs(1); + let reservation = server_h.context.gossip_producer.reserve(100, false).expect("reserve"); + let stream = pair + .server_peer + .streams + .values_mut() + .find(|s| s.p2p_id.protocol() == StreamProtocol::GossipSub) + .expect("inbound gossip stream"); + stream.state.replace(StreamState::Gossip { + read: GossipReadState::ReadingBody { + reservation, + remaining: 90, + last_read: stalled_since, + }, + write: GossipWriteState::Idle, + }); + stream.needs_spin = true; + + let PeerPair { server_ep, server_peer, .. } = &mut pair; + let mut cb = |h, e| server_ep.handle_event(h, e); + let mut on_event = |_: NetEvent| {}; + server_peer.spin(t0, &mut cb, &mut server_h.context, &mut on_event, &FxHashSet::default()); + + assert!(server_peer.connection.is_closed(), "stall must close the connection"); + assert!(server_peer.streams.is_empty(), "closing drops every stream"); + } + /// A negotiated inbound RPC stream whose request never arrives must be /// reaped by `INBOUND_RPC_IDLE_TIMEOUT` — only the server is spun past the /// deadline, so the reap can't be masked by the client's own teardown. diff --git a/crates/network/src/p2p/quic/stream.rs b/crates/network/src/p2p/quic/stream.rs index 361f54b9..9e448c8f 100644 --- a/crates/network/src/p2p/quic/stream.rs +++ b/crates/network/src/p2p/quic/stream.rs @@ -1,5 +1,6 @@ use std::io::{Error, Write}; +use bytes::Bytes; use quinn_proto::{Connection, StreamId, WriteError}; use crate::p2p::{ @@ -22,6 +23,15 @@ impl<'a> StreamIo for StreamIoImpl<'a> { } } + fn write_bytes_to_stream(&mut self, id: StreamId, data: Bytes) -> Result { + let mut stream = self.connection.send_stream(id); + match stream.write_chunks(&mut [data]) { + Ok(wrote) => Ok(wrote.bytes), + Err(WriteError::Blocked) => Ok(0), + Err(e) => Err(e.into()), + } + } + fn read_from_stream(&mut self, id: StreamId, data: &mut [u8]) -> Result { let mut stream = self.connection.recv_stream(id); let mut chunks = stream.read(true)?; diff --git a/crates/network/src/p2p/streams/gossip_in.rs b/crates/network/src/p2p/streams/gossip_in.rs index cbb6cedf..ef572008 100644 --- a/crates/network/src/p2p/streams/gossip_in.rs +++ b/crates/network/src/p2p/streams/gossip_in.rs @@ -235,6 +235,14 @@ mod tests { fn remote_addr(&self) -> SocketAddr { "127.0.0.1:0".parse().unwrap() } + + fn write_bytes_to_stream( + &mut self, + _id: StreamId, + _data: bytes::Bytes, + ) -> Result { + unreachable!("read-only test io") + } } /// A frame shorter than the 10-byte length read, pipelined hard against diff --git a/crates/network/src/p2p/streams/gossip_out.rs b/crates/network/src/p2p/streams/gossip_out.rs index fcb7ba38..9e61d411 100644 --- a/crates/network/src/p2p/streams/gossip_out.rs +++ b/crates/network/src/p2p/streams/gossip_out.rs @@ -1,17 +1,25 @@ +use std::time::Instant; + +use bytes::Bytes; use silver_common::{MAX_GOSSIP_FRAME_SIZE, P2pStreamId, TRead}; -use crate::p2p::streams::{StreamError, StreamIo}; +use crate::{ + NetworkCounters, + p2p::streams::{StreamError, StreamIo, gossip_in::GOSSIP_BODY_STALL_TIMEOUT}, +}; /// Write-side state for gossipsub: idle → varint length → body. #[derive(Debug)] pub(crate) enum GossipWriteState { Idle, - /// Writing varint length prefix. + /// Writing varint length prefix. `last_write` is the last instant any + /// bytes were accepted by the stream — the write-stall clock. WritingLength { buffer: [u8; 10], limit: usize, written: usize, tcache: TRead, + last_write: Instant, }, /// Writing body. `offset`/`length` track progress into the current /// message; the handler provides body bytes via `send_data`. @@ -19,6 +27,7 @@ pub(crate) enum GossipWriteState { offset: usize, length: usize, tcache: TRead, + last_write: Instant, }, } @@ -28,14 +37,37 @@ enum Spin { } impl GossipWriteState { + /// Instant the in-progress write times out if the peer grants no more + /// credit; `None` when idle. Mirrors `GossipReadState`'s body stall. + pub(crate) fn deadline(&self) -> Option { + match self { + Self::WritingLength { last_write, .. } | Self::Writing { last_write, .. } => { + Some(*last_write + GOSSIP_BODY_STALL_TIMEOUT) + } + Self::Idle => None, + } + } + pub fn spin( mut self, io: &mut S, p2p_id: &P2pStreamId, + now: Instant, ) -> Result { loop { - match self.spin_inner(io, p2p_id)? { - Spin::Ok(gossip_write_state) => return Ok(gossip_write_state), + match self.spin_inner(io, p2p_id, now)? { + Spin::Ok(gossip_write_state) => { + if let Some(last_write) = match &gossip_write_state { + Self::WritingLength { last_write, .. } | + Self::Writing { last_write, .. } => Some(*last_write), + Self::Idle => None, + } && now.saturating_duration_since(last_write) > GOSSIP_BODY_STALL_TIMEOUT + { + tracing::warn!(?p2p_id, "gossip body write stalled"); + return Err(StreamError::GossipWriteStall); + } + return Ok(gossip_write_state); + } Spin::Next(gossip_write_state) => { self = gossip_write_state; } @@ -46,6 +78,7 @@ impl GossipWriteState { self, io: &mut S, p2p_id: &P2pStreamId, + now: Instant, ) -> Result { match self { GossipWriteState::Idle => match io.gossip_next() { @@ -60,29 +93,162 @@ impl GossipWriteState { silver_common::encode_varint(len, &mut buffer).inspect_err(|e| { tracing::error!(?e, len, "network gossiip write failed"); })?; - Ok(Spin::Next(Self::WritingLength { buffer, limit, written: 0, tcache })) + Ok(Spin::Next(Self::WritingLength { + buffer, + limit, + written: 0, + tcache, + last_write: now, + })) } None => Ok(Spin::Ok(Self::Idle)), }, - GossipWriteState::WritingLength { buffer, limit, mut written, tcache } => { - written += io.write_to_stream(p2p_id.stream_id(), &buffer[written..limit])?; + GossipWriteState::WritingLength { + buffer, + limit, + mut written, + tcache, + mut last_write, + } => { + let n = io.write_to_stream(p2p_id.stream_id(), &buffer[written..limit])?; + written += n; + if n > 0 { + last_write = now; + } if written == limit { return Ok(Spin::Next(Self::Writing { offset: 0, length: tcache.len()?, tcache, + last_write, })); } - Ok(Spin::Ok(Self::WritingLength { buffer, limit, written, tcache })) + Ok(Spin::Ok(Self::WritingLength { buffer, limit, written, tcache, last_write })) } - GossipWriteState::Writing { mut offset, length, tcache } => { - let (buffer, _) = tcache.buffer()?; - offset += io.write_to_stream(p2p_id.stream_id(), &buffer[offset..])?; + GossipWriteState::Writing { mut offset, length, tcache, mut last_write } => { + let Some(r_offset) = tcache.with_offset(offset) else { + tracing::error!(?p2p_id, "stale tcache read @ {}, skipping", tcache.seq()); + NetworkCounters::GossipMsgSkipped.inc(); + return Ok(Spin::Next(Self::Idle)); + }; + + let bytes = Bytes::from_owner(r_offset); + let n = io.write_bytes_to_stream(p2p_id.stream_id(), bytes)?; + offset += n; + if n > 0 { + last_write = now; + } if offset == length { return Ok(Spin::Next(Self::Idle)); } - Ok(Spin::Ok(Self::Writing { offset, length, tcache })) + Ok(Spin::Ok(Self::Writing { offset, length, tcache, last_write })) } } } } + +#[cfg(test)] +mod tests { + use std::{io::Write as _, net::SocketAddr, time::Duration}; + + use quinn_proto::StreamId; + use silver_common::{StreamProtocol, TCache, TCacheProducer, TProducer, TRandomAccess}; + + use super::*; + use crate::p2p::streams::AcquiredRpcOutbound; + + /// Write-only io: hands out one queued gossip message, then accepts at + /// most `budget` bytes per write call (0 = peer granting no credit). + struct MockIo { + pending: Option, + budget: usize, + } + + impl StreamIo for MockIo { + fn write_to_stream(&mut self, _id: StreamId, data: &[u8]) -> Result { + Ok(data.len().min(self.budget)) + } + + fn write_bytes_to_stream( + &mut self, + _id: StreamId, + data: Bytes, + ) -> Result { + Ok(data.len().min(self.budget)) + } + + fn read_from_stream(&mut self, _id: StreamId, _b: &mut [u8]) -> Result { + unreachable!("write-only test io") + } + + fn close_write(&mut self, _id: StreamId) -> Result<(), StreamError> { + Ok(()) + } + + fn rpc_next(&mut self) -> Option { + None + } + + fn gossip_next(&mut self) -> Option { + self.pending.take() + } + + fn remote_addr(&self) -> SocketAddr { + "127.0.0.1:0".parse().unwrap() + } + } + + /// The read holds a raw pointer to its consumer, and the consumer one to + /// the producer's cache: box the consumer so its address survives the + /// return, and order the tuple so the consumer drops before the producer. + fn queued_msg(name: &'static str) -> (Box, TProducer, TRead) { + let mut producer = TCache::producer(name, 1 << 16); + let mut consumer = Box::new(producer.cache_ref().random_access(name, false).unwrap()); + let mut reservation = producer.reserve(100, true).unwrap(); + reservation.write_all(&[0xaa; 100]).unwrap(); + reservation.flush().unwrap(); + let read = consumer.acquire(reservation.read()); + (consumer, producer, read) + } + + #[test] + fn stalled_write_times_out() { + let p2p_id = P2pStreamId::new(0, 4, StreamProtocol::GossipSub, false); + let (_consumer, _producer, msg) = queued_msg("test_gossip_wstall"); + let mut io = MockIo { pending: Some(msg), budget: 0 }; + + let t0 = Instant::now(); + let state = GossipWriteState::Idle.spin(&mut io, &p2p_id, t0).expect("blocked write parks"); + assert!(matches!(state, GossipWriteState::WritingLength { written: 0, .. })); + assert_eq!(state.deadline(), Some(t0 + GOSSIP_BODY_STALL_TIMEOUT)); + + let state = state + .spin(&mut io, &p2p_id, t0 + GOSSIP_BODY_STALL_TIMEOUT) + .expect("at the deadline is not past it"); + let err = state + .spin(&mut io, &p2p_id, t0 + GOSSIP_BODY_STALL_TIMEOUT + Duration::from_millis(1)) + .expect_err("stalled past deadline"); + assert!(matches!(err, StreamError::GossipWriteStall)); + } + + #[test] + fn progressing_write_does_not_time_out() { + let p2p_id = P2pStreamId::new(0, 4, StreamProtocol::GossipSub, false); + let (_consumer, _producer, msg) = queued_msg("test_gossip_wprogress"); + let mut io = MockIo { pending: Some(msg), budget: 0 }; + + let t0 = Instant::now(); + let state = GossipWriteState::Idle.spin(&mut io, &p2p_id, t0).expect("blocked write parks"); + + // Credit arrives late: the length prefix and part of the body drain, + // and the stall clock restarts from this progress. + io.budget = 10; + let late = t0 + GOSSIP_BODY_STALL_TIMEOUT + Duration::from_millis(1); + let state = state.spin(&mut io, &p2p_id, late).expect("progress refreshes the deadline"); + assert!(matches!( + state, + GossipWriteState::Writing { offset: 10, length: 100, last_write, .. } if last_write == late + )); + assert_eq!(state.deadline(), Some(late + GOSSIP_BODY_STALL_TIMEOUT)); + } +} diff --git a/crates/network/src/p2p/streams/mod.rs b/crates/network/src/p2p/streams/mod.rs index 42bf5b2a..2f3db040 100644 --- a/crates/network/src/p2p/streams/mod.rs +++ b/crates/network/src/p2p/streams/mod.rs @@ -1,14 +1,15 @@ use std::{array::TryFromSliceError, fmt, net::SocketAddr}; use buffa::DecodeError; +use bytes::Bytes; use quinn_proto::{FinishError, ReadError, ReadableError, StreamId, WriteError}; use silver_common::{TCacheError, TRead}; use thiserror::Error; use crate::p2p::streams::snappy::SnappyError; -mod gossip_in; -mod gossip_out; +pub(crate) mod gossip_in; +pub(crate) mod gossip_out; mod identify_in; mod identify_out; mod negotiate; @@ -42,6 +43,7 @@ pub enum StreamError { IdentifyTooBig, ReadResponseTimeout, GossipReadStall, + GossipWriteStall, } impl fmt::Display for StreamError { @@ -52,6 +54,7 @@ impl fmt::Display for StreamError { pub trait StreamIo { fn write_to_stream(&mut self, id: StreamId, data: &[u8]) -> Result; + fn write_bytes_to_stream(&mut self, id: StreamId, data: Bytes) -> Result; fn read_from_stream(&mut self, id: StreamId, data: &mut [u8]) -> Result; fn close_write(&mut self, id: StreamId) -> Result<(), StreamError>; fn rpc_next(&mut self) -> Option; diff --git a/crates/network/src/p2p/streams/negotiate.rs b/crates/network/src/p2p/streams/negotiate.rs index 3e519e9a..9cbcc56e 100644 --- a/crates/network/src/p2p/streams/negotiate.rs +++ b/crates/network/src/p2p/streams/negotiate.rs @@ -299,6 +299,14 @@ mod tests { fn remote_addr(&self) -> std::net::SocketAddr { "127.0.0.1:12345".parse().unwrap() } + + fn write_bytes_to_stream( + &mut self, + _id: StreamId, + _data: bytes::Bytes, + ) -> Result { + Ok(0) + } } fn sid() -> StreamId { diff --git a/crates/network/src/p2p/streams/rpc/request_in.rs b/crates/network/src/p2p/streams/rpc/request_in.rs index b17eec56..6a70a232 100644 --- a/crates/network/src/p2p/streams/rpc/request_in.rs +++ b/crates/network/src/p2p/streams/rpc/request_in.rs @@ -194,6 +194,14 @@ mod tests { fn remote_addr(&self) -> SocketAddr { "127.0.0.1:0".parse().unwrap() } + + fn write_bytes_to_stream( + &mut self, + _id: StreamId, + data: bytes::Bytes, + ) -> Result { + Ok(data.len()) + } } /// Regression: a by-root request body flows through `ReadingBody` into a diff --git a/crates/network/src/p2p/streams/rpc/response_in.rs b/crates/network/src/p2p/streams/rpc/response_in.rs index 63315324..5a8a9848 100644 --- a/crates/network/src/p2p/streams/rpc/response_in.rs +++ b/crates/network/src/p2p/streams/rpc/response_in.rs @@ -369,6 +369,14 @@ mod tests { fn remote_addr(&self) -> SocketAddr { "127.0.0.1:0".parse().unwrap() } + + fn write_bytes_to_stream( + &mut self, + _id: StreamId, + data: bytes::Bytes, + ) -> Result { + Ok(data.len()) + } } #[test] diff --git a/crates/network/src/p2p/streams/state.rs b/crates/network/src/p2p/streams/state.rs index ca160140..13951148 100644 --- a/crates/network/src/p2p/streams/state.rs +++ b/crates/network/src/p2p/streams/state.rs @@ -159,9 +159,18 @@ impl StreamState { StreamState::OutgoingRpc { rpc: RpcOut::ReadResponse(read_response), .. } => { read_response.deadline() } - StreamState::Gossip { - read: GossipReadState::ReadingBody { last_read, .. }, .. - } => Some(*last_read + GOSSIP_BODY_STALL_TIMEOUT), + StreamState::Gossip { read, write } => { + let read = match read { + GossipReadState::ReadingBody { last_read, .. } => { + Some(*last_read + GOSSIP_BODY_STALL_TIMEOUT) + } + _ => None, + }; + match (read, write.deadline()) { + (Some(r), Some(w)) => Some(r.min(w)), + (r, w) => r.or(w), + } + } _ => None, } } @@ -330,7 +339,7 @@ impl StreamState { } StreamState::Gossip { mut read, mut write } => { read = read.spin(io, &mut context.gossip_producer, id, now, emit)?; - write = write.spin(io, id)?; + write = write.spin(io, id, now)?; if matches!(read, GossipReadState::Closed) && id.is_incoming() { // read closed on incoming gossip stream - terminate. diff --git a/crates/network/src/tile.rs b/crates/network/src/tile.rs index 13176ac5..e40e857b 100644 --- a/crates/network/src/tile.rs +++ b/crates/network/src/tile.rs @@ -221,6 +221,13 @@ impl NetworkTile { .into()), ); } + p2p::SendResult::ConnectionClosing => { + tracing::debug!( + peer = msg.peer_id(), + protocol = ?msg.protocol(), + "send refused: connection closing" + ); + } p2p::SendResult::UnknownPeer => { // Can happen if peer has disconnected. tracing::debug!(peer=msg.peer_id(), protocol=?msg.protocol(), "Tried to send to unknown peer"); diff --git a/crates/storage/src/store/backfill.rs b/crates/storage/src/store/backfill.rs index f6c62bb0..905b14fb 100644 --- a/crates/storage/src/store/backfill.rs +++ b/crates/storage/src/store/backfill.rs @@ -289,7 +289,7 @@ impl ColumnBackfill { let expected = match self.pending.get_mut(&block_root) { Some(expected) => expected, None => { - tracing::warn!( + tracing::debug!( block_root = hex::encode(block_root), "unrequested backfill data column sidecar" ); diff --git a/crates/surfer/src/app.rs b/crates/surfer/src/app.rs index 0b7bf7f5..e2766f92 100644 --- a/crates/surfer/src/app.rs +++ b/crates/surfer/src/app.rs @@ -332,6 +332,11 @@ impl App { self.gossip_selected = Some(self.gossip_display_order[new]); } + /// Peers pane: jump selection to the first row of the current sort. + pub fn select_top_peer(&mut self) { + self.peers_selected = self.peers_display_order.first().copied(); + } + /// Peers pane: move the sort column left/right, wrapping. pub fn adjust_peers_sort(&mut self, dir: i32) { let n = crate::render::peers_pane::COLUMNS.len() as i32; diff --git a/crates/surfer/src/main.rs b/crates/surfer/src/main.rs index b91ec3ee..47621572 100644 --- a/crates/surfer/src/main.rs +++ b/crates/surfer/src/main.rs @@ -214,6 +214,7 @@ fn handle_key(app: &mut App, code: KeyCode, app_name: &str) { KeyCode::Char('r') if app.pane == app::Pane::Peers => { app.peers_sort_desc = !app.peers_sort_desc } + KeyCode::Char('t') if app.pane == app::Pane::Peers => app.select_top_peer(), KeyCode::Char('[') => app.adjust_split(-1), KeyCode::Char(']') => app.adjust_split(1), KeyCode::Char('p') => app.flamegraph.toggle_pause(), diff --git a/crates/surfer/src/render/peers_pane.rs b/crates/surfer/src/render/peers_pane.rs index d78798fe..df5cd90f 100644 --- a/crates/surfer/src/render/peers_pane.rs +++ b/crates/surfer/src/render/peers_pane.rs @@ -79,8 +79,9 @@ fn compare(a: &Key, b: &Key) -> Ordering { pub fn draw(f: &mut Frame, area: Rect, app: &mut App) { if app.peers.is_empty() { - let block = - Block::default().borders(Borders::ALL).title(" peers — ←/→ sort column · r reverse "); + let block = Block::default() + .borders(Borders::ALL) + .title(" peers — ←/→ sort column · r reverse · t top "); let inner = block.inner(area); f.render_widget(block, area); f.render_widget( @@ -124,8 +125,8 @@ pub fn draw(f: &mut Frame, area: Rect, app: &mut App) { .map(|s| s.user_agent.as_str()) .filter(|a| !a.is_empty()); let title = match sel_agent { - Some(agent) => format!(" peers — ←/→ sort column · r reverse · {agent} "), - None => " peers — ←/→ sort column · r reverse ".to_string(), + Some(agent) => format!(" peers — ←/→ sort column · r reverse · t top · {agent} "), + None => " peers — ←/→ sort column · r reverse · t top ".to_string(), }; let block = Block::default().borders(Borders::ALL).title(title); diff --git a/scripts/start_silver_with_ethrex.sh b/scripts/start_silver_with_ethrex.sh index 00a1f45f..2784acd0 100755 --- a/scripts/start_silver_with_ethrex.sh +++ b/scripts/start_silver_with_ethrex.sh @@ -10,7 +10,7 @@ fi [ -f "$JWT" ] || openssl rand -hex 32 > "$JWT" -pgrep -f 'ethrex --network mainnet' > /dev/null || systemd-run --scope -p MemoryMax=48G --user nohup ethrex \ +pgrep -f 'ethrex --network mainnet' > /dev/null || systemd-run --scope -p MemoryMax=48G -p AllowedCpus=7-15 --user nohup ethrex \ --network mainnet --datadir /home/ubuntu/.ethrex \ --authrpc.jwtsecret "$JWT" > logs/ethrex.log 2>&1 & From be9b1689717961f5b6f1632cef0a88c835a0bcb8 Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Wed, 26 Aug 2026 11:01:12 +0100 Subject: [PATCH 2/4] fix tests --- crates/common/src/spine/tcache/consumer.rs | 25 +++++++++++----------- crates/network/src/p2p/quic/peer.rs | 3 +-- scripts/start_silver_with_ethrex.sh | 2 +- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/crates/common/src/spine/tcache/consumer.rs b/crates/common/src/spine/tcache/consumer.rs index ae8a0f5c..4a1fd9ed 100644 --- a/crates/common/src/spine/tcache/consumer.rs +++ b/crates/common/src/spine/tcache/consumer.rs @@ -165,7 +165,7 @@ impl RandomAccessConsumer { }) .and_then(|ar| { // check slot seq. - self.cache.check_seq(read.seq).then(|| ar) + self.cache.check_seq(read.seq).then_some(ar) }) } @@ -251,7 +251,7 @@ impl AcquiredRead { let consumer = unsafe { &mut *(self.consumer as *mut RandomAccessConsumer) }; consumer .acquire_strict(self.read) - .and_then(|read| Some(AcquiredWithOffset { read, offset })) + .map(|read| AcquiredWithOffset { read, offset }) } } @@ -450,19 +450,19 @@ mod tests { /// newest acquire) must land at or above the tail and be tracked. #[test] fn buckets_out_of_order_acquire_within_guard() { - // guard = (1024/10).next_multiple_of(64) = 128. + // guard = (1024/5).next_multiple_of(64) = 256. let mut b = Buckets::new(64, 1024, 0); b.acquire(0); b.release(0, ""); - b.acquire(300); + b.acquire(500); // Tail rolled over the released bucket but held 128 back from - // bucket_start(300) = 256. - assert_eq!(b.tail_seq, 128); + // bucket_start(500) = 448. + assert_eq!(b.tail_seq, 192); // Late low acquire inside the window: tracked, not dropped. b.acquire(200); assert!(b.tail_seq <= 200); b.release(200, ""); - b.release(300, ""); + b.release(500, ""); } /// An acquire below the tail (out-of-order beyond the guard window) @@ -474,7 +474,7 @@ mod tests { let mut b = Buckets::new(64, 1024, 0); // guard 128, lag threshold 921 b.acquire(0); b.release(0, ""); - b.acquire(1000); // forces tail well past bucket 0 (tail = 832) + b.acquire(1000); // forces tail well past bucket 0 (tail = 704) let tail = b.tail_seq; assert!(tail >= 128); @@ -499,7 +499,7 @@ mod tests { b.acquire(200); // bucket 3, head = 200 assert_eq!(b.tail_seq, 0); b.release(0, ""); // bucket 0 empties; head far enough ahead to advance - b.acquire(300); + b.acquire(500); assert!(b.tail_seq > 0, "tail did not advance: {}", b.tail_seq); assert!(b.tail_seq <= 200); } @@ -509,13 +509,14 @@ mod tests { let mut b = Buckets::new(64, 1024, 0); b.acquire(0); // bucket 0 b.acquire(100); // bucket 1 - b.acquire(300); // bucket 4 + b.acquire(500); // bucket 8 // Release the middle first — bucket 1 empties but tail is still at 0 b.release(100, ""); assert_eq!(b.tail_seq, 0, "tail moved while bucket 0 still held"); - // Release the head; tail jumps past bucket 0 and bucket 1 (both empty) + // Release the head; tail jumps past bucket 0 and bucket 1 and bucket 2 (all + // empty) b.release(0, ""); - b.acquire(350); + b.acquire(450); assert!(b.tail_seq >= 128, "tail did not jump: {}", b.tail_seq); } diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index 3e33d61f..65bfee94 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -1688,9 +1688,9 @@ mod tests { #[test] fn inbound_stream_negotiation() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.server_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let server_stream_id = P2pStreamId::new( @@ -1702,7 +1702,6 @@ mod tests { server_h.send_gossip(server_stream_id, b"pong", &mut pair.server_peer); wait_for(&mut pair, &mut client_h, &mut server_h, 200, |c, _| !c.received.is_empty()); - assert!(!client_h.received.is_empty(), "client never received server-initiated data"); } diff --git a/scripts/start_silver_with_ethrex.sh b/scripts/start_silver_with_ethrex.sh index 2784acd0..8ebfd9dc 100755 --- a/scripts/start_silver_with_ethrex.sh +++ b/scripts/start_silver_with_ethrex.sh @@ -10,7 +10,7 @@ fi [ -f "$JWT" ] || openssl rand -hex 32 > "$JWT" -pgrep -f 'ethrex --network mainnet' > /dev/null || systemd-run --scope -p MemoryMax=48G -p AllowedCpus=7-15 --user nohup ethrex \ +pgrep -f 'ethrex --network mainnet' > /dev/null || taskset -c 7-15 systemd-run --scope -p MemoryMax=48G --user nohup ethrex \ --network mainnet --datadir /home/ubuntu/.ethrex \ --authrpc.jwtsecret "$JWT" > logs/ethrex.log 2>&1 & From fe8f15b022feec23377849298b49082edb353bcb Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Wed, 26 Aug 2026 11:05:36 +0100 Subject: [PATCH 3/4] fmt --- crates/common/src/spine/tcache/consumer.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/crates/common/src/spine/tcache/consumer.rs b/crates/common/src/spine/tcache/consumer.rs index 4a1fd9ed..d187ea60 100644 --- a/crates/common/src/spine/tcache/consumer.rs +++ b/crates/common/src/spine/tcache/consumer.rs @@ -249,9 +249,7 @@ impl AcquiredRead { pub fn with_offset(&self, offset: usize) -> Option { let consumer = unsafe { &mut *(self.consumer as *mut RandomAccessConsumer) }; - consumer - .acquire_strict(self.read) - .map(|read| AcquiredWithOffset { read, offset }) + consumer.acquire_strict(self.read).map(|read| AcquiredWithOffset { read, offset }) } } From 61fdfabad776a9bacabb32021c23a4d46696dfbe Mon Sep 17 00:00:00 2001 From: vladimir-ea Date: Wed, 26 Aug 2026 11:23:26 +0100 Subject: [PATCH 4/4] fix drop ordering in tests --- crates/network/src/p2p/quic/peer.rs | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index 65bfee94..411da90a 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -1376,8 +1376,8 @@ mod tests { #[test] fn stream_setup_timeout_reaps_unnegotiated_stream() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let t0 = Instant::now(); pair.client_peer.open_stream(StreamProtocol::Ping).unwrap(); @@ -1417,8 +1417,8 @@ mod tests { #[test] fn stopped_outbound_goodbye_closes_connection() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let now = Instant::now(); let stream = pair.client_peer.open_stream(StreamProtocol::Goodbye).unwrap(); @@ -1436,8 +1436,8 @@ mod tests { #[test] fn stopped_outbound_gossip_keeps_connection_open() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let now = Instant::now(); let stream = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); @@ -1460,10 +1460,10 @@ mod tests { /// as stream-credit exhaustion. #[test] fn closing_connection_refuses_sends_and_drops_streams() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); let now = Instant::now(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let stream_id = P2pStreamId::new( @@ -1506,9 +1506,9 @@ mod tests { gossip_out::GossipWriteState, }; - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let stream_id = P2pStreamId::new( @@ -1555,9 +1555,9 @@ mod tests { fn inbound_rpc_timeout_reaps_unanswered_stream() { use silver_common::{RpcInbound, RpcOutbound, RpcRequest, RpcRequestOutbound}; - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let t0 = Instant::now(); let request = RpcOutbound::Request(RpcRequestOutbound { @@ -1610,9 +1610,9 @@ mod tests { /// gossip-write state machine has crossed out of `NegotiateState`. #[test] fn outbound_stream_negotiation() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let stream_id = P2pStreamId::new( @@ -1630,9 +1630,9 @@ mod tests { #[test] fn outbound_stream_data_transfer() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let stream_id = P2pStreamId::new( @@ -1653,9 +1653,9 @@ mod tests { #[test] fn bidirectional_data_transfer() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let client_stream_id = P2pStreamId::new( @@ -1711,9 +1711,9 @@ mod tests { /// enqueues. #[test] fn multiple_streams() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); let sid = pair.client_peer.open_stream(StreamProtocol::GossipSub).unwrap(); let stream_id = P2pStreamId::new( @@ -1741,9 +1741,9 @@ mod tests { /// the second frame parks, then free and verify delivery resumes. #[test] fn tcache_full_park_and_retry() { - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); + let mut pair = PeerPair::new(); // Shrink the server's inbound gossip tcache: one 6 KB frame fits, // two don't. @@ -1806,10 +1806,10 @@ mod tests { fn goodbye_delivered_before_shutdown() { use silver_common::{RpcInbound, RpcOutbound, RpcRequest, RpcRequestOutbound}; - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); let now = Instant::now(); + let mut pair = PeerPair::new(); let goodbye = RpcOutbound::Request(RpcRequestOutbound { application_id: 0, @@ -1847,10 +1847,10 @@ mod tests { RpcResponseOutbound, ssz_view::STATUS_V2_SIZE, }; - let mut pair = PeerPair::new(); let mut client_h = PeerHarness::new(); let mut server_h = PeerHarness::new(); let now = Instant::now(); + let mut pair = PeerPair::new(); let request = RpcOutbound::Request(RpcRequestOutbound { application_id: 7,