From b8cb4d12c303734bd0497ed3de826bcb8b0bc051 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Thu, 3 Sep 2026 22:14:55 +0200 Subject: [PATCH] feat(app): wrap duty callbacks with the retry executor Closes #534. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/node/mod.rs | 1 + crates/app/src/node/wire.rs | 451 +++++++++++++++++++++++++++--------- crates/app/tests/wiring.rs | 145 +++++++++++- 3 files changed, 483 insertions(+), 114 deletions(-) diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 3684fd35..5cb4556a 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -708,6 +708,7 @@ async fn run_lifecycle( parsigdb_deadliner_rx, aggsigdb: _aggsigdb, fetcher: _fetcher, + fetch_duty: _fetch_duty, inclusion_checker, validator_api_router, } = wired; diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index e4dc4c4a..655a1243 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -49,7 +49,10 @@ use pluto_eth2api::{ use pluto_featureset::{Feature, FeatureSet, Status}; use tokio_util::sync::CancellationToken; -use crate::node::AppError; +use crate::{ + node::AppError, + retry::{self, AsyncOptions, DoAsyncError}, +}; /// A `Send + Sync` boxed future. The parsigdb subscriber seams require their /// futures to be `Sync` (see `internal_subscriber`/`threshold_subscriber`), so @@ -136,6 +139,125 @@ where Arc::new(err) } +// --------------------------------------------------------------------------- +// Async retry wrapper layer +// +// Parity: charon `core.WithAsyncRetry` (`core/retry.go`), applied in +// `app.wireCoreWorkflow` as `core.WithAsyncRetry(retry.New(deadlineFunc))` +// alongside `WithTracing`/`WithTracking` (`app/app.go` at v1.7.1). It wraps +// five duty-pipeline callbacks so each one is dispatched on the retry executor +// and returns to its caller immediately: +// +// fetcher.Fetch, consensus.Participate, consensus.Propose, +// parsigex.Broadcast, bcast.Broadcast +// +// Because charon applies the retry option *outermost*, the tracker calls of +// `WithTracking` sit inside the retry loop; Pluto keeps the same nesting by +// wrapping the existing stitch closures (which already report to the tracker) +// rather than the raw component methods. +// --------------------------------------------------------------------------- + +/// A duty callback dispatched onto the async retry executor. +/// +/// Calling it spawns the wrapped work and returns immediately, mirroring +/// charon's `go retryer.DoAsync(...); return nil`. Exposed on +/// [`WiredComponents`] so the scheduler stitch can be driven directly by +/// wiring tests. +pub type DutyCallback = + Arc; + +/// How the retry executor treats a wrapped duty callback's errors. +/// +/// Charon classifies each error at run time: `net.Error`s, context errors and +/// the temporary beacon-node errors matched by `app/retry.isTemporaryBeaconErr` +/// are retried, everything else is permanent. That heuristic reads the beacon +/// node's error message, which Pluto's generated client does not preserve — +/// non-2xx responses are collapsed into message-less typed errors (e.g. any +/// non-200 attestation-data response becomes +/// `FetcherError::NilAttestationData`), so a message-substring port here would +/// never match. +/// +/// Accepted divergence: the policy is fixed per call site instead. The three +/// network-facing callbacks retry (bounded by the duty deadline and the +/// executor's exponential backoff); consensus does not, matching charon's +/// `core/retry.go` note that `ConsensusParticipate`/`ConsensusPropose` "don't +/// require retrying but they should be called async" — a failed QBFT instance +/// must not be re-run for the same duty. +#[derive(Clone, Copy)] +enum RetryPolicy { + /// Retry the call until it succeeds or the duty's deadline elapses. + Retry, + /// Run the call once; failures are logged, not retried. + Once, +} + +/// Builds the [`AsyncOptions`] shared by every wrapped duty callback. +/// +/// Deadlines come from the beacon-derived duty deadline calculator (charon's +/// `deadlineFunc`, passed to `retry.New`), and the executor is cancelled on +/// node shutdown so in-flight retries do not outlive the components they call. +fn retry_options( + deadline_calc: &Arc, + ct: &CancellationToken, +) -> AsyncOptions { + let deadline_calc = Arc::clone(deadline_calc); + AsyncOptions::default() + .with_cancellation_token(ct.clone()) + .with_deadline(move |duty: Duty| match deadline_calc.deadline(&duty) { + Ok(deadline) => deadline, + Err(err) => { + // Charon's `deadlineFunc` returns `(_, false)` for duties + // without a deadline, which leaves the retry bounded only by + // the shutdown context; a calculator failure is treated the + // same way rather than dropping the duty. + tracing::warn!( + ?err, + duty = %duty, + "retry: duty deadline unavailable, retrying until shutdown", + ); + None + } + }) +} + +/// Dispatches `call` onto the async retry executor and returns immediately. +/// +/// Parity: the body of each closure charon installs in `core.WithAsyncRetry`. +fn spawn_retried( + options: AsyncOptions, + duty: Duty, + topic: &'static str, + name: &'static str, + policy: RetryPolicy, + mut call: F, +) where + F: FnMut() -> Fut + Send + 'static, + Fut: Future> + Send + 'static, + E: std::fmt::Display + Send + 'static, +{ + tokio::spawn(retry::do_async(options, duty, topic, name, move || { + let fut = call(); + async move { + fut.await.map_err(|err| { + // `DoAsyncError` carries no payload, so the underlying + // error is logged here before it is classified. + tracing::warn!(%err, topic, name, "duty callback failed"); + match policy { + RetryPolicy::Retry => DoAsyncError::RetryableError, + RetryPolicy::Once => DoAsyncError::NonRetryableError, + } + }) + } + })); +} + +/// Returns the immediate `Ok` a wrapped callback hands back to its caller: the +/// real work has been dispatched onto the retry executor, so — exactly as in +/// charon's `WithAsyncRetry` closures — the caller is never told it failed. +fn dispatched() -> std::future::Ready> { + std::future::ready(Ok(())) +} + /// Wraps a [`DeadlineCalculator`], shifting every deadline later by a fixed /// offset. Used to derive the tracker's analyser/deleter deadlines from the /// shared duty deadline. Parity: the closures charon builds in `newTracker`. @@ -314,6 +436,10 @@ pub struct WiredComponents { pub aggsigdb: MemoryDBHandle, /// The fetcher (driven via scheduler subscriptions). pub fetcher: Arc, + /// The retry-wrapped `fetcher.fetch` duty callback the scheduler drives. + /// Returned so wiring tests can exercise the same wrapped callback the + /// scheduler does, without standing up a live duty round. + pub fetch_duty: DutyCallback, /// Networked inclusion checker; its `run` loop is spawned and supervised by /// the caller. pub inclusion_checker: Arc, @@ -572,45 +698,67 @@ pub async fn wire_core_workflow( }) }; // Stitch: fetcher.subscribe(consensus.propose), bounded by the duty - // deadline. + // deadline and dispatched onto the retry executor (charon + // `WithAsyncRetry`: `consensus`/`propose`, async but not retried). let fetch_subscriber: Subscriber = { let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); let tracker = Arc::clone(&tracker); + let retry_opts = retry_options(&deadline_calc, &ct); Arc::new(move |duty: Duty, set: UnsignedDataSet| { let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); let tracker = Arc::clone(&tracker); - Box::pin(async move { - let pubkeys: Vec = set.keys().copied().collect(); - let value = unsigneddata::unsigned_data_set_to_proto(&set)?; - // Bound consensus by the duty deadline so a stuck instance is - // cancelled (-> ConsensusTimeout) instead of running until - // shutdown. - let result = run_bounded_by_duty_deadline( - &deadline_calc, - &ct, - duty.clone(), - move |duty, dct| async move { consensus.propose(dct, duty, value).await }, - ) - .await; - - match result { - Ok(()) => { - tracker.consensus_proposed(duty, &pubkeys, None).await; - Ok(()) - } - Err(err) => { - let (reported, returned) = share_step_err(err); - tracker - .consensus_proposed(duty, &pubkeys, Some(reported)) + spawn_retried( + retry_opts.clone(), + duty.clone(), + "consensus", + "propose", + RetryPolicy::Once, + move || { + let consensus = Arc::clone(&consensus); + let ct = ct.clone(); + let deadline_calc = Arc::clone(&deadline_calc); + let tracker = Arc::clone(&tracker); + let duty = duty.clone(); + let set = set.clone(); + async move { + let pubkeys: Vec = set.keys().copied().collect(); + let value = unsigneddata::unsigned_data_set_to_proto(&set) + .map_err(|err| SharedStepError(owned_step_err(err)))?; + // Bound consensus by the duty deadline so a stuck + // instance is cancelled (-> ConsensusTimeout) instead + // of running until shutdown. + let result = + run_bounded_by_duty_deadline( + &deadline_calc, + &ct, + duty.clone(), + move |duty, dct| async move { + consensus.propose(dct, duty, value).await + }, + ) .await; - Err(returned.into()) + + match result { + Ok(()) => { + tracker.consensus_proposed(duty, &pubkeys, None).await; + Ok(()) + } + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .consensus_proposed(duty, &pubkeys, Some(reported)) + .await; + Err(returned) + } + } } - } - }) + }, + ); + Box::pin(dispatched()) }) }; @@ -671,35 +819,46 @@ pub async fn wire_core_workflow( parsigdb_deadliner, )); - // Stitch: parsigdb.subscribe_internal(parsigex.broadcast). + // Stitch: parsigdb.subscribe_internal(parsigex.broadcast), dispatched onto + // the retry executor (charon `WithAsyncRetry`: `parsigex`/`broadcast`). { let broadcast = Arc::clone(&parsigex.broadcast); let tracker = Arc::clone(&tracker); + let retry_opts = retry_options(&deadline_calc, &ct); parsigdb .subscribe_internal(parsigdb::memory::internal_subscriber( move |duty: Duty, set: ParSignedDataSet| { let broadcast = Arc::clone(&broadcast); let tracker = Arc::clone(&tracker); - async move { - match broadcast(duty.clone(), set.clone()).await { - Ok(()) => { - tracker.par_sig_ex_broadcasted(duty, &set, None).await; - Ok(()) - } - Err(err) => { - let (reported, returned) = share_step_err(err); - tracker - .par_sig_ex_broadcasted(duty, &set, Some(reported)) - .await; - Err( - parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { - source: Box::new(returned), + spawn_retried( + retry_opts.clone(), + duty.clone(), + "parsigex", + "broadcast", + RetryPolicy::Retry, + move || { + let broadcast = Arc::clone(&broadcast); + let tracker = Arc::clone(&tracker); + let duty = duty.clone(); + let set = set.clone(); + async move { + match broadcast(duty.clone(), set.clone()).await { + Ok(()) => { + tracker.par_sig_ex_broadcasted(duty, &set, None).await; + Ok(()) } - .into(), - ) + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .par_sig_ex_broadcasted(duty, &set, Some(reported)) + .await; + Err(returned) + } + } } - } - } + }, + ); + dispatched() }, )) .await; @@ -744,45 +903,65 @@ pub async fn wire_core_workflow( .await .map_err(AppError::Broadcaster)?, ); - // Stitch: sigagg.subscribe(broadcaster.broadcast). + // Stitch: sigagg.subscribe(broadcaster.broadcast), dispatched onto the + // retry executor (charon `WithAsyncRetry`: `bcast`/`broadcast`). { let broadcaster = Arc::clone(&broadcaster); let tracker = Arc::clone(&tracker); let inclusion = Arc::clone(&inclusion_checker); + let retry_opts = retry_options(&deadline_calc, &ct); aggregator.subscribe(Arc::new(move |duty: &Duty, set: &SignedDataSet| { let broadcaster = Arc::clone(&broadcaster); let tracker = Arc::clone(&tracker); let inclusion = Arc::clone(&inclusion); let duty = duty.clone(); let set = set.clone(); - Box::pin(async move { - let pubkeys: Vec = set.keys().copied().collect(); + spawn_retried( + retry_opts.clone(), + duty.clone(), + "bcast", + "broadcast", + RetryPolicy::Retry, + move || { + let broadcaster = Arc::clone(&broadcaster); + let tracker = Arc::clone(&tracker); + let inclusion = Arc::clone(&inclusion); + let duty = duty.clone(); + let set = set.clone(); + async move { + let pubkeys: Vec = set.keys().copied().collect(); - // Register for inclusion checking before broadcasting, and even - // if the broadcast fails: peers may still succeed, so the duty - // can land on-chain regardless. Parity: charon - // `core/tracking.go` `BroadcasterBroadcast`. - if let Err(err) = inclusion.submitted(&duty, &set) { - tracing::error!( - ?err, - duty = %duty, - "Internal error: failed to submit duty to inclusion checker. \ - This indicates a tracking bug that should be reported", - ); - } + // Register for inclusion checking before broadcasting, + // and even if the broadcast fails: peers may still + // succeed, so the duty can land on-chain regardless. + // Parity: charon `core/tracking.go` + // `BroadcasterBroadcast`, which sits inside the retry + // loop for the same reason. + if let Err(err) = inclusion.submitted(&duty, &set) { + tracing::error!( + ?err, + duty = %duty, + "Internal error: failed to submit duty to inclusion checker. \ + This indicates a tracking bug that should be reported", + ); + } - let step_err = match broadcaster.broadcast(duty.clone(), set).await { - Ok(()) => None, - Err(err) => { - tracing::warn!(?err, "broadcaster: broadcast"); - Some(owned_step_err(err)) + let result = broadcaster.broadcast(duty.clone(), set).await; + let (step_err, returned) = match result { + Ok(()) => (None, Ok(())), + Err(err) => { + let (reported, returned) = share_step_err(err); + (Some(reported), Err(returned)) + } + }; + tracker + .broadcaster_broadcast(duty, &pubkeys, step_err) + .await; + returned } - }; - tracker - .broadcaster_broadcast(duty, &pubkeys, step_err) - .await; - Ok(()) - }) + }, + ); + Box::pin(dispatched()) })); } let aggregator = Arc::new(aggregator); @@ -873,67 +1052,112 @@ pub async fn wire_core_workflow( // scheduler.subscribe_duty(consensus.participate), registered on the // builder before `.build()` (which blocks until chain start + sync). let mut sched_builder = SchedulerBuilder::new(); - { + // Stitch: scheduler.subscribe_duty(fetcher.fetch), dispatched onto the + // retry executor (charon `WithAsyncRetry`: `fetcher`/`fetch` — the one + // callback that is genuinely retried, so a transient beacon-node failure + // no longer drops the duty on this node). + let fetch_duty: DutyCallback = { let fetcher = Arc::clone(&fetcher); let ct = ct.clone(); let tracker = Arc::clone(&tracker); - sched_builder.subscribe_duty( - move |duty: &Duty, set: &pluto_core::types::DutyDefinitionSet| { + let retry_opts = retry_options(&deadline_calc, &ct); + Arc::new( + move |duty: Duty, set: pluto_core::types::DutyDefinitionSet| { let fetcher = Arc::clone(&fetcher); let ct = ct.clone(); let tracker = Arc::clone(&tracker); - let duty = duty.clone(); - let set = set.clone(); - async move { - let pubkeys: Vec = set.keys().copied().collect(); - match fetcher.fetch(duty.clone(), set).await { - // In-flight fetches racing shutdown fail against already - // terminated components (e.g. the aggsigdb back-edge); - // don't surface those as duty errors. - Err(err) if ct.is_cancelled() => { - tracing::debug!(?err, "fetch aborted by shutdown"); - Ok(()) - } - Ok(()) => { - tracker.fetcher_fetched(duty, &pubkeys, None).await; - Ok(()) - } - Err(err) => { - let (reported, returned) = share_step_err(err); - tracker - .fetcher_fetched(duty, &pubkeys, Some(reported)) - .await; - // `subscribe_duty` is generic over the error type, - // so the shared wrapper - // propagates as-is. - Err(returned) + spawn_retried( + retry_opts.clone(), + duty.clone(), + "fetcher", + "fetch", + RetryPolicy::Retry, + move || { + let fetcher = Arc::clone(&fetcher); + let ct = ct.clone(); + let tracker = Arc::clone(&tracker); + let duty = duty.clone(); + let set = set.clone(); + async move { + let pubkeys: Vec = set.keys().copied().collect(); + match fetcher.fetch(duty.clone(), set).await { + // In-flight fetches racing shutdown fail against + // already terminated components (e.g. the + // aggsigdb back-edge); don't surface those as + // duty errors. + Err(err) if ct.is_cancelled() => { + tracing::debug!(?err, "fetch aborted by shutdown"); + Ok(()) + } + Ok(()) => { + tracker.fetcher_fetched(duty, &pubkeys, None).await; + Ok(()) + } + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .fetcher_fetched(duty, &pubkeys, Some(reported)) + .await; + Err(returned) + } + } } - } - } + }, + ); + }, + ) + }; + { + let fetch_duty = Arc::clone(&fetch_duty); + sched_builder.subscribe_duty( + move |duty: &Duty, set: &pluto_core::types::DutyDefinitionSet| { + fetch_duty(duty.clone(), set.clone()); + dispatched::() }, "fetcher", ); } + // Stitch: scheduler.subscribe_duty(consensus.participate), dispatched onto + // the retry executor (charon `WithAsyncRetry`: `consensus`/`participate`, + // async but not retried). { let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); + let retry_opts = retry_options(&deadline_calc, &ct); sched_builder.subscribe_duty( move |duty: &Duty, _set: &pluto_core::types::DutyDefinitionSet| { let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); let duty = duty.clone(); - async move { - // Bound consensus by the duty deadline (see fetch stitch). - run_bounded_by_duty_deadline( - &deadline_calc, - &ct, - duty, - move |duty, dct| async move { consensus.participate(dct, duty).await }, - ) - .await - } + spawn_retried( + retry_opts.clone(), + duty.clone(), + "consensus", + "participate", + RetryPolicy::Once, + move || { + let consensus = Arc::clone(&consensus); + let ct = ct.clone(); + let deadline_calc = Arc::clone(&deadline_calc); + let duty = duty.clone(); + async move { + // Bound consensus by the duty deadline (see fetch + // stitch). + run_bounded_by_duty_deadline( + &deadline_calc, + &ct, + duty, + move |duty, dct| async move { + consensus.participate(dct, duty).await + }, + ) + .await + } + }, + ); + dispatched::() }, "consensus", ); @@ -1105,6 +1329,7 @@ pub async fn wire_core_workflow( parsigdb_deadliner_rx, aggsigdb, fetcher, + fetch_duty, inclusion_checker, validator_api_router, }) diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index f22ad2d8..92270efe 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -27,7 +27,11 @@ //! the wired components directly to prove the three back-edges / sign-path are //! connected. -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex as StdMutex}, + time::Duration, +}; use pluto_app::node::{ AppError, @@ -950,3 +954,142 @@ async fn multinode_parsig_exchange_reaches_submission() { ct.cancel(); } + +/// Consensus stand-in that records every `propose` call, so a test can observe +/// that a fetched duty reached the fetcher's subscriber. +#[derive(Default)] +struct RecordingConsensus { + proposed: Arc>>, +} + +impl pluto_consensus::wrapper::Consensus for RecordingConsensus { + fn protocol_id(&self) -> String { + "/pluto/test/consensus/1.0.0".to_string() + } + + fn start(&self, _ct: CancellationToken) {} + + fn participate( + &self, + _ct: CancellationToken, + _duty: Duty, + ) -> futures::future::BoxFuture<'_, pluto_consensus::wrapper::Result<()>> { + Box::pin(async { Ok(()) }) + } + + fn propose( + &self, + _ct: CancellationToken, + duty: Duty, + _value: pluto_core::corepb::v1::core::UnsignedDataSet, + ) -> futures::future::BoxFuture<'_, pluto_consensus::wrapper::Result<()>> { + self.proposed + .lock() + .expect("proposed lock") + .push(duty.clone()); + Box::pin(async { Ok(()) }) + } + + fn subscribe(&self, _subscriber: pluto_consensus::wrapper::Subscriber) {} +} + +/// (g) The duty callbacks are wrapped with the async retry executor: a +/// transient beacon-node failure on `fetcher.fetch` is retried and the duty +/// still completes (reaching the fetcher's `consensus.propose` subscriber). +/// +/// Parity: charon `core.WithAsyncRetry` (`core/retry.go`), wired in +/// `app.wireCoreWorkflow` as `core.WithAsyncRetry(retry.New(deadlineFunc))`. +/// Without the wrapper the scheduler stitch calls `fetch` exactly once, the +/// first failure drops the duty on this node, and no proposal is ever made. +#[tokio::test] +async fn retry_wrapper_recovers_transient_fetch_failure() { + const ATT_DATA_PATH: &str = "/eth/v1/validator/attestation_data"; + const SLOT: u64 = 1; + + let ct = CancellationToken::new(); + let mock = BeaconMock::builder().build().await.expect("beacon mock"); + + // One transient 503 ("beacon node is currently syncing, try again later") + // ahead of the mock's default 200. `with_priority(1)` outranks the default + // routes, `up_to_n_times(1)` retires it after the first request, so the + // retry falls through to the healthy response. + Mock::given(method("GET")) + .and(path(ATT_DATA_PATH)) + .respond_with(ResponseTemplate::new(503).set_body_string( + r#"{"code":503,"message":"Beacon node is currently syncing, try again later"}"#, + )) + .with_priority(1) + .up_to_n_times(1) + .mount(mock.server()) + .await; + + let eth2_cl = mock.client().clone(); + let pubkey = PubKey::new([9u8; PK_LEN]); + let proposed: Arc>> = Arc::default(); + let consensus = Arc::new(ConsensusWrapper::new(Arc::new(RecordingConsensus { + proposed: Arc::clone(&proposed), + }))); + + let wired = tokio::time::timeout( + GUARD, + wire_core_workflow(wire_inputs(eth2_cl, pubkey, consensus, 1), ct.clone()), + ) + .await + .expect("wire did not deadlock") + .expect("wire succeeded"); + + let def = DutyDefinitionSet::from([( + pubkey, + DutyDefinition::Attester(pluto_core::types::AttesterDutyDefinition { + pubkey, + duty: pluto_core::signeddata::AttesterDuty { + slot: SLOT, + validator_index: 2, + committee_index: 0, + committee_length: 8, + committees_at_slot: 1, + validator_committee_index: 0, + }, + }), + )]); + let duty = Duty::new_attester_duty(SlotNumber::new(SLOT)); + + // Drive the same retry-wrapped callback the scheduler drives. It is + // fire-and-forget (charon's `go retryer.DoAsync(...); return nil`), so the + // outcome is observed through the recording consensus. + (wired.fetch_duty)(duty.clone(), def); + + tokio::time::timeout(GUARD, async { + loop { + if !proposed.lock().expect("proposed lock").is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("(g) the retried fetch should complete and reach consensus.propose"); + + assert_eq!( + proposed.lock().expect("proposed lock").as_slice(), + &[duty], + "(g) the duty must complete exactly once after the retry", + ); + assert!( + count_gets(mock.server(), ATT_DATA_PATH).await >= 2, + "(g) the transient failure must have been retried", + ); + + ct.cancel(); +} + +/// Counts GETs the mock has received for `request_path`. +async fn count_gets(server: &MockServer, request_path: &str) -> usize { + server + .received_requests() + .await + .expect("requests") + .into_iter() + .filter(|r| r.method.as_str() == "GET" && r.url.path() == request_path) + .count() +}