feat(starknet_state_sync): pass new internal blocks from state sync to p2p sync client

crates/papyrus_node/src/run.rs

@@ -7,6 +7,7 @@ use std::process::exit;
 use std::sync::Arc;
 use std::time::Duration;
 
+use futures::StreamExt;
 use papyrus_base_layer::ethereum_base_layer_contract::EthereumBaseLayerConfig;
 use papyrus_common::metrics::COLLECT_PROFILING_METRICS;
 use papyrus_common::pending_classes::PendingClasses;
@@ -304,6 +305,7 @@ async fn spawn_sync_client(
                 storage_reader,
                 storage_writer,
                 p2p_sync_client_channels,
+                futures::stream::pending().boxed(),
             );
             tokio::spawn(async move { Ok(p2p_sync.run().await?) })
         }

crates/papyrus_p2p_sync/Cargo.toml

@@ -22,6 +22,7 @@ papyrus_storage.workspace = true
 rand.workspace = true
 serde.workspace = true
 starknet_api.workspace = true
+starknet_state_sync_types.workspace = true
 starknet-types-core.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

crates/papyrus_p2p_sync/src/client/mod.rs

@@ -19,6 +19,7 @@ use std::time::Duration;
 
 use class::ClassStreamBuilder;
 use futures::channel::mpsc::SendError;
+use futures::stream::BoxStream;
 use futures::Stream;
 use header::HeaderStreamBuilder;
 use papyrus_common::pending_classes::ApiContractClass;
@@ -40,11 +41,13 @@ use serde::{Deserialize, Serialize};
 use starknet_api::block::BlockNumber;
 use starknet_api::core::ClassHash;
 use starknet_api::transaction::FullTransaction;
+use starknet_state_sync_types::state_sync_types::SyncBlock;
 use state_diff::StateDiffStreamBuilder;
 use stream_builder::{DataStreamBuilder, DataStreamResult};
 use tokio_stream::StreamExt;
 use tracing::instrument;
 use transaction::TransactionStreamFactory;
+
 const STEP: u64 = 1;
 const ALLOWED_SIGNATURES_LENGTH: usize = 1;
 
