feat: initial attempt at replicator integration

Cargo.toml

@@ -57,7 +57,7 @@ async-nats = "0.46"
 async-trait = "0.1.77"
 base64 = "0.21.7"
 bincode = "1.3.3"
-bytes = "1.0"
+bytes = { version = "1.0", features = ["serde"] }
 borsh = { version = "1.5.1", features = ["derive", "unstable__schema"] }
 bs58 = "0.5.1"
 byteorder = "1.5.0"
@@ -112,6 +112,7 @@ magicblock-magic-program-api = { path = "./magicblock-magic-program-api" }
 magicblock-metrics = { path = "./magicblock-metrics" }
 magicblock-processor = { path = "./magicblock-processor" }
 magicblock-program = { path = "./programs/magicblock" }
+magicblock-replicator = { path = "./magicblock-replicator" }
 magicblock-rpc-client = { path = "./magicblock-rpc-client" }
 magicblock-services = { path = "./magicblock-services" }
 magicblock-table-mania = { path = "./magicblock-table-mania" }

magicblock-accounts-db/src/lib.rs

@@ -6,7 +6,6 @@ use index::{
 };
 use lmdb::{RwTransaction, Transaction};
 use magicblock_config::config::AccountsDbConfig;
-use parking_lot::{RwLock, RwLockWriteGuard};
 use solana_account::{
     cow::AccountBorrowed, AccountSharedData, ReadableAccount,
 };
@@ -19,10 +18,6 @@ use crate::{snapshot::SnapshotManager, traits::AccountsBank};
 
 pub type AccountsDbResult<T> = Result<T, AccountsDbError>;
 
-/// A global lock used to suspend all write operations during critical
-/// sections (like snapshots).
-pub type GlobalSyncLock = Arc<RwLock<()>>;
-
 pub const ACCOUNTSDB_DIR: &str = "accountsdb";
 
 /// The main Accounts Database.
