[CL-129] send cells

crates/bin/src/main.rs

@@ -254,7 +254,6 @@ fn main() -> Result<(), Box<dyn Error>> {
         }
     }
 
-    // Validation only: a non-Off mode fails startup rather than being ignored.
     let partial_columns =
         config.partial_columns().map_err(|error| format!("partial columns config: {error:?}"))?;
 

crates/columns/src/cell_store.rs

@@ -378,6 +378,7 @@ impl CellStore {
             block_root: *root,
             column,
             slot: context.context.slot,
+            blob_count: context.context.blob_count,
             domain: context.domain,
             available: entry.admitted.0,
             full: entry.full.as_ref().map(|f| (f.read, f.cell_offset, f.proof_offset)),

crates/common/src/cell_store/mod.rs

@@ -211,6 +211,7 @@ pub struct ColumnAvailability {
     pub block_root: [u8; 32],
     pub column: usize,
     pub slot: u64,
+    pub blob_count: usize,
     pub domain: GossipDomain,
     pub available: u128,
     pub full: Option<(TCacheRead, usize, usize)>,

crates/common/src/gossip.rs

@@ -59,10 +59,12 @@ impl GossipDomain {
 
 /// Gossipsub 1.3 extensions announcement, sent as the first RPC on
 /// every stream we write to after negotiating meshsub 1.3. Length
-/// prefix 4, then `RPC { control(3) { extensions(6) {} } }` — empty
-/// because no extension is enabled yet.
+/// prefix 4, then `RPC { control(3) { extensions(6) {} } }` when no
+/// extensions are enabled.
 pub const GOSSIP_EXTENSIONS_ANNOUNCEMENT_FRAME: &[u8] = &[4, 0x1A, 2, 0x32, 0];
 
+pub const GOSSIP_PARTIAL_EXTENSIONS_ANNOUNCEMENT_FRAME: &[u8] = &[6, 0x1A, 4, 0x32, 2, 0x50, 1];
+
 /// Eth2 gossipsub topic name. Wire topic is
 /// `/eth2/{fork_digest_hex}/{name}/ssz_snappy`; this enum covers the `{name}`
 /// portion. Subnet ids travel inline.

crates/common/src/lib.rs

@@ -8,8 +8,9 @@ pub use spine::{
 pub use crate::{
     error::Error,
     gossip::{
-        ATTESTATION_SUBNETS, GOSSIP_EXTENSIONS_ANNOUNCEMENT_FRAME, GOSSIP_TOPIC_COUNTER_SLOTS,
-        GossipDomain, GossipTopic, MAX_GOSSIP_COMPRESSED_PAYLOAD_SIZE, MAX_GOSSIP_FRAME_SIZE,
+        ATTESTATION_SUBNETS, GOSSIP_EXTENSIONS_ANNOUNCEMENT_FRAME,
+        GOSSIP_PARTIAL_EXTENSIONS_ANNOUNCEMENT_FRAME, GOSSIP_TOPIC_COUNTER_SLOTS, GossipDomain,
+        GossipTopic, MAX_GOSSIP_COMPRESSED_PAYLOAD_SIZE, MAX_GOSSIP_FRAME_SIZE,
         MAX_GOSSIP_UNCOMPRESSED_PAYLOAD_SIZE, MESSAGE_ID_LEN, MessageId, MessageIdHasher,
         SYNC_COMMITTEE_SUBNETS, gossip_topic_for_counter_slot, msg_id_invalid_snappy,
         msg_id_valid_snappy,

crates/common/src/spine/messages.rs

@@ -417,9 +417,6 @@ 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,
         /// Response targeted a stream already closed/reset, as opposed to
         /// stream-credit exhaustion opening a new request stream.
         stream_gone: bool,
@@ -440,10 +437,12 @@ pub enum PeerEvent {
         first_chunk_ms: u64,
         elapsed_ms: u64,
     },
+    /// A send was rejected or an older queued message was evicted.
+    /// This event owns send-failure accounting; stream errors are diagnostics.
     P2pOutboundMessageDropped {
         p2p_peer: usize,
         protocol: StreamProtocol,
-        rpc_request: bool,
+        msg: P2pSend,
     },
     P2pGossipTopicSubscribe {
         p2p_peer: usize,
@@ -787,7 +786,13 @@ pub enum RpcSeverity {
 #[allow(clippy::large_enum_variant)]
 pub enum P2pSend {
     Gossip(GossipMsgOut),
-    SegmentedGossip { peer_id: usize, frame: CacheFrameRef },
+    SegmentedGossip {
+        peer_id: usize,
+        frame: CacheFrameRef,
+        /// Count cells when Network finishes writing a partial response.
+        /// None for generic frames and best-effort availability withdrawals.
+        partial_cells: Option<u8>,
+    },
     Identify(usize),
     Rpc(RpcOutbound),
 }

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

@@ -167,7 +167,14 @@ impl CacheFrameRef {
             return Err(CacheFrameError::InvalidDescriptor);
         }
         let descriptor_len = buffer.len();
-        let view = CacheFrameView { read, count, wire_len, framing_start, descriptor_len };
+        let view = CacheFrameView {
+            read,
+            expires: self.expires,
+            count,
+            wire_len,
+            framing_start,
+            descriptor_len,
+        };
         let mut total = 0usize;
         for segment in view.segments() {
             if segment.length == 0 ||
@@ -192,13 +199,18 @@ impl CacheFrameRef {
 #[derive(Debug)]
 pub struct CacheFrameView {
     read: AcquiredRead,
+    expires: Instant,
     count: usize,
     wire_len: usize,
     framing_start: usize,
     descriptor_len: usize,
 }
 
 impl CacheFrameView {
+    pub fn reference(&self) -> CacheFrameRef {
+        CacheFrameRef { descriptor: self.read.read, expires: self.expires }
+    }
+
     pub fn wire_len(&self) -> usize {
         self.wire_len
     }

crates/common/src/spine/tcache/cache_frame/tests.rs

@@ -35,6 +35,10 @@ fn copy_handle_round_trips_framing_and_source_ranges() {
     )
     .unwrap();
     let view = frame.acquire(&mut consumer, now).unwrap();
+    let restored = view.reference();
+    assert_eq!(restored.read().seq(), frame.read().seq());
+    assert_eq!(restored.read().cache_ref().cache, frame.read().cache_ref().cache);
+    assert_eq!(restored.expires, frame.expires);
     assert_eq!(view.wire_len(), 8);
     assert_eq!(view.segment_count(), 3);
     let descriptor = view.descriptor_range();

crates/config/src/lib.rs

@@ -546,11 +546,11 @@ impl Config {
         self.attestation_subnet_count
     }
 
-    /// Validated partial-columns mode. Non-Off modes are unsupported
-    /// and rejected rather than silently ignored.
+    /// Receiving partial columns remains disabled until validation and request
+    /// scheduling are connected.
     pub fn partial_columns(&self) -> Result<PartialColumnsMode, Error> {
         match self.partial_columns {
-            PartialColumnsMode::Off => Ok(PartialColumnsMode::Off),
+            PartialColumnsMode::Off | PartialColumnsMode::SendOnly => Ok(self.partial_columns),
             mode => Err(Error::ConfigError(format!("partial_columns {mode:?} is not supported"))),
         }
     }
@@ -598,7 +598,6 @@ mod tests {
         assert_eq!(cfg.partial_columns().unwrap(), PartialColumnsMode::Off);
     }
 
-    /// Non-Off partial modes are unsupported.
     #[test]
     fn partial_columns_modes_are_validated() {
         let base = r#"
@@ -608,6 +607,8 @@ mod tests {
         "#;
         let cfg: Config =
             toml::from_str(&format!("{base}partial_columns = \"send_only\"")).unwrap();
+        assert_eq!(cfg.partial_columns().unwrap(), PartialColumnsMode::SendOnly);
+        let cfg: Config = toml::from_str(&format!("{base}partial_columns = \"enabled\"")).unwrap();
         let err = format!("{:?}", cfg.partial_columns().unwrap_err());
         assert!(err.contains("not supported"), "{err}");
     }

crates/control/src/cell_allocator.rs

@@ -58,6 +58,10 @@ impl CellAllocator {
         &self.producer
     }
 
+    pub fn slot_window(&self) -> (u64, Instant) {
+        (self.slot, self.slot_end)
+    }
+
     pub fn allocate(&mut self, request: AssemblyRequest) -> Result<AssemblySet, StoreError> {
         let context = request.context;
         if context.slot < self.min_slot {

crates/control/src/cell_ingress.rs

@@ -1,10 +1,10 @@
-use std::time::Instant;
+use std::{ptr, time::Instant};
 
 use flux::spine::SpineProducers;
 use fxhash::FxHashMap;
 use silver_common::{
-    ColumnOrigin, DataColumnsEvent, GossipTopic, PeerControl, SilverSpineProducers, SszCache,
-    TProducer, TRandomAccess,
+    ColumnOrigin, DataColumnsEvent, ForkName, GossipTopic, PeerControl, PeerEvent,
+    SilverSpineProducers, SszCache, TProducer, TRandomAccess,
     cell_store::{
         CellKey, CellStoreConfig, CellStoreEvent, ColumnAvailability, PendingCell, StoreError,
     },
@@ -122,15 +122,22 @@ impl CellIngress {
             CellStoreEvent::Available(update)
                 if now < update.expires && update.slot >= self.min_slot =>
             {
-                let key = (update.block_root, update.column);
-                if self.available.contains_key(&key) || self.available.len() < self.capacity {
-                    self.available.insert(key, update);
-                }
+                self.update_availability(update, now);
             }
             _ => {}
         }
     }
 
+    pub(crate) fn update_availability(&mut self, update: ColumnAvailability, now: Instant) {
+        let key = (update.block_root, update.column);
+        if now < update.expires &&
+            update.slot >= self.min_slot &&
+            (self.available.contains_key(&key) || self.available.len() < self.capacity)
+        {
+            self.available.insert(key, update);
+        }
+    }
+
     pub fn availability(
         &self,
         root: &[u8; 32],
@@ -143,6 +150,40 @@ impl CellIngress {
             .filter(|update| now < update.expires && update.slot >= self.min_slot)
     }
 
+    pub fn slot_window(&self) -> (u64, Instant) {
+        self.allocator.slot_window()
+    }
+
+    pub fn columns(&self, now: Instant) -> impl Iterator<Item = ColumnAvailability> + '_ {
+        self.available
+            .values()
+            .copied()
+            .filter(move |column| now < column.expires && column.slot >= self.min_slot)
+    }
+
+    pub fn serving_column(&self, event: &PeerEvent, now: Instant) -> Option<ColumnAvailability> {
+        let (topic, digest, full) = match event {
+            PeerEvent::SendGossip {
+                topic, domain, ssz, ssz_cache: SszCache::DataColumns, ..
+            } => (*topic, domain.digest(), Some(*ssz)),
+            PeerEvent::OutboundIHave { topic, digest, .. } => (*topic, *digest, None),
+            _ => return None,
+        };
+        let GossipTopic::DataColumnSidecar(index) = topic else { return None };
+        self.columns(now).find(|column| {
+            column.column == index as usize &&
+                column.domain.digest() == digest &&
+                column.available != 0 &&
+                (column.domain.format() != ForkName::Fulu || column.header.is_some()) &&
+                full.is_none_or(|read| {
+                    column.full.is_some_and(|(source, ..)| {
+                        source.seq() == read.seq() &&
+                            ptr::eq(&*source.cache_ref(), &*read.cache_ref())
+                    })
+                })
+        })
+    }
+
     pub fn stage_cell(
         &self,
         key: CellKey,

crates/control/src/counters.rs

@@ -1,4 +1,5 @@
 silver_common::declare_counters! {
+    #[allow(non_camel_case_types)]
     pub ControlCounters => "control" {
         TailUnavailable,
         RangesIssued,
@@ -8,5 +9,19 @@ silver_common::declare_counters! {
         RootNeedsStalled,
         RootNeedsTracked,
         RootNeedsRefused,
+        PartialMetadataReceived,
+        PartialMetadataReplaced,
+        PartialMetadataIgnored,
+        PartialStateLimited,
+        PartialFramesQueued,
+        _Reserved_PartialFramesWritten,
+        PartialFramesDropped,
+        _Reserved_PartialCellsServed,
+        PartialWithdrawals,
+        PartialExchanges,
+        _Reserved_PartialPendingFrames,
+        _Reserved_PartialResponsesSent,
+        PartialCellsRequested,
+        PartialRateLimited,
     }
 }

crates/control/src/lib.rs

@@ -2,6 +2,7 @@ pub mod cell_allocator;
 pub mod cell_ingress;
 pub mod cluster;
 mod counters;
+mod partial_exchange;
 pub mod sync_engine;
 mod tile;