Skip to content

feat(rpc): forward tx and batches submissions in full node mode #2139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mirko-von-Leipzig wants to merge 2 commits into
base: mirko/yank-mempool-subscription
Choose a base branch
from
Open

feat(rpc): forward tx and batches submissions in full node mode #2139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
96aef0b
Forward submissions in full node mode.
Mirko-von-Leipzig May 26, 2026
02a38b9
Wire it up
Mirko-von-Leipzig May 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions bin/node/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ clap = { features = ["env", "string"], workspace = true }
fs-err = { workspace = true }
humantime = { workspace = true }
miden-node-block-producer = { workspace = true }
miden-node-proto = { workspace = true }
miden-node-rpc = { workspace = true }
miden-node-store = { workspace = true }
miden-node-utils = { workspace = true }
Expand Down
44 changes: 41 additions & 3 deletions bin/node/src/commands/modes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use miden_node_proto::clients::{Builder, NtxBuilderClient, RpcClient, ValidatorClient};
use url::Url;

use super::block_producer::BlockProducerOptions;
Expand Down Expand Up @@ -27,11 +28,13 @@ impl SequencerCommand {
pub fn handle(self) -> anyhow::Result<()> {
let runtime = self.runtime.runtime_config(&self.store);
self.block_producer.validate()?;
let validator = self.external_services.validator_client();
let ntx_builder = self.external_services.ntx_builder_client();
let _ = (
runtime.rpc_listen,
runtime.data_directory,
self.external_services.validator_url,
self.external_services.ntx_builder_url,
validator,
ntx_builder,
self.block_producer.block_prover.url,
runtime.database_options,
runtime.internal_grpc_options,
Expand All @@ -58,6 +61,28 @@ pub struct SequencerExternalServiceOptions {
pub ntx_builder_url: Url,
}

impl SequencerExternalServiceOptions {
fn validator_client(&self) -> ValidatorClient {
Builder::new(self.validator_url.clone())
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<ValidatorClient>()
}

fn ntx_builder_client(&self) -> NtxBuilderClient {
Builder::new(self.ntx_builder_url.clone())
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<NtxBuilderClient>()
}
}

#[derive(clap::Args, Clone, Debug)]
pub struct FullNodeCommand {
#[command(flatten)]
Expand All @@ -73,14 +98,15 @@ pub struct FullNodeCommand {
impl FullNodeCommand {
pub fn handle(self) -> anyhow::Result<()> {
let runtime = self.runtime.runtime_config(&self.store);
let source_rpc = self.sync.source_rpc_client();
let _ = (
runtime.rpc_listen,
runtime.data_directory,
runtime.database_options,
runtime.internal_grpc_options,
runtime.external_grpc_options,
runtime.storage_options,
self.sync.block_source_url,
source_rpc,
);

anyhow::bail!(
Expand All @@ -89,3 +115,15 @@ impl FullNodeCommand {
)
}
}

impl SyncOptions {
fn source_rpc_client(&self) -> RpcClient {
Builder::new(self.block_source_url.clone())
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<RpcClient>()
}
}
2 changes: 1 addition & 1 deletion crates/rpc/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ mod server;
#[cfg(test)]
mod tests;

pub use server::Rpc;
pub use server::{Rpc, RpcMode};

// CONSTANTS
// =================================================================================================
Expand Down
133 changes: 42 additions & 91 deletions crates/rpc/src/server/api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,7 @@ use std::sync::LazyLock;
use std::time::Duration;

use anyhow::Context;
use miden_node_proto::clients::{
BlockProducerClient,
Builder,
NtxBuilderClient,
StoreRpcClient,
ValidatorClient,
};
use miden_node_proto::clients::{NtxBuilderClient, StoreRpcClient};
use miden_node_proto::decode::{read_account_id, read_account_ids, read_block_range};
use miden_node_proto::domain::account::{AccountRequest, SlotData};
use miden_node_proto::errors::ConversionError;
Expand Down Expand Up @@ -41,91 +35,32 @@ use miden_protocol::{MIN_PROOF_SECURITY_LEVEL, Word};
use miden_tx::TransactionVerifier;
use miden_tx_batch_prover::LocalBatchProver;
use tonic::{IntoRequest, Request, Response, Status};
use tracing::{Span, debug, info, info_span};
use url::Url;
use tracing::{Span, debug, info_span};

use crate::COMPONENT;
use crate::server::RpcMode;

// RPC SERVICE
// ================================================================================================

pub struct RpcService {
store: StoreRpcClient,
block_producer: Option<BlockProducerClient>,
validator: ValidatorClient,
mode: RpcMode,
ntx_builder: Option<NtxBuilderClient>,
genesis_commitment: Option<Word>,
block_commitment_cache: LruCache<BlockNumber, Word>,
}

impl RpcService {
pub(super) fn new(
store_url: Url,
block_producer_url: Option<Url>,
validator_url: Url,
ntx_builder_url: Option<Url>,
store: StoreRpcClient,
mode: RpcMode,
ntx_builder: Option<NtxBuilderClient>,
commitment_cache_capacity: NonZeroUsize,
) -> Self {
let store = {
info!(target: COMPONENT, store_endpoint = %store_url, "Initializing store client");
Builder::new(store_url)
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<StoreRpcClient>()
};

let block_producer = block_producer_url.map(|block_producer_url| {
info!(
target: COMPONENT,
block_producer_endpoint = %block_producer_url,
"Initializing block producer client",
);
Builder::new(block_producer_url)
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<BlockProducerClient>()
});

let validator = {
info!(
target: COMPONENT,
validator_endpoint = %validator_url,
"Initializing validator client",
);
Builder::new(validator_url)
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<ValidatorClient>()
};

let ntx_builder = ntx_builder_url.map(|ntx_builder_url| {
info!(
target: COMPONENT,
ntx_builder_endpoint = %ntx_builder_url,
"Initializing ntx-builder client",
);
Builder::new(ntx_builder_url)
.without_tls()
.without_timeout()
.without_metadata_version()
.without_metadata_genesis()
.with_otel_context_injection()
.connect_lazy::<NtxBuilderClient>()
});

Self {
store,
block_producer,
validator,
mode,
ntx_builder,
genesis_commitment: None,
block_commitment_cache: LruCache::new(commitment_cache_capacity),
Expand All @@ -134,8 +69,8 @@ impl RpcService {

/// Sets the genesis commitment, returning an error if it is already set.
///
/// Required since `RpcService::new()` sets up the `store` which is used to fetch the
/// `genesis_commitment`.
/// Required since the store client is used to fetch the `genesis_commitment` after
/// `RpcService` construction.
pub fn set_genesis_commitment(&mut self, commitment: Word) -> anyhow::Result<()> {
if self.genesis_commitment.is_some() {
return Err(anyhow::anyhow!("genesis commitment already set"));
Expand Down Expand Up @@ -469,12 +404,6 @@ impl api_server::Api for RpcService {
) -> Result<Response<proto::blockchain::BlockNumber>, Status> {
debug!(target: COMPONENT, request = ?request.get_ref());

let Some(block_producer) = &self.block_producer else {
return Err(Status::unavailable(
"Transaction submission not available in read-only mode",
));
};

let request = request.into_inner();

let tx = ProvenTransaction::read_from_bytes(&request.transaction).map_err(|err| {
Expand Down Expand Up @@ -535,10 +464,20 @@ impl api_server::Api for RpcService {
))
})?;

// In full node mode we forward the request to the source.
let (block_producer, validator) = match &self.mode {
RpcMode::Sequencer { block_producer, validator } => {
(block_producer.as_ref(), validator.as_ref())
},
RpcMode::FullNode { source_rpc } => {
return source_rpc.as_ref().clone().submit_proven_tx(request).await;
},
};

// Transaction inputs must be provided in order to allow for transaction re-execution via
// the Validator.
if request.transaction_inputs.is_some() {
self.validator.clone().submit_proven_transaction(request.clone()).await?;
validator.clone().submit_proven_transaction(request.clone()).await?;
} else {
return Err(Status::invalid_argument("Transaction inputs must be provided"));
}
Expand All @@ -552,10 +491,6 @@ impl api_server::Api for RpcService {
&self,
request: tonic::Request<proto::transaction::TransactionBatch>,
) -> Result<tonic::Response<proto::blockchain::BlockNumber>, Status> {
let Some(block_producer) = &self.block_producer else {
return Err(Status::unavailable("Batch submission not available in read-only mode"));
};

let request = request.into_inner();

let proven_batch = ProvenBatch::read_from_bytes(&request.batch_proof).map_err(|err| {
Expand Down Expand Up @@ -623,6 +558,16 @@ impl api_server::Api for RpcService {
return Err(Status::invalid_argument("batch proof did not match proposed batch"));
}

// In full node mode we forward the request to the source.
let (block_producer, validator) = match &self.mode {
RpcMode::Sequencer { block_producer, validator } => {
(block_producer.as_ref(), validator.as_ref())
},
RpcMode::FullNode { source_rpc } => {
return source_rpc.as_ref().clone().submit_proven_tx_batch(request).await;
},
};

Comment on lines +561 to +570
Copy link
Copy Markdown
Collaborator

@SantiagoPittella SantiagoPittella May 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess it is ok, but in this method we are going to validate the proof twice

Copy link
Copy Markdown
Collaborator Author

@Mirko-von-Leipzig Mirko-von-Leipzig May 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Verification is (relatively) cheap.

In the future we can optimise this by having an internal trusted service which allows skipping these.

// Submit each transaction to the validator.
//
// SAFETY: We checked earlier that the two iterators are the same length.
Expand All @@ -631,7 +576,7 @@ impl api_server::Api for RpcService {
transaction: tx.to_bytes(),
transaction_inputs: inputs.clone().into(),
};
self.validator.clone().submit_proven_transaction(request).await?;
validator.clone().submit_proven_transaction(request).await?;
}

block_producer.clone().submit_proven_tx_batch(request).await
Expand Down Expand Up @@ -670,15 +615,21 @@ impl api_server::Api for RpcService {

let store_status =
self.store.clone().status(Request::new(())).await.map(Response::into_inner).ok();
let block_producer_status = if let Some(block_producer) = &self.block_producer {
block_producer
let block_producer_status = match &self.mode {
RpcMode::Sequencer { block_producer, .. } => block_producer
.as_ref()
.clone()
.status(Request::new(()))
.await
.map(Response::into_inner)
.ok(),
RpcMode::FullNode { source_rpc } => source_rpc
.as_ref()
.clone()
.status(Request::new(()))
.await
.ok()
} else {
None
.and_then(|response| response.into_inner().block_producer),
};

Ok(Response::new(proto::rpc::RpcStatus {
Expand Down
Loading
Loading