fix: replication state mismatch

.gitignore

@@ -15,11 +15,12 @@ target/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+.vscode
 
 # Ledger
 test-ledger/
 test-ledger-magicblock/
-magicblock-test-storage/
+magicblock-test-storage*
 
 # Mac
 **/.DS_Store
@@ -29,9 +30,11 @@ magicblock-test-storage/
 .github/packages/npm-package/node_modules
 .github/packages/npm-package/package-lock.json
 
+# Configs
+config.json
+config.toml
+
 # AI related
 **/CLAUDE.md
 CODEBASE_MAP.md
-config.json
-config.toml
 AGENTS.md

Cargo.toml

@@ -100,7 +100,9 @@ magicblock-accounts = { path = "./magicblock-accounts" }
 magicblock-accounts-db = { path = "./magicblock-accounts-db" }
 magicblock-aperture = { path = "./magicblock-aperture" }
 magicblock-api = { path = "./magicblock-api" }
-magicblock-chainlink = { path = "./magicblock-chainlink" }
+magicblock-chainlink = { path = "./magicblock-chainlink", features = [
+  "dev-context",
+] }
 magicblock-committor-program = { path = "./magicblock-committor-program", features = [
   "no-entrypoint",
 ] }

magicblock-accounts-db/src/lib.rs

@@ -408,7 +408,7 @@ impl AccountsDb {
     }
 
     pub fn database_directory(&self) -> &Path {
-        self.snapshot_manager.database_path()
+        &self.snapshot_manager.snapshots_dir
     }
 }
 
