nautilus_network/
retry.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Retry policy for asynchronous network operations.
17
18use std::{fmt::Display, future::Future, marker::PhantomData, time::Duration};
19
20use serde::{Deserialize, Serialize};
21use tokio_util::sync::CancellationToken;
22
23use crate::{backoff::ExponentialBackoff, dst};
24
25/// Configuration for retry behavior.
26#[derive(Debug, Clone, Deserialize, Serialize)]
27#[serde(default, deny_unknown_fields)]
28pub struct RetryConfig {
29    /// Maximum number of retry attempts (total attempts = 1 initial + `max_retries`).
30    pub max_retries: u32,
31    /// Initial delay between retries in milliseconds.
32    pub initial_delay_ms: u64,
33    /// Maximum delay between retries in milliseconds.
34    pub max_delay_ms: u64,
35    /// Backoff multiplier factor.
36    pub backoff_factor: f64,
37    /// Maximum jitter in milliseconds to add to delays.
38    pub jitter_ms: u64,
39    /// Optional timeout for individual operations in milliseconds. `None` disables the timeout.
40    pub operation_timeout_ms: Option<u64>,
41    /// Whether the first retry occurs without delay.
42    ///
43    /// Connection operations typically enable this, while HTTP and order operations typically
44    /// retain a delay.
45    pub immediate_first: bool,
46    /// Optional maximum total elapsed time across all attempts and retry delays in milliseconds.
47    /// When set, this deadline also bounds an in-flight operation.
48    pub max_elapsed_ms: Option<u64>,
49}
50
51impl Default for RetryConfig {
52    fn default() -> Self {
53        Self {
54            max_retries: 3,
55            initial_delay_ms: 1_000,
56            max_delay_ms: 10_000,
57            backoff_factor: 2.0,
58            jitter_ms: 100,
59            operation_timeout_ms: Some(30_000),
60            immediate_first: false,
61            max_elapsed_ms: None,
62        }
63    }
64}
65
66/// A failure synthesized by retry machinery.
67///
68/// This type describes the retry control path only. It does not indicate whether an operation was
69/// transmitted or applied.
70#[derive(Clone, Debug, PartialEq, Eq)]
71pub enum RetryError {
72    /// The cancellation token was set.
73    Canceled,
74    /// A single operation attempt exceeded its configured timeout.
75    OperationTimeout {
76        /// Configured timeout for each attempt in milliseconds.
77        timeout_ms: u64,
78    },
79    /// The total elapsed-time budget was exhausted.
80    ElapsedBudgetExceeded {
81        /// One-based attempt position when the budget was exhausted.
82        attempt: u32,
83        /// Maximum number of attempts allowed by the retry configuration.
84        max_attempts: u32,
85        /// Last operation error when budget exhaustion followed a failed attempt.
86        last_error: Option<String>,
87    },
88    /// The retry configuration could not create a backoff state.
89    InvalidConfiguration {
90        /// Configuration validation error.
91        message: String,
92    },
93}
94
95impl Display for RetryError {
96    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
97        match self {
98            Self::Canceled => write!(f, "canceled"),
99            Self::OperationTimeout { timeout_ms } => {
100                write!(f, "Timed out after {timeout_ms}ms")
101            }
102            Self::ElapsedBudgetExceeded {
103                attempt,
104                max_attempts,
105                last_error,
106            } => {
107                write!(f, "Retry budget exceeded ({attempt}/{max_attempts})")?;
108
109                if let Some(last_error) = last_error {
110                    write!(f, ": last error: {last_error}")?;
111                }
112
113                Ok(())
114            }
115            Self::InvalidConfiguration { message } => {
116                write!(f, "Invalid configuration: {message}")
117            }
118        }
119    }
120}
121
122impl std::error::Error for RetryError {}
123
124/// A stateless, thread-safe retry manager for network operations.
125///
126/// Each execution maintains independent backoff and elapsed-time state.
127#[derive(Clone, Debug)]
128pub struct RetryManager<E> {
129    config: RetryConfig,
130    _phantom: PhantomData<E>,
131}
132
133#[bon::bon]
134impl<E> RetryManager<E>
135where
136    E: std::error::Error,
137{
138    /// Creates a new retry manager with the given configuration.
139    #[must_use]
140    pub const fn new(config: RetryConfig) -> Self {
141        Self {
142            config,
143            _phantom: PhantomData,
144        }
145    }
146
147    /// Creates a retry budget error with attempt context.
148    #[inline(always)]
149    fn budget_exceeded_error(&self, attempt: u32, last_error: Option<String>) -> RetryError {
150        RetryError::ElapsedBudgetExceeded {
151            attempt: attempt.saturating_add(1),
152            max_attempts: self.config.max_retries.saturating_add(1),
153            last_error,
154        }
155    }
156
157    /// Returns a builder for a retry-managed invocation.
158    ///
159    /// Set `retry_delay` to derive a minimum delay from an operation error, including errors
160    /// created by `create_error` for operation timeouts. The retry loop uses the greater of this
161    /// minimum and the configured exponential backoff. Retry delays do not consume the per-operation
162    /// timeout. If the effective delay cannot fit within the remaining elapsed budget, that error
163    /// is returned.
164    ///
165    /// Set `cancellation_token` to cancel the operation. Cancellation is checked at three points:
166    ///
167    /// - Before each operation attempt.
168    /// - During operation execution through `tokio::select!`.
169    /// - During retry delays.
170    ///
171    /// Cancellation mid-execution takes effect immediately by dropping the in-flight
172    /// operation future. For non-idempotent operations (e.g. an order already on the
173    /// wire) the outcome of the abandoned attempt is unknown to the caller.
174    ///
175    /// # Errors
176    ///
177    /// Returns an error if:
178    ///
179    /// - The operation returns a non-retryable error or exhausts the configured retries.
180    /// - An operation timeout terminates retry execution.
181    /// - The total elapsed-time budget expires.
182    /// - The backoff state cannot be created from the configuration.
183    /// - Cancellation is requested.
184    #[expect(
185        clippy::type_complexity,
186        reason = "bon needs one concrete optional callback type for omitted retry delays"
187    )]
188    #[builder(finish_fn = execute)]
189    pub async fn invocation<F, Fut, T>(
190        &self,
191        #[builder(start_fn)] operation_name: &str,
192        #[builder(start_fn)] operation: F,
193        #[builder(start_fn)] should_retry: impl Fn(&E) -> bool,
194        #[builder(start_fn)] create_error: impl Fn(RetryError) -> E,
195        retry_delay: Option<&(dyn Fn(&E) -> Option<Duration> + Sync)>,
196        cancellation_token: Option<&CancellationToken>,
197    ) -> Result<T, E>
198    where
199        F: FnMut() -> Fut,
200        Fut: Future<Output = Result<T, E>>,
201    {
202        self.execute_retry_loop(
203            operation_name,
204            operation,
205            should_retry,
206            |e| retry_delay.and_then(|retry_delay| retry_delay(e)),
207            create_error,
208            cancellation_token,
209        )
210        .await
211    }
212
213    async fn execute_retry_loop<F, Fut, T>(
214        &self,
215        operation_name: &str,
216        mut operation: F,
217        should_retry: impl Fn(&E) -> bool,
218        retry_delay: impl Fn(&E) -> Option<Duration>,
219        create_error: impl Fn(RetryError) -> E,
220        cancellation_token: Option<&CancellationToken>,
221    ) -> Result<T, E>
222    where
223        F: FnMut() -> Fut,
224        Fut: Future<Output = Result<T, E>>,
225    {
226        let mut backoff = ExponentialBackoff::new(
227            Duration::from_millis(self.config.initial_delay_ms),
228            Duration::from_millis(self.config.max_delay_ms),
229            self.config.backoff_factor,
230            self.config.jitter_ms,
231            self.config.immediate_first,
232        )
233        .map_err(|e| {
234            create_error(RetryError::InvalidConfiguration {
235                message: e.to_string(),
236            })
237        })?;
238
239        let mut attempt = 0;
240        let start_time = dst::time::Instant::now();
241        let max_elapsed = self.config.max_elapsed_ms.map(Duration::from_millis);
242        let deadline = max_elapsed.and_then(|duration| start_time.checked_add(duration));
243        let mut last_delayed_error = None;
244
245        loop {
246            if let Some(token) = cancellation_token
247                && token.is_cancelled()
248            {
249                log::debug!("Operation '{operation_name}' canceled after {attempt} attempts");
250                return Err(create_error(RetryError::Canceled));
251            }
252
253            if let Some(max_elapsed) = max_elapsed {
254                let elapsed = start_time.elapsed();
255                if elapsed >= max_elapsed {
256                    if let Some(e) = last_delayed_error {
257                        return Err(e);
258                    }
259
260                    return Err(create_error(self.budget_exceeded_error(attempt, None)));
261                }
262            }
263
264            last_delayed_error = None;
265
266            let attempt_future = async {
267                let result = match (self.config.operation_timeout_ms, cancellation_token) {
268                    (Some(timeout_ms), Some(token)) => {
269                        tokio::select! {
270                            biased;
271                            result = dst::time::timeout(Duration::from_millis(timeout_ms), operation()) => result,
272                            () = token.cancelled() => {
273                                log::debug!("Operation '{operation_name}' canceled during execution");
274                                return Err(create_error(RetryError::Canceled));
275                            }
276                        }
277                    }
278                    (Some(timeout_ms), None) => {
279                        dst::time::timeout(Duration::from_millis(timeout_ms), operation()).await
280                    }
281                    (None, Some(token)) => tokio::select! {
282                        biased;
283                        result = operation() => Ok(result),
284                        () = token.cancelled() => {
285                            log::debug!("Operation '{operation_name}' canceled during execution");
286                            return Err(create_error(RetryError::Canceled));
287                        }
288                    },
289                    (None, None) => Ok(operation().await),
290                };
291
292                Ok(result)
293            };
294
295            let result = if let Some(deadline) = deadline {
296                tokio::select! {
297                    biased;
298                    () = dst::time::sleep_until(deadline) => {
299                        if cancellation_token.is_some_and(CancellationToken::is_cancelled) {
300                            log::debug!("Operation '{operation_name}' canceled during execution");
301                            return Err(create_error(RetryError::Canceled));
302                        }
303                        return Err(create_error(self.budget_exceeded_error(attempt, None)));
304                    }
305                    result = attempt_future => result,
306                }
307            } else {
308                attempt_future.await
309            }?;
310
311            let (e, timed_out) = match result {
312                Ok(Ok(success)) => {
313                    if attempt > 0 {
314                        log::trace!(
315                            "Operation '{operation_name}' succeeded after {} attempts",
316                            attempt + 1
317                        );
318                    }
319
320                    return Ok(success);
321                }
322                Ok(Err(e)) => (e, false),
323                Err(_) => (
324                    create_error(RetryError::OperationTimeout {
325                        timeout_ms: self.config.operation_timeout_ms.unwrap_or(0),
326                    }),
327                    true,
328                ),
329            };
330
331            let minimum_delay = retry_delay(&e);
332
333            if !should_retry(&e) {
334                if timed_out {
335                    log::trace!("Operation '{operation_name}' non-retryable timeout: {e}");
336                } else {
337                    log::trace!("Operation '{operation_name}' non-retryable error: {e}");
338                }
339
340                return Err(e);
341            }
342
343            if attempt >= self.config.max_retries {
344                if timed_out {
345                    log::trace!(
346                        "Operation '{operation_name}' retries exhausted after timeout ({} attempts): {e}",
347                        attempt + 1
348                    );
349                } else {
350                    log::trace!(
351                        "Operation '{operation_name}' retries exhausted after {} attempts: {e}",
352                        attempt + 1
353                    );
354                }
355
356                return Err(e);
357            }
358
359            let mut delay = backoff.next_duration();
360            if let Some(minimum_delay) = minimum_delay {
361                delay = delay.max(minimum_delay);
362            }
363
364            if let Some(max_elapsed_ms) = self.config.max_elapsed_ms {
365                let elapsed = start_time.elapsed();
366                let remaining = Duration::from_millis(max_elapsed_ms).saturating_sub(elapsed);
367
368                if remaining.is_zero() {
369                    if minimum_delay.is_some() {
370                        return Err(e);
371                    }
372
373                    return Err(create_error(
374                        self.budget_exceeded_error(attempt, Some(e.to_string())),
375                    ));
376                }
377
378                if minimum_delay.is_some() && delay >= remaining {
379                    return Err(e);
380                }
381
382                delay = delay.min(remaining);
383            }
384
385            debug_assert!(
386                minimum_delay.is_none_or(|minimum_delay| delay >= minimum_delay),
387                "retry delay must honor the error-provided minimum"
388            );
389
390            if timed_out {
391                log::trace!(
392                    "Operation '{operation_name}' attempt {} timed out, retrying in {}ms: {e}",
393                    attempt + 1,
394                    delay.as_millis()
395                );
396            } else {
397                log::trace!(
398                    "Operation '{operation_name}' attempt {} failed, retrying in {}ms: {e}",
399                    attempt + 1,
400                    delay.as_millis()
401                );
402            }
403
404            // Yield even on zero-delay to avoid busy-wait loop
405            if delay.is_zero() {
406                tokio::task::yield_now().await;
407
408                if minimum_delay.is_some() {
409                    last_delayed_error = Some(e);
410                }
411
412                attempt += 1;
413                continue;
414            }
415
416            if let Some(token) = cancellation_token {
417                tokio::select! {
418                    biased;
419                    () = dst::time::sleep(delay) => {},
420                    () = token.cancelled() => {
421                        log::debug!("Operation '{operation_name}' canceled during retry delay (attempt {})", attempt + 1);
422                        return Err(create_error(RetryError::Canceled));
423                    }
424                }
425            } else {
426                dst::time::sleep(delay).await;
427            }
428
429            if minimum_delay.is_some() {
430                last_delayed_error = Some(e);
431            }
432
433            attempt += 1;
434        }
435    }
436}
437
438/// Convenience function to create a retry manager with default configuration.
439#[must_use]
440pub fn create_default_retry_manager<E>() -> RetryManager<E>
441where
442    E: std::error::Error,
443{
444    RetryManager::new(RetryConfig::default())
445}
446
447/// Convenience function to create a retry manager for HTTP operations.
448#[must_use]
449pub const fn create_http_retry_manager<E>() -> RetryManager<E>
450where
451    E: std::error::Error,
452{
453    let config = RetryConfig {
454        max_retries: 3,
455        initial_delay_ms: 1_000,
456        max_delay_ms: 10_000,
457        backoff_factor: 2.0,
458        jitter_ms: 1_000,
459        operation_timeout_ms: Some(60_000), // 60s for HTTP requests
460        immediate_first: false,
461        max_elapsed_ms: Some(180_000), // 3 minutes total budget
462    };
463
464    RetryManager::new(config)
465}
466
467/// Convenience function to create a retry manager for WebSocket operations.
468#[must_use]
469pub const fn create_websocket_retry_manager<E>() -> RetryManager<E>
470where
471    E: std::error::Error,
472{
473    let config = RetryConfig {
474        max_retries: 5,
475        initial_delay_ms: 1_000,
476        max_delay_ms: 10_000,
477        backoff_factor: 2.0,
478        jitter_ms: 1_000,
479        operation_timeout_ms: Some(30_000), // 30s for WebSocket operations
480        immediate_first: true,
481        max_elapsed_ms: Some(120_000), // 2 minutes total budget
482    };
483
484    RetryManager::new(config)
485}
486
487#[cfg(test)]
488mod test_utils {
489    use super::RetryError;
490
491    #[derive(Debug, thiserror::Error)]
492    pub(super) enum TestError {
493        #[error("Retryable error: {0}")]
494        Retryable(String),
495        #[error("Non-retryable error: {0}")]
496        NonRetryable(String),
497        #[error("Timeout error: {0}")]
498        Timeout(RetryError),
499    }
500
501    pub(super) fn should_retry_test_error(error: &TestError) -> bool {
502        matches!(error, TestError::Retryable(_))
503    }
504
505    pub(super) fn create_test_error(error: RetryError) -> TestError {
506        TestError::Timeout(error)
507    }
508}
509
510// Retry tests run under both real tokio (`#[tokio::test]`, paused-clock when
511// the test relies on virtual time advance) and madsim (`#[madsim::test]`,
512// virtual time always paused). `tokio::time::advance` has no direct madsim
513// equivalent, so explicit clock advances route through `advance_clock` below;
514// time reads and sleeps go through the `dst::time` re-export so they pick up
515// the runtime-appropriate clock. madsim auto-advances virtual time when all
516// tasks block, but `yield_until`-style busy-yield loops keep the runtime
517// non-idle, so explicit advances are still needed where they were before.
518#[cfg(test)]
519mod tests {
520    use std::sync::{
521        Arc,
522        atomic::{AtomicBool, AtomicU32, Ordering},
523    };
524
525    #[cfg(all(feature = "simulation", madsim))]
526    use madsim::task::{spawn, yield_now};
527    use rstest::rstest;
528    #[cfg(not(all(feature = "simulation", madsim)))]
529    use tokio::task::{spawn, yield_now};
530
531    use super::{test_utils::*, *};
532    use crate::dst::time;
533
534    const MAX_WAIT_ITERS: usize = 10_000;
535    const MAX_ADVANCE_ITERS: usize = 10_000;
536
537    #[cfg(all(feature = "simulation", madsim))]
538    pub(crate) async fn advance_clock(d: Duration) {
539        madsim::time::advance(d);
540        madsim::task::yield_now().await;
541    }
542
543    #[cfg(not(all(feature = "simulation", madsim)))]
544    pub(crate) async fn advance_clock(d: Duration) {
545        tokio::time::advance(d).await;
546    }
547
548    pub(crate) async fn yield_until<F>(mut condition: F)
549    where
550        F: FnMut() -> bool,
551    {
552        for _ in 0..MAX_WAIT_ITERS {
553            if condition() {
554                return;
555            }
556
557            yield_now().await;
558        }
559
560        panic!("yield_until timed out waiting for condition");
561    }
562
563    pub(crate) async fn advance_until<F>(mut condition: F)
564    where
565        F: FnMut() -> bool,
566    {
567        for _ in 0..MAX_ADVANCE_ITERS {
568            if condition() {
569                return;
570            }
571
572            advance_clock(Duration::from_millis(1)).await;
573            yield_now().await;
574        }
575
576        panic!("advance_until timed out waiting for condition");
577    }
578
579    #[rstest]
580    fn test_retry_config_default() {
581        let config = RetryConfig::default();
582        assert_eq!(config.max_retries, 3);
583        assert_eq!(config.initial_delay_ms, 1_000);
584        assert_eq!(config.max_delay_ms, 10_000);
585        // `allow` not `expect`: nightly clippy does not fire `float_cmp` inside `assert_eq!`
586        #[allow(clippy::float_cmp, reason = "test asserts the default backoff factor")]
587        {
588            assert_eq!(config.backoff_factor, 2.0);
589        }
590
591        assert_eq!(config.jitter_ms, 100);
592        assert_eq!(config.operation_timeout_ms, Some(30_000));
593        assert!(!config.immediate_first);
594        assert_eq!(config.max_elapsed_ms, None);
595    }
596
597    #[rstest]
598    #[case::canceled(RetryError::Canceled, "canceled")]
599    #[case::operation_timeout(
600        RetryError::OperationTimeout { timeout_ms: 250 },
601        "Timed out after 250ms"
602    )]
603    #[case::elapsed_budget(
604        RetryError::ElapsedBudgetExceeded {
605            attempt: 2,
606            max_attempts: 4,
607            last_error: None,
608        },
609        "Retry budget exceeded (2/4)"
610    )]
611    #[case::elapsed_budget_with_last_error(
612        RetryError::ElapsedBudgetExceeded {
613            attempt: 3,
614            max_attempts: 5,
615            last_error: Some("network unavailable".to_string()),
616        },
617        "Retry budget exceeded (3/5): last error: network unavailable"
618    )]
619    #[case::invalid_configuration(
620        RetryError::InvalidConfiguration {
621            message: "delay_initial must be non-zero".to_string(),
622        },
623        "Invalid configuration: delay_initial must be non-zero"
624    )]
625    fn test_retry_error_display(#[case] error: RetryError, #[case] expected: &str) {
626        assert_eq!(error.to_string(), expected);
627    }
628
629    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
630    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
631    async fn test_invalid_configuration_reason() {
632        let manager = RetryManager::new(RetryConfig {
633            initial_delay_ms: 0,
634            ..RetryConfig::default()
635        });
636
637        let error = manager
638            .invocation(
639                "test_invalid_configuration",
640                || async { Ok::<i32, TestError>(42) },
641                should_retry_test_error,
642                create_test_error,
643            )
644            .execute()
645            .await
646            .unwrap_err();
647
648        let TestError::Timeout(reason) = error else {
649            panic!("expected invalid configuration, was {error}");
650        };
651
652        assert_eq!(
653            reason,
654            RetryError::InvalidConfiguration {
655                message: "delay_initial must be non-zero".to_string(),
656            }
657        );
658    }
659
660    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
661    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
662    async fn test_retry_manager_success_first_attempt() {
663        let manager = RetryManager::new(RetryConfig::default());
664
665        let result = manager
666            .invocation(
667                "test_operation",
668                || async { Ok::<i32, TestError>(42) },
669                should_retry_test_error,
670                create_test_error,
671            )
672            .execute()
673            .await;
674
675        assert_eq!(result.unwrap(), 42);
676    }
677
678    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
679    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
680    async fn test_retry_manager_non_retryable_error() {
681        let manager = RetryManager::new(RetryConfig::default());
682
683        let result = manager
684            .invocation(
685                "test_operation",
686                || async { Err::<i32, TestError>(TestError::NonRetryable("test".to_string())) },
687                should_retry_test_error,
688                create_test_error,
689            )
690            .execute()
691            .await;
692
693        assert!(result.is_err());
694        assert!(matches!(result.unwrap_err(), TestError::NonRetryable(_)));
695    }
696
697    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
698    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
699    async fn test_retry_manager_retryable_error_exhausted() {
700        let config = RetryConfig {
701            max_retries: 2,
702            initial_delay_ms: 10,
703            max_delay_ms: 50,
704            backoff_factor: 2.0,
705            jitter_ms: 0,
706            operation_timeout_ms: None,
707            immediate_first: false,
708            max_elapsed_ms: None,
709        };
710
711        let manager = RetryManager::new(config);
712
713        let result = manager
714            .invocation(
715                "test_operation",
716                || async { Err::<i32, TestError>(TestError::Retryable("test".to_string())) },
717                should_retry_test_error,
718                create_test_error,
719            )
720            .execute()
721            .await;
722
723        assert!(result.is_err());
724        assert!(matches!(result.unwrap_err(), TestError::Retryable(_)));
725    }
726
727    #[rstest]
728    #[case::error(false, 200)]
729    #[case::timeout(true, 250)]
730    #[cfg_attr(
731        not(all(feature = "simulation", madsim)),
732        tokio::test(start_paused = true)
733    )]
734    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
735    async fn test_error_retry_delay_runs_outside_operation_timeout(
736        #[case] timeout: bool,
737        #[case] expected_elapsed_ms: u64,
738    ) {
739        let config = RetryConfig {
740            max_retries: 1,
741            initial_delay_ms: 10,
742            max_delay_ms: 10,
743            backoff_factor: 1.0,
744            jitter_ms: 0,
745            operation_timeout_ms: Some(50),
746            immediate_first: false,
747            max_elapsed_ms: Some(500),
748        };
749
750        let manager = RetryManager::new(config);
751        let attempts = Arc::new(AtomicU32::new(0));
752        let attempts_clone = Arc::clone(&attempts);
753        let start = time::Instant::now();
754
755        let result = manager
756            .invocation(
757                "test_error_delay",
758                move || {
759                    let attempts = Arc::clone(&attempts_clone);
760                    async move {
761                        if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
762                            if timeout {
763                                std::future::pending::<()>().await;
764                            }
765
766                            Err(TestError::Retryable("rate limit".to_string()))
767                        } else {
768                            Ok(42)
769                        }
770                    }
771                },
772                |_| true,
773                create_test_error,
774            )
775            .retry_delay(&|_| Some(Duration::from_millis(200)))
776            .execute()
777            .await;
778
779        assert_eq!(result.unwrap(), 42);
780        assert_eq!(attempts.load(Ordering::SeqCst), 2);
781        let expected_elapsed = Duration::from_millis(expected_elapsed_ms);
782        #[cfg(not(all(feature = "simulation", madsim)))]
783        assert_eq!(start.elapsed(), expected_elapsed);
784        #[cfg(all(feature = "simulation", madsim))]
785        assert!(
786            start.elapsed() >= expected_elapsed
787                && start.elapsed() < expected_elapsed + Duration::from_millis(1)
788        );
789    }
790
791    #[rstest]
792    #[cfg_attr(
793        not(all(feature = "simulation", madsim)),
794        tokio::test(start_paused = true)
795    )]
796    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
797    async fn test_error_retry_delay_observes_cancellation(#[values(false, true)] timeout: bool) {
798        let config = RetryConfig {
799            max_retries: 1,
800            initial_delay_ms: 10,
801            max_delay_ms: 10,
802            backoff_factor: 1.0,
803            jitter_ms: 0,
804            operation_timeout_ms: Some(50),
805            immediate_first: false,
806            max_elapsed_ms: Some(500),
807        };
808
809        let manager = RetryManager::new(config);
810        let attempts = Arc::new(AtomicU32::new(0));
811        let attempts_clone = Arc::clone(&attempts);
812        let token = CancellationToken::new();
813        let cancel = token.clone();
814
815        spawn(async move {
816            time::sleep(Duration::from_millis(100)).await;
817            cancel.cancel();
818        });
819
820        let error = manager
821            .invocation(
822                "test_error_delay_cancellation",
823                move || {
824                    let attempts = Arc::clone(&attempts_clone);
825                    async move {
826                        attempts.fetch_add(1, Ordering::SeqCst);
827
828                        if timeout {
829                            std::future::pending::<()>().await;
830                        }
831
832                        Err::<i32, TestError>(TestError::Retryable("rate limit".to_string()))
833                    }
834                },
835                |_| true,
836                create_test_error,
837            )
838            .retry_delay(&|_| Some(Duration::from_millis(200)))
839            .cancellation_token(&token)
840            .execute()
841            .await
842            .unwrap_err();
843
844        assert!(matches!(error, TestError::Timeout(RetryError::Canceled)));
845        assert_eq!(attempts.load(Ordering::SeqCst), 1);
846    }
847
848    #[rstest]
849    #[cfg_attr(
850        not(all(feature = "simulation", madsim)),
851        tokio::test(start_paused = true)
852    )]
853    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
854    async fn test_error_retry_delay_over_budget_returns_original_error() {
855        let config = RetryConfig {
856            max_retries: 3,
857            initial_delay_ms: 10,
858            max_delay_ms: 10,
859            backoff_factor: 1.0,
860            jitter_ms: 0,
861            operation_timeout_ms: Some(50),
862            immediate_first: false,
863            max_elapsed_ms: Some(100),
864        };
865
866        let manager = RetryManager::new(config);
867        let attempts = Arc::new(AtomicU32::new(0));
868        let attempts_clone = Arc::clone(&attempts);
869        let error = manager
870            .invocation(
871                "test_error_delay_budget",
872                move || {
873                    let attempts = Arc::clone(&attempts_clone);
874                    async move {
875                        attempts.fetch_add(1, Ordering::SeqCst);
876                        Err::<i32, TestError>(TestError::Retryable("rate limit".to_string()))
877                    }
878                },
879                should_retry_test_error,
880                create_test_error,
881            )
882            .retry_delay(&|_| Some(Duration::from_millis(200)))
883            .execute()
884            .await
885            .unwrap_err();
886
887        let TestError::Retryable(message) = error else {
888            panic!("expected original retryable error, was {error}");
889        };
890
891        assert_eq!(message, "rate limit");
892        assert_eq!(attempts.load(Ordering::SeqCst), 1);
893    }
894
895    #[rstest]
896    #[cfg_attr(
897        not(all(feature = "simulation", madsim)),
898        tokio::test(start_paused = true)
899    )]
900    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
901    async fn test_error_retry_delay_overshoot_returns_original_error() {
902        let config = RetryConfig {
903            max_retries: 3,
904            initial_delay_ms: 10,
905            max_delay_ms: 10,
906            backoff_factor: 1.0,
907            jitter_ms: 0,
908            operation_timeout_ms: Some(20),
909            immediate_first: false,
910            max_elapsed_ms: Some(100),
911        };
912
913        let manager = RetryManager::new(config);
914        let attempts = Arc::new(AtomicU32::new(0));
915        let attempts_clone = Arc::clone(&attempts);
916        let attempts_wait = Arc::clone(&attempts);
917
918        let handle = spawn(async move {
919            manager
920                .invocation(
921                    "test_error_delay_overshoot",
922                    move || {
923                        let attempts = Arc::clone(&attempts_clone);
924                        async move {
925                            attempts.fetch_add(1, Ordering::SeqCst);
926                            Err::<i32, TestError>(TestError::Retryable("rate limit".to_string()))
927                        }
928                    },
929                    should_retry_test_error,
930                    create_test_error,
931                )
932                .retry_delay(&|_| Some(Duration::from_millis(50)))
933                .execute()
934                .await
935        });
936
937        yield_until(|| attempts_wait.load(Ordering::SeqCst) == 1).await;
938        advance_clock(Duration::from_millis(100)).await;
939
940        let error = handle.await.unwrap().unwrap_err();
941
942        let TestError::Retryable(message) = error else {
943            panic!("expected original retryable error, was {error}");
944        };
945
946        assert_eq!(message, "rate limit");
947        assert_eq!(attempts.load(Ordering::SeqCst), 1);
948    }
949
950    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
951    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
952    async fn test_timeout_path() {
953        let config = RetryConfig {
954            max_retries: 2,
955            initial_delay_ms: 10,
956            max_delay_ms: 50,
957            backoff_factor: 2.0,
958            jitter_ms: 0,
959            operation_timeout_ms: Some(50),
960            immediate_first: false,
961            max_elapsed_ms: None,
962        };
963
964        let manager = RetryManager::new(config);
965
966        let result = manager
967            .invocation(
968                "test_timeout",
969                || async {
970                    time::sleep(Duration::from_millis(100)).await;
971                    Ok::<i32, TestError>(42)
972                },
973                should_retry_test_error,
974                create_test_error,
975            )
976            .execute()
977            .await;
978
979        let TestError::Timeout(reason) = result.unwrap_err() else {
980            panic!("expected operation timeout");
981        };
982
983        assert_eq!(reason, RetryError::OperationTimeout { timeout_ms: 50 });
984    }
985
986    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
987    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
988    async fn test_max_elapsed_time_budget() {
989        let config = RetryConfig {
990            max_retries: 10,
991            initial_delay_ms: 50,
992            max_delay_ms: 100,
993            backoff_factor: 2.0,
994            jitter_ms: 0,
995            operation_timeout_ms: None,
996            immediate_first: false,
997            max_elapsed_ms: Some(200),
998        };
999
1000        let manager = RetryManager::new(config);
1001
1002        let start = time::Instant::now();
1003        let result = manager
1004            .invocation(
1005                "test_budget",
1006                || async { Err::<i32, TestError>(TestError::Retryable("test".to_string())) },
1007                should_retry_test_error,
1008                create_test_error,
1009            )
1010            .execute()
1011            .await;
1012
1013        let elapsed = start.elapsed();
1014        assert!(result.is_err());
1015        assert!(matches!(result.unwrap_err(), TestError::Timeout(_)));
1016        assert!(elapsed.as_millis() >= 150);
1017        assert!(elapsed.as_millis() < 1000);
1018    }
1019
1020    #[rstest]
1021    #[case::without_operation_timeout(None)]
1022    #[case::at_operation_timeout(Some(100))]
1023    #[cfg_attr(
1024        not(all(feature = "simulation", madsim)),
1025        tokio::test(start_paused = true)
1026    )]
1027    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1028    async fn test_max_elapsed_bounds_in_flight_attempt(#[case] operation_timeout_ms: Option<u64>) {
1029        let config = RetryConfig {
1030            max_retries: 3,
1031            initial_delay_ms: 10,
1032            max_delay_ms: 20,
1033            backoff_factor: 1.0,
1034            jitter_ms: 0,
1035            operation_timeout_ms,
1036            immediate_first: false,
1037            max_elapsed_ms: Some(100),
1038        };
1039
1040        let manager = RetryManager::new(config);
1041        let attempts = Arc::new(AtomicU32::new(0));
1042        let attempts_clone = Arc::clone(&attempts);
1043        let completed = Arc::new(AtomicBool::new(false));
1044        let completed_clone = Arc::clone(&completed);
1045        let start = time::Instant::now();
1046
1047        let error = manager
1048            .invocation(
1049                "test_in_flight_budget",
1050                move || {
1051                    let attempts = Arc::clone(&attempts_clone);
1052                    let completed = Arc::clone(&completed_clone);
1053                    async move {
1054                        attempts.fetch_add(1, Ordering::SeqCst);
1055                        time::sleep(Duration::from_secs(1)).await;
1056                        completed.store(true, Ordering::SeqCst);
1057                        Ok::<i32, TestError>(42)
1058                    }
1059                },
1060                should_retry_test_error,
1061                create_test_error,
1062            )
1063            .execute()
1064            .await
1065            .unwrap_err();
1066
1067        let TestError::Timeout(reason) = error else {
1068            panic!("expected retry budget timeout, was {error}");
1069        };
1070
1071        assert_eq!(
1072            reason,
1073            RetryError::ElapsedBudgetExceeded {
1074                attempt: 1,
1075                max_attempts: 4,
1076                last_error: None,
1077            }
1078        );
1079        assert_eq!(attempts.load(Ordering::SeqCst), 1);
1080        assert!(!completed.load(Ordering::SeqCst));
1081        #[cfg(not(all(feature = "simulation", madsim)))]
1082        assert_eq!(start.elapsed(), Duration::from_millis(100));
1083        #[cfg(all(feature = "simulation", madsim))]
1084        assert!(
1085            start.elapsed() >= Duration::from_millis(100)
1086                && start.elapsed() < Duration::from_millis(101)
1087        );
1088    }
1089
1090    #[cfg_attr(
1091        not(all(feature = "simulation", madsim)),
1092        tokio::test(start_paused = true)
1093    )]
1094    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1095    async fn test_max_elapsed_bounds_later_in_flight_attempt() {
1096        let config = RetryConfig {
1097            max_retries: 3,
1098            initial_delay_ms: 10,
1099            max_delay_ms: 10,
1100            backoff_factor: 1.0,
1101            jitter_ms: 0,
1102            operation_timeout_ms: None,
1103            immediate_first: false,
1104            max_elapsed_ms: Some(100),
1105        };
1106
1107        let manager = RetryManager::new(config);
1108        let attempts = Arc::new(AtomicU32::new(0));
1109        let attempts_clone = Arc::clone(&attempts);
1110        let error = manager
1111            .invocation(
1112                "test_later_in_flight_budget",
1113                move || {
1114                    let attempt = attempts_clone.fetch_add(1, Ordering::SeqCst);
1115                    async move {
1116                        if attempt == 0 {
1117                            Err::<i32, TestError>(TestError::Retryable("first".to_string()))
1118                        } else {
1119                            std::future::pending().await
1120                        }
1121                    }
1122                },
1123                should_retry_test_error,
1124                create_test_error,
1125            )
1126            .execute()
1127            .await
1128            .unwrap_err();
1129
1130        let TestError::Timeout(reason) = error else {
1131            panic!("expected retry budget timeout, was {error}");
1132        };
1133
1134        assert_eq!(
1135            reason,
1136            RetryError::ElapsedBudgetExceeded {
1137                attempt: 2,
1138                max_attempts: 4,
1139                last_error: None,
1140            }
1141        );
1142        assert_eq!(attempts.load(Ordering::SeqCst), 2);
1143    }
1144
1145    #[cfg_attr(
1146        not(all(feature = "simulation", madsim)),
1147        tokio::test(start_paused = true)
1148    )]
1149    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1150    async fn test_cancellation_takes_precedence_when_total_deadline_is_ready() {
1151        let config = RetryConfig {
1152            max_retries: 3,
1153            initial_delay_ms: 10,
1154            max_delay_ms: 10,
1155            backoff_factor: 1.0,
1156            jitter_ms: 0,
1157            operation_timeout_ms: None,
1158            immediate_first: false,
1159            max_elapsed_ms: Some(100),
1160        };
1161
1162        let manager = RetryManager::new(config);
1163        let token = CancellationToken::new();
1164        let mut operation = Box::pin(
1165            manager
1166                .invocation(
1167                    "test_cancellation_at_deadline",
1168                    std::future::pending::<Result<i32, TestError>>,
1169                    should_retry_test_error,
1170                    create_test_error,
1171                )
1172                .cancellation_token(&token)
1173                .execute(),
1174        );
1175
1176        assert!(futures_util::poll!(&mut operation).is_pending());
1177        advance_clock(Duration::from_millis(100)).await;
1178        token.cancel();
1179
1180        let error = operation.await.unwrap_err();
1181
1182        let TestError::Timeout(reason) = error else {
1183            panic!("expected cancellation timeout, was {error}");
1184        };
1185
1186        assert_eq!(reason, RetryError::Canceled);
1187    }
1188
1189    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1190    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1191    async fn test_budget_exceeded_message_format() {
1192        let config = RetryConfig {
1193            max_retries: 5,
1194            initial_delay_ms: 10,
1195            max_delay_ms: 20,
1196            backoff_factor: 1.0,
1197            jitter_ms: 0,
1198            operation_timeout_ms: None,
1199            immediate_first: false,
1200            max_elapsed_ms: Some(35),
1201        };
1202
1203        let manager = RetryManager::new(config);
1204
1205        let result = manager
1206            .invocation(
1207                "test_budget_msg",
1208                || async { Err::<i32, TestError>(TestError::Retryable("test".to_string())) },
1209                should_retry_test_error,
1210                create_test_error,
1211            )
1212            .execute()
1213            .await;
1214
1215        assert!(result.is_err());
1216        let error_msg = result.unwrap_err().to_string();
1217
1218        assert!(error_msg.contains("Retry budget exceeded"));
1219        assert!(error_msg.contains("/6)"));
1220
1221        let prefix = "Timeout error: Retry budget exceeded (";
1222        let nums = error_msg
1223            .strip_circumfix(prefix, ")")
1224            .or_else(|| error_msg.strip_circumfix(prefix, "): last error: Retryable error: test"))
1225            .expect("error message should match retry budget format");
1226        let parts: Vec<&str> = nums.split('/').collect();
1227        assert_eq!(parts.len(), 2);
1228        let current: u32 = parts[0].parse().unwrap();
1229        let total: u32 = parts[1].parse().unwrap();
1230
1231        assert_eq!(total, 6, "Total should be max_retries + 1");
1232        assert!(current <= total, "Current attempt should not exceed total");
1233        assert!(current >= 1, "Current attempt should be at least 1");
1234    }
1235
1236    #[cfg_attr(
1237        not(all(feature = "simulation", madsim)),
1238        tokio::test(start_paused = true)
1239    )]
1240    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1241    async fn test_budget_exceeded_edge_cases() {
1242        let config = RetryConfig {
1243            max_retries: 2,
1244            initial_delay_ms: 50,
1245            max_delay_ms: 100,
1246            backoff_factor: 1.0,
1247            jitter_ms: 0,
1248            operation_timeout_ms: None,
1249            immediate_first: false,
1250            max_elapsed_ms: Some(100),
1251        };
1252
1253        let manager = RetryManager::new(config);
1254
1255        let attempt_count = Arc::new(AtomicU32::new(0));
1256        let count_clone = Arc::clone(&attempt_count);
1257
1258        let handle = spawn(async move {
1259            manager
1260                .invocation(
1261                    "test_first_attempt",
1262                    move || {
1263                        let count = Arc::clone(&count_clone);
1264                        async move {
1265                            count.fetch_add(1, Ordering::SeqCst);
1266                            Err::<i32, TestError>(TestError::Retryable("test".to_string()))
1267                        }
1268                    },
1269                    should_retry_test_error,
1270                    create_test_error,
1271                )
1272                .execute()
1273                .await
1274        });
1275
1276        // Wait for first attempt
1277        yield_until(|| attempt_count.load(Ordering::SeqCst) >= 1).await;
1278
1279        // Advance past budget to trigger check at loop start before second attempt
1280        advance_clock(Duration::from_millis(101)).await;
1281        yield_now().await;
1282
1283        let result = handle.await.unwrap();
1284        assert!(result.is_err());
1285        let error_msg = result.unwrap_err().to_string();
1286
1287        // Budget check happens at loop start, so shows (2/3) = "starting 2nd of 3 attempts"
1288        assert!(
1289            error_msg.contains("(2/3)"),
1290            "Expected (2/3) but got: {error_msg}"
1291        );
1292    }
1293
1294    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1295    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1296    async fn test_budget_exceeded_no_overflow() {
1297        let config = RetryConfig {
1298            max_retries: u32::MAX,
1299            initial_delay_ms: 10,
1300            max_delay_ms: 20,
1301            backoff_factor: 1.0,
1302            jitter_ms: 0,
1303            operation_timeout_ms: None,
1304            immediate_first: false,
1305            max_elapsed_ms: Some(1),
1306        };
1307
1308        let manager = RetryManager::new(config);
1309
1310        let result = manager
1311            .invocation(
1312                "test_overflow",
1313                || async { Err::<i32, TestError>(TestError::Retryable("test".to_string())) },
1314                should_retry_test_error,
1315                create_test_error,
1316            )
1317            .execute()
1318            .await;
1319
1320        assert!(result.is_err());
1321        let error_msg = result.unwrap_err().to_string();
1322
1323        // Should saturate at u32::MAX instead of wrapping to 0
1324        assert!(error_msg.contains("Retry budget exceeded"));
1325        assert!(error_msg.contains(&format!("/{}", u32::MAX)));
1326    }
1327
1328    #[rstest]
1329    fn test_http_retry_manager_config() {
1330        let manager = create_http_retry_manager::<TestError>();
1331        assert_eq!(manager.config.initial_delay_ms, 1_000);
1332        assert_eq!(manager.config.max_delay_ms, 10_000);
1333        #[allow(clippy::float_cmp, reason = "test asserts the preset backoff factor")]
1334        {
1335            assert_eq!(manager.config.backoff_factor, 2.0);
1336        }
1337
1338        assert_eq!(manager.config.jitter_ms, 1_000);
1339        assert_eq!(manager.config.operation_timeout_ms, Some(60_000));
1340        assert_eq!(manager.config.max_retries, 3);
1341        assert!(!manager.config.immediate_first);
1342        assert_eq!(manager.config.max_elapsed_ms, Some(180_000));
1343    }
1344
1345    #[rstest]
1346    fn test_websocket_retry_manager_config() {
1347        let manager = create_websocket_retry_manager::<TestError>();
1348        assert_eq!(manager.config.initial_delay_ms, 1_000);
1349        assert_eq!(manager.config.max_delay_ms, 10_000);
1350        #[allow(clippy::float_cmp, reason = "test asserts the preset backoff factor")]
1351        {
1352            assert_eq!(manager.config.backoff_factor, 2.0);
1353        }
1354
1355        assert_eq!(manager.config.jitter_ms, 1_000);
1356        assert_eq!(manager.config.operation_timeout_ms, Some(30_000));
1357        assert_eq!(manager.config.max_retries, 5);
1358        assert!(manager.config.immediate_first);
1359        assert_eq!(manager.config.max_elapsed_ms, Some(120_000));
1360    }
1361
1362    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1363    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1364    async fn test_timeout_respects_retry_predicate() {
1365        let config = RetryConfig {
1366            max_retries: 3,
1367            initial_delay_ms: 10,
1368            max_delay_ms: 50,
1369            backoff_factor: 2.0,
1370            jitter_ms: 0,
1371            operation_timeout_ms: Some(50),
1372            immediate_first: false,
1373            max_elapsed_ms: None,
1374        };
1375
1376        let manager = RetryManager::new(config);
1377
1378        // Test with retry predicate that rejects timeouts
1379        let should_not_retry_timeouts = |error: &TestError| !matches!(error, TestError::Timeout(_));
1380
1381        let result = manager
1382            .invocation(
1383                "test_timeout_non_retryable",
1384                || async {
1385                    time::sleep(Duration::from_millis(100)).await;
1386                    Ok::<i32, TestError>(42)
1387                },
1388                should_not_retry_timeouts,
1389                create_test_error,
1390            )
1391            .execute()
1392            .await;
1393
1394        // Should fail immediately without retries since timeout is non-retryable
1395        assert!(result.is_err());
1396        assert!(matches!(result.unwrap_err(), TestError::Timeout(_)));
1397    }
1398
1399    #[cfg_attr(
1400        not(all(feature = "simulation", madsim)),
1401        tokio::test(start_paused = true)
1402    )]
1403    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1404    async fn test_timeout_retries_when_predicate_allows() {
1405        let config = RetryConfig {
1406            max_retries: 2,
1407            initial_delay_ms: 10,
1408            max_delay_ms: 50,
1409            backoff_factor: 2.0,
1410            jitter_ms: 0,
1411            operation_timeout_ms: Some(50),
1412            immediate_first: false,
1413            max_elapsed_ms: None,
1414        };
1415
1416        let manager = RetryManager::new(config);
1417
1418        let should_retry_timeouts = |error: &TestError| matches!(error, TestError::Timeout(_));
1419        let mut attempts = 0;
1420        let result = manager
1421            .invocation(
1422                "test_timeout_retryable",
1423                || {
1424                    attempts += 1;
1425                    std::future::pending::<Result<i32, TestError>>()
1426                },
1427                should_retry_timeouts,
1428                create_test_error,
1429            )
1430            .execute()
1431            .await;
1432
1433        let TestError::Timeout(reason) = result.unwrap_err() else {
1434            panic!("expected operation timeout");
1435        };
1436
1437        assert_eq!(reason, RetryError::OperationTimeout { timeout_ms: 50 });
1438        assert_eq!(attempts, 3);
1439    }
1440
1441    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1442    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1443    async fn test_successful_retry_after_failures() {
1444        let config = RetryConfig {
1445            max_retries: 3,
1446            initial_delay_ms: 10,
1447            max_delay_ms: 50,
1448            backoff_factor: 2.0,
1449            jitter_ms: 0,
1450            operation_timeout_ms: None,
1451            immediate_first: false,
1452            max_elapsed_ms: None,
1453        };
1454
1455        let manager = RetryManager::new(config);
1456
1457        let attempt_counter = Arc::new(AtomicU32::new(0));
1458        let counter_clone = Arc::clone(&attempt_counter);
1459        let result = manager
1460            .invocation(
1461                "test_eventual_success",
1462                move || {
1463                    let counter = Arc::clone(&counter_clone);
1464                    async move {
1465                        let attempts = counter.fetch_add(1, Ordering::SeqCst);
1466                        if attempts < 2 {
1467                            Err(TestError::Retryable("temporary failure".to_string()))
1468                        } else {
1469                            Ok(42)
1470                        }
1471                    }
1472                },
1473                should_retry_test_error,
1474                create_test_error,
1475            )
1476            .execute()
1477            .await;
1478
1479        assert_eq!(result.unwrap(), 42);
1480        assert_eq!(attempt_counter.load(Ordering::SeqCst), 3);
1481    }
1482
1483    #[cfg_attr(
1484        not(all(feature = "simulation", madsim)),
1485        tokio::test(start_paused = true)
1486    )]
1487    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1488    async fn test_immediate_first_retry() {
1489        let config = RetryConfig {
1490            max_retries: 2,
1491            initial_delay_ms: 100,
1492            max_delay_ms: 200,
1493            backoff_factor: 2.0,
1494            jitter_ms: 0,
1495            operation_timeout_ms: None,
1496            immediate_first: true,
1497            max_elapsed_ms: None,
1498        };
1499
1500        let manager = RetryManager::new(config);
1501
1502        let attempt_times = Arc::new(parking_lot::Mutex::new(Vec::new()));
1503        let times_clone = Arc::clone(&attempt_times);
1504        let start = time::Instant::now();
1505
1506        let handle = spawn({
1507            let times_clone = Arc::clone(&times_clone);
1508            async move {
1509                let _ = manager
1510                    .invocation(
1511                        "test_immediate",
1512                        move || {
1513                            let times = Arc::clone(&times_clone);
1514                            async move {
1515                                times.lock().push(start.elapsed());
1516                                Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
1517                            }
1518                        },
1519                        should_retry_test_error,
1520                        create_test_error,
1521                    )
1522                    .execute()
1523                    .await;
1524            }
1525        });
1526
1527        // Allow initial attempt and immediate retry to run without advancing time
1528        yield_until(|| attempt_times.lock().len() >= 2).await;
1529
1530        // Advance time for the next backoff interval
1531        advance_clock(Duration::from_millis(100)).await;
1532        yield_now().await;
1533
1534        // Wait for the final retry to be recorded
1535        yield_until(|| attempt_times.lock().len() >= 3).await;
1536
1537        handle.await.unwrap();
1538
1539        let times = attempt_times.lock();
1540        assert_eq!(times.len(), 3); // Initial + 2 retries
1541
1542        // First retry should be immediate (within 1ms tolerance)
1543        assert!(times[1] <= Duration::from_millis(1));
1544        // Second retry should have backoff delay (at least 100ms from start)
1545        assert!(times[2] >= Duration::from_millis(100));
1546        assert!(times[2] <= Duration::from_millis(110));
1547    }
1548
1549    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1550    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1551    async fn test_operation_without_timeout() {
1552        let config = RetryConfig {
1553            max_retries: 2,
1554            initial_delay_ms: 10,
1555            max_delay_ms: 50,
1556            backoff_factor: 2.0,
1557            jitter_ms: 0,
1558            operation_timeout_ms: None, // No timeout
1559            immediate_first: false,
1560            max_elapsed_ms: None,
1561        };
1562
1563        let manager = RetryManager::new(config);
1564
1565        let start = time::Instant::now();
1566
1567        let result = manager
1568            .invocation(
1569                "test_no_timeout",
1570                || async {
1571                    time::sleep(Duration::from_millis(50)).await;
1572                    Ok::<i32, TestError>(42)
1573                },
1574                should_retry_test_error,
1575                create_test_error,
1576            )
1577            .execute()
1578            .await;
1579
1580        let elapsed = start.elapsed();
1581        assert_eq!(result.unwrap(), 42);
1582        // Should complete without timing out
1583        assert!(elapsed.as_millis() >= 30);
1584        assert!(elapsed.as_millis() < 200);
1585    }
1586
1587    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1588    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1589    async fn test_zero_retries() {
1590        let config = RetryConfig {
1591            max_retries: 0,
1592            initial_delay_ms: 10,
1593            max_delay_ms: 50,
1594            backoff_factor: 2.0,
1595            jitter_ms: 0,
1596            operation_timeout_ms: None,
1597            immediate_first: false,
1598            max_elapsed_ms: None,
1599        };
1600
1601        let manager = RetryManager::new(config);
1602
1603        let attempt_counter = Arc::new(AtomicU32::new(0));
1604        let counter_clone = Arc::clone(&attempt_counter);
1605        let result = manager
1606            .invocation(
1607                "test_no_retries",
1608                move || {
1609                    let counter = Arc::clone(&counter_clone);
1610                    async move {
1611                        counter.fetch_add(1, Ordering::SeqCst);
1612                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
1613                    }
1614                },
1615                should_retry_test_error,
1616                create_test_error,
1617            )
1618            .execute()
1619            .await;
1620
1621        assert!(result.is_err());
1622        // Should only attempt once (no retries)
1623        assert_eq!(attempt_counter.load(Ordering::SeqCst), 1);
1624    }
1625
1626    #[cfg_attr(
1627        not(all(feature = "simulation", madsim)),
1628        tokio::test(start_paused = true)
1629    )]
1630    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1631    async fn test_jitter_applied() {
1632        let config = RetryConfig {
1633            max_retries: 2,
1634            initial_delay_ms: 50,
1635            max_delay_ms: 100,
1636            backoff_factor: 2.0,
1637            jitter_ms: 50, // Significant jitter
1638            operation_timeout_ms: None,
1639            immediate_first: false,
1640            max_elapsed_ms: None,
1641        };
1642
1643        let manager = RetryManager::new(config);
1644
1645        let delays = Arc::new(parking_lot::Mutex::new(Vec::new()));
1646        let delays_clone = Arc::clone(&delays);
1647        let last_time = Arc::new(parking_lot::Mutex::new(time::Instant::now()));
1648        let last_time_clone = Arc::clone(&last_time);
1649
1650        let handle = spawn({
1651            let delays_clone = Arc::clone(&delays_clone);
1652            async move {
1653                let _ = manager
1654                    .invocation(
1655                        "test_jitter",
1656                        move || {
1657                            let delays = Arc::clone(&delays_clone);
1658                            let last_time = Arc::clone(&last_time_clone);
1659                            async move {
1660                                let now = time::Instant::now();
1661
1662                                let delay = {
1663                                    let mut last = last_time.lock();
1664                                    let d = now.duration_since(*last);
1665                                    *last = now;
1666                                    d
1667                                };
1668
1669                                delays.lock().push(delay);
1670                                Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
1671                            }
1672                        },
1673                        should_retry_test_error,
1674                        create_test_error,
1675                    )
1676                    .execute()
1677                    .await;
1678            }
1679        });
1680
1681        yield_until(|| !delays.lock().is_empty()).await;
1682        advance_until(|| delays.lock().len() >= 2).await;
1683        advance_until(|| delays.lock().len() >= 3).await;
1684
1685        handle.await.unwrap();
1686
1687        let delays = delays.lock();
1688        // Skip the first delay (initial attempt)
1689        for delay in delays.iter().skip(1) {
1690            // Each delay should be at least the base delay (50ms for first retry)
1691            assert!(delay.as_millis() >= 50);
1692            // But no more than base + jitter (allow small tolerance for step advance)
1693            assert!(delay.as_millis() <= 151);
1694        }
1695    }
1696
1697    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1698    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1699    async fn test_max_elapsed_stops_early() {
1700        let config = RetryConfig {
1701            max_retries: 100, // Very high retry count
1702            initial_delay_ms: 50,
1703            max_delay_ms: 100,
1704            backoff_factor: 1.5,
1705            jitter_ms: 0,
1706            operation_timeout_ms: None,
1707            immediate_first: false,
1708            max_elapsed_ms: Some(150), // Should stop after ~3 attempts
1709        };
1710
1711        let manager = RetryManager::new(config);
1712
1713        let attempt_counter = Arc::new(AtomicU32::new(0));
1714        let counter_clone = Arc::clone(&attempt_counter);
1715
1716        let start = time::Instant::now();
1717
1718        let result = manager
1719            .invocation(
1720                "test_elapsed_limit",
1721                move || {
1722                    let counter = Arc::clone(&counter_clone);
1723                    async move {
1724                        counter.fetch_add(1, Ordering::SeqCst);
1725                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
1726                    }
1727                },
1728                should_retry_test_error,
1729                create_test_error,
1730            )
1731            .execute()
1732            .await;
1733
1734        let elapsed = start.elapsed();
1735        assert!(result.is_err());
1736        assert!(matches!(result.unwrap_err(), TestError::Timeout(_)));
1737
1738        // Should have stopped due to time limit, not retry count
1739        let attempts = attempt_counter.load(Ordering::SeqCst);
1740        assert!(attempts < 10); // Much less than max_retries
1741        assert!(elapsed.as_millis() >= 100);
1742    }
1743
1744    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1745    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1746    async fn test_mixed_errors_retry_behavior() {
1747        let config = RetryConfig {
1748            max_retries: 5,
1749            initial_delay_ms: 10,
1750            max_delay_ms: 50,
1751            backoff_factor: 2.0,
1752            jitter_ms: 0,
1753            operation_timeout_ms: None,
1754            immediate_first: false,
1755            max_elapsed_ms: None,
1756        };
1757
1758        let manager = RetryManager::new(config);
1759
1760        let attempt_counter = Arc::new(AtomicU32::new(0));
1761        let counter_clone = Arc::clone(&attempt_counter);
1762        let result = manager
1763            .invocation(
1764                "test_mixed_errors",
1765                move || {
1766                    let counter = Arc::clone(&counter_clone);
1767                    async move {
1768                        let attempts = counter.fetch_add(1, Ordering::SeqCst);
1769                        match attempts {
1770                            0 => Err(TestError::Retryable("retry 1".to_string())),
1771                            1 => Err(TestError::Retryable("retry 2".to_string())),
1772                            2 => Err(TestError::NonRetryable("stop here".to_string())),
1773                            _ => Ok(42),
1774                        }
1775                    }
1776                },
1777                should_retry_test_error,
1778                create_test_error,
1779            )
1780            .execute()
1781            .await;
1782
1783        assert!(result.is_err());
1784        assert!(matches!(result.unwrap_err(), TestError::NonRetryable(_)));
1785        // Should stop at the non-retryable error
1786        assert_eq!(attempt_counter.load(Ordering::SeqCst), 3);
1787    }
1788
1789    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1790    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1791    async fn test_cancellation_during_retry_delay() {
1792        use tokio_util::sync::CancellationToken;
1793
1794        let config = RetryConfig {
1795            max_retries: 10,
1796            initial_delay_ms: 500, // Long delay to ensure cancellation happens during sleep
1797            max_delay_ms: 1000,
1798            backoff_factor: 2.0,
1799            jitter_ms: 0,
1800            operation_timeout_ms: None,
1801            immediate_first: false,
1802            max_elapsed_ms: None,
1803        };
1804
1805        let manager = RetryManager::new(config);
1806
1807        let token = CancellationToken::new();
1808        let token_clone = token.clone();
1809
1810        // Cancel after a short delay
1811        spawn(async move {
1812            time::sleep(Duration::from_millis(100)).await;
1813            token_clone.cancel();
1814        });
1815
1816        let attempt_counter = Arc::new(AtomicU32::new(0));
1817        let counter_clone = Arc::clone(&attempt_counter);
1818
1819        let start = time::Instant::now();
1820
1821        let result = manager
1822            .invocation(
1823                "test_cancellation",
1824                move || {
1825                    let counter = Arc::clone(&counter_clone);
1826                    async move {
1827                        counter.fetch_add(1, Ordering::SeqCst);
1828                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
1829                    }
1830                },
1831                should_retry_test_error,
1832                create_test_error,
1833            )
1834            .cancellation_token(&token)
1835            .execute()
1836            .await;
1837
1838        let elapsed = start.elapsed();
1839
1840        // Should be canceled quickly
1841        assert!(result.is_err());
1842        let error_msg = format!("{}", result.unwrap_err());
1843        assert!(error_msg.contains("canceled"));
1844
1845        // Should not have taken the full delay time
1846        assert!(elapsed.as_millis() < 600);
1847
1848        // Should have made at least one attempt
1849        let attempts = attempt_counter.load(Ordering::SeqCst);
1850        assert!(attempts >= 1);
1851    }
1852
1853    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1854    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1855    async fn test_cancellation_during_operation_execution() {
1856        use tokio_util::sync::CancellationToken;
1857
1858        let config = RetryConfig {
1859            max_retries: 5,
1860            initial_delay_ms: 50,
1861            max_delay_ms: 100,
1862            backoff_factor: 2.0,
1863            jitter_ms: 0,
1864            operation_timeout_ms: None,
1865            immediate_first: false,
1866            max_elapsed_ms: None,
1867        };
1868
1869        let manager = RetryManager::new(config);
1870
1871        let token = CancellationToken::new();
1872        let token_clone = token.clone();
1873
1874        // Cancel after a short delay
1875        spawn(async move {
1876            time::sleep(Duration::from_millis(50)).await;
1877            token_clone.cancel();
1878        });
1879
1880        let start = time::Instant::now();
1881
1882        let result = manager
1883            .invocation(
1884                "test_cancellation_during_op",
1885                || async {
1886                    // Long-running operation
1887                    time::sleep(Duration::from_millis(200)).await;
1888                    Ok::<i32, TestError>(42)
1889                },
1890                should_retry_test_error,
1891                create_test_error,
1892            )
1893            .cancellation_token(&token)
1894            .execute()
1895            .await;
1896
1897        let elapsed = start.elapsed();
1898
1899        // Should be canceled during the operation
1900        assert!(result.is_err());
1901        let error_msg = format!("{}", result.unwrap_err());
1902        assert!(error_msg.contains("canceled"));
1903
1904        // Should not have completed the long operation
1905        assert!(elapsed.as_millis() < 250);
1906    }
1907
1908    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
1909    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
1910    async fn test_cancellation_error_message() {
1911        use tokio_util::sync::CancellationToken;
1912
1913        let config = RetryConfig::default();
1914        let manager = RetryManager::new(config);
1915
1916        let token = CancellationToken::new();
1917        token.cancel(); // Pre-cancel for immediate cancellation
1918
1919        let result = manager
1920            .invocation(
1921                "test_operation",
1922                || async { Ok::<i32, TestError>(42) },
1923                should_retry_test_error,
1924                create_test_error,
1925            )
1926            .cancellation_token(&token)
1927            .execute()
1928            .await;
1929
1930        assert!(result.is_err());
1931        let error_msg = format!("{}", result.unwrap_err());
1932        assert!(error_msg.contains("canceled"));
1933    }
1934}
1935
1936#[cfg(test)]
1937mod proptest_tests {
1938    use std::sync::{
1939        Arc,
1940        atomic::{AtomicU32, Ordering},
1941    };
1942
1943    #[cfg(all(feature = "simulation", madsim))]
1944    use madsim::task::spawn;
1945    use proptest::prelude::*;
1946    // Import rstest attribute macro used within proptest! tests
1947    use rstest::rstest;
1948    #[cfg(not(all(feature = "simulation", madsim)))]
1949    use tokio::task::spawn;
1950
1951    #[cfg(not(all(feature = "simulation", madsim)))]
1952    use super::tests::{advance_until, yield_until};
1953    use super::{test_utils::*, tests::advance_clock, *};
1954    use crate::dst::time;
1955
1956    // Each proptest case constructs a runtime to drive the manager via
1957    // `block_on`. Under tokio, that runtime is paused so virtual sleeps
1958    // auto-advance; under madsim, the runtime is the deterministic sim
1959    // runtime, which also runs in virtual time. Both expose `block_on`.
1960    #[cfg(all(feature = "simulation", madsim))]
1961    fn build_paused_runtime() -> madsim::runtime::Runtime {
1962        madsim::runtime::Runtime::new()
1963    }
1964
1965    #[cfg(not(all(feature = "simulation", madsim)))]
1966    fn build_paused_runtime() -> tokio::runtime::Runtime {
1967        tokio::runtime::Builder::new_current_thread()
1968            .enable_time()
1969            .start_paused(true)
1970            .build()
1971            .unwrap()
1972    }
1973
1974    proptest! {
1975        #[rstest]
1976        fn test_retry_config_valid_ranges(
1977            max_retries in 0u32..100,
1978            initial_delay_ms in 1u64..10_000,
1979            max_delay_ms in 1u64..60_000,
1980            backoff_factor in 1.0f64..10.0,
1981            jitter_ms in 0u64..1_000,
1982            operation_timeout_ms in prop::option::of(1u64..120_000),
1983            immediate_first in any::<bool>(),
1984            max_elapsed_ms in prop::option::of(1u64..300_000)
1985        ) {
1986            // Ensure max_delay >= initial_delay for valid config
1987            let max_delay_ms = max_delay_ms.max(initial_delay_ms);
1988
1989            let config = RetryConfig {
1990                max_retries,
1991                initial_delay_ms,
1992                max_delay_ms,
1993                backoff_factor,
1994                jitter_ms,
1995                operation_timeout_ms,
1996                immediate_first,
1997                max_elapsed_ms,
1998            };
1999
2000            // Should always be able to create a RetryManager with valid config
2001            let _manager = RetryManager::<std::io::Error>::new(config);
2002        }
2003
2004        #[rstest]
2005        fn test_retry_attempts_bounded(
2006            max_retries in 0u32..5,
2007            initial_delay_ms in 1u64..10,
2008            backoff_factor in 1.0f64..2.0,
2009        ) {
2010            let rt = build_paused_runtime();
2011
2012            let config = RetryConfig {
2013                max_retries,
2014                initial_delay_ms,
2015                max_delay_ms: initial_delay_ms * 2,
2016                backoff_factor,
2017                jitter_ms: 0,
2018                operation_timeout_ms: None,
2019                immediate_first: false,
2020                max_elapsed_ms: None,
2021            };
2022
2023            let manager = RetryManager::new(config);
2024            let attempt_counter = Arc::new(AtomicU32::new(0));
2025            let counter_clone = Arc::clone(&attempt_counter);
2026
2027            let _result = rt.block_on(manager.invocation(
2028                "prop_test",
2029                move || {
2030                    let counter = Arc::clone(&counter_clone);
2031                    async move {
2032                        counter.fetch_add(1, Ordering::SeqCst);
2033                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2034                    }
2035                },
2036                |e: &TestError| matches!(e, TestError::Retryable(_)),
2037                TestError::Timeout,
2038            ).execute());
2039
2040            let attempts = attempt_counter.load(Ordering::SeqCst);
2041            // Total attempts should be 1 (initial) + max_retries
2042            prop_assert_eq!(attempts, max_retries + 1);
2043        }
2044
2045        #[rstest]
2046        fn test_error_retry_delay_obeys_selection_and_budget(
2047            backoff_ms in 1u64..500,
2048            minimum_ms in 0u64..1_000,
2049            operation_timeout_ms in 1u64..50,
2050            timeout in any::<bool>(),
2051        ) {
2052            let rt = build_paused_runtime();
2053            let selected_ms = backoff_ms.max(minimum_ms);
2054            let attempt_ms = if timeout { operation_timeout_ms } else { 0 };
2055            let elapsed_ms = attempt_ms + selected_ms;
2056            let config = |max_elapsed_ms| RetryConfig {
2057                max_retries: 1,
2058                initial_delay_ms: backoff_ms,
2059                max_delay_ms: backoff_ms,
2060                backoff_factor: 1.0,
2061                jitter_ms: 0,
2062                operation_timeout_ms: Some(operation_timeout_ms),
2063                immediate_first: false,
2064                max_elapsed_ms: Some(max_elapsed_ms),
2065            };
2066            let minimum_delay = Duration::from_millis(minimum_ms);
2067
2068            let manager = RetryManager::new(config(elapsed_ms + 1));
2069            let attempts = Arc::new(AtomicU32::new(0));
2070            let attempts_clone = Arc::clone(&attempts);
2071            let (result, elapsed) = rt.block_on(async {
2072                let start = time::Instant::now();
2073                let result = manager
2074                    .invocation(
2075                        "prop_error_delay_selection",
2076                        move || {
2077                            let attempts = Arc::clone(&attempts_clone);
2078                            async move {
2079                                if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
2080                                    if timeout {
2081                                        std::future::pending::<()>().await;
2082                                    }
2083
2084                                    Err(TestError::Retryable("rate limit".to_string()))
2085                                } else {
2086                                    Ok(42)
2087                                }
2088                            }
2089                        },
2090                        |_| true,
2091                        create_test_error,
2092                    )
2093                    .retry_delay(&|_| Some(minimum_delay))
2094                    .execute()
2095                    .await;
2096                (result, start.elapsed())
2097            });
2098
2099            prop_assert_eq!(result.unwrap(), 42);
2100            prop_assert_eq!(attempts.load(Ordering::SeqCst), 2);
2101            let expected_elapsed = Duration::from_millis(elapsed_ms);
2102            #[cfg(all(feature = "simulation", madsim))]
2103            {
2104                prop_assert!(elapsed >= expected_elapsed);
2105                prop_assert!(elapsed < expected_elapsed + Duration::from_millis(1));
2106            }
2107            #[cfg(not(all(feature = "simulation", madsim)))]
2108            prop_assert_eq!(elapsed, expected_elapsed);
2109
2110            let manager = RetryManager::new(config(elapsed_ms));
2111            let attempts = Arc::new(AtomicU32::new(0));
2112            let attempts_clone = Arc::clone(&attempts);
2113            let (error, elapsed) = rt.block_on(async {
2114                let start = time::Instant::now();
2115                let error = manager
2116                    .invocation(
2117                        "prop_error_delay_budget",
2118                        move || {
2119                            let attempts = Arc::clone(&attempts_clone);
2120                            async move {
2121                                attempts.fetch_add(1, Ordering::SeqCst);
2122
2123                                if timeout {
2124                                    std::future::pending::<()>().await;
2125                                }
2126
2127                                Err::<i32, TestError>(TestError::Retryable(
2128                                    "rate limit".to_string(),
2129                                ))
2130                            }
2131                        },
2132                        |_| true,
2133                        create_test_error,
2134                    )
2135                    .retry_delay(&|_| Some(minimum_delay))
2136                    .execute()
2137                    .await
2138                    .unwrap_err();
2139                (error, start.elapsed())
2140            });
2141
2142            match error {
2143                TestError::Retryable(message) if !timeout => {
2144                    prop_assert_eq!(message, "rate limit");
2145                }
2146                TestError::Timeout(RetryError::OperationTimeout { timeout_ms }) if timeout => {
2147                    prop_assert_eq!(timeout_ms, operation_timeout_ms);
2148                }
2149                e => prop_assert!(false, "expected original operation error, was {e}"),
2150            }
2151            prop_assert_eq!(attempts.load(Ordering::SeqCst), 1);
2152            let expected_elapsed = Duration::from_millis(attempt_ms);
2153            #[cfg(all(feature = "simulation", madsim))]
2154            if timeout {
2155                prop_assert!(elapsed >= expected_elapsed);
2156                prop_assert!(elapsed < expected_elapsed + Duration::from_millis(1));
2157            } else {
2158                prop_assert_eq!(elapsed, expected_elapsed);
2159            }
2160            #[cfg(not(all(feature = "simulation", madsim)))]
2161            prop_assert_eq!(elapsed, expected_elapsed);
2162        }
2163
2164        #[rstest]
2165        fn test_timeout_always_respected(
2166            timeout_ms in 10u64..50,
2167            operation_delay_ms in 60u64..100,
2168        ) {
2169            let rt = build_paused_runtime();
2170
2171            let config = RetryConfig {
2172                max_retries: 0, // No retries to isolate timeout behavior
2173                initial_delay_ms: 10,
2174                max_delay_ms: 100,
2175                backoff_factor: 2.0,
2176                jitter_ms: 0,
2177                operation_timeout_ms: Some(timeout_ms),
2178                immediate_first: false,
2179                max_elapsed_ms: None,
2180            };
2181
2182            let manager = RetryManager::new(config);
2183
2184            let result = rt.block_on(async {
2185                let operation_future = manager.invocation(
2186                    "timeout_test",
2187                    move || async move {
2188                        time::sleep(Duration::from_millis(operation_delay_ms)).await;
2189                        Ok::<i32, TestError>(42)
2190                    },
2191                    |_: &TestError| true,
2192                    TestError::Timeout,
2193                ).execute();
2194
2195                // Advance time to trigger timeout
2196                advance_clock(Duration::from_millis(timeout_ms + 10)).await;
2197                operation_future.await
2198            });
2199
2200            // Operation should timeout
2201            prop_assert!(result.is_err());
2202            prop_assert!(matches!(result.unwrap_err(), TestError::Timeout(_)));
2203        }
2204
2205        #[rstest]
2206        fn test_max_elapsed_always_respected(
2207            max_elapsed_ms in 20u64..50,
2208            delay_per_retry in 15u64..30,
2209            max_retries in 10u32..20,
2210        ) {
2211            let rt = build_paused_runtime();
2212
2213            // Set up config where we would exceed max_elapsed_ms before max_retries
2214            let config = RetryConfig {
2215                max_retries,
2216                initial_delay_ms: delay_per_retry,
2217                max_delay_ms: delay_per_retry * 2,
2218                backoff_factor: 1.0, // No backoff to make timing predictable
2219                jitter_ms: 0,
2220                operation_timeout_ms: None,
2221                immediate_first: false,
2222                max_elapsed_ms: Some(max_elapsed_ms),
2223            };
2224
2225            let manager = RetryManager::new(config);
2226            let attempt_counter = Arc::new(AtomicU32::new(0));
2227            let counter_clone = Arc::clone(&attempt_counter);
2228
2229            let result = rt.block_on(async {
2230                let operation_future = manager.invocation(
2231                    "elapsed_test",
2232                    move || {
2233                        let counter = Arc::clone(&counter_clone);
2234                        async move {
2235                            counter.fetch_add(1, Ordering::SeqCst);
2236                            Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2237                        }
2238                    },
2239                    |e: &TestError| matches!(e, TestError::Retryable(_)),
2240                    TestError::Timeout,
2241                ).execute();
2242
2243                // Advance time past max_elapsed_ms
2244                advance_clock(Duration::from_millis(max_elapsed_ms + delay_per_retry)).await;
2245                operation_future.await
2246            });
2247
2248            let attempts = attempt_counter.load(Ordering::SeqCst);
2249
2250            // Should have failed with timeout error
2251            prop_assert!(result.is_err());
2252            prop_assert!(matches!(result.unwrap_err(), TestError::Timeout(_)));
2253
2254            // Should have stopped before exhausting all retries
2255            prop_assert!(attempts <= max_retries + 1);
2256        }
2257
2258        #[rstest]
2259        fn test_jitter_bounds(
2260            jitter_ms in 0u64..20,
2261            base_delay_ms in 10u64..30,
2262        ) {
2263            let rt = build_paused_runtime();
2264
2265            let config = RetryConfig {
2266                max_retries: 2,
2267                initial_delay_ms: base_delay_ms,
2268                max_delay_ms: base_delay_ms * 2,
2269                backoff_factor: 1.0, // No backoff to isolate jitter
2270                jitter_ms,
2271                operation_timeout_ms: None,
2272                immediate_first: false,
2273                max_elapsed_ms: None,
2274            };
2275
2276            let manager = RetryManager::new(config);
2277            let attempt_times = Arc::new(parking_lot::Mutex::new(Vec::new()));
2278            let attempt_times_for_block = Arc::clone(&attempt_times);
2279
2280            rt.block_on(async move {
2281                #[cfg(not(all(feature = "simulation", madsim)))]
2282                let attempt_times_for_wait = Arc::clone(&attempt_times_for_block);
2283                let handle = spawn({
2284                    let attempt_times_for_task = Arc::clone(&attempt_times_for_block);
2285                    let manager = manager;
2286                    async move {
2287                        let start_time = time::Instant::now();
2288                        let _ = manager
2289                            .invocation(
2290                                "jitter_test",
2291                                move || {
2292                                    let attempt_times_inner = Arc::clone(&attempt_times_for_task);
2293                                    async move {
2294                                        attempt_times_inner
2295                                            .lock()
2296                                            .push(start_time.elapsed());
2297                                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2298                                    }
2299                                },
2300                                |e: &TestError| matches!(e, TestError::Retryable(_)),
2301                                TestError::Timeout,
2302                            ).execute()
2303                            .await;
2304                    }
2305                });
2306
2307                // Under tokio paused clock, drive virtual time forward in 1ms
2308                // ticks to release the manager's sleeps; under madsim the
2309                // runtime auto-advances when all tasks block on virtual time,
2310                // so awaiting the handle is enough and yields exact timings.
2311                #[cfg(not(all(feature = "simulation", madsim)))]
2312                {
2313                    yield_until(|| !attempt_times_for_wait.lock().is_empty()).await;
2314                    advance_until(|| attempt_times_for_wait.lock().len() >= 2).await;
2315                    advance_until(|| attempt_times_for_wait.lock().len() >= 3).await;
2316                }
2317
2318                handle.await.unwrap();
2319            });
2320
2321            let times = attempt_times.lock();
2322
2323            // We expect at least 2 attempts total (initial + at least 1 retry)
2324            prop_assert!(times.len() >= 2);
2325
2326            // First attempt should be immediate (no delay)
2327            prop_assert!(times[0].as_millis() < 5);
2328
2329            // Check subsequent retries have appropriate delays
2330            for i in 1..times.len() {
2331                let delay_from_previous = if i == 1 {
2332                    times[i].checked_sub(times[0]).unwrap()
2333                } else {
2334                    times[i].checked_sub(times[i - 1]).unwrap()
2335                };
2336
2337                // The delay floor is min(base, max - jitter): near the cap the
2338                // jittered base is lowered so the spread survives saturation
2339                let floor = base_delay_ms.min((base_delay_ms * 2).saturating_sub(jitter_ms));
2340                prop_assert!(
2341                    delay_from_previous.as_millis() >= u128::from(floor),
2342                    "Retry {} delay {}ms is less than floor {}ms",
2343                    i, delay_from_previous.as_millis(), floor
2344                );
2345
2346                // Delay should be at most base_delay + jitter
2347                prop_assert!(
2348                    delay_from_previous.as_millis() <= u128::from(base_delay_ms + jitter_ms + 1),
2349                    "Retry {} delay {}ms exceeds base {} + jitter {}",
2350                    i, delay_from_previous.as_millis(), base_delay_ms, jitter_ms
2351                );
2352            }
2353        }
2354
2355        #[rstest]
2356        fn test_immediate_first_property(
2357            immediate_first in any::<bool>(),
2358            initial_delay_ms in 10u64..30,
2359        ) {
2360            let rt = build_paused_runtime();
2361
2362            let config = RetryConfig {
2363                max_retries: 2,
2364                initial_delay_ms,
2365                max_delay_ms: initial_delay_ms * 2,
2366                backoff_factor: 2.0,
2367                jitter_ms: 0,
2368                operation_timeout_ms: None,
2369                immediate_first,
2370                max_elapsed_ms: None,
2371            };
2372
2373            let manager = RetryManager::new(config);
2374            let attempt_times = Arc::new(parking_lot::Mutex::new(Vec::new()));
2375            let attempt_times_for_block = Arc::clone(&attempt_times);
2376
2377            rt.block_on(async move {
2378                #[cfg(not(all(feature = "simulation", madsim)))]
2379                let attempt_times_for_wait = Arc::clone(&attempt_times_for_block);
2380                let handle = spawn({
2381                    let attempt_times_for_task = Arc::clone(&attempt_times_for_block);
2382                    let manager = manager;
2383                    async move {
2384                        let start = time::Instant::now();
2385                        let _ = manager
2386                            .invocation(
2387                                "immediate_test",
2388                                move || {
2389                                    let attempt_times_inner = Arc::clone(&attempt_times_for_task);
2390                                    async move {
2391                                        let elapsed = start.elapsed();
2392                                        attempt_times_inner.lock().push(elapsed);
2393                                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2394                                    }
2395                                },
2396                                |e: &TestError| matches!(e, TestError::Retryable(_)),
2397                                TestError::Timeout,
2398                            ).execute()
2399                            .await;
2400                    }
2401                });
2402
2403                // See test_jitter_bounds: madsim auto-advances virtual time
2404                // when all tasks block on it, so awaiting the handle suffices
2405                // and avoids the 1ms-tick driver's added scheduler overhead.
2406                #[cfg(not(all(feature = "simulation", madsim)))]
2407                {
2408                    yield_until(|| !attempt_times_for_wait.lock().is_empty()).await;
2409                    advance_until(|| attempt_times_for_wait.lock().len() >= 2).await;
2410                    advance_until(|| attempt_times_for_wait.lock().len() >= 3).await;
2411                }
2412
2413                handle.await.unwrap();
2414            });
2415
2416            let times = attempt_times.lock();
2417            prop_assert!(times.len() >= 2);
2418
2419            if immediate_first {
2420                // First retry should be immediate
2421                prop_assert!(times[1].as_millis() < 20,
2422                    "With immediate_first=true, first retry took {}ms",
2423                    times[1].as_millis());
2424            } else {
2425                // First retry should have delay
2426                prop_assert!(times[1].as_millis() >= u128::from(initial_delay_ms - 1),
2427                    "With immediate_first=false, first retry was too fast: {}ms",
2428                    times[1].as_millis());
2429            }
2430        }
2431
2432        #[rstest]
2433        fn test_non_retryable_stops_immediately(
2434            attempt_before_non_retryable in 0usize..3,
2435            max_retries in 3u32..5,
2436        ) {
2437            let rt = build_paused_runtime();
2438
2439            let config = RetryConfig {
2440                max_retries,
2441                initial_delay_ms: 10,
2442                max_delay_ms: 100,
2443                backoff_factor: 2.0,
2444                jitter_ms: 0,
2445                operation_timeout_ms: None,
2446                immediate_first: false,
2447                max_elapsed_ms: None,
2448            };
2449
2450            let manager = RetryManager::new(config);
2451            let attempt_counter = Arc::new(AtomicU32::new(0));
2452            let counter_clone = Arc::clone(&attempt_counter);
2453
2454            let result: Result<i32, TestError> = rt.block_on(manager.invocation(
2455                "non_retryable_test",
2456                move || {
2457                    let counter = Arc::clone(&counter_clone);
2458                    async move {
2459                        let attempts = counter.fetch_add(1, Ordering::SeqCst) as usize;
2460                        if attempts == attempt_before_non_retryable {
2461                            Err(TestError::NonRetryable("stop".to_string()))
2462                        } else {
2463                            Err(TestError::Retryable("retry".to_string()))
2464                        }
2465                    }
2466                },
2467                |e: &TestError| matches!(e, TestError::Retryable(_)),
2468                TestError::Timeout,
2469            ).execute());
2470
2471            let attempts = attempt_counter.load(Ordering::SeqCst) as usize;
2472
2473            prop_assert!(result.is_err());
2474            prop_assert!(matches!(result.unwrap_err(), TestError::NonRetryable(_)));
2475            // Should stop exactly when non-retryable error occurs
2476            prop_assert_eq!(attempts, attempt_before_non_retryable + 1);
2477        }
2478
2479        #[rstest]
2480        fn test_cancellation_stops_immediately(
2481            cancel_after_ms in 10u64..100,
2482            initial_delay_ms in 200u64..500,
2483        ) {
2484            use tokio_util::sync::CancellationToken;
2485
2486            let rt = build_paused_runtime();
2487
2488            let config = RetryConfig {
2489                max_retries: 10,
2490                initial_delay_ms,
2491                max_delay_ms: initial_delay_ms * 2,
2492                backoff_factor: 2.0,
2493                jitter_ms: 0,
2494                operation_timeout_ms: None,
2495                immediate_first: false,
2496                max_elapsed_ms: None,
2497            };
2498
2499            let manager = RetryManager::new(config);
2500            let token = CancellationToken::new();
2501            let token_clone = token.clone();
2502
2503            let result: Result<i32, TestError> = rt.block_on(async {
2504                // Spawn cancellation task
2505                spawn(async move {
2506                    time::sleep(Duration::from_millis(cancel_after_ms)).await;
2507                    token_clone.cancel();
2508                });
2509
2510                let operation_future = manager.invocation(
2511                    "cancellation_test",
2512                    || async {
2513                        Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2514                    },
2515                    |e: &TestError| matches!(e, TestError::Retryable(_)),
2516                    create_test_error,
2517                )
2518                .cancellation_token(&token)
2519                .execute();
2520
2521                // Advance time to trigger cancellation
2522                advance_clock(Duration::from_millis(cancel_after_ms + 10)).await;
2523                operation_future.await
2524            });
2525
2526            // Should be canceled
2527            prop_assert!(result.is_err());
2528            let error_msg = format!("{}", result.unwrap_err());
2529            prop_assert!(error_msg.contains("canceled"));
2530        }
2531
2532        #[rstest]
2533        fn test_budget_clamp_prevents_overshoot(
2534            max_elapsed_ms in 10u64..30,
2535            delay_per_retry in 30u64..50,
2536        ) {
2537            let rt = build_paused_runtime();
2538
2539            // Configure so that first retry delay would exceed budget
2540            let config = RetryConfig {
2541                max_retries: 5,
2542                initial_delay_ms: delay_per_retry,
2543                max_delay_ms: delay_per_retry * 2,
2544                backoff_factor: 1.0,
2545                jitter_ms: 0,
2546                operation_timeout_ms: None,
2547                immediate_first: false,
2548                max_elapsed_ms: Some(max_elapsed_ms),
2549            };
2550
2551            let manager = RetryManager::new(config);
2552            let attempts = Arc::new(AtomicU32::new(0));
2553            let attempts_for_operation = Arc::clone(&attempts);
2554
2555            let (result, elapsed) = rt.block_on(async {
2556                let started_at = time::Instant::now();
2557                let result = manager.invocation(
2558                    "budget_clamp_test",
2559                    move || {
2560                        let attempts = Arc::clone(&attempts_for_operation);
2561                        async move {
2562                            attempts.fetch_add(1, Ordering::SeqCst);
2563                            Err::<i32, TestError>(TestError::Retryable("fail".to_string()))
2564                        }
2565                    },
2566                    |e: &TestError| matches!(e, TestError::Retryable(_)),
2567                    create_test_error,
2568                ).execute().await;
2569                (result, started_at.elapsed())
2570            });
2571
2572            assert!(matches!(
2573                result,
2574                Err(TestError::Timeout(RetryError::ElapsedBudgetExceeded {
2575                    attempt: 2,
2576                    max_attempts: 6,
2577                    last_error: None,
2578                }))
2579            ));
2580            assert_eq!(attempts.load(Ordering::SeqCst), 1);
2581            #[cfg(not(all(feature = "simulation", madsim)))]
2582            assert_eq!(elapsed, Duration::from_millis(max_elapsed_ms));
2583            #[cfg(all(feature = "simulation", madsim))]
2584            assert!(
2585                elapsed >= Duration::from_millis(max_elapsed_ms)
2586                    && elapsed < Duration::from_millis(max_elapsed_ms + 1)
2587            );
2588        }
2589
2590        #[rstest]
2591        fn test_success_on_kth_attempt(
2592            k in 1usize..5,
2593            initial_delay_ms in 5u64..20,
2594        ) {
2595            let rt = build_paused_runtime();
2596
2597            let config = RetryConfig {
2598                max_retries: 10, // More than k
2599                initial_delay_ms,
2600                max_delay_ms: initial_delay_ms * 4,
2601                backoff_factor: 2.0,
2602                jitter_ms: 0,
2603                operation_timeout_ms: None,
2604                immediate_first: false,
2605                max_elapsed_ms: None,
2606            };
2607
2608            let manager = RetryManager::new(config);
2609            let attempt_counter = Arc::new(AtomicU32::new(0));
2610            let counter_clone = Arc::clone(&attempt_counter);
2611            let target_k = k;
2612
2613            let (result, _elapsed) = rt.block_on(async {
2614                let start = time::Instant::now();
2615
2616                let operation_future = manager.invocation(
2617                    "kth_attempt_test",
2618                    move || {
2619                        let counter = Arc::clone(&counter_clone);
2620                        async move {
2621                            let attempt = counter.fetch_add(1, Ordering::SeqCst) as usize;
2622                            if attempt + 1 == target_k {
2623                                Ok(42)
2624                            } else {
2625                                Err(TestError::Retryable("retry".to_string()))
2626                            }
2627                        }
2628                    },
2629                    |e: &TestError| matches!(e, TestError::Retryable(_)),
2630                    create_test_error,
2631                ).execute();
2632
2633                // Advance time to allow enough retries
2634                for _ in 0..k {
2635                    advance_clock(Duration::from_millis(initial_delay_ms * 4)).await;
2636                }
2637
2638                let result = operation_future.await;
2639                let elapsed = start.elapsed();
2640
2641                (result, elapsed)
2642            });
2643
2644            let attempts = attempt_counter.load(Ordering::SeqCst) as usize;
2645
2646            // Using paused Tokio time (start_paused + advance); assert behavior only (no wall-clock timing)
2647            prop_assert!(result.is_ok());
2648            prop_assert_eq!(result.unwrap(), 42);
2649            prop_assert_eq!(attempts, k);
2650        }
2651    }
2652}
nautilus_network/retry.rs

nautilus_network/
retry.rs
