feat: Integrate ntx-builder with validator

CHANGELOG.md

@@ -12,6 +12,7 @@
 - Added chain tip to the block producer status ([#1419](https://github.com/0xMiden/miden-node/pull/1419)).
 - The mempool's transaction capacity is now configurable ([#1433](https://github.com/0xMiden/miden-node/pull/1433)).
 - Renamed card's names in the `miden-network-monitor` binary ([#1441](https://github.com/0xMiden/miden-node/pull/1441)).
+- Integrated NTX Builder with validator via `SubmitProvenTransaction` RPC ([#1453](https://github.com/0xMiden/miden-node/pull/1453)).
 - Added pagination to `GetNetworkAccountIds` endpoint ([#1452](https://github.com/0xMiden/miden-node/pull/1452)).
 - Improved tracing in `miden-network-monitor` binary ([#1366](https://github.com/0xMiden/miden-node/pull/1366)).
 - Integrated RPC stack with Validator component for transaction validation ([#1457](https://github.com/0xMiden/miden-node/pull/1457)).

bin/node/src/commands/bundled.rs

@@ -311,6 +311,8 @@ impl BundledCommand {
             .context("Failed to parse URL")?;
 
         if should_start_ntx_builder {
+            let validator_url = Url::parse(&format!("http://{validator_address}"))
+                .context("Failed to parse URL")?;
             let id = join_set
                 .spawn(async move {
                     let block_producer_url =
@@ -319,6 +321,7 @@ impl BundledCommand {
                     NetworkTransactionBuilder::new(
                         store_ntx_builder_url,
                         block_producer_url,
+                        validator_url,
                         ntx_builder.tx_prover_url,
                         ntx_builder.ticker_interval,
                         checkpoint,

crates/ntx-builder/src/actor/execute.rs

@@ -1,5 +1,7 @@
 use std::collections::BTreeSet;
 
+use miden_node_proto::clients::ValidatorClient;
+use miden_node_proto::generated::{self as proto};
 use miden_node_utils::lru_cache::LruCache;
 use miden_node_utils::tracing::OpenTelemetrySpanExt;
 use miden_protocol::account::{
@@ -27,6 +29,7 @@ use miden_protocol::vm::FutureMaybeSend;
 use miden_protocol::{TransactionInputError, Word};
 use miden_remote_prover_client::remote_prover::tx_prover::RemoteTransactionProver;
 use miden_tx::auth::UnreachableAuth;
+use miden_tx::utils::Serializable;
 use miden_tx::{
     DataStore,
     DataStoreError,
@@ -46,7 +49,6 @@ use tracing::{Instrument, instrument};
 
 use crate::COMPONENT;
 use crate::actor::account_state::TransactionCandidate;
-use crate::block_producer::BlockProducerClient;
 use crate::store::StoreClient;
 
 #[derive(Debug, thiserror::Error)]
@@ -75,7 +77,8 @@ type NtxResult<T> = Result<T, NtxError>;
 /// Provides the context for execution [network transaction candidates](TransactionCandidate).
 #[derive(Clone)]
 pub struct NtxContext {
-    block_producer: BlockProducerClient,
+    /// Client for validating transactions via the Validator.
+    validator_client: ValidatorClient,
 
     /// The prover to delegate proofs to.
     ///
@@ -93,13 +96,13 @@ pub struct NtxContext {
 impl NtxContext {
     /// Creates a new [`NtxContext`] instance.
     pub fn new(
-        block_producer: BlockProducerClient,
+        validator_client: ValidatorClient,
         prover: Option<RemoteTransactionProver>,
         store: StoreClient,
         script_cache: LruCache<Word, NoteScript>,
     ) -> Self {
         Self {
-            block_producer,
+            validator_client,
             prover,
             store,
             script_cache,
@@ -147,7 +150,7 @@ impl NtxContext {
             .set_attribute("reference_block.number", chain_tip_header.block_num());
 
         async move {
-            async move {
+            Box::pin(async move {
                 let data_store = NtxDataStore::new(
                     account,
                     chain_tip_header,
@@ -156,14 +159,24 @@ impl NtxContext {
                     self.script_cache.clone(),
                 );
 
+                // Filter notes.
                 let notes = notes.into_iter().map(Note::from).collect::<Vec<_>>();
-                let (successful, failed) = self.filter_notes(&data_store, notes).await?;
-                let executed = Box::pin(self.execute(&data_store, successful)).await?;
-                let proven = Box::pin(self.prove(executed.into())).await?;
-                let tx_id = proven.id();
-                self.submit(proven).await?;
-                Ok((tx_id, failed))
-            }
+                let (successful_notes, failed_notes) =
+                    self.filter_notes(&data_store, notes).await?;
+
+                // Execute transaction.
+                let executed_tx = Box::pin(self.execute(&data_store, successful_notes)).await?;
+
+                // Prove transaction.
+                let tx_inputs: TransactionInputs = executed_tx.into();
+                let proven_tx = Box::pin(self.prove(tx_inputs.clone())).await?;
+                let tx_id = proven_tx.id();
+
+                // Validate proven transaction.
+                self.validate(proven_tx, tx_inputs).await?;
+
+                Ok((tx_id, failed_notes))
+            })
             .in_current_span()
             .await
             .inspect_err(|err| tracing::Span::current().set_error(err))
@@ -256,13 +269,19 @@ impl NtxContext {
         .map_err(NtxError::Proving)
     }
 
-    /// Submits the transaction to the block producer.
-    #[instrument(target = COMPONENT, name = "ntx.execute_transaction.submit", skip_all, err)]
-    async fn submit(&self, tx: ProvenTransaction) -> NtxResult<()> {
-        self.block_producer
-            .submit_proven_transaction(tx)
+    /// Validates the transaction against the Validator.
+    #[instrument(target = COMPONENT, name = "ntx.execute_transaction.validate", skip_all, err)]
+    async fn validate(&self, tx: ProvenTransaction, tx_inputs: TransactionInputs) -> NtxResult<()> {
+        let request = proto::transaction::ProvenTransaction {
+            transaction: tx.to_bytes(),
+            transaction_inputs: Some(tx_inputs.to_bytes()),
+        };
+        self.validator_client
+            .clone()
+            .submit_proven_transaction(request)
             .await
-            .map_err(NtxError::Submission)
+            .map_err(NtxError::Submission)?;
+        Ok(())
     }
 }
 

crates/ntx-builder/src/actor/mod.rs

@@ -8,6 +8,7 @@ use std::sync::Arc;
 use account_state::{NetworkAccountState, TransactionCandidate};
 use execute::NtxError;
 use futures::FutureExt;
+use miden_node_proto::clients::{Builder, ValidatorClient};
 use miden_node_proto::domain::account::NetworkAccountPrefix;
 use miden_node_proto::domain::mempool::MempoolEvent;
 use miden_node_utils::ErrorReport;
@@ -22,7 +23,6 @@ use tokio::sync::{AcquireError, RwLock, Semaphore, mpsc};
 use tokio_util::sync::CancellationToken;
 use url::Url;
 
-use crate::block_producer::BlockProducerClient;
 use crate::builder::ChainState;
 use crate::store::StoreClient;
 
@@ -52,8 +52,8 @@ pub enum ActorShutdownReason {
 pub struct AccountActorContext {
     /// Client for interacting with the store in order to load account state.
     pub store: StoreClient,
-    /// Address of the block producer gRPC server.
-    pub block_producer_url: Url,
+    /// Address of the Validator server.
+    pub validator_url: Url,
     /// Address of the remote prover. If `None`, transactions will be proven locally, which is
     // undesirable due to the performance impact.
     pub tx_prover_url: Option<Url>,
@@ -153,7 +153,7 @@ pub struct AccountActor {
     mode: ActorMode,
     event_rx: mpsc::Receiver<Arc<MempoolEvent>>,
     cancel_token: CancellationToken,
-    block_producer: BlockProducerClient,
+    validator_client: ValidatorClient,
     prover: Option<RemoteTransactionProver>,
     chain_state: Arc<RwLock<ChainState>>,
     script_cache: LruCache<Word, NoteScript>,
@@ -168,15 +168,21 @@ impl AccountActor {
         event_rx: mpsc::Receiver<Arc<MempoolEvent>>,
         cancel_token: CancellationToken,
     ) -> Self {
-        let block_producer = BlockProducerClient::new(actor_context.block_producer_url.clone());
+        let validator_client = Builder::new(actor_context.validator_url.clone())
+            .without_tls()
+            .without_timeout()
+            .without_metadata_version()
+            .without_metadata_genesis()
+            .with_otel_context_injection()
+            .connect_lazy::<ValidatorClient>();
         let prover = actor_context.tx_prover_url.clone().map(RemoteTransactionProver::new);
         Self {
             origin,
             store: actor_context.store.clone(),
             mode: ActorMode::NoViableNotes,
             event_rx,
             cancel_token,
-            block_producer,
+            validator_client,
             prover,
             chain_state: actor_context.chain_state.clone(),
             script_cache: actor_context.script_cache.clone(),
@@ -275,7 +281,7 @@ impl AccountActor {
 
         // Execute the selected transaction.
         let context = execute::NtxContext::new(
-            self.block_producer.clone(),
+            self.validator_client.clone(),
             self.prover.clone(),
             self.store.clone(),
             self.script_cache.clone(),

crates/ntx-builder/src/block_producer.rs

@@ -6,8 +6,6 @@ use miden_node_proto::domain::mempool::MempoolEvent;
 use miden_node_proto::generated::{self as proto};
 use miden_node_utils::FlattenResult;
 use miden_protocol::block::BlockNumber;
-use miden_protocol::transaction::ProvenTransaction;
-use miden_tx::utils::Serializable;
 use tokio_stream::StreamExt;
 use tonic::Status;
 use tracing::{info, instrument};
@@ -41,20 +39,6 @@ impl BlockProducerClient {
 
         Self { client: block_producer }
     }
-    #[instrument(target = COMPONENT, name = "ntx.block_producer.client.submit_proven_transaction", skip_all, err)]
-    pub async fn submit_proven_transaction(
-        &self,
-        proven_tx: ProvenTransaction,
-    ) -> Result<(), Status> {
-        let request = proto::transaction::ProvenTransaction {
-            transaction: proven_tx.to_bytes(),
-            transaction_inputs: None,
-        };
-
-        self.client.clone().submit_proven_transaction(request).await?;
-
-        Ok(())
-    }
 
     #[instrument(target = COMPONENT, name = "ntx.block_producer.client.subscribe_to_mempool", skip_all, err)]
     pub async fn subscribe_to_mempool_with_retry(

crates/ntx-builder/src/builder.rs

@@ -73,6 +73,8 @@ pub struct NetworkTransactionBuilder {
     store_url: Url,
     /// Address of the block producer gRPC server.
     block_producer_url: Url,
+    /// Address of the Validator server.
+    validator_url: Url,
     /// 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>,
@@ -101,6 +103,7 @@ impl NetworkTransactionBuilder {
     pub fn new(
         store_url: Url,
         block_producer_url: Url,
+        validator_url: Url,
         tx_prover_url: Option<Url>,
         ticker_interval: Duration,
         bp_checkpoint: Arc<Barrier>,
@@ -110,6 +113,7 @@ impl NetworkTransactionBuilder {
         Self {
             store_url,
             block_producer_url,
+            validator_url,
             tx_prover_url,
             ticker_interval,
             bp_checkpoint,
@@ -145,7 +149,7 @@ impl NetworkTransactionBuilder {
         let chain_state = Arc::new(RwLock::new(ChainState::new(chain_tip_header, chain_mmr)));
 
         let actor_context = AccountActorContext {
-            block_producer_url: self.block_producer_url.clone(),
+            validator_url: self.validator_url.clone(),
             tx_prover_url: self.tx_prover_url.clone(),
             chain_state: chain_state.clone(),
             store: store.clone(),