1use std::collections::BTreeSet;
2use std::error::Error;
3use std::future::Future;
4use std::sync::Arc;
5
6use guardian_shared::retry::{
7 ProductionRetryRuntime, RPC_TRANSPORT_SIGNALS, RetryPolicy, RetryRuntime, StructuredEvidence,
8 connect_failure_is_permanent, is_transient_error_with, run_retries,
9};
10use miden_client::note_transport::{
11 NoteInfo, NoteStream, NoteTransportClient, NoteTransportCursor, NoteTransportError,
12};
13use miden_client::rpc::domain::account::{AccountProof, GetAccountRequest};
14use miden_client::rpc::domain::account_vault::AccountVaultInfo;
15use miden_client::rpc::domain::limits::RpcLimits;
16use miden_client::rpc::domain::note::{FetchedNote, NoteSyncBlock};
17use miden_client::rpc::domain::nullifier::NullifierUpdate;
18use miden_client::rpc::domain::storage_map::StorageMapInfo;
19use miden_client::rpc::domain::sync::{ChainMmrInfo, SyncTarget};
20use miden_client::rpc::domain::transaction::TransactionRecord;
21use miden_client::rpc::{GrpcError, NetworkNoteStatusInfo, NodeRpcClient, RpcError, RpcStatusInfo};
22use miden_protocol::Word;
23use miden_protocol::account::AccountId;
24use miden_protocol::address::NetworkId;
25use miden_protocol::batch::{ProposedBatch, ProvenBatch};
26use miden_protocol::block::{BlockHeader, BlockNumber, ProvenBlock};
27use miden_protocol::crypto::merkle::mmr::MmrProof;
28use miden_protocol::note::NoteHeader;
29use miden_protocol::note::{NoteId, NoteScript, NoteTag};
30use miden_protocol::transaction::{ProvenTransaction, TransactionInputs};
31
32use crate::error::{MultisigError, Result, rpc_kind};
33
34const DEFAULT_MAX_ATTEMPTS: u32 = 2;
35
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub struct RpcRetryPolicy {
38 inner: RetryPolicy,
39}
40
41impl Default for RpcRetryPolicy {
42 fn default() -> Self {
43 Self {
44 inner: RetryPolicy::new(DEFAULT_MAX_ATTEMPTS),
45 }
46 }
47}
48
49impl RpcRetryPolicy {
50 #[must_use]
51 pub fn new(max_attempts: u32) -> Self {
52 Self {
53 inner: RetryPolicy::new(max_attempts),
54 }
55 }
56
57 #[must_use]
58 pub fn max_attempts(&self) -> u32 {
59 self.inner.max_attempts()
60 }
61
62 pub(crate) fn as_retry_policy(&self) -> &RetryPolicy {
63 &self.inner
64 }
65}
66
67#[derive(Clone, Debug, Default, PartialEq, Eq)]
68pub struct RpcConfig {
69 timeout_ms: Option<u64>,
70 retry_policy: RpcRetryPolicy,
71}
72
73impl RpcConfig {
74 #[must_use]
75 pub fn new() -> Self {
76 Self::default()
77 }
78
79 pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Result<Self> {
80 if timeout_ms == 0 {
81 return Err(MultisigError::InvalidConfig(
82 "rpc timeout must be a positive number of milliseconds".to_string(),
83 ));
84 }
85 self.timeout_ms = Some(timeout_ms);
86 Ok(self)
87 }
88
89 #[must_use]
90 pub fn with_retry_policy(mut self, retry_policy: RpcRetryPolicy) -> Self {
91 self.retry_policy = retry_policy;
92 self
93 }
94
95 #[must_use]
96 pub fn timeout_ms(&self) -> Option<u64> {
97 self.timeout_ms
98 }
99
100 #[must_use]
101 pub fn retry_policy(&self) -> &RpcRetryPolicy {
102 &self.retry_policy
103 }
104}
105
106#[derive(Clone, Debug, PartialEq, Eq)]
107pub(crate) enum RpcSelection {
108 Passthrough,
109 Configured {
110 timeout_ms: u64,
111 retry_policy: RpcRetryPolicy,
112 },
113}
114
115impl RpcConfig {
116 pub(crate) fn resolve(&self, default_timeout_ms: u64) -> RpcSelection {
117 if self.timeout_ms.is_none() && self.retry_policy.max_attempts() == 1 {
118 return RpcSelection::Passthrough;
119 }
120 RpcSelection::Configured {
121 timeout_ms: self.timeout_ms.unwrap_or(default_timeout_ms),
122 retry_policy: self.retry_policy.clone(),
123 }
124 }
125}
126
127fn grpc_error_evidence(kind: &GrpcError) -> StructuredEvidence {
128 match kind {
129 GrpcError::Cancelled
130 | GrpcError::DeadlineExceeded
131 | GrpcError::ResourceExhausted
132 | GrpcError::Unavailable => StructuredEvidence::Transient,
133 GrpcError::NotFound
134 | GrpcError::InvalidArgument
135 | GrpcError::PermissionDenied
136 | GrpcError::AlreadyExists
137 | GrpcError::FailedPrecondition
138 | GrpcError::Internal
139 | GrpcError::Unimplemented
140 | GrpcError::Unauthenticated
141 | GrpcError::Aborted
142 | GrpcError::OutOfRange
143 | GrpcError::DataLoss => StructuredEvidence::Permanent,
144 GrpcError::Unknown(_) => StructuredEvidence::Indeterminate,
145 }
146}
147
148fn rpc_link_evidence(cause: &(dyn Error + 'static)) -> StructuredEvidence {
149 cause
150 .downcast_ref::<RpcError>()
151 .and_then(rpc_kind)
152 .map(grpc_error_evidence)
153 .unwrap_or(StructuredEvidence::Indeterminate)
154}
155
156pub(crate) fn is_transient_rpc_error(error: &RpcError) -> bool {
157 is_transient_error_with(error, rpc_link_evidence, &RPC_TRANSPORT_SIGNALS)
158}
159
160fn note_transport_link_evidence(cause: &(dyn Error + 'static)) -> StructuredEvidence {
164 match cause.downcast_ref::<NoteTransportError>() {
165 Some(NoteTransportError::Connection(inner)) => {
166 if connect_failure_is_permanent(inner.as_ref()) {
167 StructuredEvidence::Permanent
168 } else {
169 StructuredEvidence::Transient
170 }
171 }
172 Some(
173 NoteTransportError::Disabled
174 | NoteTransportError::Deserialization(_)
175 | NoteTransportError::PaginationDidNotTerminate(_),
176 ) => StructuredEvidence::Permanent,
177 Some(NoteTransportError::Network(_)) | None => StructuredEvidence::Indeterminate,
178 }
179}
180
181pub(crate) fn is_transient_note_transport_error(error: &NoteTransportError) -> bool {
182 is_transient_error_with(error, note_transport_link_evidence, &RPC_TRANSPORT_SIGNALS)
183}
184
185pub(crate) struct RetryingNodeRpcClient {
186 inner: Arc<dyn NodeRpcClient>,
187 policy: RetryPolicy,
188 runtime: Arc<dyn RetryRuntime>,
189}
190
191impl RetryingNodeRpcClient {
192 pub(crate) fn new(inner: Arc<dyn NodeRpcClient>, policy: &RpcRetryPolicy) -> Self {
193 Self {
194 inner,
195 policy: policy.as_retry_policy().clone(),
196 runtime: Arc::new(ProductionRetryRuntime),
197 }
198 }
199
200 #[cfg(test)]
201 fn with_runtime(
202 inner: Arc<dyn NodeRpcClient>,
203 policy: &RpcRetryPolicy,
204 runtime: Arc<dyn RetryRuntime>,
205 ) -> Self {
206 Self {
207 inner,
208 policy: policy.as_retry_policy().clone(),
209 runtime,
210 }
211 }
212
213 async fn execute<T, F, Fut>(&self, op: F) -> std::result::Result<T, RpcError>
216 where
217 F: Fn() -> Fut + Send + Sync,
218 Fut: Future<Output = std::result::Result<T, RpcError>> + Send,
219 {
220 run_retries(
221 self.policy.max_attempts(),
222 self.runtime.as_ref(),
223 is_transient_rpc_error,
224 |_, _| {},
225 op,
226 )
227 .await
228 }
229}
230
231#[async_trait::async_trait]
232impl NodeRpcClient for RetryingNodeRpcClient {
233 async fn set_genesis_commitment(&self, commitment: Word) -> std::result::Result<(), RpcError> {
234 self.execute(|| self.inner.set_genesis_commitment(commitment))
235 .await
236 }
237
238 fn has_genesis_commitment(&self) -> Option<Word> {
239 self.inner.has_genesis_commitment()
240 }
241
242 async fn submit_proven_transaction(
245 &self,
246 proven_transaction: ProvenTransaction,
247 transaction_inputs: TransactionInputs,
248 ) -> std::result::Result<BlockNumber, RpcError> {
249 self.inner
250 .submit_proven_transaction(proven_transaction, transaction_inputs)
251 .await
252 }
253
254 async fn submit_proven_batch(
257 &self,
258 proven_batch: ProvenBatch,
259 proposed_batch: ProposedBatch,
260 transaction_inputs: Vec<TransactionInputs>,
261 ) -> std::result::Result<BlockNumber, RpcError> {
262 self.inner
263 .submit_proven_batch(proven_batch, proposed_batch, transaction_inputs)
264 .await
265 }
266
267 async fn get_block_header_by_number(
268 &self,
269 block_num: Option<BlockNumber>,
270 include_mmr_proof: bool,
271 ) -> std::result::Result<(BlockHeader, Option<MmrProof>), RpcError> {
272 self.execute(|| {
273 self.inner
274 .get_block_header_by_number(block_num, include_mmr_proof)
275 })
276 .await
277 }
278
279 async fn get_block_by_number(
280 &self,
281 block_num: BlockNumber,
282 include_proof: bool,
283 ) -> std::result::Result<ProvenBlock, RpcError> {
284 self.execute(|| self.inner.get_block_by_number(block_num, include_proof))
285 .await
286 }
287
288 async fn get_notes_by_id(
289 &self,
290 note_ids: &[NoteId],
291 ) -> std::result::Result<Vec<FetchedNote>, RpcError> {
292 self.execute(|| self.inner.get_notes_by_id(note_ids)).await
293 }
294
295 async fn sync_chain_mmr(
296 &self,
297 current_block_height: BlockNumber,
298 upper_bound: SyncTarget,
299 ) -> std::result::Result<ChainMmrInfo, RpcError> {
300 self.execute(|| self.inner.sync_chain_mmr(current_block_height, upper_bound))
301 .await
302 }
303
304 async fn sync_notes(
305 &self,
306 block_from: BlockNumber,
307 block_to: BlockNumber,
308 note_tags: &BTreeSet<NoteTag>,
309 ) -> std::result::Result<Vec<NoteSyncBlock>, RpcError> {
310 self.execute(|| self.inner.sync_notes(block_from, block_to, note_tags))
311 .await
312 }
313
314 async fn sync_nullifiers(
315 &self,
316 prefix: &[u16],
317 block_from: BlockNumber,
318 block_to: BlockNumber,
319 ) -> std::result::Result<Vec<NullifierUpdate>, RpcError> {
320 self.execute(|| self.inner.sync_nullifiers(prefix, block_from, block_to))
321 .await
322 }
323
324 async fn get_account(
325 &self,
326 account_id: AccountId,
327 request: GetAccountRequest,
328 ) -> std::result::Result<(BlockNumber, AccountProof), RpcError> {
329 self.execute(|| self.inner.get_account(account_id, request.clone()))
330 .await
331 }
332
333 async fn get_note_script_by_root(
334 &self,
335 root: Word,
336 ) -> std::result::Result<Option<NoteScript>, RpcError> {
337 self.execute(|| self.inner.get_note_script_by_root(root))
338 .await
339 }
340
341 async fn sync_storage_maps(
342 &self,
343 block_from: BlockNumber,
344 block_to: BlockNumber,
345 account_id: AccountId,
346 ) -> std::result::Result<StorageMapInfo, RpcError> {
347 self.execute(|| {
348 self.inner
349 .sync_storage_maps(block_from, block_to, account_id)
350 })
351 .await
352 }
353
354 async fn sync_account_vault(
355 &self,
356 block_from: BlockNumber,
357 block_to: BlockNumber,
358 account_id: AccountId,
359 ) -> std::result::Result<AccountVaultInfo, RpcError> {
360 self.execute(|| {
361 self.inner
362 .sync_account_vault(block_from, block_to, account_id)
363 })
364 .await
365 }
366
367 async fn sync_transactions(
368 &self,
369 block_from: BlockNumber,
370 block_to: BlockNumber,
371 account_ids: Vec<AccountId>,
372 ) -> std::result::Result<Vec<TransactionRecord>, RpcError> {
373 self.execute(|| {
374 self.inner
375 .sync_transactions(block_from, block_to, account_ids.clone())
376 })
377 .await
378 }
379
380 async fn get_network_id(&self) -> std::result::Result<NetworkId, RpcError> {
381 self.execute(|| self.inner.get_network_id()).await
382 }
383
384 async fn get_rpc_limits(&self) -> std::result::Result<RpcLimits, RpcError> {
385 self.execute(|| self.inner.get_rpc_limits()).await
386 }
387
388 fn has_rpc_limits(&self) -> Option<RpcLimits> {
389 self.inner.has_rpc_limits()
390 }
391
392 async fn set_rpc_limits(&self, limits: RpcLimits) {
393 self.inner.set_rpc_limits(limits).await;
394 }
395
396 async fn get_status_unversioned(&self) -> std::result::Result<RpcStatusInfo, RpcError> {
397 self.execute(|| self.inner.get_status_unversioned()).await
398 }
399
400 async fn get_network_note_status(
401 &self,
402 note_id: NoteId,
403 ) -> std::result::Result<NetworkNoteStatusInfo, RpcError> {
404 self.execute(|| self.inner.get_network_note_status(note_id))
405 .await
406 }
407}
408
409pub(crate) fn configured_note_transport_client(
413 inner: Arc<dyn NoteTransportClient>,
414 rpc_config: &RpcConfig,
415) -> Arc<dyn NoteTransportClient> {
416 if rpc_config.retry_policy().max_attempts() > 1 {
417 Arc::new(RetryingNoteTransportClient::new(
418 inner,
419 rpc_config.retry_policy(),
420 ))
421 } else {
422 inner
423 }
424}
425
426pub(crate) struct RetryingNoteTransportClient {
427 inner: Arc<dyn NoteTransportClient>,
428 policy: RetryPolicy,
429 runtime: Arc<dyn RetryRuntime>,
430}
431
432impl RetryingNoteTransportClient {
433 pub(crate) fn new(inner: Arc<dyn NoteTransportClient>, policy: &RpcRetryPolicy) -> Self {
434 Self {
435 inner,
436 policy: policy.as_retry_policy().clone(),
437 runtime: Arc::new(ProductionRetryRuntime),
438 }
439 }
440
441 #[cfg(test)]
442 fn with_runtime(
443 inner: Arc<dyn NoteTransportClient>,
444 policy: &RpcRetryPolicy,
445 runtime: Arc<dyn RetryRuntime>,
446 ) -> Self {
447 Self {
448 inner,
449 policy: policy.as_retry_policy().clone(),
450 runtime,
451 }
452 }
453
454 async fn retry_fetch<T, F, Fut>(&self, op: F) -> std::result::Result<T, NoteTransportError>
455 where
456 F: Fn() -> Fut + Send + Sync,
457 Fut: Future<Output = std::result::Result<T, NoteTransportError>> + Send,
458 {
459 run_retries(
460 self.policy.max_attempts(),
461 self.runtime.as_ref(),
462 is_transient_note_transport_error,
463 |_, _| {},
464 op,
465 )
466 .await
467 }
468}
469
470#[async_trait::async_trait]
471impl NoteTransportClient for RetryingNoteTransportClient {
472 async fn send_note(
476 &self,
477 header: NoteHeader,
478 details: Vec<u8>,
479 ) -> std::result::Result<(), NoteTransportError> {
480 self.inner.send_note(header, details).await
481 }
482
483 async fn fetch_notes(
484 &self,
485 tag: &[NoteTag],
486 cursor: NoteTransportCursor,
487 ) -> std::result::Result<(Vec<NoteInfo>, NoteTransportCursor), NoteTransportError> {
488 self.retry_fetch(|| self.inner.fetch_notes(tag, cursor))
489 .await
490 }
491
492 async fn stream_notes(
493 &self,
494 tag: NoteTag,
495 cursor: NoteTransportCursor,
496 ) -> std::result::Result<Box<dyn NoteStream>, NoteTransportError> {
497 self.retry_fetch(|| self.inner.stream_notes(tag, cursor))
498 .await
499 }
500}
501
502#[cfg(test)]
503mod tests {
504 use std::sync::Mutex;
505 use std::sync::atomic::{AtomicU32, Ordering};
506 use std::time::Duration;
507
508 use miden_client::rpc::RpcEndpoint;
509
510 use super::*;
511
512 fn request_error(kind: GrpcError) -> RpcError {
513 RpcError::RequestError {
514 endpoint: RpcEndpoint::GetAccount,
515 error_kind: kind,
516 endpoint_error: None,
517 source: None,
518 }
519 }
520
521 #[test]
522 fn typed_grpc_kinds_partition_into_retryable_and_permanent() {
523 let transient = [
524 GrpcError::Cancelled,
525 GrpcError::DeadlineExceeded,
526 GrpcError::ResourceExhausted,
527 GrpcError::Unavailable,
528 ];
529 for kind in transient {
530 assert!(is_transient_rpc_error(&request_error(kind)));
531 }
532
533 let permanent = [
534 GrpcError::NotFound,
535 GrpcError::InvalidArgument,
536 GrpcError::PermissionDenied,
537 GrpcError::AlreadyExists,
538 GrpcError::FailedPrecondition,
539 GrpcError::Internal,
540 GrpcError::Unimplemented,
541 GrpcError::Unauthenticated,
542 GrpcError::Aborted,
543 GrpcError::OutOfRange,
544 GrpcError::DataLoss,
545 ];
546 for kind in permanent {
547 assert!(!is_transient_rpc_error(&request_error(kind)));
548 }
549 }
550
551 #[test]
552 fn unknown_is_retryable_only_when_the_connection_failed() {
553 let transport = request_error(GrpcError::Unknown(
554 "transport error: code: 'Unknown error', message: \"transport error\", source: \
555 tonic::transport::Error(Transport, hyper::Error(Io, Kind(TimedOut)))"
556 .to_string(),
557 ));
558 assert!(is_transient_rpc_error(&transport));
559
560 let io_timeout = request_error(GrpcError::Unknown(
561 "connection error: desc = \"i/o timeout\"".to_string(),
562 ));
563 assert!(is_transient_rpc_error(&io_timeout));
564
565 let fault = request_error(GrpcError::Unknown(
566 "internal invariant violated".to_string(),
567 ));
568 assert!(!is_transient_rpc_error(&fault));
569 }
570
571 #[test]
572 fn connection_error_with_transport_source_is_retryable() {
573 let error = RpcError::ConnectionError(Box::new(std::io::Error::new(
574 std::io::ErrorKind::TimedOut,
575 "transport error: i/o timeout",
576 )));
577 assert!(is_transient_rpc_error(&error));
578 }
579
580 #[test]
581 fn deserialization_failures_are_not_retried() {
582 assert!(!is_transient_rpc_error(&RpcError::DeserializationError(
583 "unexpected field".to_string()
584 )));
585 }
586
587 #[test]
588 fn timeout_config_rejects_zero() {
589 assert!(RpcConfig::new().with_timeout_ms(0).is_err());
590 assert_eq!(
591 RpcConfig::new().with_timeout_ms(1).unwrap().timeout_ms(),
592 Some(1)
593 );
594 }
595
596 #[test]
597 fn attempt_budget_vectors_match_contract() {
598 #[derive(serde::Deserialize)]
599 #[serde(rename_all = "camelCase")]
600 struct Fixtures {
601 attempt_budgets: Vec<AttemptBudget>,
602 }
603 #[derive(serde::Deserialize)]
604 struct AttemptBudget {
605 input: Option<u32>,
606 normalized: u32,
607 }
608 let fixtures: Fixtures = serde_json::from_str(include_str!(
609 "../../../fixtures/miden-multisig-client/rpc-policy-fixtures.json"
610 ))
611 .expect("fixtures must parse");
612 for fixture in fixtures.attempt_budgets {
613 let policy = fixture.input.map(RpcRetryPolicy::new).unwrap_or_default();
614 assert_eq!(policy.max_attempts(), fixture.normalized);
615 }
616 }
617
618 #[test]
619 fn default_config_resolves_to_one_configured_retry() {
620 assert_eq!(
621 RpcConfig::new().resolve(10_000),
622 RpcSelection::Configured {
623 timeout_ms: 10_000,
624 retry_policy: RpcRetryPolicy::new(2),
625 }
626 );
627 }
628
629 #[test]
630 fn an_explicit_single_attempt_policy_opts_out_entirely() {
631 assert_eq!(
632 RpcConfig::new()
633 .with_retry_policy(RpcRetryPolicy::new(1))
634 .resolve(10_000),
635 RpcSelection::Passthrough
636 );
637 }
638
639 #[test]
640 fn configured_timeout_or_retries_resolve_to_configured() {
641 assert_eq!(
642 RpcConfig::new()
643 .with_timeout_ms(5_000)
644 .unwrap()
645 .resolve(10_000),
646 RpcSelection::Configured {
647 timeout_ms: 5_000,
648 retry_policy: RpcRetryPolicy::default(),
649 }
650 );
651 assert_eq!(
652 RpcConfig::new()
653 .with_retry_policy(RpcRetryPolicy::new(3))
654 .resolve(10_000),
655 RpcSelection::Configured {
656 timeout_ms: 10_000,
657 retry_policy: RpcRetryPolicy::new(3),
658 }
659 );
660 }
661
662 #[derive(Default)]
663 struct RecordingRuntime {
664 sleeps: Mutex<Vec<Duration>>,
665 }
666
667 #[async_trait::async_trait]
668 impl RetryRuntime for RecordingRuntime {
669 async fn sleep(&self, duration: Duration) {
670 self.sleeps.lock().unwrap().push(duration);
671 }
672
673 fn unit_random(&self) -> f64 {
674 0.5
675 }
676 }
677
678 struct ScriptedNetworkIdInner {
679 failures_before_success: AtomicU32,
680 calls: AtomicU32,
681 error: fn() -> RpcError,
682 }
683
684 #[async_trait::async_trait]
685 impl NodeRpcClient for ScriptedNetworkIdInner {
686 async fn get_network_id(&self) -> std::result::Result<NetworkId, RpcError> {
687 self.calls.fetch_add(1, Ordering::SeqCst);
688 let remaining = self.failures_before_success.load(Ordering::SeqCst);
689 if remaining > 0 {
690 self.failures_before_success
691 .store(remaining - 1, Ordering::SeqCst);
692 return Err((self.error)());
693 }
694 Ok(NetworkId::Devnet)
695 }
696
697 fn has_genesis_commitment(&self) -> Option<Word> {
698 None
699 }
700 fn has_rpc_limits(&self) -> Option<RpcLimits> {
701 None
702 }
703 async fn set_rpc_limits(&self, _: RpcLimits) {}
704 async fn set_genesis_commitment(&self, _: Word) -> std::result::Result<(), RpcError> {
705 unimplemented!()
706 }
707 async fn submit_proven_transaction(
708 &self,
709 _: ProvenTransaction,
710 _: TransactionInputs,
711 ) -> std::result::Result<BlockNumber, RpcError> {
712 unimplemented!()
713 }
714 async fn submit_proven_batch(
715 &self,
716 _: ProvenBatch,
717 _: ProposedBatch,
718 _: Vec<TransactionInputs>,
719 ) -> std::result::Result<BlockNumber, RpcError> {
720 unimplemented!()
721 }
722 async fn get_block_header_by_number(
723 &self,
724 _: Option<BlockNumber>,
725 _: bool,
726 ) -> std::result::Result<(BlockHeader, Option<MmrProof>), RpcError> {
727 unimplemented!()
728 }
729 async fn get_block_by_number(
730 &self,
731 _: BlockNumber,
732 _: bool,
733 ) -> std::result::Result<ProvenBlock, RpcError> {
734 unimplemented!()
735 }
736 async fn get_notes_by_id(
737 &self,
738 _: &[NoteId],
739 ) -> std::result::Result<Vec<FetchedNote>, RpcError> {
740 unimplemented!()
741 }
742 async fn sync_chain_mmr(
743 &self,
744 _: BlockNumber,
745 _: SyncTarget,
746 ) -> std::result::Result<ChainMmrInfo, RpcError> {
747 unimplemented!()
748 }
749 async fn sync_notes(
750 &self,
751 _: BlockNumber,
752 _: BlockNumber,
753 _: &BTreeSet<NoteTag>,
754 ) -> std::result::Result<Vec<NoteSyncBlock>, RpcError> {
755 unimplemented!()
756 }
757 async fn sync_nullifiers(
758 &self,
759 _: &[u16],
760 _: BlockNumber,
761 _: BlockNumber,
762 ) -> std::result::Result<Vec<NullifierUpdate>, RpcError> {
763 unimplemented!()
764 }
765 async fn get_account(
766 &self,
767 _: AccountId,
768 _: GetAccountRequest,
769 ) -> std::result::Result<(BlockNumber, AccountProof), RpcError> {
770 unimplemented!()
771 }
772 async fn get_note_script_by_root(
773 &self,
774 _: Word,
775 ) -> std::result::Result<Option<NoteScript>, RpcError> {
776 unimplemented!()
777 }
778 async fn sync_storage_maps(
779 &self,
780 _: BlockNumber,
781 _: BlockNumber,
782 _: AccountId,
783 ) -> std::result::Result<StorageMapInfo, RpcError> {
784 unimplemented!()
785 }
786 async fn sync_account_vault(
787 &self,
788 _: BlockNumber,
789 _: BlockNumber,
790 _: AccountId,
791 ) -> std::result::Result<AccountVaultInfo, RpcError> {
792 unimplemented!()
793 }
794 async fn sync_transactions(
795 &self,
796 _: BlockNumber,
797 _: BlockNumber,
798 _: Vec<AccountId>,
799 ) -> std::result::Result<Vec<TransactionRecord>, RpcError> {
800 unimplemented!()
801 }
802 async fn get_rpc_limits(&self) -> std::result::Result<RpcLimits, RpcError> {
803 unimplemented!()
804 }
805 async fn get_status_unversioned(&self) -> std::result::Result<RpcStatusInfo, RpcError> {
806 unimplemented!()
807 }
808 async fn get_network_note_status(
809 &self,
810 _: NoteId,
811 ) -> std::result::Result<NetworkNoteStatusInfo, RpcError> {
812 unimplemented!()
813 }
814 }
815
816 fn rate_limit_error() -> RpcError {
817 RpcError::RequestError {
818 endpoint: RpcEndpoint::SyncChainMmr,
819 error_kind: GrpcError::ResourceExhausted,
820 endpoint_error: None,
821 source: None,
822 }
823 }
824
825 #[tokio::test]
826 async fn a_rate_limited_read_retries_until_it_succeeds() {
827 let inner = Arc::new(ScriptedNetworkIdInner {
828 failures_before_success: AtomicU32::new(2),
829 calls: AtomicU32::new(0),
830 error: rate_limit_error,
831 });
832 let runtime = Arc::new(RecordingRuntime::default());
833 let client = RetryingNodeRpcClient::with_runtime(
834 inner.clone(),
835 &RpcRetryPolicy::new(3),
836 runtime.clone(),
837 );
838
839 let network = client.get_network_id().await.unwrap();
840
841 assert_eq!(network, NetworkId::Devnet);
842 assert_eq!(inner.calls.load(Ordering::SeqCst), 3);
843 assert_eq!(
844 runtime.sleeps.lock().unwrap().as_slice(),
845 [Duration::from_millis(500), Duration::from_millis(1000)]
846 );
847 }
848
849 #[tokio::test]
850 async fn an_exhausted_budget_returns_the_final_upstream_error_unchanged() {
851 let inner = Arc::new(ScriptedNetworkIdInner {
852 failures_before_success: AtomicU32::new(10),
853 calls: AtomicU32::new(0),
854 error: rate_limit_error,
855 });
856 let runtime = Arc::new(RecordingRuntime::default());
857 let client = RetryingNodeRpcClient::with_runtime(
858 inner.clone(),
859 &RpcRetryPolicy::new(2),
860 runtime.clone(),
861 );
862
863 let error = client.get_network_id().await.unwrap_err();
864
865 assert!(matches!(
866 error,
867 RpcError::RequestError {
868 error_kind: GrpcError::ResourceExhausted,
869 ..
870 }
871 ));
872 assert_eq!(inner.calls.load(Ordering::SeqCst), 2);
873 }
874
875 #[tokio::test]
876 async fn a_permanent_failure_does_not_retry_or_sleep() {
877 let inner = Arc::new(ScriptedNetworkIdInner {
878 failures_before_success: AtomicU32::new(10),
879 calls: AtomicU32::new(0),
880 error: || request_error(GrpcError::InvalidArgument),
881 });
882 let runtime = Arc::new(RecordingRuntime::default());
883 let client = RetryingNodeRpcClient::with_runtime(
884 inner.clone(),
885 &RpcRetryPolicy::new(5),
886 runtime.clone(),
887 );
888
889 client.get_network_id().await.unwrap_err();
890
891 assert_eq!(inner.calls.load(Ordering::SeqCst), 1);
892 assert!(runtime.sleeps.lock().unwrap().is_empty());
893 }
894
895 #[tokio::test]
896 async fn a_default_policy_retries_a_transient_read_once() {
897 let inner = Arc::new(ScriptedNetworkIdInner {
898 failures_before_success: AtomicU32::new(1),
899 calls: AtomicU32::new(0),
900 error: rate_limit_error,
901 });
902 let runtime = Arc::new(RecordingRuntime::default());
903 let client = RetryingNodeRpcClient::with_runtime(
904 inner.clone(),
905 &RpcRetryPolicy::default(),
906 runtime.clone(),
907 );
908
909 client.get_network_id().await.unwrap();
910
911 assert_eq!(inner.calls.load(Ordering::SeqCst), 2);
912 assert_eq!(runtime.sleeps.lock().unwrap().len(), 1);
913 }
914
915 #[tokio::test]
916 async fn an_explicit_single_attempt_policy_never_retries() {
917 let inner = Arc::new(ScriptedNetworkIdInner {
918 failures_before_success: AtomicU32::new(1),
919 calls: AtomicU32::new(0),
920 error: rate_limit_error,
921 });
922 let runtime = Arc::new(RecordingRuntime::default());
923 let client = RetryingNodeRpcClient::with_runtime(
924 inner.clone(),
925 &RpcRetryPolicy::new(1),
926 runtime.clone(),
927 );
928
929 client.get_network_id().await.unwrap_err();
930
931 assert_eq!(inner.calls.load(Ordering::SeqCst), 1);
932 assert!(runtime.sleeps.lock().unwrap().is_empty());
933 }
934
935 #[tokio::test]
936 async fn concurrent_initializations_survive_a_rate_limit_window() {
937 let runtime = Arc::new(RecordingRuntime::default());
938
939 let clients: Vec<_> = (0..64)
940 .map(|_| {
941 let inner = Arc::new(ScriptedNetworkIdInner {
942 failures_before_success: AtomicU32::new(2),
943 calls: AtomicU32::new(0),
944 error: rate_limit_error,
945 });
946 RetryingNodeRpcClient::with_runtime(inner, &RpcRetryPolicy::new(4), runtime.clone())
947 })
948 .collect();
949
950 let results =
951 futures::future::join_all(clients.iter().map(|client| client.get_network_id())).await;
952
953 assert_eq!(results.len(), 64);
954 assert!(results.iter().all(|result| result.is_ok()));
955 assert_eq!(runtime.sleeps.lock().unwrap().len(), 128);
956 }
957
958 struct ScriptedNoteTransportInner {
959 failures_before_success: AtomicU32,
960 fetch_calls: AtomicU32,
961 send_calls: AtomicU32,
962 error: fn() -> NoteTransportError,
963 }
964
965 #[async_trait::async_trait]
966 impl NoteTransportClient for ScriptedNoteTransportInner {
967 async fn send_note(
968 &self,
969 _: NoteHeader,
970 _: Vec<u8>,
971 ) -> std::result::Result<(), NoteTransportError> {
972 self.send_calls.fetch_add(1, Ordering::SeqCst);
973 Err((self.error)())
974 }
975
976 async fn fetch_notes(
977 &self,
978 _: &[NoteTag],
979 cursor: NoteTransportCursor,
980 ) -> std::result::Result<(Vec<NoteInfo>, NoteTransportCursor), NoteTransportError> {
981 self.fetch_calls.fetch_add(1, Ordering::SeqCst);
982 let remaining = self.failures_before_success.load(Ordering::SeqCst);
983 if remaining > 0 {
984 self.failures_before_success
985 .store(remaining - 1, Ordering::SeqCst);
986 return Err((self.error)());
987 }
988 Ok((Vec::new(), cursor))
989 }
990
991 async fn stream_notes(
992 &self,
993 _: NoteTag,
994 _: NoteTransportCursor,
995 ) -> std::result::Result<Box<dyn NoteStream>, NoteTransportError> {
996 unimplemented!()
997 }
998 }
999
1000 fn test_note_header() -> NoteHeader {
1001 let sender = AccountId::from_hex("0x7b7b7b7a7b7b7b017b7b7b7b7b7b7b").unwrap();
1002 let metadata = miden_protocol::note::NoteMetadata::new(
1003 miden_protocol::note::PartialNoteMetadata::new(
1004 sender,
1005 miden_protocol::note::NoteType::Private,
1006 ),
1007 &miden_protocol::note::NoteAttachments::default(),
1008 );
1009 NoteHeader::new(
1010 miden_protocol::note::NoteDetailsCommitment::from_raw_commitments(
1011 Word::default(),
1012 Word::default(),
1013 ),
1014 metadata,
1015 )
1016 }
1017
1018 fn note_fetch_timeout() -> NoteTransportError {
1019 NoteTransportError::Network(
1020 "Fetch notes failed: Status { code: Cancelled, message: \"Timeout expired\" }"
1021 .to_string(),
1022 )
1023 }
1024
1025 #[test]
1026 fn note_transport_errors_classify_by_variant_and_text() {
1027 assert!(is_transient_note_transport_error(¬e_fetch_timeout()));
1028 assert!(is_transient_note_transport_error(
1029 &NoteTransportError::Connection(Box::new(std::io::Error::new(
1030 std::io::ErrorKind::TimedOut,
1031 "i/o timeout",
1032 )))
1033 ));
1034 assert!(!is_transient_note_transport_error(
1035 &NoteTransportError::Disabled
1036 ));
1037 assert!(!is_transient_note_transport_error(
1038 &NoteTransportError::Network("note not recognized by relay".to_string())
1039 ));
1040 assert!(!is_transient_note_transport_error(
1041 &NoteTransportError::PaginationDidNotTerminate(64)
1042 ));
1043 }
1044
1045 #[tokio::test]
1046 async fn a_timed_out_note_fetch_retries_until_it_succeeds() {
1047 let inner = Arc::new(ScriptedNoteTransportInner {
1048 failures_before_success: AtomicU32::new(2),
1049 fetch_calls: AtomicU32::new(0),
1050 send_calls: AtomicU32::new(0),
1051 error: note_fetch_timeout,
1052 });
1053 let runtime = Arc::new(RecordingRuntime::default());
1054 let client = RetryingNoteTransportClient::with_runtime(
1055 inner.clone(),
1056 &RpcRetryPolicy::new(3),
1057 runtime.clone(),
1058 );
1059
1060 client
1061 .fetch_notes(&[], NoteTransportCursor::from(0))
1062 .await
1063 .unwrap();
1064
1065 assert_eq!(inner.fetch_calls.load(Ordering::SeqCst), 3);
1066 assert_eq!(runtime.sleeps.lock().unwrap().len(), 2);
1067 }
1068
1069 #[tokio::test]
1070 async fn a_note_send_is_never_retried_even_when_transient() {
1071 let inner = Arc::new(ScriptedNoteTransportInner {
1072 failures_before_success: AtomicU32::new(10),
1073 fetch_calls: AtomicU32::new(0),
1074 send_calls: AtomicU32::new(0),
1075 error: note_fetch_timeout,
1076 });
1077 let runtime = Arc::new(RecordingRuntime::default());
1078 let client = RetryingNoteTransportClient::with_runtime(
1079 inner.clone(),
1080 &RpcRetryPolicy::new(5),
1081 runtime.clone(),
1082 );
1083
1084 let error = client
1085 .send_note(test_note_header(), Vec::new())
1086 .await
1087 .unwrap_err();
1088
1089 assert!(is_transient_note_transport_error(&error));
1090 assert_eq!(inner.send_calls.load(Ordering::SeqCst), 1);
1091 assert!(runtime.sleeps.lock().unwrap().is_empty());
1092 }
1093
1094 #[test]
1095 fn connection_failures_classify_by_wrapped_cause() {
1096 let refused = NoteTransportError::Connection(Box::new(std::io::Error::new(
1097 std::io::ErrorKind::ConnectionRefused,
1098 "connection refused",
1099 )));
1100 assert!(is_transient_note_transport_error(&refused));
1101
1102 let bad_cert = NoteTransportError::Connection(Box::new(std::io::Error::other(
1103 "invalid peer certificate: UnknownIssuer",
1104 )));
1105 assert!(!is_transient_note_transport_error(&bad_cert));
1106
1107 let bad_uri = NoteTransportError::Connection(Box::new(std::io::Error::other(
1108 "transport error: invalid URI",
1109 )));
1110 assert!(!is_transient_note_transport_error(&bad_uri));
1111
1112 let tls_config = NoteTransportError::Connection(Box::new(std::io::Error::other(
1113 "tls config error: no roots",
1114 )));
1115 assert!(!is_transient_note_transport_error(&tls_config));
1116 }
1117
1118 #[tokio::test]
1119 async fn configured_note_transport_installs_retries_only_when_asked() {
1120 let retried_inner = Arc::new(ScriptedNoteTransportInner {
1121 failures_before_success: AtomicU32::new(1),
1122 fetch_calls: AtomicU32::new(0),
1123 send_calls: AtomicU32::new(0),
1124 error: note_fetch_timeout,
1125 });
1126 let wrapped = configured_note_transport_client(
1127 retried_inner.clone(),
1128 &RpcConfig::new().with_retry_policy(RpcRetryPolicy::new(2)),
1129 );
1130 wrapped
1131 .fetch_notes(&[], NoteTransportCursor::from(0))
1132 .await
1133 .unwrap();
1134 assert_eq!(retried_inner.fetch_calls.load(Ordering::SeqCst), 2);
1135
1136 let passthrough_inner = Arc::new(ScriptedNoteTransportInner {
1137 failures_before_success: AtomicU32::new(1),
1138 fetch_calls: AtomicU32::new(0),
1139 send_calls: AtomicU32::new(0),
1140 error: note_fetch_timeout,
1141 });
1142 let passthrough = configured_note_transport_client(
1143 passthrough_inner.clone(),
1144 &RpcConfig::new().with_retry_policy(RpcRetryPolicy::new(1)),
1145 );
1146 passthrough
1147 .fetch_notes(&[], NoteTransportCursor::from(0))
1148 .await
1149 .unwrap_err();
1150 assert_eq!(passthrough_inner.fetch_calls.load(Ordering::SeqCst), 1);
1151
1152 let send_inner = Arc::new(ScriptedNoteTransportInner {
1153 failures_before_success: AtomicU32::new(10),
1154 fetch_calls: AtomicU32::new(0),
1155 send_calls: AtomicU32::new(0),
1156 error: note_fetch_timeout,
1157 });
1158 let wrapped_send = configured_note_transport_client(
1159 send_inner.clone(),
1160 &RpcConfig::new().with_retry_policy(RpcRetryPolicy::new(5)),
1161 );
1162 wrapped_send
1163 .send_note(test_note_header(), Vec::new())
1164 .await
1165 .unwrap_err();
1166 assert_eq!(send_inner.send_calls.load(Ordering::SeqCst), 1);
1167 }
1168}