remove allocation on outbound gossip

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`).

crates/bin/src/main.rs

@@ -140,7 +140,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         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))),

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,

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;

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<RandomAccessConsumer, Error> {
+        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<RandomAccessConsumer, Error> {
+        self.ra_consumer(name, auto_free, true)
+    }
+
+    fn ra_consumer(
+        &self,
+        name: &'static str,
+        auto_free: bool,
+        strict: bool,
     ) -> Result<RandomAccessConsumer, Error> {
         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::<Slot>();
 

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<AcquiredRead> {
+        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_some(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,11 @@ impl AcquiredRead {
         }
         consumer.cache.read(self.read.seq).map(|(data, _, ts)| (data, ts))
     }
+
+    pub fn with_offset(&self, offset: usize) -> Option<AcquiredWithOffset> {
+        let consumer = unsafe { &mut *(self.consumer as *mut RandomAccessConsumer) };
+        consumer.acquire_strict(self.read).map(|read| AcquiredWithOffset { read, offset })
+    }
 }
 
 impl Deref for AcquiredRead {
@@ -252,6 +278,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 +310,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,20 +342,24 @@ 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
         // live bucket (the matching release is dropped below tail). Skip;
         // 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 +389,7 @@ impl Buckets {
             }
             self.tail_seq += self.bucket_size;
         }
+        true
     }
 
     fn release(&mut self, seq: u64, name: &str) {
@@ -383,19 +448,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)
@@ -407,7 +472,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);
 
@@ -432,7 +497,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);
     }
@@ -442,13 +507,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);
     }
 

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,
     }
 }
 

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,
 }