rama_net/client/proxy/route/
failure_cache.rs

1use core::{
2    fmt,
3    sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
4    time::Duration,
5};
6use std::sync::Arc;
7
8use moka::{policy::EvictionPolicy, sync::Cache};
9use rama_core::{
10    Layer, Service,
11    error::{BoxError, BoxErrorExt as _},
12    extensions::ExtensionsRef,
13    telemetry::tracing,
14};
15use rama_utils::{macros::define_inner_service_accessors, time::now_monotonic_nanos};
16
17use crate::client::{
18    ConnectionError, ConnectionErrorDomain, ConnectionErrorKind, ConnectorService,
19    EstablishedClientConnection,
20};
21use crate::{
22    AuthorityInputExt, Protocol, ProtocolInputExt, address::HostWithPort, user::ProxyCredential,
23};
24
25use super::ProxyRoute;
26
27/// Scope used to identify temporarily failing proxy routes.
28#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
29#[non_exhaustive]
30pub enum ProxyRouteFailureCacheScope {
31    /// Keep failures isolated per proxy route and final destination.
32    #[default]
33    PerDestination,
34    /// Share failures for a proxy route across every final destination.
35    PerProxy,
36}
37
38/// Configuration for a [`ProxyRouteFailureCache`].
39#[derive(Debug, Clone)]
40#[non_exhaustive]
41pub struct ProxyRouteFailureCacheConfig {
42    /// Duration of the first temporary block. Defaults to one minute.
43    pub initial_backoff: Duration,
44    /// Maximum temporary block after repeated failures. Defaults to 30 minutes.
45    pub max_backoff: Duration,
46    /// Maximum time reserved for one half-open health probe. Defaults to 30 seconds.
47    pub probe_lease: Duration,
48    /// Maximum number of remembered route keys. Defaults to 1,024.
49    pub max_entries: u64,
50    /// Whether failures are isolated by final destination. Defaults to
51    /// [`ProxyRouteFailureCacheScope::PerDestination`].
52    pub scope: ProxyRouteFailureCacheScope,
53}
54
55impl Default for ProxyRouteFailureCacheConfig {
56    fn default() -> Self {
57        Self {
58            initial_backoff: Duration::from_secs(60),
59            max_backoff: Duration::from_mins(30),
60            probe_lease: Duration::from_secs(30),
61            max_entries: 1_024,
62            scope: ProxyRouteFailureCacheScope::PerDestination,
63        }
64    }
65}
66
67impl ProxyRouteFailureCacheConfig {
68    fn validate(&self) -> Result<(), BoxError> {
69        if self.initial_backoff.is_zero() {
70            return Err(BoxError::from_static_str(
71                "proxy route failure cache initial backoff must be non-zero",
72            ));
73        }
74        if self.max_backoff < self.initial_backoff {
75            return Err(BoxError::from_static_str(
76                "proxy route failure cache max backoff must not be smaller than its initial backoff",
77            ));
78        }
79        if self.probe_lease.is_zero() {
80            return Err(BoxError::from_static_str(
81                "proxy route failure cache probe lease must be non-zero",
82            ));
83        }
84        if self.max_entries == 0 {
85            return Err(BoxError::from_static_str(
86                "proxy route failure cache capacity must be non-zero",
87            ));
88        }
89        Ok(())
90    }
91
92    fn backoff(&self, previous_failures: u32) -> Duration {
93        let multiplier = 1u32
94            .checked_shl(previous_failures.min(31))
95            .unwrap_or(u32::MAX);
96        self.initial_backoff
97            .saturating_mul(multiplier)
98            .min(self.max_backoff)
99    }
100}
101
102#[derive(Clone, PartialEq, Eq, Hash)]
103struct FailureCacheKey {
104    protocol: Option<Protocol>,
105    proxy: HostWithPort,
106    basic_username: Option<String>,
107    bearer_credential: bool,
108    destination_protocol: Option<Protocol>,
109    destination: Option<HostWithPort>,
110}
111
112type SharedFailureCacheKey = Arc<FailureCacheKey>;
113
114#[derive(Default)]
115struct FailureEntry {
116    blocked_until: AtomicU64,
117    probe_until: AtomicU64,
118    failure_count: AtomicU32,
119    active_attempts: AtomicU32,
120    succeeded: AtomicBool,
121}
122
123impl FailureEntry {
124    fn mark_live(&self) {
125        // Publish success before clearing the failure state. A racing failure
126        // either observes it before updating the deadline or clears its update
127        // in the post-CAS success check.
128        self.succeeded.store(true, Ordering::Release);
129        self.blocked_until.store(0, Ordering::Release);
130        self.probe_until.store(0, Ordering::Release);
131        self.failure_count.store(0, Ordering::Release);
132    }
133}
134
135struct AttemptPermit {
136    entries: Arc<Cache<SharedFailureCacheKey, Arc<FailureEntry>>>,
137    key: SharedFailureCacheKey,
138    entry: Arc<FailureEntry>,
139    started_time: u64,
140    probe_lease: Option<u64>,
141    remove_on_drop: bool,
142}
143
144impl AttemptPermit {
145    fn release_probe(&mut self) {
146        if let Some(lease) = self.probe_lease.take() {
147            let _release_result = self.entry.probe_until.compare_exchange(
148                lease,
149                0,
150                Ordering::AcqRel,
151                Ordering::Acquire,
152            );
153        }
154    }
155
156    fn mark_live(&mut self) {
157        self.entry.mark_live();
158        self.release_probe();
159        self.entries.invalidate(&self.key);
160        self.remove_on_drop = false;
161    }
162}
163
164impl Drop for AttemptPermit {
165    fn drop(&mut self) {
166        self.release_probe();
167        let previous_attempts = self.entry.active_attempts.fetch_sub(1, Ordering::AcqRel);
168        debug_assert!(previous_attempts > 0);
169        if self.remove_on_drop
170            && previous_attempts == 1
171            && self.entry.failure_count.load(Ordering::Acquire) == 0
172        {
173            self.entries.invalidate(&self.key);
174        }
175    }
176}
177
178enum CacheDecision {
179    Attempt(AttemptPermit),
180    Blocked(Duration),
181}
182
183/// Shared negative cache for temporarily failing proxy routes.
184///
185/// The cache is bounded and safe to clone across connector services. Healthy
186/// routes create only transient state while an attempt is in flight; a success
187/// or non-cacheable failure removes it, so retained entries represent negative
188/// backoff state. State transitions for one route key use atomics and do not
189/// contend with unrelated proxy routes or destinations. After a backoff expires,
190/// at most one caller receives a half-open probe permit for that key.
191///
192/// Keys contain the proxy protocol and address, an optional Basic routing
193/// username, and optionally the final destination protocol and address. Basic
194/// passwords and bearer tokens are never retained in a key.
195#[derive(Clone)]
196pub struct ProxyRouteFailureCache {
197    entries: Arc<Cache<SharedFailureCacheKey, Arc<FailureEntry>>>,
198    config: Arc<ProxyRouteFailureCacheConfig>,
199}
200
201impl fmt::Debug for ProxyRouteFailureCache {
202    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
203        f.debug_struct("ProxyRouteFailureCache")
204            .field("config", &self.config)
205            .field("entry_count", &self.entries.entry_count())
206            .finish()
207    }
208}
209
210impl Default for ProxyRouteFailureCache {
211    fn default() -> Self {
212        Self::build(ProxyRouteFailureCacheConfig::default())
213    }
214}
215
216impl ProxyRouteFailureCache {
217    /// Create a failure cache using the given validated configuration.
218    pub fn try_new(config: ProxyRouteFailureCacheConfig) -> Result<Self, BoxError> {
219        config.validate()?;
220        Ok(Self::build(config))
221    }
222
223    fn build(config: ProxyRouteFailureCacheConfig) -> Self {
224        let idle = config.max_backoff.saturating_mul(2);
225        let entries = Arc::new(
226            Cache::builder()
227                .max_capacity(config.max_entries)
228                .initial_capacity(config.max_entries.min(128) as usize)
229                .eviction_policy(EvictionPolicy::lru())
230                .time_to_idle(idle)
231                .build(),
232        );
233        Self {
234            entries,
235            config: Arc::new(config),
236        }
237    }
238
239    fn begin<Input>(&self, input: &Input) -> Option<CacheDecision>
240    where
241        Input: AuthorityInputExt + ExtensionsRef + ProtocolInputExt,
242    {
243        let route = input.extensions().get_arc::<ProxyRoute>()?;
244        let ProxyRoute::Proxy(proxy) = route.as_ref() else {
245            return None;
246        };
247        let (destination_protocol, destination) = match self.config.scope {
248            ProxyRouteFailureCacheScope::PerDestination => (
249                input.protocol(),
250                Some(
251                    input
252                        .authority()?
253                        .into_host_with_port(input.protocol_default_port())?,
254                ),
255            ),
256            ProxyRouteFailureCacheScope::PerProxy => (None, None),
257        };
258        let (basic_username, bearer_credential) = match proxy.credential.as_ref() {
259            Some(ProxyCredential::Basic(basic)) => (Some(basic.username()), false),
260            Some(ProxyCredential::Bearer(_)) => (None, true),
261            None => (None, false),
262        };
263        let key = Arc::new(FailureCacheKey {
264            protocol: proxy.protocol.clone(),
265            proxy: proxy.address.clone(),
266            basic_username: basic_username.map(ToOwned::to_owned),
267            bearer_credential,
268            destination_protocol: destination_protocol.cloned(),
269            destination,
270        });
271        // Install transient state before starting the first attempt so a
272        // concurrent success can publish its completion before an older
273        // failing attempt tries to block the route. Successful state is
274        // invalidated instead of being retained in the negative cache.
275        let entry = self
276            .entries
277            .get_with(key.clone(), || Arc::new(FailureEntry::default()));
278
279        let failure_count = entry.failure_count.load(Ordering::Acquire);
280        let blocked_until = entry.blocked_until.load(Ordering::Acquire);
281        if failure_count == 0 && blocked_until == 0 {
282            entry.active_attempts.fetch_add(1, Ordering::AcqRel);
283            return Some(CacheDecision::Attempt(AttemptPermit {
284                entries: self.entries.clone(),
285                key,
286                started_time: now_monotonic_nanos(),
287                entry,
288                probe_lease: None,
289                remove_on_drop: true,
290            }));
291        }
292
293        let mut now = now_monotonic_nanos();
294        if let Some(remaining) = remaining_duration(blocked_until, now) {
295            return Some(CacheDecision::Blocked(remaining));
296        }
297
298        let lease_duration = duration_nanos(self.config.probe_lease);
299        loop {
300            now = now_monotonic_nanos();
301            let probe_until = entry.probe_until.load(Ordering::Acquire);
302            if let Some(remaining) = remaining_duration(probe_until, now) {
303                return Some(CacheDecision::Blocked(remaining));
304            }
305            let new_probe_until = now.saturating_add(lease_duration);
306            if entry
307                .probe_until
308                .compare_exchange(
309                    probe_until,
310                    new_probe_until,
311                    Ordering::AcqRel,
312                    Ordering::Acquire,
313                )
314                .is_ok()
315            {
316                entry.active_attempts.fetch_add(1, Ordering::AcqRel);
317                return Some(CacheDecision::Attempt(AttemptPermit {
318                    entries: self.entries.clone(),
319                    key,
320                    started_time: now_monotonic_nanos(),
321                    entry,
322                    probe_lease: Some(new_probe_until),
323                    remove_on_drop: false,
324                }));
325            }
326        }
327    }
328
329    fn mark_failure(&self, permit: &mut AttemptPermit) {
330        let entry = &permit.entry;
331        let started_time = permit.started_time;
332        loop {
333            if entry.succeeded.load(Ordering::Acquire) {
334                break;
335            }
336
337            let current_deadline = entry.blocked_until.load(Ordering::Acquire);
338            if current_deadline > started_time {
339                break;
340            }
341
342            let previous_failures = entry.failure_count.load(Ordering::Relaxed);
343            let new_deadline = now_monotonic_nanos()
344                .saturating_add(duration_nanos(self.config.backoff(previous_failures)));
345            if entry
346                .blocked_until
347                .compare_exchange(
348                    current_deadline,
349                    new_deadline,
350                    Ordering::AcqRel,
351                    Ordering::Acquire,
352                )
353                .is_ok()
354            {
355                entry
356                    .failure_count
357                    .store(previous_failures.saturating_add(1), Ordering::Release);
358                if entry.succeeded.load(Ordering::Acquire) {
359                    entry.mark_live();
360                }
361                break;
362            }
363        }
364        permit.remove_on_drop = false;
365        permit.release_probe();
366    }
367
368    /// Return the configured cache policy.
369    #[must_use]
370    pub fn config(&self) -> &ProxyRouteFailureCacheConfig {
371        &self.config
372    }
373
374    /// Return an approximate number of retained cache entries.
375    #[must_use]
376    pub fn entry_count(&self) -> u64 {
377        self.entries.entry_count()
378    }
379
380    /// Clear every retained proxy failure and backoff state.
381    pub fn invalidate_all(&self) {
382        self.entries.invalidate_all();
383    }
384}
385
386fn duration_nanos(duration: Duration) -> u64 {
387    duration.as_nanos().try_into().unwrap_or(u64::MAX)
388}
389
390fn remaining_duration(deadline: u64, now: u64) -> Option<Duration> {
391    deadline
392        .checked_sub(now)
393        .filter(|remaining| *remaining != 0)
394        .map(Duration::from_nanos)
395}
396
397fn should_cache_failure(error: &ConnectionError) -> bool {
398    error.domain() == ConnectionErrorDomain::Transport
399        && matches!(
400            error.kind(),
401            ConnectionErrorKind::Unavailable
402                | ConnectionErrorKind::Timeout
403                | ConnectionErrorKind::Protocol
404        )
405}
406
407/// Error returned when a proxy route is still inside its failure backoff.
408#[derive(Debug)]
409pub struct ProxyRouteFailureCachedError {
410    retry_after: Duration,
411}
412
413impl ProxyRouteFailureCachedError {
414    /// Remaining duration before this route can be probed again.
415    #[must_use]
416    pub const fn retry_after(&self) -> Duration {
417        self.retry_after
418    }
419}
420
421impl fmt::Display for ProxyRouteFailureCachedError {
422    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
423        write!(
424            f,
425            "proxy route is temporarily blocked after a connection failure (retry in {:?})",
426            self.retry_after
427        )
428    }
429}
430
431impl core::error::Error for ProxyRouteFailureCachedError {}
432
433/// Connector wrapper that applies a shared [`ProxyRouteFailureCache`].
434///
435/// This service reads only a singular selected [`ProxyRoute`]. Place it inside
436/// a connection pool so an existing pooled connection is considered before a
437/// new connection is suppressed by the negative cache.
438///
439/// Transport `Unavailable`, `Timeout`, and `Protocol` failures are cached.
440/// A success or any non-cacheable failure clears existing backoff state so the
441/// next request can try the route again. Direct and plural routes are ignored.
442#[derive(Debug, Clone)]
443pub struct ProxyRouteFailureCacheConnector<S> {
444    inner: S,
445    cache: ProxyRouteFailureCache,
446}
447
448impl<S> ProxyRouteFailureCacheConnector<S> {
449    /// Wrap a connector with a shared proxy route failure cache.
450    #[must_use]
451    pub const fn new(inner: S, cache: ProxyRouteFailureCache) -> Self {
452        Self { inner, cache }
453    }
454
455    /// Return the configured cache.
456    #[must_use]
457    pub const fn cache(&self) -> &ProxyRouteFailureCache {
458        &self.cache
459    }
460
461    define_inner_service_accessors!();
462}
463
464impl<S, Input> Service<Input> for ProxyRouteFailureCacheConnector<S>
465where
466    S: ConnectorService<Input>,
467    Input: AuthorityInputExt + ExtensionsRef + ProtocolInputExt + Send + 'static,
468{
469    type Output = EstablishedClientConnection<S::Connection, Input>;
470    type Error = ConnectionError;
471
472    async fn serve(&self, input: Input) -> Result<Self::Output, Self::Error> {
473        let cache = &self.cache;
474        let mut permit = match cache.begin(&input) {
475            None => return self.inner.connect(input).await,
476            Some(CacheDecision::Attempt(permit)) => permit,
477            Some(CacheDecision::Blocked(retry_after)) => {
478                tracing::debug!(?retry_after, "skip temporarily failing proxy route",);
479                return Err(ConnectionError::transport(
480                    ProxyRouteFailureCachedError { retry_after },
481                    ConnectionErrorKind::Unavailable,
482                ));
483            }
484        };
485
486        match self.inner.connect(input).await {
487            Ok(established) => {
488                permit.mark_live();
489                Ok(established)
490            }
491            Err(error) if should_cache_failure(&error) => {
492                cache.mark_failure(&mut permit);
493                Err(error)
494            }
495            Err(error) => {
496                permit.mark_live();
497                Err(error)
498            }
499        }
500    }
501}
502
503/// Layer that applies a shared [`ProxyRouteFailureCache`] to a connector.
504#[derive(Debug, Clone)]
505pub struct ProxyRouteFailureCacheLayer {
506    cache: ProxyRouteFailureCache,
507}
508
509impl ProxyRouteFailureCacheLayer {
510    /// Create a layer using the given shared failure cache.
511    #[must_use]
512    pub const fn new(cache: ProxyRouteFailureCache) -> Self {
513        Self { cache }
514    }
515
516    /// Return the shared failure cache.
517    #[must_use]
518    pub const fn cache(&self) -> &ProxyRouteFailureCache {
519        &self.cache
520    }
521}
522
523impl<S> Layer<S> for ProxyRouteFailureCacheLayer {
524    type Service = ProxyRouteFailureCacheConnector<S>;
525
526    fn layer(&self, inner: S) -> Self::Service {
527        ProxyRouteFailureCacheConnector::new(inner, self.cache.clone())
528    }
529
530    fn into_layer(self, inner: S) -> Self::Service {
531        ProxyRouteFailureCacheConnector::new(inner, self.cache)
532    }
533}
534
535#[cfg(test)]
536mod tests {
537    use core::future::Future;
538    use core::sync::atomic::{AtomicUsize, Ordering};
539    use std::{sync::Arc, time::Duration};
540
541    use rama_core::{ServiceInput, service::service_fn};
542    use tokio::sync::{Barrier, Notify};
543
544    use crate::client::{ConnectRequest, ProxyRoute, ProxyRoutes};
545
546    use super::*;
547
548    fn cache(scope: ProxyRouteFailureCacheScope) -> ProxyRouteFailureCache {
549        ProxyRouteFailureCache::try_new(ProxyRouteFailureCacheConfig {
550            initial_backoff: Duration::from_millis(20),
551            max_backoff: Duration::from_millis(80),
552            probe_lease: Duration::from_secs(1),
553            max_entries: 32,
554            scope,
555        })
556        .unwrap()
557    }
558
559    fn proxy(username: Option<&str>) -> ProxyRoute {
560        let address = match username {
561            Some(username) => format!("http://{username}:secret@proxy.example:8080"),
562            None => "http://proxy.example:8080".to_owned(),
563        };
564        ProxyRoute::Proxy(address.parse().unwrap())
565    }
566
567    fn request(destination: &str, route: ProxyRoute) -> ConnectRequest {
568        let request = ConnectRequest::new(destination.parse().unwrap());
569        request.extensions.insert(route);
570        request
571    }
572
573    fn unavailable() -> ConnectionError {
574        ConnectionError::transport(
575            BoxError::from_static_str("proxy unavailable"),
576            ConnectionErrorKind::Unavailable,
577        )
578    }
579
580    async fn within_test_timeout<F: Future>(future: F) -> F::Output {
581        tokio::time::timeout(Duration::from_secs(5), future)
582            .await
583            .expect("concurrent failure-cache test operation should complete")
584    }
585
586    fn begin_attempt(
587        failure_cache: &ProxyRouteFailureCache,
588        request: &ConnectRequest,
589    ) -> AttemptPermit {
590        match failure_cache.begin(request) {
591            Some(CacheDecision::Attempt(permit)) => permit,
592            Some(CacheDecision::Blocked(_)) => panic!("route was unexpectedly blocked"),
593            None => panic!("proxy route was unexpectedly ignored"),
594        }
595    }
596
597    #[tokio::test]
598    async fn repeated_failure_is_suppressed_for_same_destination() {
599        let attempts = Arc::new(AtomicUsize::new(0));
600        let inner = service_fn({
601            let attempts = attempts.clone();
602            move |_input: ConnectRequest| {
603                attempts.fetch_add(1, Ordering::SeqCst);
604                async {
605                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
606                        unavailable(),
607                    )
608                }
609            }
610        });
611        let connector = ProxyRouteFailureCacheConnector::new(
612            inner,
613            cache(ProxyRouteFailureCacheScope::PerDestination),
614        );
615
616        let _first_error = connector
617            .serve(request("one.example:443", proxy(None)))
618            .await
619            .unwrap_err();
620        let error = connector
621            .serve(request("one.example:443", proxy(None)))
622            .await
623            .unwrap_err();
624
625        assert_eq!(attempts.load(Ordering::SeqCst), 1);
626        assert_eq!(error.domain(), ConnectionErrorDomain::Transport);
627        assert_eq!(error.kind(), ConnectionErrorKind::Unavailable);
628        assert!(
629            error
630                .get_ref()
631                .downcast_ref::<ProxyRouteFailureCachedError>()
632                .is_some()
633        );
634    }
635
636    #[tokio::test]
637    async fn cached_error_does_not_expose_route_or_credentials() {
638        let inner = service_fn(|_input: ConnectRequest| async {
639            Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(unavailable())
640        });
641        let connector = ProxyRouteFailureCacheConnector::new(
642            inner,
643            cache(ProxyRouteFailureCacheScope::PerDestination),
644        );
645
646        let _first_error = connector
647            .serve(request("one.example:443", proxy(Some("alice"))))
648            .await
649            .unwrap_err();
650        let error = connector
651            .serve(request("one.example:443", proxy(Some("alice"))))
652            .await
653            .unwrap_err();
654        let rendered = format!("{error:?} {error}");
655
656        assert!(rendered.contains("temporarily blocked"));
657        assert!(!rendered.contains("alice"));
658        assert!(!rendered.contains("secret"));
659        assert!(!rendered.contains("proxy.example"));
660    }
661
662    #[tokio::test]
663    async fn timeout_and_protocol_failures_are_cached() {
664        for kind in [ConnectionErrorKind::Timeout, ConnectionErrorKind::Protocol] {
665            let attempts = Arc::new(AtomicUsize::new(0));
666            let inner = service_fn({
667                let attempts = attempts.clone();
668                move |_input: ConnectRequest| {
669                    attempts.fetch_add(1, Ordering::SeqCst);
670                    async move {
671                        Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
672                            ConnectionError::transport(
673                                BoxError::from_static_str("cacheable proxy failure"),
674                                kind,
675                            ),
676                        )
677                    }
678                }
679            });
680            let connector = ProxyRouteFailureCacheConnector::new(
681                inner,
682                cache(ProxyRouteFailureCacheScope::PerDestination),
683            );
684
685            let _first_error = connector
686                .serve(request("one.example:443", proxy(None)))
687                .await
688                .unwrap_err();
689            let cached = connector
690                .serve(request("one.example:443", proxy(None)))
691                .await
692                .unwrap_err();
693
694            assert_eq!(attempts.load(Ordering::SeqCst), 1, "{kind}");
695            assert!(
696                cached
697                    .get_ref()
698                    .downcast_ref::<ProxyRouteFailureCachedError>()
699                    .is_some(),
700                "{kind}"
701            );
702        }
703    }
704
705    #[tokio::test(start_paused = true)]
706    async fn first_failure_is_cached_when_monotonic_time_is_zero() {
707        let attempts = Arc::new(AtomicUsize::new(0));
708        let inner = service_fn({
709            let attempts = attempts.clone();
710            move |_input: ConnectRequest| {
711                attempts.fetch_add(1, Ordering::SeqCst);
712                async {
713                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
714                        unavailable(),
715                    )
716                }
717            }
718        });
719        let connector = ProxyRouteFailureCacheConnector::new(
720            inner,
721            cache(ProxyRouteFailureCacheScope::PerDestination),
722        );
723
724        let _first_error = connector
725            .serve(request("one.example:443", proxy(None)))
726            .await
727            .unwrap_err();
728        let second_error = connector
729            .serve(request("one.example:443", proxy(None)))
730            .await
731            .unwrap_err();
732
733        assert_eq!(attempts.load(Ordering::SeqCst), 1);
734        assert!(
735            second_error
736                .get_ref()
737                .downcast_ref::<ProxyRouteFailureCachedError>()
738                .is_some()
739        );
740    }
741
742    #[tokio::test]
743    async fn per_destination_scope_does_not_poison_another_target() {
744        let attempts = Arc::new(AtomicUsize::new(0));
745        let inner = service_fn({
746            let attempts = attempts.clone();
747            move |_input: ConnectRequest| {
748                attempts.fetch_add(1, Ordering::SeqCst);
749                async {
750                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
751                        unavailable(),
752                    )
753                }
754            }
755        });
756        let connector = ProxyRouteFailureCacheConnector::new(
757            inner,
758            cache(ProxyRouteFailureCacheScope::PerDestination),
759        );
760
761        let _first_error = connector
762            .serve(request("one.example:443", proxy(None)))
763            .await
764            .unwrap_err();
765        let _second_error = connector
766            .serve(request("two.example:443", proxy(None)))
767            .await
768            .unwrap_err();
769
770        assert_eq!(attempts.load(Ordering::SeqCst), 2);
771    }
772
773    #[tokio::test]
774    async fn per_destination_scope_distinguishes_application_protocol() {
775        let attempts = Arc::new(AtomicUsize::new(0));
776        let inner = service_fn({
777            let attempts = attempts.clone();
778            move |_input: ConnectRequest| {
779                attempts.fetch_add(1, Ordering::SeqCst);
780                async {
781                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
782                        unavailable(),
783                    )
784                }
785            }
786        });
787        let connector = ProxyRouteFailureCacheConnector::new(
788            inner,
789            cache(ProxyRouteFailureCacheScope::PerDestination),
790        );
791
792        let _http_error = connector
793            .serve(
794                request("one.example:443", proxy(None)).with_application_protocol(Protocol::HTTP),
795            )
796            .await
797            .unwrap_err();
798        let _https_error = connector
799            .serve(
800                request("one.example:443", proxy(None)).with_application_protocol(Protocol::HTTPS),
801            )
802            .await
803            .unwrap_err();
804
805        assert_eq!(attempts.load(Ordering::SeqCst), 2);
806    }
807
808    #[tokio::test]
809    async fn per_proxy_scope_shares_failure_across_targets() {
810        let attempts = Arc::new(AtomicUsize::new(0));
811        let inner = service_fn({
812            let attempts = attempts.clone();
813            move |_input: ConnectRequest| {
814                attempts.fetch_add(1, Ordering::SeqCst);
815                async {
816                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
817                        unavailable(),
818                    )
819                }
820            }
821        });
822        let connector = ProxyRouteFailureCacheConnector::new(
823            inner,
824            cache(ProxyRouteFailureCacheScope::PerProxy),
825        );
826
827        let _first_error = connector
828            .serve(request("one.example:443", proxy(None)))
829            .await
830            .unwrap_err();
831        let _second_error = connector
832            .serve(request("two.example:443", proxy(None)))
833            .await
834            .unwrap_err();
835
836        assert_eq!(attempts.load(Ordering::SeqCst), 1);
837    }
838
839    #[tokio::test]
840    async fn routing_usernames_have_independent_failure_state() {
841        let attempts = Arc::new(AtomicUsize::new(0));
842        let inner = service_fn({
843            let attempts = attempts.clone();
844            move |_input: ConnectRequest| {
845                attempts.fetch_add(1, Ordering::SeqCst);
846                async {
847                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
848                        unavailable(),
849                    )
850                }
851            }
852        });
853        let connector = ProxyRouteFailureCacheConnector::new(
854            inner,
855            cache(ProxyRouteFailureCacheScope::PerDestination),
856        );
857
858        let _alice_error = connector
859            .serve(request("one.example:443", proxy(Some("alice"))))
860            .await
861            .unwrap_err();
862        let _bob_error = connector
863            .serve(request("one.example:443", proxy(Some("bob"))))
864            .await
865            .unwrap_err();
866
867        assert_eq!(attempts.load(Ordering::SeqCst), 2);
868    }
869
870    #[tokio::test]
871    async fn direct_routes_are_never_cached() {
872        let attempts = Arc::new(AtomicUsize::new(0));
873        let inner = service_fn({
874            let attempts = attempts.clone();
875            move |_input: ConnectRequest| {
876                attempts.fetch_add(1, Ordering::SeqCst);
877                async {
878                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
879                        unavailable(),
880                    )
881                }
882            }
883        });
884        let connector = ProxyRouteFailureCacheConnector::new(
885            inner,
886            cache(ProxyRouteFailureCacheScope::PerDestination),
887        );
888
889        for _ in 0..2 {
890            let _error = connector
891                .serve(request("one.example:443", ProxyRoute::Direct))
892                .await
893                .unwrap_err();
894        }
895
896        assert_eq!(attempts.load(Ordering::SeqCst), 2);
897    }
898
899    #[tokio::test]
900    async fn plural_routes_are_never_used_as_cache_keys() {
901        let attempts = Arc::new(AtomicUsize::new(0));
902        let inner = service_fn({
903            let attempts = attempts.clone();
904            move |_input: ConnectRequest| {
905                attempts.fetch_add(1, Ordering::SeqCst);
906                async {
907                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
908                        unavailable(),
909                    )
910                }
911            }
912        });
913        let connector = ProxyRouteFailureCacheConnector::new(
914            inner,
915            cache(ProxyRouteFailureCacheScope::PerDestination),
916        );
917
918        for _ in 0..2 {
919            let request = ConnectRequest::new("one.example:443".parse().unwrap());
920            request.extensions.insert(ProxyRoutes::new([proxy(None)]));
921            let _error = connector.serve(request).await.unwrap_err();
922        }
923
924        assert_eq!(attempts.load(Ordering::SeqCst), 2);
925    }
926
927    #[tokio::test]
928    async fn non_cacheable_failures_do_not_block_later_attempts() {
929        for (domain, kind) in [
930            (
931                ConnectionErrorDomain::Transport,
932                ConnectionErrorKind::Rejected,
933            ),
934            (ConnectionErrorDomain::Transport, ConnectionErrorKind::Other),
935            (
936                ConnectionErrorDomain::Transport,
937                ConnectionErrorKind::Authentication,
938            ),
939            (
940                ConnectionErrorDomain::Application,
941                ConnectionErrorKind::Protocol,
942            ),
943            (
944                ConnectionErrorDomain::Local,
945                ConnectionErrorKind::InvalidInput,
946            ),
947            (ConnectionErrorDomain::Unknown, ConnectionErrorKind::Other),
948        ] {
949            let attempts = Arc::new(AtomicUsize::new(0));
950            let inner = service_fn({
951                let attempts = attempts.clone();
952                move |_input: ConnectRequest| {
953                    attempts.fetch_add(1, Ordering::SeqCst);
954                    async move {
955                        Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
956                            ConnectionError::new(
957                                BoxError::from_static_str("not cacheable"),
958                                domain,
959                                kind,
960                            ),
961                        )
962                    }
963                }
964            });
965            let connector = ProxyRouteFailureCacheConnector::new(
966                inner,
967                cache(ProxyRouteFailureCacheScope::PerDestination),
968            );
969
970            for _ in 0..2 {
971                let _error = connector
972                    .serve(request("one.example:443", proxy(None)))
973                    .await
974                    .unwrap_err();
975            }
976
977            assert_eq!(attempts.load(Ordering::SeqCst), 2, "{domain}/{kind}");
978        }
979    }
980
981    #[tokio::test]
982    async fn success_clears_failure_backoff() {
983        let attempts = Arc::new(AtomicUsize::new(0));
984        let inner = service_fn({
985            let attempts = attempts.clone();
986            move |input: ConnectRequest| {
987                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
988                async move {
989                    if attempt == 0 {
990                        Err(unavailable())
991                    } else {
992                        Ok(EstablishedClientConnection {
993                            input,
994                            conn: ServiceInput::new(()),
995                        })
996                    }
997                }
998            }
999        });
1000        let connector = ProxyRouteFailureCacheConnector::new(
1001            inner,
1002            cache(ProxyRouteFailureCacheScope::PerDestination),
1003        );
1004
1005        let _first_error = connector
1006            .serve(request("one.example:443", proxy(None)))
1007            .await
1008            .unwrap_err();
1009        tokio::time::sleep(Duration::from_millis(40)).await;
1010        let _first_connection = connector
1011            .serve(request("one.example:443", proxy(None)))
1012            .await
1013            .unwrap();
1014        let _second_connection = connector
1015            .serve(request("one.example:443", proxy(None)))
1016            .await
1017            .unwrap();
1018
1019        assert_eq!(attempts.load(Ordering::SeqCst), 3);
1020    }
1021
1022    #[tokio::test]
1023    async fn successful_routes_are_not_retained() {
1024        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1025        let inner = service_fn(|input: ConnectRequest| async {
1026            Ok::<_, ConnectionError>(EstablishedClientConnection {
1027                input,
1028                conn: ServiceInput::new(()),
1029            })
1030        });
1031        let connector = ProxyRouteFailureCacheConnector::new(inner, failure_cache.clone());
1032
1033        let _connection = connector
1034            .serve(request("one.example:443", proxy(None)))
1035            .await
1036            .unwrap();
1037        failure_cache.entries.run_pending_tasks();
1038
1039        assert_eq!(failure_cache.entry_count(), 0);
1040    }
1041
1042    #[test]
1043    fn cancelled_attempt_is_not_retained() {
1044        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1045        let request = request("one.example:443", proxy(None));
1046
1047        let permit = begin_attempt(&failure_cache, &request);
1048        failure_cache.entries.run_pending_tasks();
1049        assert_eq!(failure_cache.entry_count(), 1);
1050
1051        drop(permit);
1052        failure_cache.entries.run_pending_tasks();
1053        assert_eq!(failure_cache.entry_count(), 0);
1054    }
1055
1056    #[test]
1057    fn published_failure_deadline_blocks_before_count_update() {
1058        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1059        let request = request("one.example:443", proxy(None));
1060        let permit = begin_attempt(&failure_cache, &request);
1061        permit.entry.blocked_until.store(
1062            now_monotonic_nanos().saturating_add(duration_nanos(Duration::from_secs(1))),
1063            Ordering::Release,
1064        );
1065
1066        assert!(matches!(
1067            failure_cache.begin(&request),
1068            Some(CacheDecision::Blocked(_))
1069        ));
1070    }
1071
1072    #[test]
1073    fn success_state_wins_over_an_in_flight_failure() {
1074        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1075        let request = request("one.example:443", proxy(None));
1076        let mut permit = begin_attempt(&failure_cache, &request);
1077        let entry = permit.entry.clone();
1078
1079        entry.mark_live();
1080        failure_cache.mark_failure(&mut permit);
1081
1082        assert!(entry.succeeded.load(Ordering::Acquire));
1083        assert_eq!(entry.blocked_until.load(Ordering::Acquire), 0);
1084        assert_eq!(entry.probe_until.load(Ordering::Acquire), 0);
1085        assert_eq!(entry.failure_count.load(Ordering::Acquire), 0);
1086    }
1087
1088    #[test]
1089    fn newer_failure_deadline_is_not_replaced() {
1090        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1091        let request = request("one.example:443", proxy(None));
1092        let mut permit = begin_attempt(&failure_cache, &request);
1093        let entry = permit.entry.clone();
1094        let newer_deadline = permit.started_time.saturating_add(1);
1095        entry.blocked_until.store(newer_deadline, Ordering::Release);
1096        entry.failure_count.store(1, Ordering::Release);
1097
1098        failure_cache.mark_failure(&mut permit);
1099
1100        assert_eq!(entry.blocked_until.load(Ordering::Acquire), newer_deadline);
1101        assert_eq!(entry.failure_count.load(Ordering::Acquire), 1);
1102    }
1103
1104    #[test]
1105    fn equal_failure_deadline_can_be_replaced() {
1106        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1107        let request = request("one.example:443", proxy(None));
1108        let mut permit = begin_attempt(&failure_cache, &request);
1109        let entry = permit.entry.clone();
1110        entry
1111            .blocked_until
1112            .store(permit.started_time, Ordering::Release);
1113        entry.failure_count.store(1, Ordering::Release);
1114
1115        failure_cache.mark_failure(&mut permit);
1116
1117        assert!(entry.blocked_until.load(Ordering::Acquire) > permit.started_time);
1118        assert_eq!(entry.failure_count.load(Ordering::Acquire), 2);
1119    }
1120
1121    #[tokio::test]
1122    async fn concurrent_new_success_prevents_older_failure_from_blocking() {
1123        let attempts = Arc::new(AtomicUsize::new(0));
1124        let failure_started = Arc::new(Notify::new());
1125        let release_failure = Arc::new(Notify::new());
1126        let inner = service_fn({
1127            let attempts = attempts.clone();
1128            let failure_started = failure_started.clone();
1129            let release_failure = release_failure.clone();
1130            move |input: ConnectRequest| {
1131                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
1132                let failure_started = failure_started.clone();
1133                let release_failure = release_failure.clone();
1134                async move {
1135                    if attempt == 0 {
1136                        failure_started.notify_one();
1137                        release_failure.notified().await;
1138                        Err(unavailable())
1139                    } else {
1140                        Ok(EstablishedClientConnection {
1141                            input,
1142                            conn: ServiceInput::new(()),
1143                        })
1144                    }
1145                }
1146            }
1147        });
1148        let connector = ProxyRouteFailureCacheConnector::new(
1149            inner,
1150            cache(ProxyRouteFailureCacheScope::PerDestination),
1151        );
1152
1153        let failure_notification = failure_started.notified();
1154        let failing_attempt = tokio::spawn({
1155            let connector = connector.clone();
1156            async move {
1157                connector
1158                    .serve(request("one.example:443", proxy(None)))
1159                    .await
1160            }
1161        });
1162        within_test_timeout(failure_notification).await;
1163
1164        let _concurrent_success =
1165            within_test_timeout(connector.serve(request("one.example:443", proxy(None))))
1166                .await
1167                .unwrap();
1168        release_failure.notify_one();
1169        let _older_error = within_test_timeout(failing_attempt)
1170            .await
1171            .unwrap()
1172            .unwrap_err();
1173
1174        let _later_success =
1175            within_test_timeout(connector.serve(request("one.example:443", proxy(None))))
1176                .await
1177                .expect("the older failure must not block a route that succeeded concurrently");
1178        assert_eq!(attempts.load(Ordering::SeqCst), 3);
1179    }
1180
1181    #[tokio::test]
1182    async fn cancelled_attempt_does_not_discard_concurrent_failure() {
1183        let attempts = Arc::new(AtomicUsize::new(0));
1184        let failure_started = Arc::new(Notify::new());
1185        let cancelled_started = Arc::new(Notify::new());
1186        let release_failure = Arc::new(Notify::new());
1187        let hold_cancelled = Arc::new(Notify::new());
1188        let inner = service_fn({
1189            let attempts = attempts.clone();
1190            let failure_started = failure_started.clone();
1191            let cancelled_started = cancelled_started.clone();
1192            let release_failure = release_failure.clone();
1193            let hold_cancelled = hold_cancelled.clone();
1194            move |_input: ConnectRequest| {
1195                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
1196                let failure_started = failure_started.clone();
1197                let cancelled_started = cancelled_started.clone();
1198                let release_failure = release_failure.clone();
1199                let hold_cancelled = hold_cancelled.clone();
1200                async move {
1201                    if attempt == 0 {
1202                        failure_started.notify_one();
1203                        release_failure.notified().await;
1204                    } else {
1205                        cancelled_started.notify_one();
1206                        hold_cancelled.notified().await;
1207                    }
1208                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
1209                        unavailable(),
1210                    )
1211                }
1212            }
1213        });
1214        let connector = ProxyRouteFailureCacheConnector::new(
1215            inner,
1216            cache(ProxyRouteFailureCacheScope::PerDestination),
1217        );
1218
1219        let failure_notification = failure_started.notified();
1220        let failing_attempt = tokio::spawn({
1221            let connector = connector.clone();
1222            async move {
1223                connector
1224                    .serve(request("one.example:443", proxy(None)))
1225                    .await
1226            }
1227        });
1228        within_test_timeout(failure_notification).await;
1229
1230        let cancelled_notification = cancelled_started.notified();
1231        let cancelled_attempt = tokio::spawn({
1232            let connector = connector.clone();
1233            async move {
1234                connector
1235                    .serve(request("one.example:443", proxy(None)))
1236                    .await
1237            }
1238        });
1239        within_test_timeout(cancelled_notification).await;
1240
1241        release_failure.notify_one();
1242        let _failure = within_test_timeout(failing_attempt)
1243            .await
1244            .unwrap()
1245            .unwrap_err();
1246        cancelled_attempt.abort();
1247        let _cancelled = within_test_timeout(cancelled_attempt).await.unwrap_err();
1248
1249        let cached = within_test_timeout(connector.serve(request("one.example:443", proxy(None))))
1250            .await
1251            .unwrap_err();
1252        assert_eq!(attempts.load(Ordering::SeqCst), 2);
1253        assert!(
1254            cached
1255                .get_ref()
1256                .downcast_ref::<ProxyRouteFailureCachedError>()
1257                .is_some()
1258        );
1259    }
1260
1261    #[tokio::test]
1262    async fn expired_entry_allows_only_one_concurrent_probe() {
1263        const TASKS: usize = 32;
1264
1265        let attempts = Arc::new(AtomicUsize::new(0));
1266        let probe_started = Arc::new(Notify::new());
1267        let release_probe = Arc::new(Notify::new());
1268        let inner = service_fn({
1269            let attempts = attempts.clone();
1270            let probe_started = probe_started.clone();
1271            let release_probe = release_probe.clone();
1272            move |_input: ConnectRequest| {
1273                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
1274                let probe_started = probe_started.clone();
1275                let release_probe = release_probe.clone();
1276                async move {
1277                    if attempt > 0 {
1278                        probe_started.notify_one();
1279                        release_probe.notified().await;
1280                    }
1281                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
1282                        unavailable(),
1283                    )
1284                }
1285            }
1286        });
1287        let connector = ProxyRouteFailureCacheConnector::new(
1288            inner,
1289            cache(ProxyRouteFailureCacheScope::PerDestination),
1290        );
1291
1292        let _initial_error = connector
1293            .serve(request("one.example:443", proxy(None)))
1294            .await
1295            .unwrap_err();
1296        tokio::time::sleep(Duration::from_millis(40)).await;
1297
1298        let barrier = Arc::new(Barrier::new(TASKS + 1));
1299        let mut tasks = Vec::with_capacity(TASKS);
1300        for _ in 0..TASKS {
1301            let connector = connector.clone();
1302            let barrier = barrier.clone();
1303            tasks.push(tokio::spawn(async move {
1304                barrier.wait().await;
1305                connector
1306                    .serve(request("one.example:443", proxy(None)))
1307                    .await
1308            }));
1309        }
1310        let probe_notification = probe_started.notified();
1311        within_test_timeout(async {
1312            barrier.wait().await;
1313            probe_notification.await;
1314        })
1315        .await;
1316        tokio::task::yield_now().await;
1317        release_probe.notify_one();
1318
1319        for task in tasks {
1320            let _error = within_test_timeout(task).await.unwrap().unwrap_err();
1321        }
1322        assert_eq!(attempts.load(Ordering::SeqCst), 2);
1323    }
1324
1325    #[tokio::test]
1326    async fn cancelling_half_open_probe_releases_its_lease() {
1327        let attempts = Arc::new(AtomicUsize::new(0));
1328        let probe_started = Arc::new(Notify::new());
1329        let hold_probe = Arc::new(Notify::new());
1330        let inner = service_fn({
1331            let attempts = attempts.clone();
1332            let probe_started = probe_started.clone();
1333            let hold_probe = hold_probe.clone();
1334            move |_input: ConnectRequest| {
1335                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
1336                let probe_started = probe_started.clone();
1337                let hold_probe = hold_probe.clone();
1338                async move {
1339                    if attempt == 1 {
1340                        probe_started.notify_one();
1341                        hold_probe.notified().await;
1342                    }
1343                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
1344                        unavailable(),
1345                    )
1346                }
1347            }
1348        });
1349        let connector = ProxyRouteFailureCacheConnector::new(
1350            inner,
1351            cache(ProxyRouteFailureCacheScope::PerDestination),
1352        );
1353
1354        let _initial_error = connector
1355            .serve(request("one.example:443", proxy(None)))
1356            .await
1357            .unwrap_err();
1358        tokio::time::sleep(Duration::from_millis(40)).await;
1359
1360        let probe_notification = probe_started.notified();
1361        let task = tokio::spawn({
1362            let connector = connector.clone();
1363            async move {
1364                connector
1365                    .serve(request("one.example:443", proxy(None)))
1366                    .await
1367            }
1368        });
1369        within_test_timeout(probe_notification).await;
1370        task.abort();
1371        let _join_error = within_test_timeout(task).await.unwrap_err();
1372
1373        let _probe_error = tokio::time::timeout(
1374            Duration::from_millis(100),
1375            connector.serve(request("one.example:443", proxy(None))),
1376        )
1377        .await
1378        .expect("cancelled probe must release its lease")
1379        .unwrap_err();
1380        assert_eq!(attempts.load(Ordering::SeqCst), 3);
1381    }
1382
1383    #[tokio::test]
1384    async fn cache_never_retains_more_than_its_capacity() {
1385        const CAPACITY: u64 = 4;
1386
1387        let failure_cache = ProxyRouteFailureCache::try_new(ProxyRouteFailureCacheConfig {
1388            initial_backoff: Duration::from_secs(1),
1389            max_backoff: Duration::from_secs(1),
1390            probe_lease: Duration::from_secs(1),
1391            max_entries: CAPACITY,
1392            scope: ProxyRouteFailureCacheScope::PerDestination,
1393        })
1394        .unwrap();
1395        let inner = service_fn(|_input: ConnectRequest| async {
1396            Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(unavailable())
1397        });
1398        let connector = ProxyRouteFailureCacheConnector::new(inner, failure_cache.clone());
1399
1400        for index in 0..16 {
1401            let _error = connector
1402                .serve(request(
1403                    &format!("destination-{index}.example:443"),
1404                    proxy(None),
1405                ))
1406                .await
1407                .unwrap_err();
1408        }
1409        failure_cache.entries.run_pending_tasks();
1410
1411        assert!(failure_cache.entry_count() <= CAPACITY);
1412    }
1413
1414    #[tokio::test]
1415    async fn invalidating_cache_clears_retained_failure() {
1416        let attempts = Arc::new(AtomicUsize::new(0));
1417        let inner = service_fn({
1418            let attempts = attempts.clone();
1419            move |_input: ConnectRequest| {
1420                attempts.fetch_add(1, Ordering::SeqCst);
1421                async {
1422                    Err::<EstablishedClientConnection<ServiceInput<()>, ConnectRequest>, _>(
1423                        unavailable(),
1424                    )
1425                }
1426            }
1427        });
1428        let failure_cache = cache(ProxyRouteFailureCacheScope::PerDestination);
1429        let connector = ProxyRouteFailureCacheConnector::new(inner, failure_cache.clone());
1430
1431        let _first_error = connector
1432            .serve(request("one.example:443", proxy(None)))
1433            .await
1434            .unwrap_err();
1435        failure_cache.entries.run_pending_tasks();
1436        assert_eq!(failure_cache.entry_count(), 1);
1437
1438        failure_cache.invalidate_all();
1439        let _second_error = connector
1440            .serve(request("one.example:443", proxy(None)))
1441            .await
1442            .unwrap_err();
1443        assert_eq!(attempts.load(Ordering::SeqCst), 2);
1444    }
1445
1446    #[test]
1447    fn rejects_invalid_configuration() {
1448        for config in [
1449            ProxyRouteFailureCacheConfig {
1450                initial_backoff: Duration::ZERO,
1451                ..Default::default()
1452            },
1453            ProxyRouteFailureCacheConfig {
1454                initial_backoff: Duration::from_secs(2),
1455                max_backoff: Duration::from_secs(1),
1456                ..Default::default()
1457            },
1458            ProxyRouteFailureCacheConfig {
1459                probe_lease: Duration::ZERO,
1460                ..Default::default()
1461            },
1462            ProxyRouteFailureCacheConfig {
1463                max_entries: 0,
1464                ..Default::default()
1465            },
1466        ] {
1467            ProxyRouteFailureCache::try_new(config).unwrap_err();
1468        }
1469    }
1470
1471    #[test]
1472    fn default_configuration_matches_easy_client_policy() {
1473        let custom_cache = cache(ProxyRouteFailureCacheScope::PerProxy);
1474        assert_eq!(
1475            custom_cache.config().scope,
1476            ProxyRouteFailureCacheScope::PerProxy
1477        );
1478        assert_eq!(
1479            custom_cache.config().initial_backoff,
1480            Duration::from_millis(20)
1481        );
1482
1483        let failure_cache = ProxyRouteFailureCache::default();
1484        let config = failure_cache.config();
1485        assert_eq!(config.scope, ProxyRouteFailureCacheScope::PerDestination);
1486        assert_eq!(config.initial_backoff, Duration::from_secs(60));
1487        assert_eq!(config.max_backoff, Duration::from_mins(30));
1488        assert_eq!(config.probe_lease, Duration::from_secs(30));
1489        assert_eq!(config.max_entries, 1_024);
1490        assert!(format!("{failure_cache:?}").contains("ProxyRouteFailureCache"));
1491    }
1492
1493    #[test]
1494    fn remaining_duration_excludes_the_deadline_itself() {
1495        assert_eq!(remaining_duration(11, 10), Some(Duration::from_nanos(1)));
1496        assert_eq!(remaining_duration(10, 10), None);
1497        assert_eq!(remaining_duration(9, 10), None);
1498    }
1499
1500    #[test]
1501    fn backoff_doubles_until_configured_cap() {
1502        let config = ProxyRouteFailureCacheConfig {
1503            initial_backoff: Duration::from_secs(1),
1504            max_backoff: Duration::from_secs(5),
1505            ..Default::default()
1506        };
1507        assert_eq!(config.backoff(0), Duration::from_secs(1));
1508        assert_eq!(config.backoff(1), Duration::from_secs(2));
1509        assert_eq!(config.backoff(2), Duration::from_secs(4));
1510        assert_eq!(config.backoff(3), Duration::from_secs(5));
1511        assert_eq!(config.backoff(u32::MAX), Duration::from_secs(5));
1512    }
1513}
rama_net/client/proxy/route/failure_cache.rs

rama_net/client/proxy/route/
failure_cache.rs
