feat: add stagekind enum and initialise syncstage table (#2426)

Cargo.lock

@@ -5363,6 +5363,7 @@ dependencies = [
  "reth-primitives",
  "reth-provider",
  "reth-staged-sync",
+ "reth-stages",
  "reth-tracing",
  "secp256k1 0.26.0",
  "serde",

crates/staged-sync/Cargo.toml

@@ -17,6 +17,7 @@ reth-downloaders = { path = "../../crates/net/downloaders" }
 reth-primitives = { path = "../../crates/primitives" }
 reth-provider = { path = "../../crates/storage/provider", features = ["test-utils"] }
 reth-net-nat = { path = "../../crates/net/nat" }
+reth-stages = { path = "../stages" }
 
 # io
 serde = "1.0"

crates/staged-sync/src/utils/init.rs

@@ -6,6 +6,7 @@ use reth_db::{
     transaction::{DbTx, DbTxMut},
 };
 use reth_primitives::{keccak256, Account, Bytecode, ChainSpec, StorageEntry, H256};
+use reth_stages::StageKind;
 use std::{path::Path, sync::Arc};
 use tracing::debug;
 
@@ -68,6 +69,11 @@ pub fn init_genesis<DB: Database>(
     tx.put::<tables::HeaderTD>(0, header.difficulty.into())?;
     tx.put::<tables::Headers>(0, header)?;
 
+    // insert sync stage
+    for stage in StageKind::ALL.iter() {
+        tx.put::<tables::SyncStage>(stage.to_string(), 0)?;
+    }
+
     tx.commit()?;
     Ok(hash)
 }
@@ -115,14 +121,12 @@ pub fn insert_genesis_state<DB: Database>(
 
 #[cfg(test)]
 mod tests {
-
-    use std::sync::Arc;
-
     use super::{init_genesis, InitDatabaseError};
     use reth_db::mdbx::test_utils::create_test_rw_db;
     use reth_primitives::{
         GOERLI, GOERLI_GENESIS, MAINNET, MAINNET_GENESIS, SEPOLIA, SEPOLIA_GENESIS,
     };
+    use std::sync::Arc;
 
     #[test]
     fn success_init_genesis_mainnet() {

crates/stages/src/id.rs

@@ -1,4 +1,8 @@
-use crate::stages::{BODIES, FINISH, HEADERS};
+use crate::stages::{
+    ACCOUNT_HASHING, BODIES, EXECUTION, FINISH, HEADERS, INDEX_ACCOUNT_HISTORY,
+    INDEX_STORAGE_HISTORY, MERKLE_EXECUTION, MERKLE_UNWIND, SENDER_RECOVERY, TOTAL_DIFFICULTY,
+    TRANSACTION_LOOKUP,
+};
 use reth_db::{
     tables::SyncStage,
     transaction::{DbTx, DbTxMut},
@@ -7,6 +11,69 @@ use reth_db::{
 use reth_primitives::BlockNumber;
 use std::fmt::Display;
 
+/// All known stages
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[allow(missing_docs)]
+pub enum StageKind {
+    Headers,
+    Bodies,
+    SenderRecovery,
+    TotalDifficulty,
+    AccountHashing,
+    StorageHashing,
+    IndexAccountHistory,
+    IndexStorageHistory,
+    MerkleExecution,
+    MerkleUnwind,
+    Execution,
+    TransactionLookup,
+    Finish,
+}
+
+impl StageKind {
+    /// All supported Stages
+    pub const ALL: [StageKind; 13] = [
+        StageKind::Headers,
+        StageKind::Bodies,
+        StageKind::SenderRecovery,
+        StageKind::TotalDifficulty,
+        StageKind::AccountHashing,
+        StageKind::StorageHashing,
+        StageKind::IndexAccountHistory,
+        StageKind::IndexStorageHistory,
+        StageKind::MerkleExecution,
+        StageKind::MerkleUnwind,
+        StageKind::Execution,
+        StageKind::TransactionLookup,
+        StageKind::Finish,
+    ];
+
+    /// Returns the ID of this stage.
+    pub fn id(&self) -> StageId {
+        match self {
+            StageKind::Headers => HEADERS,
+            StageKind::Bodies => BODIES,
+            StageKind::SenderRecovery => SENDER_RECOVERY,
+            StageKind::TotalDifficulty => TOTAL_DIFFICULTY,
+            StageKind::AccountHashing => ACCOUNT_HASHING,
+            StageKind::StorageHashing => ACCOUNT_HASHING,
+            StageKind::IndexAccountHistory => INDEX_ACCOUNT_HISTORY,
+            StageKind::IndexStorageHistory => INDEX_STORAGE_HISTORY,
+            StageKind::MerkleExecution => MERKLE_EXECUTION,
+            StageKind::MerkleUnwind => MERKLE_UNWIND,
+            StageKind::Execution => EXECUTION,
+            StageKind::TransactionLookup => TRANSACTION_LOOKUP,
+            StageKind::Finish => FINISH,
+        }
+    }
+}
+
+impl Display for StageKind {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.id())
+    }
+}
+
 /// The ID of a stage.
 ///
 /// Each stage ID must be unique.

crates/stages/src/pipeline/mod.rs

@@ -149,7 +149,7 @@ where
     }
 
     /// Registers progress metrics for each registered stage
-    fn register_metrics(&mut self, db: Arc<DB>) {
+    pub fn register_metrics(&mut self, db: Arc<DB>) {
         for stage in &self.stages {
             let stage_id = stage.id();
             self.metrics.stage_checkpoint(

crates/storage/provider/src/transaction.rs

@@ -1024,7 +1024,7 @@ where
         &self,
         block_number: BlockNumber,
     ) -> Result<(), TransactionError> {
-        // iterate over
+        // iterate over all existing stages in the table and update its progress.
         let mut cursor = self.cursor_write::<tables::SyncStage>()?;
         while let Some((stage_name, _)) = cursor.next()? {
             cursor.upsert(stage_name, block_number)?