reddb_server/
backup_bootstrap.rs

1//! Env-driven `BackupBootstrap` (issue #517).
2//!
3//! Parses the canonical `REDDB_BACKUP_*` env contract into a
4//! [`BackupConfig`]. Pure function — env access is injected as a
5//! closure so unit tests need no real process env. The `red` binary
6//! calls [`from_env`] at boot; the returned `Option<BackupConfig>`
7//! drives `Options::with_remote_backend` + `with_atomic_remote_backend`
8//! wiring and the archiver / checkpointer task intervals.
9//!
10//! Contract:
11//!   * `REDDB_BACKUP_S3_ENDPOINT`   (required)
12//!   * `REDDB_BACKUP_S3_BUCKET`     (required)
13//!   * `REDDB_BACKUP_S3_PREFIX`     (required)
14//!   * `REDDB_BACKUP_S3_ACCESS_KEY_ID`     (required)
15//!   * `REDDB_BACKUP_S3_SECRET_ACCESS_KEY` (required)
16//!   * `REDDB_BACKUP_S3_CREDENTIALS_FILE` (optional alternative to both keys;
17//!     absolute path to atomically rotated, scoped temporary credentials)
18//!   * `REDDB_BACKUP_S3_REGION`     (default `auto`)
19//!   * `REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS` (default 3600, must be > 0)
20//!   * `REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS`  (default 30,   must be > 0)
21//!   * `REDDB_BACKUP_PAUSE_ON_LAG_SECS`        (default 0 = disabled; > 0 enables
22//!     graceful read-only mode when WAL archive lag exceeds the threshold —
23//!     issue #519)
24//!
25//! Resolution:
26//!   * All required vars absent → `Ok(None)` (standalone; identical to
27//!     today's behaviour).
28//!   * All required vars present → `Ok(Some(BackupConfig))`.
29//!   * Partial config (at least one required present, at least one
30//!     missing) → `Err` naming the missing var.
31//!   * Non-numeric / zero interval → `Err`.
32
33/// Parsed configuration produced by [`from_env`]. Carries everything
34/// the `red` binary needs to construct an `S3Backend` and the two
35/// background tasks (archiver + checkpointer).
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct BackupConfig {
38    pub endpoint: String,
39    pub bucket: String,
40    pub region: String,
41    pub access_key_id: String,
42    pub secret_access_key: String,
43    pub credentials_file: Option<String>,
44    pub prefix: String,
45    pub checkpoint_interval_secs: u64,
46    pub wal_flush_interval_secs: u64,
47    /// Issue #519 — when > 0, the engine monitors archive lag (`now -
48    /// last_successful_archive_at`) and transitions to a graceful
49    /// read-only mode when the lag exceeds this threshold. `0` keeps
50    /// the legacy behaviour (writes always accepted while local volume
51    /// has room, regardless of remote backend health).
52    pub pause_on_lag_secs: u64,
53}
54
55const REQUIRED_VARS: &[&str] = &[
56    "REDDB_BACKUP_S3_ENDPOINT",
57    "REDDB_BACKUP_S3_BUCKET",
58    "REDDB_BACKUP_S3_PREFIX",
59    "REDDB_BACKUP_S3_ACCESS_KEY_ID",
60    "REDDB_BACKUP_S3_SECRET_ACCESS_KEY",
61];
62
63const REGION_VAR: &str = "REDDB_BACKUP_S3_REGION";
64const CREDENTIALS_FILE_VAR: &str = "REDDB_BACKUP_S3_CREDENTIALS_FILE";
65const CHECKPOINT_VAR: &str = "REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS";
66const WAL_FLUSH_VAR: &str = "REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS";
67const PAUSE_ON_LAG_VAR: &str = "REDDB_BACKUP_PAUSE_ON_LAG_SECS";
68
69const DEFAULT_REGION: &str = "auto";
70const DEFAULT_CHECKPOINT_SECS: u64 = 3600;
71const DEFAULT_WAL_FLUSH_SECS: u64 = 30;
72const DEFAULT_PAUSE_ON_LAG_SECS: u64 = 0;
73
74/// Parse the `REDDB_BACKUP_*` env contract using the supplied
75/// env-var lookup. See module docs for the contract.
76pub fn from_env<F>(env: F) -> Result<Option<BackupConfig>, String>
77where
78    F: Fn(&str) -> Option<String>,
79{
80    let credentials_file = env(CREDENTIALS_FILE_VAR).filter(|v| !v.trim().is_empty());
81    if let Some(path) = &credentials_file {
82        if !std::path::Path::new(path).is_absolute() {
83            return Err(format!("{CREDENTIALS_FILE_VAR} must be an absolute path"));
84        }
85        if REQUIRED_VARS[3..]
86            .iter()
87            .any(|name| env(name).is_some_and(|v| !v.trim().is_empty()))
88        {
89            return Err("backup credentials file and static keys are mutually exclusive".into());
90        }
91    }
92    let required_vars = if credentials_file.is_some() {
93        &REQUIRED_VARS[..3]
94    } else {
95        REQUIRED_VARS
96    };
97    let presence: Vec<(&str, Option<String>)> = required_vars
98        .iter()
99        .map(|name| (*name, env(name).filter(|v| !v.trim().is_empty())))
100        .collect();
101
102    let present_count = presence.iter().filter(|(_, v)| v.is_some()).count();
103
104    if present_count == 0 && credentials_file.is_none() {
105        return Ok(None);
106    }
107
108    if present_count < required_vars.len() {
109        let missing: Vec<&str> = presence
110            .iter()
111            .filter_map(|(n, v)| v.is_none().then_some(*n))
112            .collect();
113        return Err(format!(
114            "partial REDDB_BACKUP_S3_* config; missing: {}",
115            missing.join(", ")
116        ));
117    }
118
119    let mut required = presence.into_iter().map(|(_, v)| v.unwrap());
120    let endpoint = required.next().unwrap();
121    let bucket = required.next().unwrap();
122    let prefix = required.next().unwrap();
123    let access_key_id = required.next().unwrap_or_default();
124    let secret_access_key = required.next().unwrap_or_default();
125
126    let region = env(REGION_VAR)
127        .filter(|v| !v.trim().is_empty())
128        .unwrap_or_else(|| DEFAULT_REGION.to_string());
129
130    let checkpoint_interval_secs = parse_interval(&env, CHECKPOINT_VAR, DEFAULT_CHECKPOINT_SECS)?;
131    let wal_flush_interval_secs = parse_interval(&env, WAL_FLUSH_VAR, DEFAULT_WAL_FLUSH_SECS)?;
132    let pause_on_lag_secs = parse_pause_on_lag(&env, DEFAULT_PAUSE_ON_LAG_SECS)?;
133
134    Ok(Some(BackupConfig {
135        endpoint,
136        bucket,
137        region,
138        access_key_id,
139        secret_access_key,
140        credentials_file,
141        prefix,
142        checkpoint_interval_secs,
143        wal_flush_interval_secs,
144        pause_on_lag_secs,
145    }))
146}
147
148fn parse_pause_on_lag<F>(env: &F, default: u64) -> Result<u64, String>
149where
150    F: Fn(&str) -> Option<String>,
151{
152    let Some(raw) = env(PAUSE_ON_LAG_VAR).filter(|v| !v.trim().is_empty()) else {
153        return Ok(default);
154    };
155    let trimmed = raw.trim();
156    let parsed: i128 = trimmed
157        .parse()
158        .map_err(|_| format!("{PAUSE_ON_LAG_VAR} must be a non-negative integer; got {raw:?}"))?;
159    if parsed < 0 {
160        return Err(format!(
161            "{PAUSE_ON_LAG_VAR} must be >= 0; got {parsed} (negative not allowed)"
162        ));
163    }
164    let as_u64 = u64::try_from(parsed)
165        .map_err(|_| format!("{PAUSE_ON_LAG_VAR} exceeds u64 range; got {parsed}"))?;
166    Ok(as_u64)
167}
168
169fn parse_interval<F>(env: &F, name: &str, default: u64) -> Result<u64, String>
170where
171    F: Fn(&str) -> Option<String>,
172{
173    let Some(raw) = env(name).filter(|v| !v.trim().is_empty()) else {
174        return Ok(default);
175    };
176    let trimmed = raw.trim();
177    let parsed: i128 = trimmed
178        .parse()
179        .map_err(|_| format!("{name} must be a positive integer; got {raw:?}"))?;
180    if parsed <= 0 {
181        return Err(format!(
182            "{name} must be > 0; got {parsed} (zero/negative not allowed)"
183        ));
184    }
185    let as_u64 =
186        u64::try_from(parsed).map_err(|_| format!("{name} exceeds u64 range; got {parsed}"))?;
187    Ok(as_u64)
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193    use std::collections::HashMap;
194
195    #[cfg(windows)]
196    const TEST_CREDENTIALS_PATH: &str = r"C:\var\run\secrets\backup\credentials.json";
197    #[cfg(not(windows))]
198    const TEST_CREDENTIALS_PATH: &str = "/var/run/secrets/backup/credentials.json";
199
200    #[test]
201    fn mounted_credentials_replace_static_keys_without_fallback() {
202        let mut map: HashMap<&str, &str> = [
203            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
204            ("REDDB_BACKUP_S3_BUCKET", "synthetic-bucket"),
205            ("REDDB_BACKUP_S3_PREFIX", "org/db/"),
206            ("REDDB_BACKUP_S3_CREDENTIALS_FILE", TEST_CREDENTIALS_PATH),
207        ]
208        .into_iter()
209        .collect();
210        let lookup_map = |key: &str| map.get(key).map(|v| v.to_string());
211        let config = from_env(lookup_map)
212            .expect("valid file mode")
213            .expect("backup configured");
214        assert_eq!(
215            config.credentials_file.as_deref(),
216            Some(TEST_CREDENTIALS_PATH)
217        );
218        assert!(config.access_key_id.is_empty() && config.secret_access_key.is_empty());
219        map.insert("REDDB_BACKUP_S3_ACCESS_KEY_ID", "must-not-fall-back");
220        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
221        map.remove("REDDB_BACKUP_S3_ACCESS_KEY_ID");
222        map.insert(
223            "REDDB_BACKUP_S3_CREDENTIALS_FILE",
224            "relative/credentials.json",
225        );
226        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
227        map.insert("REDDB_BACKUP_S3_CREDENTIALS_FILE", TEST_CREDENTIALS_PATH);
228        map.remove("REDDB_BACKUP_S3_BUCKET");
229        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
230    }
231
232    #[test]
233    fn credentials_file_alone_is_partial_configuration() {
234        assert!(
235            from_env(|key| (key == CREDENTIALS_FILE_VAR).then(|| TEST_CREDENTIALS_PATH.into()))
236                .is_err()
237        );
238    }
239
240    fn lookup<'a>(
241        map: &'a HashMap<&'static str, &'static str>,
242    ) -> impl Fn(&str) -> Option<String> + 'a {
243        move |k| map.get(k).map(|s| s.to_string())
244    }
245
246    #[test]
247    fn none_present_yields_none() {
248        let map: HashMap<&'static str, &'static str> = HashMap::new();
249        let got = from_env(lookup(&map)).unwrap();
250        assert!(got.is_none());
251    }
252
253    #[test]
254    fn all_required_present_yields_config_with_defaults() {
255        let map: HashMap<&'static str, &'static str> = [
256            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
257            ("REDDB_BACKUP_S3_BUCKET", "buck"),
258            ("REDDB_BACKUP_S3_PREFIX", "clusters/dev/"),
259            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
260            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
261        ]
262        .into_iter()
263        .collect();
264        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
265        assert_eq!(cfg.endpoint, "https://s3.example.com");
266        assert_eq!(cfg.bucket, "buck");
267        assert_eq!(cfg.prefix, "clusters/dev/");
268        assert_eq!(cfg.access_key_id, "AK");
269        assert_eq!(cfg.secret_access_key, "SK");
270        assert_eq!(cfg.region, DEFAULT_REGION);
271        assert_eq!(cfg.checkpoint_interval_secs, DEFAULT_CHECKPOINT_SECS);
272        assert_eq!(cfg.wal_flush_interval_secs, DEFAULT_WAL_FLUSH_SECS);
273        assert_eq!(cfg.pause_on_lag_secs, DEFAULT_PAUSE_ON_LAG_SECS);
274    }
275
276    #[test]
277    fn pause_on_lag_is_parsed_when_present() {
278        let map: HashMap<&'static str, &'static str> = [
279            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
280            ("REDDB_BACKUP_S3_BUCKET", "b"),
281            ("REDDB_BACKUP_S3_PREFIX", "p/"),
282            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
283            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
284            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "300"),
285        ]
286        .into_iter()
287        .collect();
288        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
289        assert_eq!(cfg.pause_on_lag_secs, 300);
290    }
291
292    #[test]
293    fn pause_on_lag_zero_is_disabled() {
294        let map: HashMap<&'static str, &'static str> = [
295            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
296            ("REDDB_BACKUP_S3_BUCKET", "b"),
297            ("REDDB_BACKUP_S3_PREFIX", "p/"),
298            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
299            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
300            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "0"),
301        ]
302        .into_iter()
303        .collect();
304        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
305        assert_eq!(cfg.pause_on_lag_secs, 0);
306    }
307
308    #[test]
309    fn pause_on_lag_negative_is_error() {
310        let map: HashMap<&'static str, &'static str> = [
311            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
312            ("REDDB_BACKUP_S3_BUCKET", "b"),
313            ("REDDB_BACKUP_S3_PREFIX", "p/"),
314            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
315            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
316            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "-1"),
317        ]
318        .into_iter()
319        .collect();
320        let err = from_env(lookup(&map)).unwrap_err();
321        assert!(err.contains("REDDB_BACKUP_PAUSE_ON_LAG_SECS"), "{err}");
322    }
323
324    #[test]
325    fn pause_on_lag_non_numeric_is_error() {
326        let map: HashMap<&'static str, &'static str> = [
327            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
328            ("REDDB_BACKUP_S3_BUCKET", "b"),
329            ("REDDB_BACKUP_S3_PREFIX", "p/"),
330            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
331            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
332            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "soon"),
333        ]
334        .into_iter()
335        .collect();
336        let err = from_env(lookup(&map)).unwrap_err();
337        assert!(err.contains("REDDB_BACKUP_PAUSE_ON_LAG_SECS"), "{err}");
338        assert!(err.contains("non-negative"), "{err}");
339    }
340
341    #[test]
342    fn all_required_present_with_explicit_overrides() {
343        let map: HashMap<&'static str, &'static str> = [
344            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
345            ("REDDB_BACKUP_S3_BUCKET", "b"),
346            ("REDDB_BACKUP_S3_PREFIX", "p/"),
347            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
348            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
349            ("REDDB_BACKUP_S3_REGION", "us-east-1"),
350            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "60"),
351            ("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS", "5"),
352        ]
353        .into_iter()
354        .collect();
355        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
356        assert_eq!(cfg.region, "us-east-1");
357        assert_eq!(cfg.checkpoint_interval_secs, 60);
358        assert_eq!(cfg.wal_flush_interval_secs, 5);
359    }
360
361    #[test]
362    fn partial_config_names_missing_var() {
363        let map: HashMap<&'static str, &'static str> = [
364            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
365            ("REDDB_BACKUP_S3_BUCKET", "b"),
366        ]
367        .into_iter()
368        .collect();
369        let err = from_env(lookup(&map)).unwrap_err();
370        assert!(err.contains("REDDB_BACKUP_S3_PREFIX"), "{err}");
371        assert!(err.contains("REDDB_BACKUP_S3_ACCESS_KEY_ID"), "{err}");
372        assert!(err.contains("REDDB_BACKUP_S3_SECRET_ACCESS_KEY"), "{err}");
373    }
374
375    #[test]
376    fn whitespace_only_required_treated_as_missing() {
377        let map: HashMap<&'static str, &'static str> = [
378            ("REDDB_BACKUP_S3_ENDPOINT", "   "),
379            ("REDDB_BACKUP_S3_BUCKET", "b"),
380            ("REDDB_BACKUP_S3_PREFIX", "p/"),
381            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
382            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
383        ]
384        .into_iter()
385        .collect();
386        let err = from_env(lookup(&map)).unwrap_err();
387        assert!(err.contains("REDDB_BACKUP_S3_ENDPOINT"), "{err}");
388    }
389
390    #[test]
391    fn non_numeric_interval_is_error() {
392        let map: HashMap<&'static str, &'static str> = [
393            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
394            ("REDDB_BACKUP_S3_BUCKET", "b"),
395            ("REDDB_BACKUP_S3_PREFIX", "p/"),
396            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
397            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
398            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "abc"),
399        ]
400        .into_iter()
401        .collect();
402        let err = from_env(lookup(&map)).unwrap_err();
403        assert!(
404            err.contains("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS"),
405            "{err}"
406        );
407        assert!(err.contains("positive integer"), "{err}");
408    }
409
410    #[test]
411    fn zero_interval_is_error() {
412        let map: HashMap<&'static str, &'static str> = [
413            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
414            ("REDDB_BACKUP_S3_BUCKET", "b"),
415            ("REDDB_BACKUP_S3_PREFIX", "p/"),
416            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
417            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
418            ("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS", "0"),
419        ]
420        .into_iter()
421        .collect();
422        let err = from_env(lookup(&map)).unwrap_err();
423        assert!(
424            err.contains("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS"),
425            "{err}"
426        );
427        assert!(err.contains("> 0"), "{err}");
428    }
429
430    #[test]
431    fn negative_interval_is_error() {
432        let map: HashMap<&'static str, &'static str> = [
433            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
434            ("REDDB_BACKUP_S3_BUCKET", "b"),
435            ("REDDB_BACKUP_S3_PREFIX", "p/"),
436            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
437            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
438            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "-10"),
439        ]
440        .into_iter()
441        .collect();
442        let err = from_env(lookup(&map)).unwrap_err();
443        assert!(
444            err.contains("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS"),
445            "{err}"
446        );
447    }
448}
reddb_server/backup_bootstrap.rs

reddb_server/
backup_bootstrap.rs
