feat: option to skip ledger replay

Cargo.toml

@@ -72,6 +72,7 @@ convert_case = "0.8.0"
 conjunto-transwise = { git = "https://github.com/magicblock-labs/conjunto.git", rev = "bf82b45" }
 console-subscriber = "0.2.0"
 crossbeam-channel = "0.5.11"
+const_format = "0.2.34"
 ed25519-dalek = "1.0.1"
 enum-iterator = "1.5.0"
 env_logger = "0.11.2"

magicblock-accounts-db/Cargo.toml

@@ -21,6 +21,7 @@ solana-account = { workspace = true }
 parking_lot = "0.12"
 
 # misc
+const_format = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 thiserror = { workspace = true }
 log = { workspace = true }

magicblock-accounts-db/src/lib.rs

@@ -1,5 +1,6 @@
 use std::{path::Path, sync::Arc};
 
+use const_format::concatcp;
 use error::AccountsDbError;
 use index::AccountsDbIndex;
 use log::{error, warn};
@@ -19,7 +20,8 @@ pub type AdbResult<T> = Result<T, AccountsDbError>;
 /// some critical operation is in action, e.g. snapshotting
 pub type StWLock = Arc<RwLock<()>>;
 
-const ACCOUNTSDB_SUB_DIR: &str = "accountsdb/main";
+pub const ACCOUNTSDB_DIR: &str = "accountsdb";
+const ACCOUNTSDB_SUB_DIR: &str = concatcp!(ACCOUNTSDB_DIR, "/main");
 
 pub struct AccountsDb {
     /// Main accounts storage, where actual account records are kept

magicblock-api/src/fund_account.rs

@@ -1,6 +1,7 @@
 use std::path::Path;
 
 use magicblock_bank::bank::Bank;
+use magicblock_config::LedgerResumeStrategy;
 use magicblock_core::magic_program;
 use solana_sdk::{
     account::Account, clock::Epoch, pubkey::Pubkey, signature::Keypair,
@@ -56,9 +57,9 @@ pub(crate) fn fund_validator_identity(bank: &Bank, validator_id: &Pubkey) {
 pub(crate) fn funded_faucet(
     bank: &Bank,
     ledger_path: &Path,
-    create_new: bool,
+    resume_strategy: &LedgerResumeStrategy,
 ) -> ApiResult<Keypair> {
-    let faucet_keypair = if create_new {
+    let faucet_keypair = if resume_strategy.is_removing_ledger() {
         let faucet_keypair = Keypair::new();
         write_faucet_keypair_to_ledger(ledger_path, &faucet_keypair)?;
         faucet_keypair

magicblock-api/src/ledger.rs

@@ -6,36 +6,48 @@ use std::{
 
 use fd_lock::{RwLock, RwLockWriteGuard};
 use log::*;
+use magicblock_accounts_db::ACCOUNTSDB_DIR;
+use magicblock_config::LedgerResumeStrategy;
 use magicblock_ledger::Ledger;
-use solana_sdk::{signature::Keypair, signer::EncodableKey};
+use solana_sdk::{clock::Slot, signature::Keypair, signer::EncodableKey};
 
-use crate::{
-    errors::{ApiError, ApiResult},
-    utils::fs::remove_directory_contents_if_exists,
-};
+use crate::errors::{ApiError, ApiResult};
 
 // -----------------
 // Init
 // -----------------
-pub(crate) fn init(ledger_path: PathBuf, reset: bool) -> ApiResult<Ledger> {
-    if reset {
-        remove_directory_contents_if_exists(ledger_path.as_path()).map_err(
-            |err| {
-                error!(
-                    "Error: Unable to remove {}: {}",
-                    ledger_path.display(),
-                    err
-                );
-                ApiError::UnableToCleanLedgerDirectory(
-                    ledger_path.display().to_string(),
-                )
-            },
-        )?;
+pub(crate) fn init(
+    ledger_path: PathBuf,
+    resume_strategy: &LedgerResumeStrategy,
+) -> ApiResult<(Ledger, Slot)> {
+    // Save the last slot from the previous ledger to restart from it
+    let last_slot = if resume_strategy.is_resuming() {
+        let previous_ledger = Ledger::open(ledger_path.as_path())?;
+        previous_ledger.get_max_blockhash().map(|(slot, _)| slot)?
+    } else {
+        Slot::default()
+    };
+
+    if resume_strategy.is_removing_ledger() {
+        remove_ledger_directory_if_exists(
+            ledger_path.as_path(),
+            resume_strategy,
+        )
+        .map_err(|err| {
+            error!(
+                "Error: Unable to remove {}: {}",
+                ledger_path.display(),
+                err
+            );
+            ApiError::UnableToCleanLedgerDirectory(
+                ledger_path.display().to_string(),
+            )
+        })?;
     }
 
     fs::create_dir_all(&ledger_path)?;
 
-    Ok(Ledger::open(ledger_path.as_path())?)
+    Ok((Ledger::open(ledger_path.as_path())?, last_slot))
 }
 
 // -----------------
@@ -165,3 +177,40 @@ pub(crate) fn ledger_parent_dir(ledger_path: &Path) -> ApiResult<PathBuf> {
     })?;
     Ok(parent.to_path_buf())
 }
+
+fn remove_ledger_directory_if_exists(
+    dir: &Path,
+    resume_strategy: &LedgerResumeStrategy,
+) -> Result<(), std::io::Error> {
+    if !dir.exists() {
+        return Ok(());
+    }
+    for entry in fs::read_dir(dir)? {
+        let entry = entry?;
+
+        // When resuming, keep the accounts db
+        if entry.file_name() == ACCOUNTSDB_DIR && resume_strategy.is_resuming()
+        {
+            continue;
+        }
+
+        // When resuming, keep the validator keypair
+        if let Ok(validator_keypair_path) = validator_keypair_path(dir) {
+            if resume_strategy.is_resuming()
+                && validator_keypair_path
+                    .file_name()
+                    .map(|key_path| key_path == entry.file_name())
+                    .unwrap_or(false)
+            {
+                continue;
+            }
+        }
+
+        if entry.metadata()?.is_dir() {
+            fs::remove_dir_all(entry.path())?
+        } else {
+            fs::remove_file(entry.path())?
+        }
+    }
+    Ok(())
+}

magicblock-api/src/lib.rs

@@ -8,7 +8,6 @@ pub mod ledger;
 pub mod magic_validator;
 mod slot;
 mod tickers;
-mod utils;
 
 pub use init_geyser_service::InitGeyserServiceConfig;
 pub use magicblock_config::EphemeralConfig;

magicblock-api/src/magic_validator.rs

@@ -40,8 +40,8 @@ use magicblock_committor_service::{
     config::ChainConfig, CommittorService, ComputeBudgetConfig,
 };
 use magicblock_config::{
-    AccountsDbConfig, EphemeralConfig, LifecycleMode, PrepareLookupTables,
-    ProgramConfig,
+    AccountsDbConfig, EphemeralConfig, LedgerConfig, LedgerResumeStrategy,
+    LifecycleMode, PrepareLookupTables, ProgramConfig,
 };
 use magicblock_geyser_plugin::rpc::GeyserRpcService;
 use magicblock_ledger::{
@@ -196,14 +196,12 @@ impl MagicValidator {
             config.validator_config.validator.base_fees,
         );
 
-        let ledger = Self::init_ledger(
-            config.validator_config.ledger.path.as_ref(),
-            config.validator_config.ledger.reset,
-        )?;
+        let (ledger, last_slot) =
+            Self::init_ledger(&config.validator_config.ledger)?;
         Self::sync_validator_keypair_with_ledger(
             ledger.ledger_path(),
             &identity_keypair,
-            config.validator_config.ledger.reset,
+            &config.validator_config.ledger.resume_strategy,
         )?;
 
         // SAFETY:
@@ -223,7 +221,7 @@ impl MagicValidator {
             config.validator_config.validator.millis_per_slot,
             validator_pubkey,
             ledger_parent_path,
-            ledger.get_max_blockhash().map(|(slot, _)| slot)?,
+            last_slot,
         )?;
 
         let ledger_truncator = LedgerTruncator::new(
@@ -238,7 +236,7 @@ impl MagicValidator {
         let faucet_keypair = funded_faucet(
             &bank,
             ledger.ledger_path().as_path(),
-            config.validator_config.ledger.reset,
+            &config.validator_config.ledger.resume_strategy,
         )?;
 
         load_programs_into_bank(
@@ -520,28 +518,28 @@ impl MagicValidator {
     }
 
     fn init_ledger(
-        ledger_path: Option<&String>,
-        reset: bool,
-    ) -> ApiResult<Arc<Ledger>> {
-        let ledger_path = match ledger_path {
+        ledger_config: &LedgerConfig,
+    ) -> ApiResult<(Arc<Ledger>, Slot)> {
+        let ledger_path = match &ledger_config.path {
             Some(ledger_path) => PathBuf::from(ledger_path),
             None => {
                 let ledger_path = TempDir::new()?;
                 ledger_path.path().to_path_buf()
             }
         };
-        let ledger = ledger::init(ledger_path, reset)?;
+        let (ledger, last_slot) =
+            ledger::init(ledger_path, &ledger_config.resume_strategy)?;
         let ledger_shared = Arc::new(ledger);
         init_persister(ledger_shared.clone());
-        Ok(ledger_shared)
+        Ok((ledger_shared, last_slot))
     }
 
     fn sync_validator_keypair_with_ledger(
         ledger_path: &Path,
         validator_keypair: &Keypair,
-        reset_ledger: bool,
+        resume_strategy: &LedgerResumeStrategy,
     ) -> ApiResult<()> {
-        if reset_ledger {
+        if !resume_strategy.is_resuming() {
             write_validator_keypair_to_ledger(ledger_path, validator_keypair)?;
         } else {
             let ledger_validator_keypair =
@@ -581,7 +579,7 @@ impl MagicValidator {
     // Start/Stop
     // -----------------
     fn maybe_process_ledger(&self) -> ApiResult<()> {
-        if self.config.ledger.reset {
+        if !self.config.ledger.resume_strategy.is_replaying() {
             return Ok(());
         }
         let slot_to_continue_at = process_ledger(&self.ledger, &self.bank)?;
@@ -814,7 +812,7 @@ impl MagicValidator {
                 }
             }
 
-            if !self.config.ledger.reset {
+            if self.config.ledger.resume_strategy.is_replaying() {
                 let remote_account_cloner_worker =
                     remote_account_cloner_worker.clone();
                 tokio::spawn(async move {

magicblock-config/src/accounts.rs

@@ -28,7 +28,7 @@ pub struct AccountsConfig {
     pub lifecycle: LifecycleMode,
     #[serde(default)]
     #[command(flatten)]
-    pub commit: CommitStrategy,
+    pub commit: CommitStrategyConfig,
     #[clap_from_serde_skip]
     #[arg(help = "The list of allowed programs to load.")]
     #[serde(default)]
@@ -152,9 +152,11 @@ pub enum LifecycleMode {
 // -----------------
 #[clap_prefix("commit")]
 #[clap_from_serde]
-#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Args)]
+#[derive(
+    Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Args, Mergeable,
+)]
 #[serde(deny_unknown_fields)]
-pub struct CommitStrategy {
+pub struct CommitStrategyConfig {
     #[derive_env_var]
     #[serde(default = "default_frequency_millis")]
     pub frequency_millis: u64,
@@ -179,7 +181,7 @@ fn default_compute_unit_price() -> u64 {
     1_000_000 // 1_000_000 micro-lamports == 1 Lamport
 }
 
-impl Default for CommitStrategy {
+impl Default for CommitStrategyConfig {
     fn default() -> Self {
         Self {
             frequency_millis: default_frequency_millis(),
@@ -283,7 +285,7 @@ mod tests {
                 ws_url: None,
             },
             lifecycle: LifecycleMode::Ephemeral,
-            commit: CommitStrategy {
+            commit: CommitStrategyConfig {
                 frequency_millis: 123,
                 compute_unit_price: 123,
             },
@@ -315,7 +317,7 @@ mod tests {
                 ws_url: None,
             },
             lifecycle: LifecycleMode::Ephemeral,
-            commit: CommitStrategy {
+            commit: CommitStrategyConfig {
                 frequency_millis: 123,
                 compute_unit_price: 123,
             },
@@ -344,7 +346,7 @@ mod tests {
                 ws_url: Some(vec![Url::parse("wss://0.0.0.0:7999").unwrap()]),
             },
             lifecycle: LifecycleMode::Offline,
-            commit: CommitStrategy {
+            commit: CommitStrategyConfig {
                 frequency_millis: 1234,
                 compute_unit_price: 1234,
             },
@@ -372,7 +374,7 @@ mod tests {
                 ws_url: None,
             },
             lifecycle: LifecycleMode::Ephemeral,
-            commit: CommitStrategy {
+            commit: CommitStrategyConfig {
                 frequency_millis: 123,
                 compute_unit_price: 123,
             },