miden_node_proto/clients/
mod.rs

1//! gRPC client builder utilities for Miden node.
2//!
3//! This module provides a unified type-safe [`Builder`] for creating various gRPC clients with
4//! explicit configuration decisions for TLS, timeout, and metadata.
5//!
6//! # Examples
7//!
8//! ```rust
9//! # use miden_node_proto::clients::{Builder, WantsTls, RpcClient};
10//! # use url::Url;
11//!
12//! # async fn example() -> anyhow::Result<()> {
13//! // Create an RPC client with OTEL and TLS
14//! let url = Url::parse("https://example.com:8080")?;
15//! let client: RpcClient = Builder::new(url)
16//!     .with_tls()?                   // or `.without_tls()`
17//!     .without_timeout()             // or `.with_timeout(Duration::from_secs(10))`
18//!     .without_metadata_version()    // or `.with_metadata_version("1.0".into())`
19//!     .without_metadata_genesis()    // or `.with_metadata_genesis(genesis)`
20//!     .without_auth_header()         // or `.with_auth_header_value(AsciiMetadataValue::from_static("value"))`
21//!     .with_otel_context_injection() // or `.without_otel_context_injection()`
22//!     .connect::<RpcClient>()
23//!     .await?;
24//! # Ok(())
25//! # }
26//! ```
27
28use std::marker::PhantomData;
29use std::ops::{Deref, DerefMut};
30use std::str::FromStr;
31use std::time::Duration;
32
33use http::header::ACCEPT;
34use miden_node_tracing::grpc::OtelInterceptor;
35use miden_node_tracing::{debug, info, warn};
36use miden_protocol::Word;
37use miden_protocol::batch::ProposedBatch;
38use miden_protocol::utils::serde::Serializable;
39use tonic::metadata::AsciiMetadataValue;
40use tonic::service::interceptor::InterceptedService;
41use tonic::transport::{Channel, ClientTlsConfig, Endpoint, Error as TransportError};
42use tonic::{Request, Status};
43use url::Url;
44
45use crate::generated;
46
47#[derive(Clone)]
48pub struct Interceptor {
49    otel: Option<OtelInterceptor>,
50    accept: AsciiMetadataValue,
51    auth_header_value: Option<AsciiMetadataValue>,
52}
53
54impl Default for Interceptor {
55    fn default() -> Self {
56        Self {
57            otel: None,
58            accept: AsciiMetadataValue::from_static(Self::MEDIA_TYPE),
59            auth_header_value: None,
60        }
61    }
62}
63
64impl Interceptor {
65    const MEDIA_TYPE: &str = "application/vnd.miden";
66    const VERSION: &str = "version";
67    const GENESIS: &str = "genesis";
68    const NETWORK_TX_AUTH_HEADER_NAME: &str = "x-miden-network-tx-auth";
69
70    fn new(
71        enable_otel: bool,
72        version: Option<&str>,
73        genesis: Option<&str>,
74        auth_header: Option<AsciiMetadataValue>,
75    ) -> Self {
76        if let Some(version) = version
77            && !version.is_ascii()
78        {
79            panic!("version contains non-ascii values: {version}");
80        }
81
82        if let Some(genesis) = genesis
83            && !genesis.is_ascii()
84        {
85            panic!("genesis contains non-ascii values: {genesis}");
86        }
87
88        let accept = match (version, genesis) {
89            (None, None) => Self::MEDIA_TYPE.to_string(),
90            (None, Some(genesis)) => format!("{}; {}={genesis}", Self::MEDIA_TYPE, Self::GENESIS),
91            (Some(version), None) => format!("{}; {}={version}", Self::MEDIA_TYPE, Self::VERSION),
92            (Some(version), Some(genesis)) => format!(
93                "{}; {}={version}, {}={genesis}",
94                Self::MEDIA_TYPE,
95                Self::VERSION,
96                Self::GENESIS
97            ),
98        };
99        Self {
100            otel: enable_otel.then_some(OtelInterceptor),
101            // SAFETY: we checked that all values are ascii at the top of the function.
102            accept: AsciiMetadataValue::from_str(&accept).unwrap(),
103            auth_header_value: auth_header,
104        }
105    }
106}
107
108impl tonic::service::Interceptor for Interceptor {
109    fn call(&mut self, mut request: tonic::Request<()>) -> Result<Request<()>, Status> {
110        if let Some(mut otel) = self.otel {
111            request = otel.call(request)?;
112        }
113
114        if request.metadata().get(ACCEPT.as_str()).is_none() {
115            request.metadata_mut().insert(ACCEPT.as_str(), self.accept.clone());
116        }
117
118        if let Some(value) = &self.auth_header_value {
119            request.metadata_mut().insert(Self::NETWORK_TX_AUTH_HEADER_NAME, value.clone());
120        }
121
122        Ok(request)
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[test]
131    fn interceptor_preserves_existing_accept_metadata() {
132        let original_accept =
133            AsciiMetadataValue::from_static("application/vnd.miden; version=1.2; genesis=0x1234");
134        let mut request = Request::new(());
135        request.metadata_mut().insert(ACCEPT.as_str(), original_accept.clone());
136
137        let mut interceptor = Interceptor::new(false, Some("9.9"), Some("0xabcd"), None);
138        let request = tonic::service::Interceptor::call(&mut interceptor, request)
139            .expect("interceptor should succeed");
140
141        assert_eq!(request.metadata().get(ACCEPT.as_str()), Some(&original_accept));
142    }
143
144    #[test]
145    fn interceptor_inserts_accept_metadata_when_missing() {
146        let mut interceptor = Interceptor::new(false, Some("9.9"), Some("0xabcd"), None);
147
148        let request = tonic::service::Interceptor::call(&mut interceptor, Request::new(()))
149            .expect("interceptor should succeed");
150
151        assert_eq!(
152            request.metadata().get(ACCEPT.as_str()).and_then(|value| value.to_str().ok()),
153            Some("application/vnd.miden; version=9.9, genesis=0xabcd"),
154        );
155    }
156
157    #[tokio::test]
158    async fn connection_monitor_stops_when_cancelled() {
159        let builder = Builder::new(Url::parse("http://127.0.0.1:1").unwrap())
160            .without_tls()
161            .without_timeout()
162            .without_metadata_version()
163            .without_metadata_genesis()
164            .without_auth_header()
165            .without_otel_context_injection();
166        let shutdown = miden_node_utils::shutdown::CancellationToken::new();
167        shutdown.cancel();
168
169        tokio::time::timeout(
170            Duration::from_millis(100),
171            builder.monitor::<RpcClient>("test-dependency", shutdown),
172        )
173        .await
174        .expect("cancelled monitor should return promptly");
175    }
176}
177
178// TYPE ALIASES TO AID LEGIBILITY
179// ================================================================================================
180
181type InterceptedChannel = InterceptedService<Channel, Interceptor>;
182type GeneratedRpcClient = generated::rpc::api_client::ApiClient<InterceptedChannel>;
183type GeneratedProxyStatusClient =
184    generated::remote_prover::proxy_status_api_client::ProxyStatusApiClient<InterceptedChannel>;
185type GeneratedProverClient = generated::remote_prover::api_client::ApiClient<InterceptedChannel>;
186type GeneratedValidatorClient = generated::validator::api_client::ApiClient<InterceptedChannel>;
187type GeneratedNtxBuilderClient = generated::ntx_builder::api_client::ApiClient<InterceptedChannel>;
188type GeneratedSequencerClient = generated::sequencer::api_client::ApiClient<InterceptedChannel>;
189type GeneratedProvenTransaction = generated::transaction::ProvenTransaction;
190type SealedTransactionInputs = generated::transaction::SealedTransactionInputs;
191
192// gRPC CLIENTS
193// ================================================================================================
194
195#[derive(Debug, Clone)]
196pub struct RpcClient(GeneratedRpcClient);
197#[derive(Debug, Clone)]
198pub struct RemoteProverProxyStatusClient(GeneratedProxyStatusClient);
199#[derive(Debug, Clone)]
200pub struct RemoteProverClient(GeneratedProverClient);
201#[derive(Debug, Clone)]
202pub struct ValidatorClient(GeneratedValidatorClient);
203#[derive(Debug, Clone)]
204pub struct NtxBuilderClient(GeneratedNtxBuilderClient);
205#[derive(Debug, Clone)]
206pub struct SequencerClient(GeneratedSequencerClient);
207
208impl DerefMut for RpcClient {
209    fn deref_mut(&mut self) -> &mut Self::Target {
210        &mut self.0
211    }
212}
213
214impl Deref for RpcClient {
215    type Target = GeneratedRpcClient;
216
217    fn deref(&self) -> &Self::Target {
218        &self.0
219    }
220}
221
222impl DerefMut for RemoteProverProxyStatusClient {
223    fn deref_mut(&mut self) -> &mut Self::Target {
224        &mut self.0
225    }
226}
227
228impl Deref for RemoteProverProxyStatusClient {
229    type Target = GeneratedProxyStatusClient;
230
231    fn deref(&self) -> &Self::Target {
232        &self.0
233    }
234}
235
236impl DerefMut for RemoteProverClient {
237    fn deref_mut(&mut self) -> &mut Self::Target {
238        &mut self.0
239    }
240}
241
242impl Deref for RemoteProverClient {
243    type Target = GeneratedProverClient;
244
245    fn deref(&self) -> &Self::Target {
246        &self.0
247    }
248}
249
250impl DerefMut for ValidatorClient {
251    fn deref_mut(&mut self) -> &mut Self::Target {
252        &mut self.0
253    }
254}
255
256impl Deref for ValidatorClient {
257    type Target = GeneratedValidatorClient;
258
259    fn deref(&self) -> &Self::Target {
260        &self.0
261    }
262}
263
264impl DerefMut for NtxBuilderClient {
265    fn deref_mut(&mut self) -> &mut Self::Target {
266        &mut self.0
267    }
268}
269
270impl Deref for NtxBuilderClient {
271    type Target = GeneratedNtxBuilderClient;
272
273    fn deref(&self) -> &Self::Target {
274        &self.0
275    }
276}
277
278impl DerefMut for SequencerClient {
279    fn deref_mut(&mut self) -> &mut Self::Target {
280        &mut self.0
281    }
282}
283
284impl Deref for SequencerClient {
285    type Target = GeneratedSequencerClient;
286
287    fn deref(&self) -> &Self::Target {
288        &self.0
289    }
290}
291
292// GRPC CLIENT BUILDER TRAIT
293// ================================================================================================
294
295/// Trait for building gRPC clients from a common [`Builder`] configuration.
296pub trait GrpcClient {
297    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self;
298}
299
300impl GrpcClient for RpcClient {
301    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
302        Self(GeneratedRpcClient::new(InterceptedService::new(channel, interceptor)))
303    }
304}
305
306impl GrpcClient for RemoteProverProxyStatusClient {
307    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
308        Self(GeneratedProxyStatusClient::new(InterceptedService::new(channel, interceptor)))
309    }
310}
311
312impl GrpcClient for RemoteProverClient {
313    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
314        Self(GeneratedProverClient::new(InterceptedService::new(channel, interceptor)))
315    }
316}
317
318impl GrpcClient for ValidatorClient {
319    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
320        Self(GeneratedValidatorClient::new(InterceptedService::new(channel, interceptor)))
321    }
322}
323
324impl GrpcClient for NtxBuilderClient {
325    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
326        Self(GeneratedNtxBuilderClient::new(InterceptedService::new(channel, interceptor)))
327    }
328}
329
330impl GrpcClient for SequencerClient {
331    fn with_interceptor(channel: Channel, interceptor: Interceptor) -> Self {
332        Self(GeneratedSequencerClient::new(InterceptedService::new(channel, interceptor)))
333    }
334}
335
336// STRICT TYPE-SAFE BUILDER (NO DEFAULTS)
337// ================================================================================================
338
339/// A type-safe builder that forces the caller to make an explicit decision for each
340/// configuration item (TLS, timeout, metadata version, metadata genesis) before connecting.
341///
342/// This builder replaces the previous defaulted builder. Callers must explicitly choose TLS,
343/// timeout, and metadata options before connecting.
344///
345/// Usage example:
346///
347/// ```rust
348/// # use miden_node_proto::clients::{Builder, WantsTls, RpcClient};
349/// # use url::Url;
350/// # use std::time::Duration;
351///
352/// # async fn example() -> anyhow::Result<()> {
353/// let url = Url::parse("https://rpc.example.com:8080")?;
354/// let client: RpcClient = Builder::new(url)
355///     .with_tls()?                          // or `.without_tls()`
356///     .with_timeout(Duration::from_secs(5)) // or `.without_timeout()`
357///     .with_metadata_version("1.0".into())  // or `.without_metadata_version()`
358///     .without_metadata_genesis()           // or `.with_metadata_genesis(genesis)`
359///     .without_auth_header()                // or `.with_auth_header_value(AsciiMetadataValue::from_static("value"))`
360///     .with_otel_context_injection()        // or `.without_otel_context_injection()`
361///     .connect::<RpcClient>()
362///     .await?;
363/// # Ok(())
364/// # }
365/// ```
366#[derive(Clone, Debug)]
367pub struct Builder<State> {
368    endpoint: Endpoint,
369    endpoint_url: Url,
370    metadata_version: Option<String>,
371    metadata_genesis: Option<Word>,
372    metadata_auth_header_value: Option<AsciiMetadataValue>,
373    enable_otel: bool,
374    _state: PhantomData<State>,
375}
376
377#[derive(Copy, Clone, Debug)]
378pub struct WantsTls;
379#[derive(Copy, Clone, Debug)]
380pub struct WantsTimeout;
381#[derive(Copy, Clone, Debug)]
382pub struct WantsVersion;
383#[derive(Copy, Clone, Debug)]
384pub struct WantsGenesis;
385#[derive(Copy, Clone, Debug)]
386pub struct WantsOTel;
387#[derive(Copy, Clone, Debug)]
388pub struct WantsConnection;
389
390impl<State> Builder<State> {
391    /// Convenience function to cast the state type and carry internal configuration forward.
392    fn next_state<Next>(self) -> Builder<Next> {
393        Builder {
394            endpoint: self.endpoint,
395            endpoint_url: self.endpoint_url,
396            metadata_version: self.metadata_version,
397            metadata_genesis: self.metadata_genesis,
398            metadata_auth_header_value: self.metadata_auth_header_value,
399            enable_otel: self.enable_otel,
400            _state: PhantomData::<Next>,
401        }
402    }
403}
404
405/// Client HTTP/2 keepalive interval.
406const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(20);
407/// Client HTTP/2 keepalive: how long to wait for a PING ack before considering the connection dead.
408const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
409/// OS-level TCP keepalive backstop for direct (non-proxied) connections.
410const TCP_KEEPALIVE: Duration = Duration::from_secs(30);
411
412impl Builder<WantsTls> {
413    /// Create a new strict builder from a gRPC endpoint URL such as `http://localhost:8080` or
414    /// `https://api.example.com:443`.
415    pub fn new(url: Url) -> Builder<WantsTls> {
416        let endpoint = Endpoint::from_shared(String::from(url.clone()))
417            .expect("Url type always results in valid endpoint")
418            // Detect silently dropped connections so long-lived streams can't hang forever; see the
419            // keepalive constants above.
420            .http2_keep_alive_interval(HTTP2_KEEPALIVE_INTERVAL)
421            .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT)
422            .keep_alive_while_idle(true)
423            .tcp_keepalive(Some(TCP_KEEPALIVE));
424
425        Builder {
426            endpoint,
427            endpoint_url: url,
428            metadata_version: None,
429            metadata_genesis: None,
430            metadata_auth_header_value: None,
431            enable_otel: false,
432            _state: PhantomData,
433        }
434    }
435
436    /// Explicitly disable TLS.
437    pub fn without_tls(self) -> Builder<WantsTimeout> {
438        self.next_state()
439    }
440
441    /// Explicitly enable TLS.
442    pub fn with_tls(mut self) -> Result<Builder<WantsTimeout>, TransportError> {
443        self.endpoint = self.endpoint.tls_config(ClientTlsConfig::new().with_native_roots())?;
444
445        Ok(self.next_state())
446    }
447}
448
449impl Builder<WantsTimeout> {
450    /// Explicitly disable request timeout.
451    pub fn without_timeout(self) -> Builder<WantsVersion> {
452        self.next_state()
453    }
454
455    /// Explicitly configure a request timeout.
456    pub fn with_timeout(mut self, duration: Duration) -> Builder<WantsVersion> {
457        self.endpoint = self.endpoint.timeout(duration);
458        self.next_state()
459    }
460}
461
462impl Builder<WantsVersion> {
463    /// Do not include version in request metadata.
464    pub fn without_metadata_version(mut self) -> Builder<WantsGenesis> {
465        self.metadata_version = None;
466        self.next_state()
467    }
468
469    /// Include a specific version string in request metadata.
470    pub fn with_metadata_version(mut self, version: String) -> Builder<WantsGenesis> {
471        self.metadata_version = Some(version);
472        self.next_state()
473    }
474}
475
476impl Builder<WantsGenesis> {
477    /// Do not include genesis commitment in request metadata.
478    pub fn without_metadata_genesis(mut self) -> Builder<WantsOTel> {
479        self.metadata_genesis = None;
480        self.next_state()
481    }
482
483    /// Include a specific genesis commitment in request metadata.
484    pub fn with_metadata_genesis(mut self, genesis: Word) -> Builder<WantsOTel> {
485        self.metadata_genesis = Some(genesis);
486        self.next_state()
487    }
488}
489
490impl Builder<WantsOTel> {
491    /// Do not include any additional metadata header in request metadata.
492    #[must_use]
493    pub fn without_auth_header(mut self) -> Self {
494        self.metadata_auth_header_value = None;
495        self
496    }
497
498    /// Include an additional ASCII metadata header in request metadata.
499    #[must_use]
500    pub fn with_auth_header_value(mut self, value: AsciiMetadataValue) -> Self {
501        self.metadata_auth_header_value = Some(value);
502        self
503    }
504
505    /// Enables OpenTelemetry context propagation via gRPC.
506    ///
507    /// This is used to by OpenTelemetry to connect traces across network boundaries. The server on
508    /// the other end must be configured to receive and use the injected trace context.
509    pub fn with_otel_context_injection(mut self) -> Builder<WantsConnection> {
510        self.enable_otel = true;
511        self.next_state()
512    }
513
514    /// Disables OpenTelemetry context propagation. This should be disabled when interfacing with
515    /// external third party gRPC servers.
516    pub fn without_otel_context_injection(mut self) -> Builder<WantsConnection> {
517        self.enable_otel = false;
518        self.next_state()
519    }
520}
521
522impl Builder<WantsConnection> {
523    /// Establish an eager connection and return a fully configured client.
524    pub async fn connect<T>(self) -> Result<T, TransportError>
525    where
526        T: GrpcClient,
527    {
528        let channel = self.endpoint.connect().await?;
529        Ok(self.connect_with_channel::<T>(channel))
530    }
531
532    /// Establish a lazy connection and return a client that will connect on first use.
533    pub fn connect_lazy<T>(self) -> T
534    where
535        T: GrpcClient,
536    {
537        let channel = self.endpoint.connect_lazy();
538        self.connect_with_channel::<T>(channel)
539    }
540
541    /// Monitors whether the configured endpoint can establish a transport connection.
542    ///
543    /// The monitor is non-blocking with respect to service startup. It warns on the first failed
544    /// attempt, retries with capped exponential backoff, and reports when the dependency becomes
545    /// reachable. Once connected it waits for shutdown instead of creating duplicate connections.
546    pub async fn monitor<T>(
547        self,
548        dependency_name: &'static str,
549        shutdown: miden_node_utils::shutdown::CancellationToken,
550    ) where
551        T: GrpcClient + Send + 'static,
552    {
553        const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
554        const RETRY_MIN: Duration = Duration::from_secs(1);
555        const RETRY_MAX: Duration = Duration::from_secs(30);
556
557        use miden_node_utils::retry::BackoffBuilder;
558
559        let endpoint = miden_node_utils::formatting::format_endpoint(&self.endpoint_url);
560        let mut backoff = miden_node_utils::retry::exponential(RETRY_MIN, RETRY_MAX).build();
561        let mut first_failure = true;
562
563        loop {
564            let attempt = tokio::time::timeout(CONNECT_TIMEOUT, self.clone().connect::<T>());
565            let result = tokio::select! {
566                () = shutdown.cancelled() => return,
567                result = attempt => result,
568            };
569
570            match result {
571                Ok(Ok(_client)) => {
572                    info!(
573                        "Configured service reachable",
574                        dependency.name = dependency_name,
575                        dependency.endpoint = endpoint.as_str()
576                    );
577                    shutdown.cancelled().await;
578                    return;
579                },
580                Ok(Err(err)) if first_failure => {
581                    warn!(
582                        &err,
583                        "Configured service unreachable",
584                        dependency.name = dependency_name,
585                        dependency.endpoint = endpoint.as_str()
586                    );
587                },
588                Err(_elapsed) if first_failure => {
589                    warn!(
590                        "Configured service connection timed out",
591                        dependency.name = dependency_name,
592                        dependency.endpoint = endpoint.as_str(),
593                        timeout.ms = CONNECT_TIMEOUT.as_millis() as u64
594                    );
595                },
596                Ok(Err(err)) => {
597                    debug!(
598                        &err,
599                        "Configured service still unreachable",
600                        dependency.name = dependency_name,
601                        dependency.endpoint = endpoint.as_str()
602                    );
603                },
604                Err(_elapsed) => {
605                    debug!(
606                        "Configured service connection still timing out",
607                        dependency.name = dependency_name,
608                        dependency.endpoint = endpoint.as_str(),
609                        timeout.ms = CONNECT_TIMEOUT.as_millis() as u64
610                    );
611                },
612            }
613            first_failure = false;
614
615            let retry_delay = backoff.next().unwrap_or(RETRY_MAX);
616            tokio::select! {
617                () = shutdown.cancelled() => return,
618                () = tokio::time::sleep(retry_delay) => {},
619            }
620        }
621    }
622
623    fn connect_with_channel<T>(self, channel: Channel) -> T
624    where
625        T: GrpcClient,
626    {
627        let metadata_genesis = self.metadata_genesis.map(|genesis| genesis.to_hex());
628        let interceptor = Interceptor::new(
629            self.enable_otel,
630            self.metadata_version.as_deref(),
631            metadata_genesis.as_deref(),
632            self.metadata_auth_header_value,
633        );
634        T::with_interceptor(channel, interceptor)
635    }
636}
637
638impl ValidatorClient {
639    /// Submits each transaction in the batch to the validator for re-execution.
640    ///
641    /// # Errors
642    ///
643    /// - If `sealed_transaction_inputs` does not match the batch's transactions in length
644    pub async fn submit_batch(
645        &mut self,
646        proposed_batch: &ProposedBatch,
647        sealed_transaction_inputs: &[SealedTransactionInputs],
648    ) -> Result<(), Status> {
649        if proposed_batch.transactions().len() != sealed_transaction_inputs.len() {
650            return Err(Status::invalid_argument(
651                "transaction inputs do not match the batch's transactions",
652            ));
653        }
654        for (tx, inputs) in proposed_batch.transactions().iter().zip(sealed_transaction_inputs) {
655            let proven_tx = GeneratedProvenTransaction {
656                transaction: tx.to_bytes(),
657                sealed_transaction_inputs: Some(inputs.clone()),
658            };
659            self.submit_proven_transaction(proven_tx).await?;
660        }
661        Ok(())
662    }
663}
miden_node_proto/clients/mod.rs

miden_node_proto/clients/
mod.rs