@@ -39,10 +34,6 @@ pub struct AccountsDb {
     index: AccountsDbIndex,
     /// Manages snapshots and state restoration.
     snapshot_manager: Arc<SnapshotManager>,
-    /// Global lock ensures atomic snapshots by pausing writes.
-    /// Note: Reads are generally wait-free/lock-free via mmap,
-    /// unless they require index cursor stability.
-    write_lock: GlobalSyncLock,
 }
 
 impl AccountsDb {
@@ -79,7 +70,6 @@ impl AccountsDb {
             storage,
             index,
             snapshot_manager,
-            write_lock: GlobalSyncLock::default(),
         };
 
         // Recover state if the requested slot is older than our current state
@@ -291,30 +281,22 @@ impl AccountsDb {
     ///
     /// Returns the state checksum computed at snapshot time.
     /// The checksum can be used to verify state consistency across nodes.
-    pub fn take_snapshot(self: &Arc<Self>, slot: u64) -> u64 {
-        let this = self.clone();
-
-        // Phase 1: Create snapshot directory (with write lock)
-        let locked = this.write_lock.write();
-        this.flush();
-        // SAFETY:
-        // we have acquired the write lock above
-        let checksum = unsafe { this.checksum() };
-        let used_storage = this.storage.active_segment();
-
-        let snapshot_dir = this
-            .snapshot_manager
-            .create_snapshot_dir(slot, used_storage);
-        drop(locked);
-        thread::spawn(move || {
-            // Phase 2: Archive directory (no lock needed)
-            let _ = snapshot_dir
-                .and_then(|dir| {
-                    this.snapshot_manager.archive_and_register(&dir)
-                })
-                .log_err(|| "failed to create accountsdb snapshot");
-        });
-        checksum
+    ///
+    ///
+    /// # Safety
+    /// the caller must ensure that no state transitions are taking
+    /// place concurrently when this operation is in progress
+    pub unsafe fn take_snapshot(&self, slot: u64) -> AccountsDbResult<u64> {
+        // Create snapshot directory (potential deep copy)
+        self.flush();
+        let checksum = self.checksum();
+        let used_storage = self.storage.active_segment();
+
+        let manager = self.snapshot_manager.clone();
+        let dir = manager.create_snapshot_dir(slot, used_storage)?;
+        // Archive directory in background to avoid blocking the caller
+        thread::spawn(move || manager.archive_and_register(&dir));
+        Ok(checksum)
     }
 
     /// Ensures the database state is at most `slot`.
@@ -336,9 +318,6 @@ impl AccountsDb {
             "Current slot ahead of target, rolling back"
         );
 
-        // Block all writes during restoration
-        let _guard = self.write_lock.write();
-
         let restored_slot = self
             .snapshot_manager
             .restore_from_snapshot(target_slot)
@@ -372,10 +351,6 @@ impl AccountsDb {
         self.index.flush();
     }
 
-    pub fn write_lock(&self) -> GlobalSyncLock {
-        self.write_lock.clone()
-    }
-
     /// Inserts an external snapshot archive received over the network.
     ///
     /// If the snapshot slot is newer than the current DB slot, immediately
@@ -418,8 +393,8 @@ impl AccountsDb {
     /// suitable for verifying state consistency across nodes.
     ///
     /// # Safety
-    /// the caller must acquire the write lock on accountsdb, so that
-    /// the state doesn't change during checksum computation
+    /// the caller must ensure that no concurrent write access is being performed on
+    /// accountsdb, so that the state doesn't change during checksum computation
     pub unsafe fn checksum(&self) -> u64 {
         let mut hasher = xxhash3_64::Hasher::new();
         for (pubkey, acc) in self.iter_all() {
@@ -432,15 +407,6 @@ impl AccountsDb {
         hasher.finish()
     }
 
-    /// Acquires exclusive write access to the database.
-    ///
-    /// The returned guard blocks all other write operations while held.
-    /// Use this when you need to ensure the database state doesn't change
-    /// during operations like checksum computation.
-    pub fn lock_database(&self) -> RwLockWriteGuard<'_, ()> {
-        self.write_lock.write()
-    }
-
     pub fn database_directory(&self) -> &Path {
         self.snapshot_manager.database_path()
     }

magicblock-accounts-db/src/tests.rs

@@ -585,12 +585,6 @@ fn test_checksum_deterministic_across_dbs() {
         db2.insert_account(&pubkey, &account).unwrap();
     }
 
-    // Acquire write locks before computing checksums
-    let lock1 = db1.write_lock();
-    let lock2 = db2.write_lock();
-    let _guard1 = lock1.write();
-    let _guard2 = lock2.write();
-
     assert_eq!(
         unsafe { db1.checksum() },
         unsafe { db2.checksum() },
@@ -610,17 +604,13 @@ fn test_checksum_detects_state_change() {
         })
         .collect();
 
-    let lock = env.write_lock();
-    let _guard = lock.write();
     let original_checksum = unsafe { env.checksum() };
-    drop(_guard);
 
     // Modify a single account's data
     accounts[5].1.data_as_mut_slice()[0] ^= 0xFF;
     env.insert_account(&accounts[5].0, &accounts[5].1).unwrap();
 
     {
-        let _guard = lock.write();
         assert_ne!(
             unsafe { env.checksum() },
             original_checksum,
@@ -634,7 +624,6 @@ fn test_checksum_detects_state_change() {
         .unwrap();
 
     {
-        let _guard = lock.write();
         assert_ne!(
             unsafe { env.checksum() },
             original_checksum,
@@ -715,15 +704,15 @@ impl TestEnv {
 
     /// Takes a snapshot and waits for archiving to complete.
     fn take_snapshot_and_wait(&self, slot: u64) -> u64 {
-        let checksum = self.adb.take_snapshot(slot);
+        let checksum = unsafe { self.adb.take_snapshot(slot) };
         // Wait for background archiving to complete
         let mut retries = 0;
         while !self.adb.snapshot_exists(slot) && retries < 200 {
             std::thread::sleep(std::time::Duration::from_millis(50));
             retries += 1;
         }
         assert!(self.adb.snapshot_exists(slot), "Snapshot should exist");
-        checksum
+        checksum.expect("failed to take accountsdb snapshot")
     }
 
     fn restore_to_slot(mut self, slot: u64) -> Self {

magicblock-aperture/src/encoder.rs

@@ -3,18 +3,15 @@ use std::fmt::Debug;
 use hyper::body::Bytes;
 use json::Serialize;
 use magicblock_core::{
-    link::{
-        accounts::LockedAccount,
-        transactions::{TransactionResult, TransactionStatus},
-    },
+    link::{accounts::LockedAccount, transactions::TransactionStatus},
     Slot,
 };
 use solana_account::ReadableAccount;
 use solana_account_decoder::{
     encode_ui_account, UiAccountEncoding, UiDataSliceConfig,
 };
 use solana_pubkey::Pubkey;
-use solana_transaction_error::TransactionError;
+use solana_transaction_error::{TransactionError, TransactionResult};
 
 use crate::{
     requests::{params::SerdeSignature, payload::NotificationPayload},
@@ -110,7 +107,7 @@ impl Encoder for ProgramAccountEncoder {
 pub(crate) struct TransactionResultEncoder;
 
 impl Encoder for TransactionResultEncoder {
-    type Data = TransactionResult;
+    type Data = TransactionResult<()>;
 
     fn encode(
         &self,

magicblock-aperture/src/requests/http/mod.rs

@@ -213,7 +213,10 @@ impl HttpDispatcher {
         }
 
         let txn = transaction.sanitize(sigverify)?;
-        Ok(WithEncoded { txn, encoded })
+        Ok(WithEncoded {
+            txn,
+            encoded: encoded.into(),
+        })
     }
 
     /// Ensures all accounts required for a transaction are present in the `AccountsDb`.

magicblock-aperture/src/state/transactions.rs

@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
-use magicblock_core::{link::transactions::TransactionResult, Slot};
+use magicblock_core::Slot;
 use solana_signature::Signature;
+use solana_transaction_error::TransactionResult;
 
 use super::ExpiringCache;
 
@@ -18,5 +19,5 @@ pub(crate) struct SignatureResult {
     /// The slot in which the transaction was processed.
     pub slot: Slot,
     /// The result of the transaction (e.g., success or an error).
-    pub result: TransactionResult,
+    pub result: TransactionResult<()>,
 }

magicblock-api/Cargo.toml

@@ -29,6 +29,7 @@ magicblock-metrics = { workspace = true }
 magicblock-processor = { workspace = true }
 magicblock-program = { workspace = true }
 magicblock-services = { workspace = true }
+magicblock-replicator = { workspace = true }
 magicblock-task-scheduler = { workspace = true }
 magicblock-validator-admin = { workspace = true }
 

magicblock-api/src/errors.rs

@@ -116,6 +116,9 @@ pub enum ApiError {
     FailedToSanitizeTransaction(
         #[from] solana_transaction_error::TransactionError,
     ),
+
+    #[error("Replication service failed: {0}")]
+    Replication(#[from] magicblock_replicator::Error),
 }
 
 impl From<magicblock_accounts::errors::AccountsError> for ApiError {