backup/
lib.rs

1//! Continuous backup of a mushroomdb store to S3-compatible object storage.
2//!
3//! A [`BackupWorker`] runs beside the server. Every `wal_interval_ms` it ships
4//! the WAL bytes fsynced since the last tick as one immutable segment,
5//! `{prefix}/wal/{generation}/{start:020}-{end:020}`; when a snapshot is
6//! written it uploads `snapshot.bin`, any new `wal.N.archive` and the side
7//! files; and it writes manifests (`{prefix}/manifest/{commit_seq:020}.json`,
8//! pointed to by `{prefix}/LATEST`) naming consistent restore points.
9//!
10//! The recovery point is the segment cadence, not the snapshot interval: a
11//! commit is off-box durable once the segment holding it is uploaded.
12//!
13//! Before it uploads anything the worker takes the prefix's [`BucketLease`],
14//! an advisory single-writer lock, and while another instance holds a fresh
15//! one it uploads nothing and reports itself unhealthy (see [`lease`]).
16//!
17//! [`restore_from_s3`] reads a restore point back — the newest, or the one
18//! holding a point in time — verifies every object, and installs it as a
19//! store with a new backup lineage (see [`restore`]). [`boot::restore_on_boot`]
20//! does the same into an empty data directory when a server starts, and
21//! hands the bucket lease it took to the worker (see [`boot`]).
22//!
23//! The worker never takes an engine lock and never writes to the store; a
24//! failed upload is retried with back-off and reported through
25//! [`BackupHandle`], never to the write path.
26
27/// The object-store crate this one builds on, for callers that read the bucket.
28pub use object_store;
29
30pub mod boot;
31pub mod disk;
32pub mod lease;
33pub mod manifest;
34pub mod restore;
35pub mod segment;
36pub mod store;
37pub mod worker;
38
39pub use boot::{restore_on_boot, BootError, BootLease, BootRestore};
40pub use lease::{BucketLease, Exclusion, LeaseError};
41pub use manifest::{FileEntry, Manifest, SegmentEntry, MANIFEST_VERSION};
42pub use restore::{
43    restore_from_s3, restore_from_store, RestoreS3Error, RestoreS3Outcome, RestoreTarget,
44};
45pub use store::{build_store, BackupConfig, S3Location};
46pub use worker::{
47    BackupCounters, BackupHandle, BackupTrigger, BackupWorker, WalSource, SIDE_FILES,
48};
49
50/// Why a backup step or its configuration failed.
51#[derive(Debug)]
52pub enum BackupError {
53    /// The environment asked for a backup that cannot be set up.
54    Config(String),
55    /// The object store refused or failed a request.
56    Store(object_store::Error),
57    /// Reading the store's files failed.
58    Io(std::io::Error),
59    /// A manifest in the bucket is missing or unreadable.
60    Manifest(String),
61    /// The store's files are between states; the step is retried.
62    Inconsistent(String),
63    /// An object-store request took longer than `op_timeout_ms`.
64    Timeout(String),
65    /// The bucket lease does not allow this write: it is lost, or it could
66    /// not be renewed in time and the holder has fenced itself.
67    Lease(String),
68    /// The prefix belongs to another store lineage, or this store was
69    /// rewound; the worker refuses it.
70    Lineage(String),
71    /// The bucket refused a conditional write: another writer is active.
72    Conflict(String),
73}
74
75impl std::fmt::Display for BackupError {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        match self {
78            BackupError::Config(m) => write!(f, "backup configuration: {m}"),
79            // A refusal's body can carry the access key id and signing
80            // material; none of it reaches a log.
81            BackupError::Store(e) => write!(f, "object store: {}", redact(&e.to_string())),
82            BackupError::Timeout(what) => write!(f, "{what} did not finish within the deadline"),
83            BackupError::Io(e) => write!(f, "reading the store: {e}"),
84            BackupError::Manifest(m) => write!(f, "manifest: {m}"),
85            BackupError::Inconsistent(m) => write!(f, "{m}"),
86            BackupError::Lease(m) => write!(f, "bucket lease: {m}"),
87            BackupError::Lineage(m) | BackupError::Conflict(m) => write!(f, "{m}"),
88        }
89    }
90}
91
92impl std::error::Error for BackupError {}
93
94impl From<object_store::Error> for BackupError {
95    fn from(e: object_store::Error) -> Self {
96        BackupError::Store(e)
97    }
98}
99
100impl From<std::io::Error> for BackupError {
101    fn from(e: std::io::Error) -> Self {
102        BackupError::Io(e)
103    }
104}
105
106/// XML elements of an S3 error body whose content identifies the caller or
107/// the signature.
108const REDACTED_ELEMENTS: [&str; 7] = [
109    "AWSAccessKeyId",
110    "AccessKeyId",
111    "SignatureProvided",
112    "StringToSign",
113    "StringToSignBytes",
114    "CanonicalRequest",
115    "CanonicalRequestBytes",
116];
117
118/// Query parameters and header fields whose value is signing material.
119const REDACTED_FIELDS: [&str; 6] = [
120    "X-Amz-Credential=",
121    "X-Amz-Signature=",
122    "X-Amz-Security-Token=",
123    "Credential=",
124    "Signature=",
125    "x-amz-security-token:",
126];
127
128/// `text` with access-key ids, signatures and other credential material
129/// replaced by `[redacted]`: the content of [`REDACTED_ELEMENTS`] XML
130/// elements, the value after each of [`REDACTED_FIELDS`], and anything shaped
131/// like an AWS access key id (`AKIA…`, `ASIA…`: four letters and sixteen more
132/// upper-case letters or digits).
133pub fn redact(text: &str) -> String {
134    let mut out = text.to_string();
135    for el in REDACTED_ELEMENTS {
136        let (open, close) = (format!("<{el}>"), format!("</{el}>"));
137        let mut from = 0;
138        while let Some(i) = out[from..].find(&open).map(|i| i + from + open.len()) {
139            let Some(j) = out[i..].find(&close).map(|j| j + i) else {
140                break;
141            };
142            out.replace_range(i..j, "[redacted]");
143            from = i + "[redacted]".len() + close.len();
144        }
145    }
146    for field in REDACTED_FIELDS {
147        let mut from = 0;
148        while let Some(i) = out[from..].find(field).map(|i| i + from + field.len()) {
149            let end = out[i..]
150                .find(|c: char| c == '&' || c == ',' || c == '"' || c == '<' || c.is_whitespace())
151                .map_or(out.len(), |j| j + i);
152            out.replace_range(i..end, "[redacted]");
153            from = i + "[redacted]".len();
154        }
155    }
156    redact_key_ids(&out)
157}
158
159fn redact_key_ids(text: &str) -> String {
160    const PREFIXES: [&str; 6] = ["AKIA", "ASIA", "AIDA", "AROA", "AGPA", "ANPA"];
161    let bytes = text.as_bytes();
162    let mut out = String::with_capacity(text.len());
163    let mut i = 0;
164    while i < bytes.len() {
165        let boundary_before = i == 0 || !bytes[i - 1].is_ascii_alphanumeric();
166        let candidate = bytes.get(i..i + 20).filter(|w| {
167            PREFIXES.iter().any(|p| w.starts_with(p.as_bytes()))
168                && w.iter()
169                    .all(|b| b.is_ascii_uppercase() || b.is_ascii_digit())
170        });
171        let boundary_after = bytes.get(i + 20).is_none_or(|b| !b.is_ascii_alphanumeric());
172        if boundary_before && boundary_after && candidate.is_some() {
173            out.push_str("[redacted]");
174            i += 20;
175        } else {
176            let ch = text[i..].chars().next().expect("in bounds");
177            out.push(ch);
178            i += ch.len_utf8();
179        }
180    }
181    out
182}
183
184#[cfg(test)]
185mod tests {
186    use super::*;
187
188    const BODY: &str = "<?xml version=\"1.0\"?><Error><Code>SignatureDoesNotMatch</Code>\
189        <AWSAccessKeyId>AKIAIOSFODNN7EXAMPLE</AWSAccessKeyId>\
190        <StringToSign>AWS4-HMAC-SHA256\n20261020T000000Z</StringToSign>\
191        <SignatureProvided>abc123</SignatureProvided><RequestId>R1</RequestId></Error>";
192
193    #[test]
194    fn an_s3_refusal_body_loses_its_credentials() {
195        let r = redact(BODY);
196        for secret in ["AKIAIOSFODNN7EXAMPLE", "abc123", "AWS4-HMAC-SHA256"] {
197            assert!(!r.contains(secret), "{secret} survived: {r}");
198        }
199        assert!(
200            r.contains("SignatureDoesNotMatch"),
201            "the useful part stays: {r}"
202        );
203        assert!(r.contains("<RequestId>R1</RequestId>"));
204    }
205
206    #[test]
207    fn signing_parameters_and_bare_key_ids_are_redacted() {
208        let r = redact(
209            "GET /b?X-Amz-Credential=AKIAIOSFODNN7EXAMPLE%2F2026&X-Amz-Signature=deadbeef ok \
210             Authorization: AWS4-HMAC-SHA256 Credential=ASIAABCDEFGHIJKLMNOP/x, Signature=ff",
211        );
212        for secret in [
213            "AKIAIOSFODNN7EXAMPLE",
214            "deadbeef",
215            "ASIAABCDEFGHIJKLMNOP",
216            "=ff",
217        ] {
218            assert!(!r.contains(secret), "{secret} survived: {r}");
219        }
220        assert!(r.contains(" ok "));
221        // A bare id in prose, too; a longer word that merely starts the same is not.
222        assert_eq!(redact("key AKIAIOSFODNN7EXAMPLE."), "key [redacted].");
223        assert_eq!(redact("AKIAIOSFODNN7EXAMPLEXX"), "AKIAIOSFODNN7EXAMPLEXX");
224    }
225
226    #[test]
227    fn a_store_error_displays_redacted() {
228        let e = BackupError::Store(object_store::Error::Generic {
229            store: "S3",
230            source: Box::new(std::io::Error::other(BODY)),
231        });
232        let shown = e.to_string();
233        assert!(!shown.contains("AKIAIOSFODNN7EXAMPLE"), "{shown}");
234        assert!(shown.contains("[redacted]"));
235    }
236}
backup/lib.rs

backup/
lib.rs
