miden_node_ntx_builder/
builder.rs

1use std::pin::Pin;
2use std::sync::Arc;
3
4use anyhow::Context;
5use futures::Stream;
6use miden_node_proto::domain::account::NetworkAccountId;
7use miden_node_proto::domain::mempool::MempoolEvent;
8use miden_protocol::account::delta::AccountUpdateDetails;
9use tokio::net::TcpListener;
10use tokio::sync::mpsc;
11use tokio::task::JoinSet;
12use tokio_stream::StreamExt;
13use tonic::Status;
14
15use crate::NtxBuilderConfig;
16use crate::actor::{AccountActorContext, AccountOrigin, ActorRequest};
17use crate::chain_state::SharedChainState;
18use crate::clients::StoreClient;
19use crate::coordinator::Coordinator;
20use crate::db::Db;
21use crate::server::NtxBuilderRpcServer;
22
23// NETWORK TRANSACTION BUILDER
24// ================================================================================================
25
26/// A boxed, pinned stream of mempool events with a `'static` lifetime.
27///
28/// Boxing gives the stream a `'static` lifetime by ensuring it owns all its data, avoiding
29/// complex lifetime annotations that would otherwise be required when storing `impl TryStream`.
30pub(crate) type MempoolEventStream =
31    Pin<Box<dyn Stream<Item = Result<MempoolEvent, Status>> + Send>>;
32
33/// Network transaction builder component.
34///
35/// The network transaction builder is in charge of building transactions that consume notes
36/// against network accounts. These notes are identified and communicated by the block producer.
37/// The service maintains a list of unconsumed notes and periodically executes and proves
38/// transactions that consume them (reaching out to the store to retrieve state as necessary).
39///
40/// The builder manages the tasks for every network account on the chain through the coordinator.
41///
42/// Create an instance using [`NtxBuilderConfig::build()`].
43pub struct NetworkTransactionBuilder {
44    /// Configuration for the builder.
45    config: NtxBuilderConfig,
46    /// Coordinator for managing actor tasks.
47    coordinator: Coordinator,
48    /// Client for the store gRPC API.
49    store: StoreClient,
50    /// Database for persistent state.
51    db: Db,
52    /// Shared chain state updated by the event loop and read by actors.
53    chain_state: Arc<SharedChainState>,
54    /// Context shared with all account actors.
55    actor_context: AccountActorContext,
56    /// Stream of mempool events from the block producer.
57    mempool_events: MempoolEventStream,
58    /// Database update requests from account actors.
59    ///
60    /// We keep database writes centralized so this is how actors communicate
61    /// items to write.
62    actor_request_rx: mpsc::Receiver<ActorRequest>,
63}
64
65impl NetworkTransactionBuilder {
66    #[expect(clippy::too_many_arguments)]
67    pub(crate) fn new(
68        config: NtxBuilderConfig,
69        coordinator: Coordinator,
70        store: StoreClient,
71        db: Db,
72        chain_state: Arc<SharedChainState>,
73        actor_context: AccountActorContext,
74        mempool_events: MempoolEventStream,
75        actor_request_rx: mpsc::Receiver<ActorRequest>,
76    ) -> Self {
77        Self {
78            config,
79            coordinator,
80            store,
81            db,
82            chain_state,
83            actor_context,
84            mempool_events,
85            actor_request_rx,
86        }
87    }
88
89    /// Runs the network transaction builder event loop until a fatal error occurs.
90    ///
91    /// If a `TcpListener` is provided, a gRPC server is also spawned to expose the
92    /// `GetNoteError` endpoint.
93    ///
94    /// This method:
95    /// 1. Optionally starts a gRPC server for note error queries
96    /// 2. Spawns a background task to load existing network accounts from the store
97    /// 3. Runs the main event loop, processing mempool events and managing actors
98    ///
99    /// # Errors
100    ///
101    /// Returns an error if:
102    /// - The mempool event stream ends unexpectedly
103    /// - An actor encounters a fatal error
104    /// - The account loader task fails
105    /// - The gRPC server fails
106    pub async fn run(self, listener: Option<TcpListener>) -> anyhow::Result<()> {
107        let mut join_set = JoinSet::new();
108
109        // Start the gRPC server if a listener is provided.
110        if let Some(listener) = listener {
111            let server = NtxBuilderRpcServer::new(self.db.clone());
112            join_set.spawn(async move {
113                server.serve(listener).await.context("ntx-builder gRPC server failed")
114            });
115        }
116
117        join_set.spawn(self.run_event_loop());
118
119        // Wait for either the event loop or the gRPC server to complete.
120        // Any completion is treated as fatal.
121        if let Some(result) = join_set.join_next().await {
122            result.context("ntx-builder task panicked")??;
123        }
124
125        Ok(())
126    }
127
128    /// Runs the main event loop.
129    async fn run_event_loop(mut self) -> anyhow::Result<()> {
130        // Spawn a background task to load network accounts from the store.
131        // Accounts are sent through a channel and processed in the main event loop.
132        let (account_tx, mut account_rx) =
133            mpsc::channel::<NetworkAccountId>(self.config.account_channel_capacity);
134        let account_loader_store = self.store.clone();
135        let mut account_loader_handle = tokio::spawn(async move {
136            account_loader_store
137                .stream_network_account_ids(account_tx)
138                .await
139                .context("failed to load network accounts from store")
140        });
141
142        // Main event loop.
143        loop {
144            tokio::select! {
145                // Handle actor result. If a timed-out actor needs respawning, do so.
146                result = self.coordinator.next() => {
147                    if let Some(account_id) = result? {
148                        self.coordinator
149                            .spawn_actor(AccountOrigin::store(account_id), &self.actor_context);
150                    }
151                },
152                // Handle mempool events.
153                event = self.mempool_events.next() => {
154                    let event = event
155                        .context("mempool event stream ended")?
156                        .context("mempool event stream failed")?;
157
158                    self.handle_mempool_event(event).await?;
159                },
160                // Handle account batches loaded from the store.
161                // Once all accounts are loaded, the channel closes and this branch
162                // becomes inactive (recv returns None and we stop matching).
163                Some(account_id) = account_rx.recv() => {
164                    self.handle_loaded_account(account_id).await?;
165                },
166                // Handle requests from actors.
167                Some(request) = self.actor_request_rx.recv() => {
168                    self.handle_actor_request(request).await?;
169                },
170                // Handle account loader task completion/failure.
171                // If the task fails, we abort since the builder would be in a degraded state
172                // where existing notes against network accounts won't be processed.
173                result = &mut account_loader_handle => {
174                    result
175                        .context("account loader task panicked")
176                        .flatten()?;
177
178                    tracing::info!("account loading from store completed");
179                    account_loader_handle = tokio::spawn(std::future::pending());
180                },
181            }
182        }
183    }
184
185    /// Handles account IDs loaded from the store by syncing state to DB and spawning actors.
186    #[tracing::instrument(name = "ntx.builder.handle_loaded_account", skip(self, account_id))]
187    async fn handle_loaded_account(
188        &mut self,
189        account_id: NetworkAccountId,
190    ) -> Result<(), anyhow::Error> {
191        // Fetch account from store and write to DB.
192        let account = self
193            .store
194            .get_network_account(account_id)
195            .await
196            .context("failed to load account from store")?
197            .context("account should exist in store")?;
198
199        let block_num = self.chain_state.chain_tip_block_number();
200        let notes = self
201            .store
202            .get_unconsumed_network_notes(account_id, block_num.as_u32())
203            .await
204            .context("failed to load notes from store")?;
205
206        // Write account and notes to DB.
207        self.db
208            .sync_account_from_store(account_id, account.clone(), notes.clone())
209            .await
210            .context("failed to sync account to DB")?;
211
212        self.coordinator
213            .spawn_actor(AccountOrigin::store(account_id), &self.actor_context);
214        Ok(())
215    }
216
217    /// Handles mempool events by writing to DB first, then notifying actors.
218    #[tracing::instrument(name = "ntx.builder.handle_mempool_event", skip(self, event))]
219    async fn handle_mempool_event(&mut self, event: MempoolEvent) -> Result<(), anyhow::Error> {
220        match &event {
221            MempoolEvent::TransactionAdded { account_delta, .. } => {
222                // Write event effects to DB first.
223                self.coordinator
224                    .write_event(&event)
225                    .await
226                    .context("failed to write TransactionAdded to DB")?;
227
228                // Handle account deltas in case an account is being created.
229                if let Some(AccountUpdateDetails::Delta(delta)) = account_delta {
230                    // Handle account deltas for network accounts only.
231                    if let Some(network_account) = AccountOrigin::transaction(delta) {
232                        // Spawn new actors if a transaction creates a new network account.
233                        let is_creating_account = delta.is_full_state();
234                        if is_creating_account {
235                            self.coordinator.spawn_actor(network_account, &self.actor_context);
236                        }
237                    }
238                }
239                let inactive_targets = self.coordinator.send_targeted(&event);
240                for account_id in inactive_targets {
241                    self.coordinator
242                        .spawn_actor(AccountOrigin::store(account_id), &self.actor_context);
243                }
244                Ok(())
245            },
246            // Update chain state and notify affected actors.
247            MempoolEvent::BlockCommitted { header, .. } => {
248                // Write event effects to DB first.
249                let result = self
250                    .coordinator
251                    .write_event(&event)
252                    .await
253                    .context("failed to write BlockCommitted to DB")?;
254
255                self.chain_state
256                    .update_chain_tip(header.as_ref().clone(), self.config.max_block_count);
257                self.coordinator.notify_accounts(&result.accounts_to_notify);
258                Ok(())
259            },
260            // Notify affected actors (reverted account actors will self-cancel when they
261            // detect their account has been removed from the DB).
262            MempoolEvent::TransactionsReverted(_) => {
263                // Write event effects to DB first.
264                let result = self
265                    .coordinator
266                    .write_event(&event)
267                    .await
268                    .context("failed to write TransactionsReverted to DB")?;
269
270                self.coordinator.notify_accounts(&result.accounts_to_notify);
271                Ok(())
272            },
273        }
274    }
275
276    /// Processes a request from an account actor.
277    async fn handle_actor_request(&mut self, request: ActorRequest) -> Result<(), anyhow::Error> {
278        match request {
279            ActorRequest::NotesFailed { failed_notes, block_num, ack_tx } => {
280                self.db
281                    .notes_failed(failed_notes, block_num)
282                    .await
283                    .context("failed to mark notes as failed")?;
284                let _ = ack_tx.send(());
285            },
286            ActorRequest::CacheNoteScript { script_root, script } => {
287                self.db
288                    .insert_note_script(script_root, &script)
289                    .await
290                    .context("failed to cache note script")?;
291            },
292        }
293        Ok(())
294    }
295}
miden_node_ntx_builder/builder.rs

miden_node_ntx_builder/
builder.rs
