apollo_state_sync: spwan storage reader server for state sync (#11719)

Author: nadin-Starkware

crates/apollo_state_sync/src/runner/mod.rs

@@ -58,6 +58,7 @@ use papyrus_common::pending_classes::PendingClasses;
 use starknet_api::block::{BlockHash, BlockHashAndNumber};
 use starknet_api::felt;
 use tokio::sync::RwLock;
+use tokio::task::AbortHandle;
 use tracing::instrument::Instrument;
 use tracing::{debug, info_span};
 
@@ -70,7 +71,9 @@ pub struct StateSyncRunner {
     new_block_dev_null_future: BoxFuture<'static, Never>,
     rpc_server_future: BoxFuture<'static, ()>,
     register_metrics_future: BoxFuture<'static, ()>,
-    pub storage_reader_server: Option<GenericStorageReaderServer>,
+    // Kept alive to maintain the server running.
+    #[allow(dead_code)]
+    storage_reader_server_handle: Option<AbortHandle>,
 }
 
 #[async_trait]
@@ -106,7 +109,7 @@ pub struct StateSyncResources {
     pub shared_highest_block: Arc<RwLock<Option<BlockHashAndNumber>>>,
     pub pending_data: Arc<RwLock<PendingData>>,
     pub pending_classes: Arc<RwLock<PendingClasses>>,
-    pub storage_reader_server: Option<GenericStorageReaderServer>,
+    pub storage_reader_server_handle: Option<AbortHandle>,
 }
 
 impl StateSyncResources {
@@ -118,6 +121,8 @@ impl StateSyncResources {
                 storage_reader_server_config,
             )
             .expect("StateSyncRunner failed opening storage");
+        let storage_reader_server_handle =
+            GenericStorageReaderServer::spawn_if_enabled(storage_reader_server);
         let shared_highest_block = Arc::new(RwLock::new(None));
         let pending_data = Arc::new(RwLock::new(PendingData {
             // The pending data might change later to DeprecatedPendingBlock, depending on the
@@ -135,7 +140,7 @@ impl StateSyncResources {
             shared_highest_block,
             pending_data,
             pending_classes,
-            storage_reader_server,
+            storage_reader_server_handle,
         }
     }
 }
@@ -162,7 +167,7 @@ impl StateSyncRunner {
             shared_highest_block,
             pending_data,
             pending_classes,
-            storage_reader_server,
+            storage_reader_server_handle,
         } = StateSyncResources::new(&storage_config, storage_reader_server_config);
 
         let register_metrics_future = register_metrics(storage_reader.clone()).boxed();
@@ -224,7 +229,7 @@ impl StateSyncRunner {
                     new_block_dev_null_future: pending().boxed(),
                     rpc_server_future,
                     register_metrics_future,
-                    storage_reader_server,
+                    storage_reader_server_handle,
                 },
                 storage_reader,
             );
@@ -335,7 +340,7 @@ impl StateSyncRunner {
                 new_block_dev_null_future,
                 rpc_server_future,
                 register_metrics_future,
-                storage_reader_server,
+                storage_reader_server_handle,
             },
             storage_reader,
         )

crates/apollo_state_sync/src/runner/test.rs

@@ -24,7 +24,7 @@ fn run_panics_when_network_future_returns() {
         new_block_dev_null_future,
         rpc_server_future,
         register_metrics_future,
-        storage_reader_server: None,
+        storage_reader_server_handle: None,
     };
     state_sync_runner.start().now_or_never().unwrap();
 }
@@ -48,7 +48,7 @@ fn run_panics_when_network_future_returns_error() {
         new_block_dev_null_future,
         rpc_server_future,
         register_metrics_future,
-        storage_reader_server: None,
+        storage_reader_server_handle: None,
     };
     state_sync_runner.start().now_or_never().unwrap();
 }
@@ -71,7 +71,7 @@ fn run_panics_when_sync_client_future_returns_error() {
         new_block_dev_null_future,
         rpc_server_future,
         register_metrics_future,
-        storage_reader_server: None,
+        storage_reader_server_handle: None,
     };
     state_sync_runner.start().now_or_never().unwrap();
 }