[CL-129] partial payloads: wire cell store into tiles

crates/bin/src/main.rs

@@ -17,12 +17,15 @@ use rand::RngCore;
 use silver_application_boundary::ApplicationBoundaryTile;
 use silver_beacon_state::{BeaconStateTile, SlotTicker};
 use silver_beacon_state_data::{BeaconState, SLOTS_PER_EPOCH};
-use silver_columns::tile::{ColumnConsumers, DataColumnsTile};
+use silver_columns::{
+    cell_store::CellStoreConfig,
+    tile::{ColumnConsumers, DataColumnsTile},
+};
 #[cfg(feature = "alloc-profile")]
 use silver_common::metrics::CountingAllocator;
 use silver_common::{
-    APP_NAME, Enr, ProtoIdentify, SilverSpine, TCache, TCacheProducer, profiler::enable_profiler,
-    tracing::initialise_tracing_log,
+    APP_NAME, Enr, ProtoIdentify, SilverSpine, TCache, TCacheProducer,
+    cells::GOSSIP_DELIVERY_RETENTION, profiler::enable_profiler, tracing::initialise_tracing_log,
 };
 use silver_config::Config;
 use silver_control::{Controller, cluster::AttestationClusterConfig, sync_engine::SyncEngine};
@@ -150,6 +153,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     tracing::info!(enr = local_enr.to_base64(), "local ENR on startup");
 
     let chain_config = config.chain_config();
+    let spec = Arc::new(chain_config.spec.clone());
     sleep_until_genesis(chain_config.genesis_unix_secs);
     let ticker = SlotTicker::new(
         chain_config.genesis_unix_secs,
@@ -216,8 +220,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         trusted_ips,
     );
     let identify = config.identify()?;
-    let p2p_context = Context {
-        data_columns_consumer: None,
+    let mut p2p_context = Context {
         gossip_producer: incoming_gossip_producer,
         gossip_consumer: outgoing_gossip_producer
             .cache_ref()
@@ -228,6 +231,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         cluster_nodes: cluster_nodes.map(ClusterNodes::new),
         cluster_inbound_producer,
         cluster_outbound_consumer,
+        data_columns_consumer: None,
     };
 
     let now = Instant::now();
@@ -249,15 +253,23 @@ fn main() -> Result<(), Box<dyn Error>> {
         }
     }
 
+    let cell_config =
+        CellStoreConfig::new(spec.clone(), das_custody_groups, GOSSIP_DELIVERY_RETENTION)
+            .map_err(|error| format!("cell store configuration: {error:?}"))?;
+    let data_columns_producer = TCache::producer("data_columns", cell_config.cache_capacity());
+    let columns_consumer =
+        data_columns_producer.cache_ref().retained_random_access("columns_cells")?;
+    p2p_context.data_columns_consumer =
+        Some(Box::new(data_columns_producer.cache_ref().retained_random_access("network_cells")?));
+    let (cell_slot, cell_slot_start) = ticker.current_slot_start();
+
     let network_tile = NetworkTile::new(discv5_addr, discv5, p2p_addr, p2p_endpoint, p2p_context)?;
 
     let (checkpoint, checkpoint_pubkeys) = load_checkpoint(&config)?;
     let booting_from_local_checkpoint = !checkpoint.is_empty();
 
     tracing::info!("booting from local checkpoint: {booting_from_local_checkpoint}");
 
-    let spec = Arc::new(chain_config.spec.clone());
-
     let gossip_handler = GossipHandler::new(
         incoming_gossip_consumer,
         ssz_gossip_producer,
@@ -289,6 +301,9 @@ fn main() -> Result<(), Box<dyn Error>> {
             spec.clone(),
         ),
     )?;
+    control_tile = control_tile
+        .with_data_columns_cache(cell_config, data_columns_producer, cell_slot, cell_slot_start)
+        .map_err(|error| format!("cell store construction: {error:?}"))?;
     control_tile.set_pending_subnet_topics(
         silver_common::attnet_subnets(subnets)
             .map(silver_common::GossipTopic::BeaconAttestation)
@@ -349,7 +364,8 @@ fn main() -> Result<(), Box<dyn Error>> {
             chain_config.slot_duration(),
             chain_config.playload_lookahead(),
         ),
-    );
+    )
+    .with_data_columns_consumer(columns_consumer);
 
     let beacon_api_binds =
         config.beacon_api_bind().iter().map(String::as_str).map(Bind::parse).collect::<Vec<_>>();

crates/columns/src/tile.rs

