chore: load network accounts asynchronously in NTX Builder

CHANGELOG.md

@@ -20,6 +20,7 @@
 - Added `GetLimits` endpoint to the RPC server ([#1410](https://github.com/0xMiden/miden-node/pull/1410)).
 - Added gRPC-Web probe support to the `miden-network-monitor` binary ([#1484](https://github.com/0xMiden/miden-node/pull/1484)).
 - Add DB schema change check ([#1268](https://github.com/0xMiden/miden-node/pull/1485)).
+- Decoupled ntx-builder from block-producer startup by loading network accounts asynchronously via a background task ([#????](https://github.com/0xMiden/miden-node/pull/????)).
 
 ### Changes
 

bin/node/src/commands/block_producer.rs

@@ -1,10 +1,8 @@
-use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
 use miden_node_block_producer::BlockProducer;
 use miden_node_utils::grpc::UrlExt;
-use tokio::sync::Barrier;
 use url::Url;
 
 use super::{ENV_BLOCK_PRODUCER_URL, ENV_STORE_BLOCK_PRODUCER_URL};
@@ -93,7 +91,6 @@ impl BlockProducerCommand {
             block_interval: block_producer.block_interval,
             max_txs_per_batch: block_producer.max_txs_per_batch,
             max_batches_per_block: block_producer.max_batches_per_block,
-            production_checkpoint: Arc::new(Barrier::new(1)),
             grpc_timeout,
             mempool_tx_capacity: block_producer.mempool_tx_capacity,
         }

bin/node/src/commands/bundled.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 use std::path::PathBuf;
-use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
@@ -14,7 +13,6 @@ use miden_protocol::block::BlockSigner;
 use miden_protocol::crypto::dsa::ecdsa_k256_keccak::SecretKey;
 use miden_protocol::utils::Deserializable;
 use tokio::net::TcpListener;
-use tokio::sync::Barrier;
 use tokio::task::JoinSet;
 use url::Url;
 
@@ -222,19 +220,11 @@ impl BundledCommand {
             })
             .id();
 
-        // A sync point between the ntx-builder and block-producer components.
         let should_start_ntx_builder = !ntx_builder.disabled;
-        let checkpoint = if should_start_ntx_builder {
-            Barrier::new(2)
-        } else {
-            Barrier::new(1)
-        };
-        let checkpoint = Arc::new(checkpoint);
 
         // Start block-producer. The block-producer's endpoint is available after loading completes.
         let block_producer_id = join_set
             .spawn({
-                let checkpoint = Arc::clone(&checkpoint);
                 let store_url = Url::parse(&format!("http://{store_block_producer_address}"))
                     .context("Failed to parse URL")?;
                 let validator_url = Url::parse(&format!("http://{validator_address}"))
@@ -250,7 +240,6 @@ impl BundledCommand {
                         block_interval: block_producer.block_interval,
                         max_batches_per_block: block_producer.max_batches_per_block,
                         max_txs_per_batch: block_producer.max_txs_per_batch,
-                        production_checkpoint: checkpoint,
                         grpc_timeout,
                         mempool_tx_capacity: block_producer.mempool_tx_capacity,
                     }
@@ -320,8 +309,6 @@ impl BundledCommand {
                         store_ntx_builder_url,
                         block_producer_url,
                         ntx_builder.tx_prover_url,
-                        ntx_builder.ticker_interval,
-                        checkpoint,
                         ntx_builder.script_cache_size,
                     )
                     .run()

crates/block-producer/src/server/mod.rs

@@ -18,7 +18,7 @@ use miden_protocol::block::BlockNumber;
 use miden_protocol::transaction::ProvenTransaction;
 use miden_protocol::utils::serde::Deserializable;
 use tokio::net::TcpListener;
-use tokio::sync::{Barrier, Mutex, RwLock};
+use tokio::sync::{Mutex, RwLock};
 use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
 use tonic::Status;
 use tower_http::trace::TraceLayer;
@@ -65,11 +65,6 @@ pub struct BlockProducer {
     pub max_txs_per_batch: usize,
     /// The maximum number of batches per block.
     pub max_batches_per_block: usize,
-    /// Block production only begins after this checkpoint barrier completes.
-    ///
-    /// The block-producers gRPC endpoint will be available before this point, so this lets the
-    /// mempool synchronize its event stream without risking a race condition.
-    pub production_checkpoint: Arc<Barrier>,
     /// Server-side timeout for an individual gRPC request.
     ///
     /// If the handler takes longer than this duration, the server cancels the call.
@@ -155,12 +150,7 @@ impl BlockProducer {
         // any complete or fail, we can shutdown the rest (somewhat) gracefully.
         let mut tasks = tokio::task::JoinSet::new();
 
-        // Launch the gRPC server and wait at the checkpoint for any other components to be in sync.
-        //
-        // This is used to ensure the ntx-builder can subscribe to the mempool events without
-        // playing catch up caused by block-production.
-        //
-        // This is a temporary work-around until the ntx-builder can resync on the fly.
+        // Launch the gRPC server.
         let rpc_id = tasks
             .spawn({
                 let mempool = mempool.clone();
@@ -171,7 +161,6 @@ impl BlockProducer {
                 }
             })
             .id();
-        self.production_checkpoint.wait().await;
 
         let batch_builder_id = tasks
             .spawn({

crates/ntx-builder/src/builder.rs

@@ -1,20 +1,19 @@
 use std::num::NonZeroUsize;
 use std::sync::Arc;
-use std::time::Duration;
 
 use anyhow::Context;
 use futures::TryStreamExt;
 use miden_node_proto::domain::account::NetworkAccountPrefix;
 use miden_node_proto::domain::mempool::MempoolEvent;
 use miden_node_utils::lru_cache::LruCache;
 use miden_protocol::Word;
+use miden_protocol::account::AccountId;
 use miden_protocol::account::delta::AccountUpdateDetails;
 use miden_protocol::block::BlockHeader;
 use miden_protocol::crypto::merkle::mmr::PartialMmr;
 use miden_protocol::note::NoteScript;
 use miden_protocol::transaction::PartialBlockchain;
-use tokio::sync::{Barrier, RwLock};
-use tokio::time;
+use tokio::sync::{RwLock, mpsc};
 use url::Url;
 
 use crate::MAX_IN_PROGRESS_TXS;
@@ -76,13 +75,6 @@ pub struct NetworkTransactionBuilder {
     /// Address of the remote prover. If `None`, transactions will be proven locally, which is
     /// undesirable due to the performance impact.
     tx_prover_url: Option<Url>,
-    /// Interval for checking pending notes and executing network transactions.
-    ticker_interval: Duration,
-    /// A checkpoint used to sync start-up process with the block-producer.
-    ///
-    /// This informs the block-producer when we have subscribed to mempool events and that it is
-    /// safe to begin block-production.
-    bp_checkpoint: Arc<Barrier>,
     /// Shared LRU cache for storing retrieved note scripts to avoid repeated store calls.
     /// This cache is shared across all account actors.
     script_cache: LruCache<Word, NoteScript>,
@@ -91,13 +83,16 @@ pub struct NetworkTransactionBuilder {
 }
 
 impl NetworkTransactionBuilder {
+    /// Channel capacity for account loading. Each item in the channel is a batch of account IDs
+    /// loaded from a single page. A small capacity is sufficient since each page can contain
+    /// up to ~289,000 accounts.
+    const ACCOUNT_CHANNEL_CAPACITY: usize = 4;
+
     /// Creates a new instance of the network transaction builder.
     pub fn new(
         store_url: Url,
         block_producer_url: Url,
         tx_prover_url: Option<Url>,
-        ticker_interval: Duration,
-        bp_checkpoint: Arc<Barrier>,
         script_cache_size: NonZeroUsize,
     ) -> Self {
         let script_cache = LruCache::new(script_cache_size);
@@ -106,8 +101,6 @@ impl NetworkTransactionBuilder {
             store_url,
             block_producer_url,
             tx_prover_url,
-            ticker_interval,
-            bp_checkpoint,
             script_cache,
             coordinator,
         }
@@ -127,15 +120,6 @@ impl NetworkTransactionBuilder {
             .await
             .context("failed to subscribe to mempool events")?;
 
-        // Unlock the block-producer's block production. The block-producer is prevented from
-        // producing blocks until we have subscribed to mempool events.
-        //
-        // This is a temporary work-around until the ntx-builder can resync on the fly.
-        self.bp_checkpoint.wait().await;
-
-        let mut interval = tokio::time::interval(self.ticker_interval);
-        interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
-
         // Create chain state that will be updated by the coordinator and read by actors.
         let chain_state = Arc::new(RwLock::new(ChainState::new(chain_tip_header, chain_mmr)));
 
@@ -147,15 +131,16 @@ impl NetworkTransactionBuilder {
             script_cache: self.script_cache.clone(),
         };
 
-        // Create initial set of actors based on all known network accounts.
-        let account_ids = store.get_network_account_ids().await?;
-        for account_id in account_ids {
-            if let Ok(account_prefix) = NetworkAccountPrefix::try_from(account_id) {
-                self.coordinator
-                    .spawn_actor(AccountOrigin::store(account_prefix), &actor_context)
-                    .await?;
+        // Spawn a background task to load network accounts from the store.
+        // Accounts are sent through a channel in batches and processed in the main event loop.
+        let (account_tx, mut account_rx) =
+            mpsc::channel::<Vec<AccountId>>(Self::ACCOUNT_CHANNEL_CAPACITY);
+        let account_loader_store = store.clone();
+        tokio::spawn(async move {
+            if let Err(err) = account_loader_store.stream_network_account_ids(account_tx).await {
+                tracing::error!(%err, "failed to load network accounts from store");
             }
-        }
+        });
 
         // Main loop which manages actors and passes mempool events to them.
         loop {
@@ -176,8 +161,35 @@ impl NetworkTransactionBuilder {
                         chain_state.clone(),
                     ).await?;
                 },
+                // Handle account batches loaded from the store.
+                // Once all accounts are loaded, the channel closes and this branch
+                // becomes inactive (recv returns None and we stop matching).
+                Some(account_ids) = account_rx.recv() => {
+                    self.handle_loaded_accounts(account_ids, &actor_context).await?;
+                },
+            }
+        }
+    }
+
+    /// Handles a batch of account IDs loaded from the store by spawning actors for them.
+    #[tracing::instrument(
+        name = "ntx.builder.handle_loaded_accounts",
+        skip(self, account_ids, actor_context),
+        fields(count = account_ids.len())
+    )]
+    async fn handle_loaded_accounts(
+        &mut self,
+        account_ids: Vec<AccountId>,
+        actor_context: &AccountActorContext,
+    ) -> Result<(), anyhow::Error> {
+        for account_id in account_ids {
+            if let Ok(account_prefix) = NetworkAccountPrefix::try_from(account_id) {
+                self.coordinator
+                    .spawn_actor(AccountOrigin::store(account_prefix), actor_context)
+                    .await?;
             }
         }
+        Ok(())
     }
 
     /// Handles mempool events by sending them to actors via the coordinator and/or spawning new

crates/ntx-builder/src/store.rs

@@ -173,21 +173,24 @@ impl StoreClient {
         Ok(all_notes)
     }
 
-    /// Get all network account IDs.
+    /// Streams network account IDs to the provided sender, page-by-page.
     ///
-    /// Since the `GetNetworkAccountIds` method is paginated, we loop through all pages until we
-    /// reach the end.
+    /// This method is designed to be run in a background task, sending accounts to the main event
+    /// loop as they are loaded. This allows the ntx-builder to start processing mempool events
+    /// without waiting for all accounts to be preloaded.
     ///
     /// Each page can return up to `MAX_RESPONSE_PAYLOAD_BYTES / AccountId::SERIALIZED_SIZE`
-    /// accounts (~289,000). With `100_000` iterations, which is assumed to be sufficient for the
-    /// foreseeable future.
-    #[instrument(target = COMPONENT, name = "store.client.get_network_account_ids", skip_all, err)]
-    pub async fn get_network_account_ids(&self) -> Result<Vec<AccountId>, StoreError> {
-        const MAX_ITERATIONS: u32 = 100_000;
+    /// accounts (~289,000). With 1000 iterations, this supports up to ~524 million network
+    /// accounts, which is assumed to be sufficient for the foreseeable future.
+    #[instrument(target = COMPONENT, name = "store.client.stream_network_account_ids", skip_all, err)]
+    pub async fn stream_network_account_ids(
+        &self,
+        sender: tokio::sync::mpsc::Sender<Vec<AccountId>>,
+    ) -> Result<(), StoreError> {
+        const MAX_ITERATIONS: u32 = 1000;
 
         let mut block_range = BlockNumber::from(0)..=BlockNumber::from(u32::MAX);
 
-        let mut ids = Vec::new();
         let mut iterations_count = 0;
 
         loop {
@@ -198,14 +201,14 @@ impl StoreClient {
                 .await?
                 .into_inner();
 
-            let accounts: Result<Vec<AccountId>, ConversionError> = response
+            let accounts = response
                 .account_ids
                 .into_iter()
                 .map(|account_id| {
                     AccountId::read_from_bytes(&account_id.id)
                         .map_err(|err| ConversionError::deserialization_error("account_id", err))
                 })
-                .collect();
+                .collect::<Result<Vec<AccountId>, ConversionError>>()?;
 
             let pagination_info = response.pagination_info.ok_or(
                 ConversionError::MissingFieldInProtobufRepresentation {
@@ -214,7 +217,13 @@ impl StoreClient {
                 },
             )?;
 
-            ids.extend(accounts?);
+            if !accounts.is_empty() {
+                // Send the entire batch at once. If the receiver is dropped, stop loading.
+                if sender.send(accounts).await.is_err() {
+                    return Ok(());
+                }
+            }
+
             iterations_count += 1;
             block_range =
                 BlockNumber::from(pagination_info.block_num)..=BlockNumber::from(u32::MAX);
@@ -228,7 +237,7 @@ impl StoreClient {
             }
         }
 
-        Ok(ids)
+        Ok(())
     }
 
     #[instrument(target = COMPONENT, name = "store.client.get_note_script_by_root", skip_all, err)]