@@ -226,6 +229,8 @@ pub struct P2PSyncClient {
     storage_reader: StorageReader,
     storage_writer: StorageWriter,
     p2p_sync_channels: P2PSyncClientChannels,
+    #[allow(dead_code)]
+    internal_blocks_receiver: BoxStream<'static, (BlockNumber, SyncBlock)>,
 }
 
 impl P2PSyncClient {
@@ -234,8 +239,9 @@ impl P2PSyncClient {
         storage_reader: StorageReader,
         storage_writer: StorageWriter,
         p2p_sync_channels: P2PSyncClientChannels,
+        internal_blocks_receiver: BoxStream<'static, (BlockNumber, SyncBlock)>,
     ) -> Self {
-        Self { config, storage_reader, storage_writer, p2p_sync_channels }
+        Self { config, storage_reader, storage_writer, p2p_sync_channels, internal_blocks_receiver }
     }
 
     #[instrument(skip(self), level = "debug", err)]

crates/papyrus_p2p_sync/src/client/test_utils.rs

@@ -112,6 +112,7 @@ pub fn setup() -> TestArgs {
         storage_reader.clone(),
         storage_writer,
         p2p_sync_channels,
+        futures::stream::pending().boxed(),
     );
     TestArgs {
         p2p_sync,
@@ -194,6 +195,7 @@ pub async fn run_test(max_query_lengths: HashMap<DataType, u64>, actions: Vec<Ac
         storage_reader.clone(),
         storage_writer,
         p2p_sync_channels,
+        futures::stream::pending().boxed(),
     );
 
     let mut headers_current_query_responses_manager = None;

crates/starknet_state_sync/src/lib.rs

@@ -2,6 +2,8 @@ pub mod config;
 pub mod runner;
 
 use async_trait::async_trait;
+use futures::channel::mpsc::{channel, Sender};
+use futures::SinkExt;
 use papyrus_storage::body::BodyStorageReader;
 use papyrus_storage::state::StateStorageReader;
 use papyrus_storage::StorageReader;
@@ -12,18 +14,23 @@ use starknet_state_sync_types::communication::{
     StateSyncResponse,
     StateSyncResult,
 };
+use starknet_state_sync_types::errors::StateSyncError;
 use starknet_state_sync_types::state_sync_types::SyncBlock;
 
 use crate::config::StateSyncConfig;
 use crate::runner::StateSyncRunner;
 
+const BUFFER_SIZE: usize = 100000;
+
 pub fn create_state_sync_and_runner(config: StateSyncConfig) -> (StateSync, StateSyncRunner) {
-    let (state_sync_runner, storage_reader) = StateSyncRunner::new(config);
-    (StateSync { storage_reader }, state_sync_runner)
+    let (new_block_sender, new_block_receiver) = channel(BUFFER_SIZE);
+    let (state_sync_runner, storage_reader) = StateSyncRunner::new(config, new_block_receiver);
+    (StateSync { storage_reader, new_block_sender }, state_sync_runner)
 }
 
 pub struct StateSync {
     storage_reader: StorageReader,
+    new_block_sender: Sender<(BlockNumber, SyncBlock)>,
 }
 
 // TODO(shahak): Have StateSyncRunner call StateSync instead of the opposite once we stop supporting
@@ -35,8 +42,13 @@ impl ComponentRequestHandler<StateSyncRequest, StateSyncResponse> for StateSync
             StateSyncRequest::GetBlock(block_number) => {
                 StateSyncResponse::GetBlock(self.get_block(block_number))
             }
-            StateSyncRequest::AddNewBlock(_block_number, _sync_block) => {
-                todo!()
+            StateSyncRequest::AddNewBlock(block_number, sync_block) => {
+                StateSyncResponse::AddNewBlock(
+                    self.new_block_sender
+                        .send((block_number, sync_block))
+                        .await
+                        .map_err(StateSyncError::from),
+                )
             }
         }
     }

crates/starknet_state_sync/src/runner/mod.rs

@@ -2,15 +2,18 @@
 mod test;
 
 use async_trait::async_trait;
+use futures::channel::mpsc::Receiver;
 use futures::future::BoxFuture;
-use futures::FutureExt;
+use futures::{FutureExt, StreamExt};
 use papyrus_network::network_manager::{self, NetworkError};
 use papyrus_p2p_sync::client::{P2PSyncClient, P2PSyncClientChannels, P2PSyncClientError};
 use papyrus_p2p_sync::server::{P2PSyncServer, P2PSyncServerChannels};
 use papyrus_p2p_sync::{Protocol, BUFFER_SIZE};
 use papyrus_storage::{open_storage, StorageReader};
+use starknet_api::block::BlockNumber;
 use starknet_sequencer_infra::component_definitions::ComponentStarter;
 use starknet_sequencer_infra::errors::ComponentError;
+use starknet_state_sync_types::state_sync_types::SyncBlock;
 
 use crate::config::StateSyncConfig;
 
@@ -37,7 +40,10 @@ impl ComponentStarter for StateSyncRunner {
 }
 
 impl StateSyncRunner {
-    pub fn new(config: StateSyncConfig) -> (Self, StorageReader) {
+    pub fn new(
+        config: StateSyncConfig,
+        new_block_receiver: Receiver<(BlockNumber, SyncBlock)>,
+    ) -> (Self, StorageReader) {
         let (storage_reader, storage_writer) =
             open_storage(config.storage_config).expect("StateSyncRunner failed opening storage");
 
@@ -65,6 +71,7 @@ impl StateSyncRunner {
             storage_reader.clone(),
             storage_writer,
             p2p_sync_client_channels,
+            new_block_receiver.boxed(),
         );
 
         let header_server_receiver = network_manager

crates/starknet_state_sync_types/Cargo.toml

@@ -10,6 +10,7 @@ workspace = true
 
 [dependencies]
 async-trait.workspace = true
+futures.workspace = true
 papyrus_proc_macros.workspace = true
 papyrus_storage.workspace = true
 serde = { workspace = true, features = ["derive"] }

crates/starknet_state_sync_types/src/errors.rs

@@ -1,3 +1,4 @@
+use futures::channel::mpsc::SendError;
 use papyrus_storage::StorageError;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
@@ -10,10 +11,18 @@ pub enum StateSyncError {
     // We put the string of the error instead.
     #[error("Unexpected storage error: {0}")]
     StorageError(String),
+    #[error("Error while sending SyncBlock from StateSync to P2pSyncClient")]
+    SendError(String),
 }
 
 impl From<StorageError> for StateSyncError {
     fn from(error: StorageError) -> Self {
         StateSyncError::StorageError(error.to_string())
     }
 }
+
+impl From<SendError> for StateSyncError {
+    fn from(error: SendError) -> Self {
+        StateSyncError::SendError(error.to_string())
+    }
+}