feat(starknet_state_sync): add revert_up_to_and_including option

config/sequencer/default_config.json

@@ -1364,6 +1364,16 @@
     "privacy": "Public",
     "value": 50
   },
+  "state_sync_config.revert_config.revert_up_to_and_including": {
+    "description": "The component will revert blocks up to this block number (including).",
+    "pointer_target": "revert_config.revert_up_to_and_including",
+    "privacy": "Public"
+  },
+  "state_sync_config.revert_config.should_revert": {
+    "description": "Whether the component should revert block and do nothing else.",
+    "pointer_target": "revert_config.should_revert",
+    "privacy": "Public"
+  },
   "state_sync_config.storage_config.db_config.chain_id": {
     "description": "The chain to follow. For more details see https://docs.starknet.io/documentation/architecture_and_concepts/Blocks/transactions/#chain-id.",
     "pointer_target": "chain_id",

crates/starknet_sequencer_node/src/config/node_config.rs

@@ -117,7 +117,10 @@ pub static CONFIG_POINTERS: LazyLock<ConfigPointers> = LazyLock::new(|| {
     let mut common_execution_config = generate_struct_pointer(
         "revert_config".to_owned(),
         &RevertConfig::default(),
-        set_pointing_param_paths(&["consensus_manager_config.revert_config"]),
+        set_pointing_param_paths(&[
+            "state_sync_config.revert_config",
+            "consensus_manager_config.revert_config",
+        ]),
     );
     pointers.append(&mut common_execution_config);
     pointers

crates/starknet_state_sync/Cargo.toml

@@ -9,6 +9,7 @@ repository.workspace = true
 workspace = true
 
 [dependencies]
+apollo_reverts.workspace = true
 async-trait.workspace = true
 futures.workspace = true
 papyrus_common.workspace = true

crates/starknet_state_sync/src/config.rs

@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
 use std::path::PathBuf;
 use std::result;
 
+use apollo_reverts::RevertConfig;
 use papyrus_config::dumping::{append_sub_config_name, ser_optional_sub_config, SerializeConfig};
 use papyrus_config::{ParamPath, SerializedParam};
 use papyrus_network::NetworkConfig;
@@ -27,19 +28,26 @@ pub struct StateSyncConfig {
     pub central_sync_client_config: Option<CentralSyncClientConfig>,
     #[validate]
     pub network_config: NetworkConfig,
+    #[validate]
+    pub revert_config: RevertConfig,
 }
 
 impl SerializeConfig for StateSyncConfig {
     fn dump(&self) -> BTreeMap<ParamPath, SerializedParam> {
-        vec![
-            append_sub_config_name(self.storage_config.dump(), "storage_config"),
-            append_sub_config_name(self.network_config.dump(), "network_config"),
-            ser_optional_sub_config(&self.p2p_sync_client_config, "p2p_sync_client_config"),
-            ser_optional_sub_config(&self.central_sync_client_config, "central_sync_client_config"),
-        ]
-        .into_iter()
-        .flatten()
-        .collect()
+        let mut config = BTreeMap::new();
+
+        config.extend(append_sub_config_name(self.storage_config.dump(), "storage_config"));
+        config.extend(append_sub_config_name(self.network_config.dump(), "network_config"));
+        config.extend(append_sub_config_name(self.revert_config.dump(), "revert_config"));
+        config.extend(ser_optional_sub_config(
+            &self.p2p_sync_client_config,
+            "p2p_sync_client_config",
+        ));
+        config.extend(ser_optional_sub_config(
+            &self.central_sync_client_config,
+            "central_sync_client_config",
+        ));
+        config
     }
 }
 
@@ -67,6 +75,7 @@ impl Default for StateSyncConfig {
             p2p_sync_client_config: Some(P2pSyncClientConfig::default()),
             central_sync_client_config: None,
             network_config: NetworkConfig { tcp_port: STATE_SYNC_TCP_PORT, ..Default::default() },
+            revert_config: RevertConfig::default(),
         }
     }
 }

crates/starknet_state_sync/src/runner/mod.rs

@@ -3,9 +3,10 @@ mod test;
 
 use std::sync::Arc;
 
+use apollo_reverts::{revert_block, revert_blocks_and_eternal_pending};
 use async_trait::async_trait;
 use futures::channel::mpsc::Receiver;
-use futures::future::{self, BoxFuture};
+use futures::future::{self, pending, BoxFuture};
 use futures::never::Never;
 use futures::{FutureExt, StreamExt};
 use papyrus_common::pending_classes::PendingClasses;
@@ -18,6 +19,7 @@ use papyrus_p2p_sync::client::{
 };
 use papyrus_p2p_sync::server::{P2pSyncServer, P2pSyncServerChannels};
 use papyrus_p2p_sync::{Protocol, BUFFER_SIZE};
+use papyrus_storage::header::HeaderStorageReader;
 use papyrus_storage::{open_storage, StorageReader, StorageWriter};
 use papyrus_sync::sources::central::{CentralError, CentralSource};
 use papyrus_sync::sources::pending::PendingSource;
@@ -88,8 +90,46 @@ impl StateSyncRunner {
             p2p_sync_client_config,
             central_sync_client_config,
             network_config,
+            revert_config,
         } = config;
 
+        let (storage_reader, mut storage_writer) =
+            open_storage(storage_config).expect("StateSyncRunner failed opening storage");
+
+        if revert_config.should_revert {
+            let revert_up_to_and_including = revert_config.revert_up_to_and_including;
+            // We assume that sync always writes the headers before any other block data.
+            let current_header_marker = storage_reader
+                .begin_ro_txn()
+                .expect("Should be able to begin read only transaction")
+                .get_header_marker()
+                .expect("Should have a header marker");
+
+            let revert_block_fn = move |current_block_number| {
+                // TODO(alonl): update metrics
+                revert_block(&mut storage_writer, current_block_number);
+                async {}
+            };
+
+            return (
+                Self {
+                    network_future: pending().boxed(),
+                    p2p_sync_client_future: revert_blocks_and_eternal_pending(
+                        current_header_marker,
+                        revert_up_to_and_including,
+                        revert_block_fn,
+                        "State Sync",
+                    )
+                    .map(|_never| unreachable!("Never should never be constructed"))
+                    .boxed(),
+                    p2p_sync_server_future: pending().boxed(),
+                    central_sync_client_future: pending().boxed(),
+                    new_block_dev_null_future: pending().boxed(),
+                },
+                storage_reader,
+            );
+        }
+
         let network_manager_metrics = Some(metrics::NetworkMetrics {
             num_connected_peers: STATE_SYNC_NUM_CONNECTED_PEERS,
             num_active_inbound_sessions: STATE_SYNC_NUM_ACTIVE_INBOUND_SESSIONS,
@@ -101,9 +141,6 @@ impl StateSyncRunner {
             network_manager_metrics,
         );
 
-        let (storage_reader, storage_writer) =
-            open_storage(storage_config).expect("StateSyncRunner failed opening storage");
-
         // Creating the sync server future
         let p2p_sync_server = Self::new_p2p_state_sync_server(
             storage_reader.clone(),