@@ -434,7 +434,7 @@ impl AccountsBank for AccountsDb {
     /// Removes accounts matching a predicate.
     fn remove_where(
         &self,
-        predicate: impl Fn(&Pubkey, &AccountSharedData) -> bool,
+        mut predicate: impl FnMut(&Pubkey, &AccountSharedData) -> bool,
     ) -> AccountsDbResult<usize> {
         let to_remove = self
             .iter_all()

magicblock-accounts-db/src/snapshot.rs

@@ -66,7 +66,7 @@ impl SnapshotStrategy {
 #[derive(Debug)]
 pub struct SnapshotManager {
     db_path: PathBuf,
-    snapshots_dir: PathBuf,
+    pub(crate) snapshots_dir: PathBuf,
     strategy: SnapshotStrategy,
     /// Ordered registry of archive paths (oldest to newest).
     registry: Mutex<VecDeque<PathBuf>>,
@@ -129,7 +129,9 @@ impl SnapshotManager {
         let archive_path = snapshot_dir.with_extension(ARCHIVE_EXT);
         let tmp_path = archive_path.with_extension("tmp");
 
-        info!(archive_path = %archive_path.display(), "Archiving snapshot");
+        if let Some(archive) = archive_path.file_name() {
+            info!(?archive, "Archiving snapshot");
+        }
 
         // Write to temporary file first
         let file = File::create(&tmp_path).log_err(|| {
@@ -146,6 +148,7 @@ impl SnapshotManager {
         let file = enc.finish().log_err(|| "Failed to finish gzip archive")?;
         file.sync_all()
             .log_err(|| "Failed to sync archive to disk")?;
+        drop(file);
 
         // Atomically rename to final path
         fs::rename(&tmp_path, &archive_path).log_err(|| {
@@ -178,7 +181,9 @@ impl SnapshotManager {
         current_slot: u64,
     ) -> AccountsDbResult<bool> {
         // Validate archive structure
-        Self::validate_archive(archive_bytes)?;
+        Self::validate_archive(archive_bytes).log_err(|| {
+            format!("snapshot archive bytes are corrupted, slot: {slot}")
+        })?;
 
         let archive_path = self.slot_to_archive_path(slot);
         if archive_path.exists() {
@@ -331,7 +336,6 @@ impl SnapshotManager {
 
     /// Registers an archive in the registry.
     fn register_archive(&self, archive_path: PathBuf) {
-        info!(archive_path = %archive_path.display(), "Snapshot registered");
         self.registry.lock().push_back(archive_path);
     }
 

magicblock-accounts-db/src/traits.rs

@@ -8,7 +8,7 @@ pub trait AccountsBank: Send + Sync + 'static {
     fn remove_account(&self, pubkey: &Pubkey);
     fn remove_where(
         &self,
-        predicate: impl Fn(&Pubkey, &AccountSharedData) -> bool,
+        predicate: impl FnMut(&Pubkey, &AccountSharedData) -> bool,
     ) -> AccountsDbResult<usize>;
 
     fn remove_account_conditionally(

magicblock-aperture/src/lib.rs

@@ -1,3 +1,5 @@
+use std::net::SocketAddr;
+
 use error::{ApertureError, RpcError};
 use magicblock_config::config::aperture::ApertureConfig;
 use magicblock_core::link::DispatchEndpoints;
@@ -17,17 +19,21 @@ pub async fn initialize_aperture(
     dispatch: &DispatchEndpoints,
     cancel: CancellationToken,
 ) -> ApertureResult<JsonRpcServer> {
-    // Start up an event processor tasks, which will handle forwarding of any validator
-    // originating event to client subscribers, or use them to update server's caches
-    EventProcessor::start(config, &state, dispatch, cancel.clone())?;
-    let server = JsonRpcServer::new(config, state, dispatch, cancel).await?;
+    let server =
+        JsonRpcServer::new(config, state.clone(), dispatch, cancel.clone())
+            .await?;
+    // Start event processors only after the server has bound its sockets so a
+    // bind failure cannot leak background tasks during retries in tests/startup.
+    EventProcessor::start(config, &state, dispatch, cancel)?;
     Ok(server)
 }
 
 /// An entrypoint to startup JSON-RPC server, for both HTTP and WS requests
 pub struct JsonRpcServer {
     http: HttpServer,
     websocket: WebsocketServer,
+    http_addr: SocketAddr,
+    ws_addr: SocketAddr,
 }
 
 impl JsonRpcServer {
@@ -39,18 +45,42 @@ impl JsonRpcServer {
         cancel: CancellationToken,
     ) -> ApertureResult<Self> {
         // try to bind to socket before spawning anything (handy in tests)
-        let mut addr = config.listen.0;
-        let http = TcpListener::bind(addr).await.map_err(RpcError::internal)?;
-        addr.set_port(addr.port() + 1);
-        let ws = TcpListener::bind(addr).await.map_err(RpcError::internal)?;
+        let http = TcpListener::bind(config.listen.0)
+            .await
+            .map_err(RpcError::internal)?;
+        let http_addr = http.local_addr().map_err(RpcError::internal)?;
+
+        let mut ws_addr = http_addr;
+        if config.listen.0.port() == 0 {
+            ws_addr.set_port(0);
+        } else {
+            ws_addr.set_port(config.listen.0.port() + 1);
+        }
+        let ws = TcpListener::bind(ws_addr)
+            .await
+            .map_err(RpcError::internal)?;
+        let ws_addr = ws.local_addr().map_err(RpcError::internal)?;
 
         // initialize HTTP and Websocket servers
         let websocket = {
             let cancel = cancel.clone();
             WebsocketServer::new(ws, &state, cancel).await?
         };
         let http = HttpServer::new(http, state, cancel, dispatch).await?;
-        Ok(Self { http, websocket })
+        Ok(Self {
+            http,
+            websocket,
+            http_addr,
+            ws_addr,
+        })
+    }
+
+    pub fn http_addr(&self) -> SocketAddr {
+        self.http_addr
+    }
+
+    pub fn ws_addr(&self) -> SocketAddr {
+        self.ws_addr
     }
 
     /// Run JSON-RPC server indefinitely, until cancel token is used to signal shut down

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

@@ -1,5 +1,3 @@
-use magicblock_core::link::transactions::with_encoded;
-
 use super::prelude::*;
 
 impl HttpDispatcher {
@@ -10,39 +8,37 @@ impl HttpDispatcher {
     /// the faucet is not enabled on the node.
     pub(crate) async fn request_airdrop(
         &self,
-        request: &mut JsonRequest,
+        _request: &mut JsonRequest,
     ) -> HandlerResult {
         // Airdrops are only supported if a faucet keypair is configured.
         // Which is never the case with *ephemeral* running mode of the validator
-        let Some(ref faucet) = self.context.faucet else {
-            return Err(RpcError::invalid_request(
-                "free airdrop faucet is disabled",
-            ));
-        };
+        Err(RpcError::invalid_request("free airdrop faucet is disabled"))
+        // TODO(bmuddha): allow free airdrops when other modes are fully reintroduced
+        // https://github.com/magicblock-labs/magicblock-validator/issues/1093
 
-        let (pubkey, lamports) =
-            parse_params!(request.params()?, Serde32Bytes, u64);
-        let pubkey = some_or_err!(pubkey);
-        let lamports = some_or_err!(lamports);
-        if lamports == 0 {
-            return Err(RpcError::invalid_params("lamports must be > 0"));
-        }
+        // let (pubkey, lamports) =
+        //     parse_params!(request.params()?, Serde32Bytes, u64);
+        // let pubkey = some_or_err!(pubkey);
+        // let lamports = some_or_err!(lamports);
+        // if lamports == 0 {
+        //     return Err(RpcError::invalid_params("lamports must be > 0"));
+        // }
 
-        // Build and execute the airdrop transfer transaction.
-        let txn = solana_system_transaction::transfer(
-            faucet,
-            &pubkey,
-            lamports,
-            self.blocks.get_latest().hash,
-        );
-        // we just signed the transaction, it must have a signature
-        let signature =
-            SerdeSignature(txn.signatures.first().cloned().unwrap_or_default());
+        // // Build and execute the airdrop transfer transaction.
+        // let txn = solana_system_transaction::transfer(
+        //     faucet,
+        //     &pubkey,
+        //     lamports,
+        //     self.blocks.get_latest().hash,
+        // );
+        // // we just signed the transaction, it must have a signature
+        // let signature =
+        //     SerdeSignature(txn.signatures.first().cloned().unwrap_or_default());
 
-        self.transactions_scheduler
-            .execute(with_encoded(txn)?)
-            .await?;
+        // self.transactions_scheduler
+        //     .execute(with_encoded(txn)?)
+        //     .await?;
 
-        Ok(ResponsePayload::encode_no_context(&request.id, signature))
+        // Ok(ResponsePayload::encode_no_context(&request.id, signature))
     }
 }

magicblock-aperture/src/state/mod.rs

@@ -14,7 +14,6 @@ use magicblock_chainlink::{
 };
 use magicblock_ledger::Ledger;
 use solana_feature_set::FeatureSet;
-use solana_keypair::Keypair;
 use solana_pubkey::Pubkey;
 use subscriptions::SubscriptionsDb;
 use transactions::TransactionsCache;
@@ -31,6 +30,7 @@ pub type ChainlinkImpl = Chainlink<
 /// This struct aggregates thread-safe handles (`Arc`) and concurrently accessible
 /// components (caches, databases) that need to be available across various parts
 /// of the application, such as RPC handlers and event processors.
+#[derive(Clone)]
 pub struct SharedState {
     /// The public key of the validator node.
     pub(crate) context: NodeContext,
@@ -51,12 +51,10 @@ pub struct SharedState {
 }
 
 /// Holds the core configuration and runtime parameters that define the node's operational context.
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct NodeContext {
     /// The public key of the validator node.
     pub identity: Pubkey,
-    /// The keypair for the optional faucet, used to airdrop tokens.
-    pub faucet: Option<Keypair>,
     /// Base fee charged for transaction execution per signature.
     pub base_fee: u64,
     /// Runtime features activated for this node (used to compute fees)

magicblock-aperture/src/tests.rs

@@ -42,14 +42,8 @@ fn ws_channel() -> (WsConnectionChannel, Receiver<Bytes>) {
 
 fn chainlink(accounts_db: &Arc<AccountsDb>) -> ChainlinkImpl {
     let cfg = ChainLinkConfig::default();
-    ChainlinkImpl::try_new(
-        accounts_db,
-        None,
-        Pubkey::new_unique(),
-        Pubkey::new_unique(),
-        &cfg,
-    )
-    .expect("Failed to create Chainlink")
+    ChainlinkImpl::try_new(accounts_db, None, Pubkey::new_unique(), &cfg)
+        .expect("Failed to create Chainlink")
 }
 
 mod event_processor {

magicblock-aperture/tests/mocked.rs

@@ -158,13 +158,13 @@ async fn test_get_epoch_info() {
         .expect("get_epoch_info request failed");
 
     assert_eq!(epoch_info.epoch, 0, "epoch should be 0");
-    // The absolute_slot should be at most 1 behind the current slot
-    // due to auto-advancement timing
+    // The absolute_slot should be at most 3 behind the current slot
+    // due to auto-advancement timing (50ms block time)
     let current_slot = env.latest_slot();
     assert!(
         epoch_info.absolute_slot <= current_slot
-            && epoch_info.absolute_slot >= current_slot.saturating_sub(1),
-        "absolute_slot {} should be within 1 of current slot {}",
+            && epoch_info.absolute_slot >= current_slot.saturating_sub(3),
+        "absolute_slot {} should be within 3 of current slot {}",
         epoch_info.absolute_slot,
         current_slot
     );