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    #[test]
196    fn mounted_credentials_replace_static_keys_without_fallback() {
197        let mut map: HashMap<&str, &str> = [
198            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
199            ("REDDB_BACKUP_S3_BUCKET", "synthetic-bucket"),
200            ("REDDB_BACKUP_S3_PREFIX", "org/db/"),
201            (
202                "REDDB_BACKUP_S3_CREDENTIALS_FILE",
203                "/var/run/secrets/backup/credentials.json",
204            ),
205        ]
206        .into_iter()
207        .collect();
208        let lookup_map = |key: &str| map.get(key).map(|v| v.to_string());
209        let config = from_env(lookup_map)
210            .expect("valid file mode")
211            .expect("backup configured");
212        assert_eq!(
213            config.credentials_file.as_deref(),
214            Some("/var/run/secrets/backup/credentials.json")
215        );
216        assert!(config.access_key_id.is_empty() && config.secret_access_key.is_empty());
217        map.insert("REDDB_BACKUP_S3_ACCESS_KEY_ID", "must-not-fall-back");
218        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
219        map.remove("REDDB_BACKUP_S3_ACCESS_KEY_ID");
220        map.insert(
221            "REDDB_BACKUP_S3_CREDENTIALS_FILE",
222            "relative/credentials.json",
223        );
224        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
225        map.insert(
226            "REDDB_BACKUP_S3_CREDENTIALS_FILE",
227            "/absolute/credentials.json",
228        );
229        map.remove("REDDB_BACKUP_S3_BUCKET");
230        assert!(from_env(|key| map.get(key).map(|v| v.to_string())).is_err());
231    }
232
233    #[test]
234    fn credentials_file_alone_is_partial_configuration() {
235        assert!(
236            from_env(|key| (key == CREDENTIALS_FILE_VAR).then(|| "/absolute/file.json".into()))
237                .is_err()
238        );
239    }
240
241    fn lookup<'a>(
242        map: &'a HashMap<&'static str, &'static str>,
243    ) -> impl Fn(&str) -> Option<String> + 'a {
244        move |k| map.get(k).map(|s| s.to_string())
245    }
246
247    #[test]
248    fn none_present_yields_none() {
249        let map: HashMap<&'static str, &'static str> = HashMap::new();
250        let got = from_env(lookup(&map)).unwrap();
251        assert!(got.is_none());
252    }
253
254    #[test]
255    fn all_required_present_yields_config_with_defaults() {
256        let map: HashMap<&'static str, &'static str> = [
257            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
258            ("REDDB_BACKUP_S3_BUCKET", "buck"),
259            ("REDDB_BACKUP_S3_PREFIX", "clusters/dev/"),
260            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
261            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
262        ]
263        .into_iter()
264        .collect();
265        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
266        assert_eq!(cfg.endpoint, "https://s3.example.com");
267        assert_eq!(cfg.bucket, "buck");
268        assert_eq!(cfg.prefix, "clusters/dev/");
269        assert_eq!(cfg.access_key_id, "AK");
270        assert_eq!(cfg.secret_access_key, "SK");
271        assert_eq!(cfg.region, DEFAULT_REGION);
272        assert_eq!(cfg.checkpoint_interval_secs, DEFAULT_CHECKPOINT_SECS);
273        assert_eq!(cfg.wal_flush_interval_secs, DEFAULT_WAL_FLUSH_SECS);
274        assert_eq!(cfg.pause_on_lag_secs, DEFAULT_PAUSE_ON_LAG_SECS);
275    }
276
277    #[test]
278    fn pause_on_lag_is_parsed_when_present() {
279        let map: HashMap<&'static str, &'static str> = [
280            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
281            ("REDDB_BACKUP_S3_BUCKET", "b"),
282            ("REDDB_BACKUP_S3_PREFIX", "p/"),
283            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
284            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
285            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "300"),
286        ]
287        .into_iter()
288        .collect();
289        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
290        assert_eq!(cfg.pause_on_lag_secs, 300);
291    }
292
293    #[test]
294    fn pause_on_lag_zero_is_disabled() {
295        let map: HashMap<&'static str, &'static str> = [
296            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
297            ("REDDB_BACKUP_S3_BUCKET", "b"),
298            ("REDDB_BACKUP_S3_PREFIX", "p/"),
299            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
300            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
301            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "0"),
302        ]
303        .into_iter()
304        .collect();
305        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
306        assert_eq!(cfg.pause_on_lag_secs, 0);
307    }
308
309    #[test]
310    fn pause_on_lag_negative_is_error() {
311        let map: HashMap<&'static str, &'static str> = [
312            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
313            ("REDDB_BACKUP_S3_BUCKET", "b"),
314            ("REDDB_BACKUP_S3_PREFIX", "p/"),
315            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
316            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
317            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "-1"),
318        ]
319        .into_iter()
320        .collect();
321        let err = from_env(lookup(&map)).unwrap_err();
322        assert!(err.contains("REDDB_BACKUP_PAUSE_ON_LAG_SECS"), "{err}");
323    }
324
325    #[test]
326    fn pause_on_lag_non_numeric_is_error() {
327        let map: HashMap<&'static str, &'static str> = [
328            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
329            ("REDDB_BACKUP_S3_BUCKET", "b"),
330            ("REDDB_BACKUP_S3_PREFIX", "p/"),
331            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
332            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
333            ("REDDB_BACKUP_PAUSE_ON_LAG_SECS", "soon"),
334        ]
335        .into_iter()
336        .collect();
337        let err = from_env(lookup(&map)).unwrap_err();
338        assert!(err.contains("REDDB_BACKUP_PAUSE_ON_LAG_SECS"), "{err}");
339        assert!(err.contains("non-negative"), "{err}");
340    }
341
342    #[test]
343    fn all_required_present_with_explicit_overrides() {
344        let map: HashMap<&'static str, &'static str> = [
345            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
346            ("REDDB_BACKUP_S3_BUCKET", "b"),
347            ("REDDB_BACKUP_S3_PREFIX", "p/"),
348            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
349            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
350            ("REDDB_BACKUP_S3_REGION", "us-east-1"),
351            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "60"),
352            ("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS", "5"),
353        ]
354        .into_iter()
355        .collect();
356        let cfg = from_env(lookup(&map)).unwrap().expect("Some");
357        assert_eq!(cfg.region, "us-east-1");
358        assert_eq!(cfg.checkpoint_interval_secs, 60);
359        assert_eq!(cfg.wal_flush_interval_secs, 5);
360    }
361
362    #[test]
363    fn partial_config_names_missing_var() {
364        let map: HashMap<&'static str, &'static str> = [
365            ("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.com"),
366            ("REDDB_BACKUP_S3_BUCKET", "b"),
367        ]
368        .into_iter()
369        .collect();
370        let err = from_env(lookup(&map)).unwrap_err();
371        assert!(err.contains("REDDB_BACKUP_S3_PREFIX"), "{err}");
372        assert!(err.contains("REDDB_BACKUP_S3_ACCESS_KEY_ID"), "{err}");
373        assert!(err.contains("REDDB_BACKUP_S3_SECRET_ACCESS_KEY"), "{err}");
374    }
375
376    #[test]
377    fn whitespace_only_required_treated_as_missing() {
378        let map: HashMap<&'static str, &'static str> = [
379            ("REDDB_BACKUP_S3_ENDPOINT", "   "),
380            ("REDDB_BACKUP_S3_BUCKET", "b"),
381            ("REDDB_BACKUP_S3_PREFIX", "p/"),
382            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
383            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
384        ]
385        .into_iter()
386        .collect();
387        let err = from_env(lookup(&map)).unwrap_err();
388        assert!(err.contains("REDDB_BACKUP_S3_ENDPOINT"), "{err}");
389    }
390
391    #[test]
392    fn non_numeric_interval_is_error() {
393        let map: HashMap<&'static str, &'static str> = [
394            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
395            ("REDDB_BACKUP_S3_BUCKET", "b"),
396            ("REDDB_BACKUP_S3_PREFIX", "p/"),
397            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
398            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
399            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "abc"),
400        ]
401        .into_iter()
402        .collect();
403        let err = from_env(lookup(&map)).unwrap_err();
404        assert!(
405            err.contains("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS"),
406            "{err}"
407        );
408        assert!(err.contains("positive integer"), "{err}");
409    }
410
411    #[test]
412    fn zero_interval_is_error() {
413        let map: HashMap<&'static str, &'static str> = [
414            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
415            ("REDDB_BACKUP_S3_BUCKET", "b"),
416            ("REDDB_BACKUP_S3_PREFIX", "p/"),
417            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
418            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
419            ("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS", "0"),
420        ]
421        .into_iter()
422        .collect();
423        let err = from_env(lookup(&map)).unwrap_err();
424        assert!(
425            err.contains("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS"),
426            "{err}"
427        );
428        assert!(err.contains("> 0"), "{err}");
429    }
430
431    #[test]
432    fn negative_interval_is_error() {
433        let map: HashMap<&'static str, &'static str> = [
434            ("REDDB_BACKUP_S3_ENDPOINT", "https://x"),
435            ("REDDB_BACKUP_S3_BUCKET", "b"),
436            ("REDDB_BACKUP_S3_PREFIX", "p/"),
437            ("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK"),
438            ("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK"),
439            ("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS", "-10"),
440        ]
441        .into_iter()
442        .collect();
443        let err = from_env(lookup(&map)).unwrap_err();
444        assert!(
445            err.contains("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS"),
446            "{err}"
447        );
448    }
449}
reddb_server/backup_bootstrap.rs

reddb_server/
backup_bootstrap.rs
