diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 8468c94d..da835165 100644 --- a/crates/bin/src/main.rs +++ b/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> { 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> { 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> { cluster_nodes: cluster_nodes.map(ClusterNodes::new), cluster_inbound_producer, cluster_outbound_consumer, + data_columns_consumer: None, }; let now = Instant::now(); @@ -249,6 +253,16 @@ fn main() -> Result<(), Box> { } } + 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)?; @@ -256,8 +270,6 @@ fn main() -> Result<(), Box> { 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> { 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> { 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::>(); diff --git a/crates/columns/src/tile.rs b/crates/columns/src/tile.rs index 77e4d313..42d1e241 100644 --- a/crates/columns/src/tile.rs +++ b/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>, } 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 for DataColumnsTile { fn loop_body(&mut self, adapter: &mut SpineAdapter) { 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() => { diff --git a/crates/control/src/tile.rs b/crates/control/src/tile.rs index 269f6f21..990fdb58 100644 --- a/crates/control/src/tile.rs +++ b/crates/control/src/tile.rs @@ -1,11 +1,13 @@ 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}; @@ -13,6 +15,7 @@ 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, + cell_ingress: Option, } 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.cell_ingress = Some(CellIngress::new(config, producer, slot, slot_start)?); + Ok(self) + } + pub fn set_pending_subnet_topics(&mut self, topics: Vec) { 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 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 diff --git a/crates/network/src/p2p/quic/gossip_frame/tests.rs b/crates/network/src/p2p/quic/gossip_frame/tests.rs index 9610011f..0672f2e3 100644 --- a/crates/network/src/p2p/quic/gossip_frame/tests.rs +++ b/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();