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)).
 
 ### Changes
 

bin/node/src/commands/bundled.rs

@@ -275,6 +275,7 @@ impl BundledCommand {
                     NetworkTransactionBuilder::new(
                         store_ntx_builder_url,
                         block_producer_url,
+                        rpc_url,
                         ntx_builder.tx_prover_url,
                         ntx_builder.ticker_interval,
                         checkpoint,

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

@@ -46,7 +46,7 @@ use tracing::{Instrument, instrument};
 
 use crate::COMPONENT;
 use crate::actor::account_state::TransactionCandidate;
-use crate::block_producer::BlockProducerClient;
+use crate::rpc::RpcClient;
 use crate::store::StoreClient;
 
 #[derive(Debug, thiserror::Error)]
@@ -75,7 +75,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 submitting transactions to the network.
+    rpc_client: RpcClient,
 
     /// The prover to delegate proofs to.
     ///
@@ -93,17 +94,12 @@ pub struct NtxContext {
 impl NtxContext {
     /// Creates a new [`NtxContext`] instance.
     pub fn new(
-        block_producer: BlockProducerClient,
+        rpc_client: RpcClient,
         prover: Option<RemoteTransactionProver>,
         store: StoreClient,
         script_cache: LruCache<Word, NoteScript>,
     ) -> Self {
-        Self {
-            block_producer,
-            prover,
-            store,
-            script_cache,
-        }
+        Self { rpc_client, prover, store, script_cache }
     }
 
     /// Executes a transaction end-to-end: filtering, executing, proving, and submitted to the block
@@ -147,7 +143,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,
@@ -157,13 +153,15 @@ impl NtxContext {
                 );
 
                 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?;
+                let executed_tx = Box::pin(self.execute(&data_store, successful_notes)).await?;
+                let tx_inputs: TransactionInputs = executed_tx.into();
+                let proven_tx = Box::pin(self.prove(tx_inputs.clone())).await?;
+                let tx_id = proven_tx.id();
+                self.submit(proven_tx, tx_inputs).await?;
+                Ok((tx_id, failed_notes))
+            })
             .in_current_span()
             .await
             .inspect_err(|err| tracing::Span::current().set_error(err))
@@ -256,11 +254,11 @@ impl NtxContext {
         .map_err(NtxError::Proving)
     }
 
-    /// Submits the transaction to the block producer.
+    /// Submits the transaction to the RPC server with transaction inputs.
     #[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)
+    async fn submit(&self, tx: ProvenTransaction, tx_inputs: TransactionInputs) -> NtxResult<()> {
+        self.rpc_client
+            .submit_proven_transaction(tx, tx_inputs)
             .await
             .map_err(NtxError::Submission)
     }

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

@@ -22,8 +22,8 @@ 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::rpc::RpcClient;
 use crate::store::StoreClient;
 
 // ACTOR SHUTDOWN REASON
@@ -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 RPC gRPC server.
+    pub rpc_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,
+    rpc_client: RpcClient,
     prover: Option<RemoteTransactionProver>,
     chain_state: Arc<RwLock<ChainState>>,
     script_cache: LruCache<Word, NoteScript>,
@@ -168,15 +168,15 @@ impl AccountActor {
         event_rx: mpsc::Receiver<Arc<MempoolEvent>>,
         cancel_token: CancellationToken,
     ) -> Self {
-        let block_producer = BlockProducerClient::new(actor_context.block_producer_url.clone());
+        let rpc_client = RpcClient::new(actor_context.rpc_url.clone());
         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,
+            rpc_client,
             prover,
             chain_state: actor_context.chain_state.clone(),
             script_cache: actor_context.script_cache.clone(),
@@ -275,7 +275,7 @@ impl AccountActor {
 
         // Execute the selected transaction.
         let context = execute::NtxContext::new(
-            self.block_producer.clone(),
+            self.rpc_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_objects::block::BlockNumber;
-use miden_objects::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 RPC gRPC server.
+    rpc_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,
+        rpc_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,
+            rpc_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(),
+            rpc_url: self.rpc_url.clone(),
             tx_prover_url: self.tx_prover_url.clone(),
             chain_state: chain_state.clone(),
             store: store.clone(),

crates/ntx-builder/src/lib.rs

@@ -4,6 +4,7 @@ mod actor;
 mod block_producer;
 mod builder;
 mod coordinator;
+mod rpc;
 mod store;
 
 pub use builder::NetworkTransactionBuilder;

crates/ntx-builder/src/rpc.rs

@@ -0,0 +1,52 @@
+use miden_node_proto::clients::{Builder, RpcClient as InnerRpcClient};
+use miden_node_proto::generated::{self as proto};
+use miden_objects::transaction::{ProvenTransaction, TransactionInputs};
+use miden_tx::utils::Serializable;
+use tonic::Status;
+use tracing::{info, instrument};
+use url::Url;
+
+use crate::COMPONENT;
+
+// RPC CLIENT
+// ================================================================================================
+
+/// Interface to the RPC server's gRPC API.
+#[derive(Clone, Debug)]
+pub struct RpcClient {
+    client: InnerRpcClient,
+}
+
+impl RpcClient {
+    /// Creates a new RPC client with a lazy connection.
+    pub fn new(rpc_url: Url) -> Self {
+        info!(target: COMPONENT, rpc_endpoint = %rpc_url, "Initializing RPC client with lazy connection");
+
+        let rpc = Builder::new(rpc_url)
+            .without_tls()
+            .without_timeout()
+            .without_metadata_version()
+            .without_metadata_genesis()
+            .with_otel_context_injection()
+            .connect_lazy::<InnerRpcClient>();
+
+        Self { client: rpc }
+    }
+
+    /// Submits a proven transaction with transaction inputs to the RPC server.
+    #[instrument(target = COMPONENT, name = "ntx.rpc.client.submit_proven_transaction", skip_all, err)]
+    pub async fn submit_proven_transaction(
+        &self,
+        proven_tx: ProvenTransaction,
+        tx_inputs: TransactionInputs,
+    ) -> Result<(), Status> {
+        let request = proto::transaction::ProvenTransaction {
+            transaction: proven_tx.to_bytes(),
+            transaction_inputs: Some(tx_inputs.to_bytes()),
+        };
+
+        self.client.clone().submit_proven_transaction(request).await?;
+
+        Ok(())
+    }
+}