arqen 0.16.0

Rust-first backend toolkit for HTTP services, agents, automation, jobs, and storage adapters
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! Retry-aware startup operations for remote Thingd deployments.

use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use crate::core::{AppError, ErrorKind};
use crate::thingd::ThingdBackend;

/// Bounded retry settings for application startup work such as catalog seeding.
#[derive(Debug, Clone)]
pub struct BootstrapPolicy {
    pub max_attempts: u32,
    pub initial_backoff: Duration,
    pub max_backoff: Duration,
}

impl Default for BootstrapPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 12,
            initial_backoff: Duration::from_secs(1),
            max_backoff: Duration::from_secs(30),
        }
    }
}

/// Retry an operation while Thingd is unavailable during startup.
pub async fn retry<F, Fut, T>(policy: BootstrapPolicy, mut operation: F) -> Result<T, AppError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, AppError>>,
{
    let attempts = policy.max_attempts.max(1);
    let mut backoff = policy.initial_backoff;
    for attempt in 1..=attempts {
        match operation().await {
            Ok(value) => return Ok(value),
            Err(error) if attempt < attempts && retryable(&error) => {
                tokio::time::sleep(backoff).await;
                backoff = backoff.saturating_mul(2).min(policy.max_backoff);
            }
            Err(error) => return Err(error),
        }
    }
    unreachable!("bootstrap attempts are always at least one")
}

/// Retry the standard Thingd seed operation until the remote store is ready.
pub async fn seed_with_retry(
    backend: Arc<dyn ThingdBackend>,
    policy: BootstrapPolicy,
) -> Result<(), AppError> {
    retry(policy, move || {
        let backend = Arc::clone(&backend);
        async move { backend.seed().await }
    })
    .await
}

fn retryable(error: &AppError) -> bool {
    matches!(
        error.kind,
        ErrorKind::Unavailable | ErrorKind::Timeout | ErrorKind::Dependency
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};

    #[tokio::test]
    async fn retries_transient_failures() {
        let calls = Arc::new(AtomicU32::new(0));
        let result = retry(
            BootstrapPolicy {
                max_attempts: 3,
                initial_backoff: Duration::ZERO,
                max_backoff: Duration::ZERO,
            },
            {
                let calls = Arc::clone(&calls);
                move || {
                    let attempt = calls.fetch_add(1, Ordering::Relaxed);
                    async move {
                        if attempt < 2 {
                            Err(AppError::new(ErrorKind::Unavailable, "not ready"))
                        } else {
                            Ok("ready")
                        }
                    }
                }
            },
        )
        .await
        .unwrap();
        assert_eq!(result, "ready");
        assert_eq!(calls.load(Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn does_not_retry_permanent_errors() {
        let calls = Arc::new(AtomicU32::new(0));
        let error = retry(
            BootstrapPolicy {
                max_attempts: 3,
                initial_backoff: Duration::ZERO,
                max_backoff: Duration::ZERO,
            },
            {
                let calls = Arc::clone(&calls);
                move || {
                    calls.fetch_add(1, Ordering::Relaxed);
                    async { Err::<(), _>(AppError::new(ErrorKind::Validation, "invalid seed")) }
                }
            },
        )
        .await
        .unwrap_err();
        assert_eq!(error.kind, ErrorKind::Validation);
        assert_eq!(calls.load(Ordering::Relaxed), 1);
    }
}