cache data columns with missing parent

crates/beacon_state/data/src/view.rs

@@ -202,6 +202,7 @@ impl BeaconStateReader {
     /// inactivity boxes) are safe to read optimistically. The pending /
     /// longtail bases are realloc-prone `Vec`s — reading their CONTENT here
     /// can race a finalize realloc; those reads need the lock-guarded path.
+    #[timed]
     pub fn read<F, R>(&self, reader: &F) -> Option<R>
     where
         F: Fn(StateReadView<'_>) -> R,

crates/common/src/spine/messages.rs

@@ -300,13 +300,17 @@ pub enum PeerEvent {
     P2pCannotCreateStream {
         p2p_peer: usize,
         protocol: StreamProtocol,
+        /// Failed send was an outbound RPC request: the PM must release the
+        /// `outbound_in_flight` slot admitted for it, else it leaks.
+        rpc_request: bool,
     },
     P2pStreamClosed {
         stream_id: P2pStreamId,
     },
     P2pOutboundMessageDropped {
         p2p_peer: usize,
         protocol: StreamProtocol,
+        rpc_request: bool,
     },
     P2pGossipTopicSubscribe {
         p2p_peer: usize,

crates/common/src/spine/tcache/consumer.rs

@@ -166,9 +166,9 @@ impl RandomAccessConsumer {
                     );
                     tail = if self.last_head > tail { self.last_head } else { head };
                     self.active.roll_to(tail);
+                    self.last_read = Nanos::now();
                 }
                 self.last_head = head;
-                self.last_read = Nanos::now();
             }
 
             cache_head.tails[self.index].store(tail, Ordering::Release);

crates/common/src/spine/tcache/producer.rs

@@ -66,7 +66,8 @@ impl TCacheProducer for Producer {
     fn reserve(&mut self, len: usize, auto_commit: bool) -> Option<Reservation> {
         let tcache = unsafe { &*self.cache };
         if tcache.reserve_len(self.seq, len) > self.space as usize ||
-            self.seq - self.published_seq > (tcache.len >> 1) as u64
+            self.seq - self.published_seq > (tcache.len >> 4) as u64
+        // for 32MB buffer, publish head for every 2MB reserved
         {
             // try reclaim space.
             self.publish_head();
@@ -216,6 +217,12 @@ impl Write for Reservation {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         let buffer = self.cache.write(self.seq).map_err(std::io::Error::other)?;
         if buf.len() + self.offset > buffer.len() {
+            tracing::error!(
+                reservation_len = buffer.len(),
+                offset = self.offset,
+                data_len = buf.len(),
+                "tried to write > reservation"
+            );
             return Err(std::io::ErrorKind::FileTooLarge.into());
         }
         buffer[self.offset..self.offset + buf.len()].copy_from_slice(buf);

crates/control/src/sync_engine/select.rs

@@ -23,11 +23,21 @@ pub(super) fn select_target(ctx: &Ctx, current: SyncUpdate, force_resync: bool)
         }
     }
 
+    let mut force_head_resync = force_resync;
     if let SyncUpdate::SyncingHead { head_root, head_slot } = current {
         let reached = local_head_slot >= head_slot;
         if !reached && !ctx.peers.is_rejected(&head_root) && ctx.peers.backs_head(&head_root) {
             return SyncUpdate::SyncingHead { head_root, head_slot };
         }
+        // Pin reached (or lost its backers) while still behind the wall: the
+        // pin was stale the moment the chain moved on. Chase a fresh peer head
+        // with the lag gate suppressed — falling through would land in
+        // Following inside the `head_lag_threshold_slots` dead band, leaving
+        // the gap to gossip recovery. Head sync ends only when no backed peer
+        // head is ahead of us.
+        if local_head_slot < wall_slot {
+            force_head_resync = true;
+        }
     }
 
     if let Some((epoch, root)) = ctx.peers.best_finalized_target(
@@ -41,7 +51,7 @@ pub(super) fn select_target(ctx: &Ctx, current: SyncUpdate, force_resync: bool)
     }
 
     if let Some((head_root, head_slot)) =
-        ctx.peers.best_head_target(local_head_slot, wall_slot, &ctx.cfg, force_resync)
+        ctx.peers.best_head_target(local_head_slot, wall_slot, &ctx.cfg, force_head_resync)
     {
         return SyncUpdate::SyncingHead { head_root, head_slot };
     }

crates/gossip/src/message.rs

@@ -26,6 +26,7 @@ pub(super) fn handle_incoming(
     emit: &mut impl FnMut(GossipHandlerEvent),
 ) -> Result<(), Error> {
     validate_compressed_payload_size(snappy_data.len()).inspect_err(|_| {
+        tracing::warn!(?stream_id, topic_string, "invalid gossip frame");
         emit(GossipHandlerEvent::PeerEvent(PeerEvent::P2pGossipInvalidFrame {
             p2p_peer: stream_id.peer(),
         }));
@@ -57,15 +58,18 @@ pub(super) fn handle_incoming(
         .reserve(len, false)
         .ok_or(Error::BufferTooSmall)
         .inspect_err(|e| {
-            tracing::error!(?e, len, "failed to reserve incoming gossip SSZ");
+            tracing::error!(?e, len, topic_string, "failed to reserve incoming gossip SSZ");
         })?;
 
     let msg_id = decompress_to_reservation(
         incoming_gossip_publish,
         snappy_data,
         &mut reservation,
         topic_string,
-    )?;
+    )
+    .inspect_err(|e| {
+        tracing::error!(?stream_id, ?e, topic_string, "failed to decompress gossip msg")
+    })?;
 
     if !dedup_cache.insert(fast_id, msg_id) {
         // Second dedup check. Different snappy bytes can decompress to the same message

crates/network/src/p2p/streams/gossip_in.rs

@@ -10,7 +10,9 @@ use crate::p2p::streams::{StreamError, StreamIo};
 /// Read-side state for gossipsub: varint length prefix then body.
 #[derive(Debug)]
 pub(crate) enum GossipReadState {
-    /// Reading varint length prefix (byte by byte).
+    /// Reading varint length prefix. The read is chunked, so `buf` may hold
+    /// bytes past the varint — body bytes, and for frames shorter than the
+    /// buffer even the next frame's start.
     ReadingLength { buf: [u8; 10], read: usize },
     /// Have read length but buffer needs to be allocated.
     AllocBody { length: usize, buf: [u8; 10], buf_start: usize, buf_end: usize },
@@ -94,17 +96,36 @@ impl GossipReadState {
                     // write stream_id as header
                     let _ = reservation.write(p2p_id.as_ref())?;
 
-                    // write any remainder bytes
+                    // The length read may have pulled bytes past this frame's
+                    // body; the excess belongs to the next frame.
+                    let take = (buf_end - buf_start).min(length);
                     let mut remaining = length;
-                    if buf_end > buf_start {
-                        remaining -= reservation.write(&buf[buf_start..buf_end])?;
+                    if take > 0 {
+                        remaining -= reservation.write(&buf[buf_start..buf_start + take])?;
+                    }
+                    if remaining == 0 {
+                        assert!(reservation.is_committed());
+                        tracing::warn!(
+                            ?p2p_id,
+                            length,
+                            frame = %format_args!("{:02x?}", &buf[buf_start..buf_start + take]),
+                            "tiny gossip frame"
+                        );
+                        let excess = buf_end - buf_start - take;
+                        let mut next = [0u8; 10];
+                        next[..excess].copy_from_slice(&buf[buf_start + take..buf_end]);
+                        return Ok(Spin::Next(Self::ReadingLength { buf: next, read: excess }));
                     }
                     return Ok(Spin::Next(Self::ReadingBody { reservation, remaining }));
                 }
                 Ok(Spin::Ok(Self::AllocBody { length, buf, buf_start, buf_end }))
             }
             GossipReadState::ReadingBody { mut reservation, mut remaining } => {
-                let n = io.read_from_stream(p2p_id.stream_id(), reservation.remaining_buffer()?)?;
+                let n = io
+                    .read_from_stream(p2p_id.stream_id(), reservation.remaining_buffer()?)
+                    .inspect_err(|e| {
+                        tracing::error!(?e, ?p2p_id, remaining, "reservation write failed");
+                    })?;
                 reservation.increment_offset(n);
                 remaining -= n;
                 if remaining == 0 {
@@ -128,7 +149,81 @@ fn checked_frame_length(length: u64) -> Result<usize, StreamError> {
 
 #[cfg(test)]
 mod tests {
+    use std::net::SocketAddr;
+
+    use quinn_proto::StreamId;
+    use silver_common::{StreamProtocol, TCache, TRead};
+
     use super::*;
+    use crate::p2p::streams::AcquiredRpcOutbound;
+
+    struct MockIo {
+        data: Vec<u8>,
+        pos: usize,
+    }
+
+    impl StreamIo for MockIo {
+        fn write_to_stream(&mut self, _id: StreamId, _data: &[u8]) -> Result<usize, StreamError> {
+            unreachable!("read-only test io")
+        }
+
+        fn read_from_stream(
+            &mut self,
+            _id: StreamId,
+            buf: &mut [u8],
+        ) -> Result<usize, StreamError> {
+            let n = (self.data.len() - self.pos).min(buf.len());
+            buf[..n].copy_from_slice(&self.data[self.pos..self.pos + n]);
+            self.pos += n;
+            Ok(n)
+        }
+
+        fn close_write(&mut self, _id: StreamId) -> Result<(), StreamError> {
+            Ok(())
+        }
+
+        fn rpc_next(&mut self) -> Option<AcquiredRpcOutbound> {
+            None
+        }
+
+        fn gossip_next(&mut self) -> Option<TRead> {
+            None
+        }
+
+        fn remote_addr(&self) -> SocketAddr {
+            "127.0.0.1:0".parse().unwrap()
+        }
+    }
+
+    /// A frame shorter than the 10-byte length read, pipelined hard against
+    /// the next frame: the length read grabs the whole frame plus the next
+    /// frame's start. Regression: the remainder write overflowed the
+    /// reservation (live: `FileTooLarge` teardown right after a meshsub
+    /// 1.3.0→1.2.0 renegotiation, whose peers pipeline a 6-byte first frame).
+    #[test]
+    fn pipelined_small_frame_crosses_length_read() {
+        let mut wire = vec![6u8];
+        wire.extend_from_slice(b"aaaaaa");
+        wire.push(12);
+        wire.extend_from_slice(b"bbbbbbbbbbbb");
+
+        let mut producer = TCache::producer("test_gossip_pipelined", 1 << 16);
+        let mut consumer = producer.cache_ref().consumer("t").expect("consumer");
+        let p2p_id = P2pStreamId::new(0, 4, StreamProtocol::GossipSub, true);
+        let mut io = MockIo { data: wire, pos: 0 };
+
+        let state = GossipReadState::default()
+            .spin(&mut io, &mut producer, &p2p_id)
+            .expect("pipelined small frame must not error");
+        assert!(matches!(state, GossipReadState::ReadingLength { read: 0, .. }));
+
+        let header = size_of::<P2pStreamId>();
+        let (frame, _) = consumer.read().expect("first frame");
+        assert_eq!(&frame[header..], b"aaaaaa");
+        consumer.free();
+        let (frame, _) = consumer.read().expect("second frame");
+        assert_eq!(&frame[header..], b"bbbbbbbbbbbb");
+    }
 
     #[test]
     fn frame_length_boundaries() {

crates/network/src/tile.rs

@@ -167,6 +167,7 @@ impl NetworkTile {
             }
 
             if !adapter.consume_one(|msg: P2pSend, producers| {
+                let rpc_request = matches!(&msg, P2pSend::Rpc(RpcOutbound::Request(_)));
                 let result = match msg {
                     P2pSend::Gossip(gossip_msg_out) => {
                         gossips += 1;
@@ -188,6 +189,7 @@ impl NetworkTile {
                             &(PeerEvent::P2pCannotCreateStream {
                                 p2p_peer: msg.peer_id(),
                                 protocol: msg.protocol(),
+                                rpc_request,
                             }
                             .into()),
                         );
@@ -197,6 +199,7 @@ impl NetworkTile {
                             &(PeerEvent::P2pOutboundMessageDropped {
                                 p2p_peer: msg.peer_id(),
                                 protocol: msg.protocol(),
+                                rpc_request,
                             }
                             .into()),
                         );

crates/peer/src/manager.rs

@@ -485,10 +485,16 @@ impl PeerManager {
                 self.dialing.remove(&peer_id);
                 self.on_disconnected(p2p_peer, now, "transport", emit);
             }
-            PeerEvent::P2pCannotCreateStream { p2p_peer, .. } => {
+            PeerEvent::P2pCannotCreateStream { p2p_peer, protocol, rpc_request } => {
+                if rpc_request {
+                    self.release_outbound_in_flight(p2p_peer, protocol);
+                }
                 self.add_behaviour_penalty(p2p_peer, 1.0, "cannot create stream");
             }
-            PeerEvent::P2pOutboundMessageDropped { p2p_peer, .. } => {
+            PeerEvent::P2pOutboundMessageDropped { p2p_peer, protocol, rpc_request } => {
+                if rpc_request {
+                    self.release_outbound_in_flight(p2p_peer, protocol);
+                }
                 self.add_behaviour_penalty(p2p_peer, 1.0, "outbound message dropped");
             }
             PeerEvent::P2pStreamClosed { stream_id } => {
@@ -1310,6 +1316,24 @@ impl PeerManager {
         }
     }
 
+    /// A request admitted by `admit_outbound_request` never reached the wire:
+    /// no response terminator or stream close will ever fire for it, so the
+    /// in-flight slot must be released here or it leaks until disconnect
+    /// (2 leaks brick the peer for that protocol —
+    /// `MAX_RPC_PROTOCOL_IN_FLIGHT`).
+    fn release_outbound_in_flight(&mut self, conn: usize, protocol: StreamProtocol) {
+        if let Some(peer) = self.peers.get_mut(&conn) {
+            let ord = protocol.ordinal() as usize;
+            peer.outbound_in_flight[ord] = peer.outbound_in_flight[ord].saturating_sub(1);
+            tracing::debug!(
+                conn,
+                ?protocol,
+                in_flight = peer.outbound_in_flight[ord],
+                "released in-flight slot for failed send"
+            );
+        }
+    }
+
     fn add_behaviour_penalty(&mut self, conn: usize, delta: f64, offence: &'static str) {
         if let Some(peer) = self.peers.get_mut(&conn) {
             peer.behaviour_penalty += delta;

crates/storage/src/store.rs

@@ -390,6 +390,7 @@ impl Store {
         });
     }
 
+    #[timed]
     pub(super) fn add_block(
         &mut self,
         block_root: [u8; 32],
@@ -401,10 +402,7 @@ impl Store {
         // settled — finalized blocks already live in the flat store and
         // anything else is an orphan. Dedup repeats. Canonicity is resolved
         // by the head walk at query time and by finalization promotion.
-        if slot <= self.finalized_slot ||
-            self.unfinalized.contains(&block_root) ||
-            self.root_index.contains_key(&block_root)
-        {
+        if slot <= self.finalized_slot || self.has_block(&block_root) {
             return;
         }
         self.unfinalized.insert(block_root, slot, parent_root);
@@ -537,6 +535,13 @@ impl Store {
         self.head_slot
     }
 
+    /// BS-accepted block, any fork: every `PersistBlock` lands in
+    /// `unfinalized` (or `root_index` once promoted), so membership here is
+    /// "validated", independent of the current head chain.
+    pub(super) fn has_block(&self, root: &[u8; 32]) -> bool {
+        self.unfinalized.contains(root) || self.root_index.contains_key(root)
+    }
+
     pub(super) fn finalized_slot(&self) -> u64 {
         self.finalized_slot
     }