@@ -15,6 +15,7 @@ use silver_common::{
     EngineResp, GossipTopic, IngestionTime, NewGossipMsg, Origin, P2pStreamId, PeerEvent,
     RequestId, RpcInbound, RpcSeverity, SilverSpine, SilverSpineProducers, StreamProtocol,
     SyncNeed, SyncUpdate, TCacheRead, TProducer, TRandomAccess, TRead, Wheel,
+    cells::RetentionEvent,
     column_util::{self as util, KzgScratch},
     ssz_view::{NUMBER_OF_COLUMNS, SignedBeaconBlockView, StatusView},
     ticker::SlotTicker,
@@ -88,6 +89,7 @@ pub struct DataColumnsTile {
     // Declared last so it drops last: a parked column's read releases through
     // the consumer it was acquired from.
     consumers: ColumnConsumers,
+    data_columns_consumer: Option<Box<TRandomAccess>>,
 }
 
 impl DataColumnsTile {
@@ -114,9 +116,16 @@ impl DataColumnsTile {
             el_fetcher: ElBlobFetcher::new(engine_resp_consumer),
             el_column_producer,
             kzg_scratch: KzgScratch::default(),
+            data_columns_consumer: None,
         }
     }
 
+    pub fn with_data_columns_consumer(mut self, consumer: TRandomAccess) -> Self {
+        assert!(consumer.is_retained());
+        self.data_columns_consumer = Some(Box::new(consumer));
+        self
+    }
+
     #[timed]
     fn beacon_block(
         &mut self,
@@ -676,6 +685,11 @@ impl Tile<SilverSpine> for DataColumnsTile {
 
     fn loop_body(&mut self, adapter: &mut SpineAdapter<SilverSpine>) {
         self.consumers.free();
+        if let Some(consumer) = &mut self.data_columns_consumer {
+            adapter.consume(|event: RetentionEvent, _| {
+                consumer.advance_retention(event.retain_from);
+            });
+        }
 
         adapter.consume(|gossip: NewGossipMsg, producers| match gossip.topic {
             silver_common::GossipTopic::BeaconBlock if self.sync_state.is_synced() => {

crates/control/src/tile.rs

@@ -1,18 +1,21 @@
 use std::time::{Duration, Instant};
 
 use flux::{spine::SpineAdapter, tile::Tile};
+use silver_columns::cell_store::{CellStoreConfig, StoreError};
 use silver_common::{
     BeaconApiRequest, BeaconStateEvent, GossipTopic, LOCAL_GOSSIP_STREAM_ID, Nanos, P2pSend,
     PeerControl, PeerEvent, PeerStats, RpcInbound, RpcOutbound, RpcRequest, RpcRequestOutbound,
-    RpcResponse, RpcResponseInbound, SilverSpine, SilverSpineProducers, SyncNeed, SyncUpdate,
-    TMultiProducer, TProducer, TRandomAccess,
+    RpcResponse, RpcResponseInbound, SLOTS_PER_EPOCH, SilverSpine, SilverSpineProducers, SyncNeed,
+    SyncUpdate, TMultiProducer, TProducer, TRandomAccess,
+    cells::CellStoreEvent,
     ssz_view::{METADATA_SIZE, STATUS_V2_SIZE, StatusView},
 };
 use silver_gossip::{GossipHandler, GossipHandlerEvent};
 use silver_peer::PeerManager;
 
 use self::attestation_cluster::AttestationClusterHandler;
 use crate::{
+    cell_ingress::CellIngress,
     cluster::{AttestationClusterConfig, ClusterError},
     sync_engine::{SyncAction, SyncEngine},
 };
@@ -52,6 +55,7 @@ pub struct Controller {
     /// meshes would earn P3 deficit at peers since nothing validates or
     /// forwards until then. Drained into the PM on the first transition.
     pending_subnet_topics: Vec<GossipTopic>,
+    cell_ingress: Option<CellIngress>,
 }
 
 impl Controller {
@@ -90,9 +94,21 @@ impl Controller {
             last_peer_persist: now,
             auto_ping: true,
             pending_subnet_topics: Vec::new(),
+            cell_ingress: None,
         })
     }
 
+    pub fn with_data_columns_cache(
+        mut self,
+        config: CellStoreConfig,
+        producer: TProducer,
+        slot: u64,
+        slot_start: Instant,
+    ) -> Result<Self, StoreError> {
+        self.cell_ingress = Some(CellIngress::new(config, producer, slot, slot_start)?);
+        Ok(self)
+    }
+
     pub fn set_pending_subnet_topics(&mut self, topics: Vec<GossipTopic>) {
         self.pending_subnet_topics = topics;
     }
@@ -116,6 +132,9 @@ impl Controller {
 
     fn handle_latest_status(&mut self, latest_status_event: Option<([u8; 92], u64, u64)>) -> bool {
         if let Some((ssz, latest_block_slot, wall_slot)) = latest_status_event {
+            if let Some(ingress) = &mut self.cell_ingress {
+                ingress.set_min_slot(StatusView::finalized_epoch(&ssz) * SLOTS_PER_EPOCH);
+            }
             tracing::debug!(wall_slot, latest_block_slot, "new status set");
             // PM still tracks our Status (peer-Status validation) + applied head
             // (custody-peer eligibility); the wall slot is the engine's only.
@@ -158,6 +177,12 @@ impl Tile<SilverSpine> for Controller {
         let now = Instant::now();
         self.rpc_ssz_consumer.free();
         self.attestation_cluster.free();
+        if let Some(ingress) = &mut self.cell_ingress {
+            ingress.spin(now, &adapter.producers);
+            adapter.consume(|event: CellStoreEvent, producers| {
+                ingress.handle(event, now, producers);
+            });
+        }
 
         // Local status must land before the sync drive below: issuance is
         // capped against the imported head, and a one-loop-stale watermark

crates/network/src/p2p/quic/gossip_frame/tests.rs

@@ -426,7 +426,6 @@ fn framing_fragments_share_one_lazy_owner_until_the_last_ack() {
     )
     .unwrap();
     let before = ALLOCATIONS.with(Cell::get);
-    let before = ALLOCATIONS.with(Cell::get);
     let mut writer = h.acquire(reference).unwrap().into_writer();
     assert_eq!(ALLOCATIONS.with(Cell::get) - before, 0);
     let descriptor_len = writer.frame.segments.descriptor_len();