fix: resolve corrupted accounts (#306)

## Description

- Resolve corrupted accounts on restart

## Note

This change assume that we ensure undelegation are landing 

<!-- greptile_comment -->

## Greptile Summary

Enhanced account corruption handling and ledger replay functionality,
focusing on improved restart reliability and account state management.

- Added debug logging in
`magicblock-accounts-db/src/persist/accounts_persister.rs` to trace
storage file loading and corruption detection
- Increased account hydration concurrency from 30 to 100 in
`magicblock-account-cloner/src/remote_account_cloner_worker.rs` for
better performance
- Added ownership validation in
`magicblock-account-cloner/src/remote_account_cloner_worker.rs` to
prevent unnecessary re-cloning of delegated accounts
- Implemented scheduled commits clearing in
`magicblock-api/src/magic_validator.rs` to avoid duplicate commits
during ledger processing
- Added slot advancement logic in
`magicblock-api/src/magic_validator.rs` to ensure correct transaction
ordering after restart



<!-- /greptile_comment -->

Author: GabrielePicco

magicblock-account-cloner/src/remote_account_cloner_worker.rs

@@ -276,7 +276,7 @@ where
         // TODO(GabrielePicco): Make the concurrency configurable
         stream
             .map(Ok::<_, AccountClonerError>)
-            .try_for_each_concurrent(30, |(pubkey, owner)| async move {
+            .try_for_each_concurrent(100, |(pubkey, owner)| async move {
                 trace!("Hydrating '{}'", pubkey);
                 let res = self
                     .do_clone_and_update_cache(
@@ -613,7 +613,14 @@ where
                         at_slot: account_chain_snapshot.at_slot,
                     });
                 }
-                if !stage.should_clone_delegated_account(delegation_record) {
+                if !stage.should_clone_delegated_account(delegation_record)
+                    && self
+                        .internal_account_provider
+                        .get_account(pubkey)
+                        .is_some_and(|acc| {
+                            acc.owner().eq(&delegation_record.owner)
+                        })
+                {
                     // NOTE: the account was already cloned when the initial instance of this
                     // validator ran. We don't want to clone it again during ledger replay, however
                     // we want to use it as a delegated + cloned account, thus we respond in the
@@ -625,6 +632,7 @@ where
                         signature: Signature::new_unique(),
                     });
                 }
+
                 self.do_clone_delegated_account(
                     pubkey,
                     // TODO(GabrielePicco): Avoid cloning

magicblock-accounts-db/src/persist/accounts_persister.rs

@@ -316,6 +316,7 @@ impl AccountsPersister {
                     StorageAccess::Mmap,
                 ) {
                     Ok((append_vec, num_accounts)) => {
+                        debug!("Loading storage from {:?}", file);
                         let accounts = AccountsFile::AppendVec(append_vec);
                         let storage = AccountStorageEntry::new_existing(
                             slot,

magicblock-api/src/magic_validator.rs

@@ -10,6 +10,25 @@ use std::{
     time::Duration,
 };
 
+use crate::{
+    accounts::create_accounts_run_and_snapshot_dirs,
+    errors::{ApiError, ApiResult},
+    external_config::try_convert_accounts_config,
+    fund_account::{
+        fund_magic_context, fund_validator_identity, funded_faucet,
+    },
+    geyser_transaction_notify_listener::GeyserTransactionNotifyListener,
+    init_geyser_service::{init_geyser_service, InitGeyserServiceConfig},
+    ledger::{
+        self, ledger_parent_dir, read_validator_keypair_from_ledger,
+        write_validator_keypair_to_ledger,
+    },
+    slot::advance_slot_and_update_ledger,
+    tickers::{
+        init_commit_accounts_ticker, init_slot_ticker,
+        init_system_metrics_ticker,
+    },
+};
 use conjunto_transwise::RpcProviderConfig;
 use log::*;
 use magicblock_account_cloner::{
@@ -59,26 +78,6 @@ use solana_sdk::{
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;
 
-use crate::{
-    accounts::create_accounts_run_and_snapshot_dirs,
-    errors::{ApiError, ApiResult},
-    external_config::try_convert_accounts_config,
-    fund_account::{
-        fund_magic_context, fund_validator_identity, funded_faucet,
-    },
-    geyser_transaction_notify_listener::GeyserTransactionNotifyListener,
-    init_geyser_service::{init_geyser_service, InitGeyserServiceConfig},
-    ledger::{
-        self, ledger_parent_dir, read_validator_keypair_from_ledger,
-        write_validator_keypair_to_ledger,
-    },
-    slot::advance_slot_and_update_ledger,
-    tickers::{
-        init_commit_accounts_ticker, init_slot_ticker,
-        init_system_metrics_ticker,
-    },
-};
-
 // -----------------
 // MagicValidatorConfig
 // -----------------