hotfix: small fixes for the replay (#304)

## Description

This PR introduces optimizations and reliability improvements for ledger
replay:

- Start from the last (not corrupted snapshot), delete corrupted ones
- Added hard linking for account files in `accounts_persister.rs` to
prevent file removal during ledger access and replay
- Increased RemoteAccountUpdatesWorker connection refresh interval from
5 to 50 minutes in `magic_validator.rs` for better stability (also,
there are issues if the ws shard disconnect during replay)
- Added `FailedToGetSubscriptionSlot` error handling in
`account_cloner.rs` for subscription slot retrieval failures
- Increased account hydration concurrency from 10 to 30 and fetch
retries from 10 to 50 in `remote_account_cloner_worker.rs` for improved
reliability


<!-- /greptile_comment -->

Author: GabrielePicco

magicblock-account-cloner/src/account_cloner.rs

@@ -35,6 +35,9 @@ pub enum AccountClonerError {
 
     #[error("FailedToFetchSatisfactorySlot")]
     FailedToFetchSatisfactorySlot,
+
+    #[error("FailedToGetSubscriptionSlot")]
+    FailedToGetSubscriptionSlot,
 }
 
 pub type AccountClonerResult<T> = Result<T, AccountClonerError>;

magicblock-account-cloner/src/remote_account_cloner_worker.rs

@@ -134,7 +134,7 @@ where
     ) -> Self {
         let (clone_request_sender, clone_request_receiver) =
             unbounded_channel();
-        let fetch_retries = 10;
+        let fetch_retries = 50;
         Self {
             internal_account_provider,
             account_fetcher,
@@ -273,9 +273,10 @@ where
         // retry resulting in overall slower hydration.
         // If the optimal rate here is desired we might make this configurable in the
         // future.
+        // TODO(GabrielePicco): Make the concurrency configurable
         stream
             .map(Ok::<_, AccountClonerError>)
-            .try_for_each_concurrent(10, |(pubkey, owner)| async move {
+            .try_for_each_concurrent(30, |(pubkey, owner)| async move {
                 trace!("Hydrating '{}'", pubkey);
                 let res = self
                     .do_clone_and_update_cache(
@@ -446,9 +447,15 @@ where
                         }
                         // If we failed to fetch too many time, stop here
                         if fetch_count >= self.fetch_retries {
-                            return Err(
-                                AccountClonerError::FailedToFetchSatisfactorySlot,
-                            );
+                            return if min_context_slot.is_none() {
+                                Err(
+                                    AccountClonerError::FailedToGetSubscriptionSlot,
+                                )
+                            } else {
+                                Err(
+                                    AccountClonerError::FailedToFetchSatisfactorySlot,
+                                )
+                            };
                         }
                     }
                     Err(error) => {
@@ -459,7 +466,7 @@ where
                     }
                 };
                 // Wait a bit in the hopes of the min_context_slot becoming available (about half a slot)
-                sleep(Duration::from_millis(300)).await;
+                sleep(Duration::from_millis(400)).await;
             }
         } else {
             self.fetch_account_chain_snapshot(pubkey, None).await?

magicblock-account-cloner/tests/remote_account_cloner.rs

@@ -278,7 +278,7 @@ async fn test_clone_fails_stale_undelegated_account_when_ephemeral() {
         result,
         Err(AccountClonerError::FailedToFetchSatisfactorySlot)
     ));
-    assert_eq!(account_fetcher.get_fetch_count(&undelegated_account), 10); // Must have retried
+    assert_eq!(account_fetcher.get_fetch_count(&undelegated_account), 50); // Must have retried
     assert!(account_updates.has_account_monitoring(&undelegated_account));
     assert!(account_dumper.was_untouched(&undelegated_account));
     // Cleanup everything correctly

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

@@ -254,7 +254,6 @@ impl AccountsPersister {
             .first()
             .ok_or(AccountsDbError::NoStoragePathProvided)?;
 
-        // Read all files sorted slot/append_vec_id and return the last one
         let files = fs::read_dir(path)?;
         let mut files: Vec<_> = files
             .filter_map(|entry| {
@@ -285,56 +284,61 @@ impl AccountsPersister {
             })
             .collect();
 
-        files.sort_by(
-            |(_, slot_a, id_a): &(PathBuf, Slot, AppendVecId),
-             (_, slot_b, id_b): &(PathBuf, Slot, AppendVecId)| {
-                // Sorting in reverse order
-                if slot_a == slot_b {
-                    id_b.cmp(id_a)
-                } else {
-                    slot_b.cmp(slot_a)
-                }
-            },
-        );
+        files.sort_by(|(_, slot_a, id_a), (_, slot_b, id_b)| {
+            if slot_a == slot_b {
+                id_b.cmp(id_a)
+            } else {
+                slot_b.cmp(slot_a)
+            }
+        });
 
-        let matching_file = {
-            let mut matching_file = None;
-            for (file, slot, id) in files {
-                if slot <= max_slot {
-                    matching_file.replace((file, slot, id));
-                    break;
+        for (file, slot, id) in files {
+            if slot <= max_slot {
+                let link = file.with_extension("link");
+                if let Err(err) = fs::hard_link(&file, &link) {
+                    warn!("Failed to create hard link for {:?}: {}", file, err);
+                    continue;
+                }
+                let file = link;
+
+                let file_size = match fs::metadata(&file) {
+                    Ok(metadata) => metadata.len() as usize,
+                    Err(err) => {
+                        warn!("Failed to get metadata for {:?}: {}", file, err);
+                        let _ = fs::remove_file(&file);
+                        continue;
+                    }
+                };
+
+                match AppendVec::new_from_file(
+                    &file,
+                    file_size,
+                    StorageAccess::Mmap,
+                ) {
+                    Ok((append_vec, num_accounts)) => {
+                        let accounts = AccountsFile::AppendVec(append_vec);
+                        let storage = AccountStorageEntry::new_existing(
+                            slot,
+                            id,
+                            accounts,
+                            num_accounts,
+                        );
+                        return Ok(Some((storage, slot)));
+                    }
+                    Err(err) => {
+                        warn!("Failed to load append vec from {:?}: {}. Deleting corrupted file.", file, err);
+                        let _ = fs::remove_file(&file);
+                    }
                 }
             }
-            matching_file
-        };
-        let Some((file, slot, id)) = matching_file else {
-            warn!(
-                "No storage found with slot <= {} inside {}",
-                max_slot,
-                path.display().to_string(),
-            );
-            return Ok(None);
-        };
-
-        // When we drop the AppendVec the underlying file is removed from the
-        // filesystem. There is no way to configure this via public methods.
-        // Thus we copy the file before using it for the AppendVec. This way
-        // we prevent account files being removed when we point a tool at the ledger
-        // or replay it.
-        let file = {
-            let copy = file.with_extension("copy");
-            fs::copy(&file, &copy)?;
-            copy
-        };
+        }
 
-        // Create a AccountStorageEntry from the file
-        let file_size = fs::metadata(&file)?.len() as usize;
-        let (append_vec, num_accounts) =
-            AppendVec::new_from_file(&file, file_size, StorageAccess::Mmap)?;
-        let accounts = AccountsFile::AppendVec(append_vec);
-        let storage =
-            AccountStorageEntry::new_existing(slot, id, accounts, num_accounts);
-        Ok(Some((storage, slot)))
+        warn!(
+            "No valid storage found with slot <= {} inside {}",
+            max_slot,
+            path.display().to_string(),
+        );
+        Ok(None)
     }
 
     // -----------------

magicblock-api/src/magic_validator.rs

@@ -242,8 +242,8 @@ impl MagicValidator {
                 remote_rpc_config.clone(),
                 remote_rpc_config.clone(),
             ],
-            // We'll kill/refresh one connection every 5 minutes
-            Duration::from_secs(60 * 5),
+            // We'll kill/refresh one connection every 50 minutes
+            Duration::from_secs(60 * 50),
         );
 
         let transaction_status_sender = TransactionStatusSender {