backup/
disk.rs

1//! Everything the worker reads from the store directory, as blocking calls.
2//!
3//! The worker runs each of these on Tokio's blocking pool
4//! ([`tokio::task::spawn_blocking`]), never on a runtime thread: a segment
5//! read can be 64 MiB, and a side file grows with the store's history.
6
7use std::io;
8use std::path::Path;
9use std::time::UNIX_EPOCH;
10
11/// A file's identity as `(len, mtime_nanos)` — the form the engine uses for
12/// `snapshot.bin` ([`mushroomdb::WalPosition::snapshot`]).
13pub type Ident = (u64, u64);
14
15/// [`Ident`] of a file's metadata, computed as the engine computes it.
16pub fn ident_of(meta: &std::fs::Metadata) -> Ident {
17    let mtime_nanos = meta
18        .modified()
19        .ok()
20        .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
21        .map(|d| d.as_nanos() as u64)
22        .unwrap_or(0);
23    (meta.len(), mtime_nanos)
24}
25
26fn not_found_is_none<T>(r: io::Result<T>) -> io::Result<Option<T>> {
27    match r {
28        Ok(v) => Ok(Some(v)),
29        Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
30        Err(e) => Err(e),
31    }
32}
33
34/// What one look at the store directory says.
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub struct DiskState {
37    /// `snapshot.bin`'s identity; `None` when there is none.
38    pub snapshot: Option<Ident>,
39    /// `wal.bin`'s length in bytes (0 when absent).
40    pub wal_len: u64,
41    /// Which file `wal.bin` is ([`mushroomdb::WalPosition::wal`]'s form).
42    pub wal: Option<u64>,
43    /// Whether `snapshot.intent` holds anything: a snapshot is running, or
44    /// one stopped half-way.
45    pub intent_nonempty: bool,
46}
47
48pub fn observe(dir: &Path) -> io::Result<DiskState> {
49    let snapshot =
50        not_found_is_none(std::fs::metadata(dir.join("snapshot.bin")))?.map(|m| ident_of(&m));
51    let wal_meta = not_found_is_none(std::fs::metadata(dir.join("wal.bin")))?;
52    let wal_len = wal_meta.as_ref().map_or(0, |m| m.len());
53    let wal = wal_meta.as_ref().and_then(file_ident);
54    let intent_nonempty = not_found_is_none(std::fs::metadata(dir.join("snapshot.intent")))?
55        .is_some_and(|m| m.len() > 0);
56    Ok(DiskState {
57        snapshot,
58        wal_len,
59        wal,
60        intent_nonempty,
61    })
62}
63
64impl DiskState {
65    /// Nothing that pairs a snapshot with a WAL changed between `self` and a
66    /// later look `after`: no snapshot was running at either, the same
67    /// snapshot and the same WAL file are there, and the WAL only grew.
68    pub fn unchanged_until(&self, after: &DiskState) -> bool {
69        !self.intent_nonempty
70            && !after.intent_nonempty
71            && self.snapshot == after.snapshot
72            && self.wal == after.wal
73            && after.wal_len >= self.wal_len
74    }
75}
76
77/// The inode number on Unix; `None` elsewhere. The same identity the engine
78/// records for `wal.bin`.
79fn file_ident(meta: &std::fs::Metadata) -> Option<u64> {
80    #[cfg(unix)]
81    {
82        use std::os::unix::fs::MetadataExt as _;
83        Some(meta.ino())
84    }
85    #[cfg(not(unix))]
86    {
87        let _ = meta;
88        None
89    }
90}
91
92/// `snapshot.bin` opened, with the identity of the file actually opened.
93pub fn open_snapshot(dir: &Path) -> io::Result<Option<(std::fs::File, Ident)>> {
94    let Some(f) = not_found_is_none(std::fs::File::open(dir.join("snapshot.bin")))? else {
95        return Ok(None);
96    };
97    let ident = ident_of(&f.metadata()?);
98    Ok(Some((f, ident)))
99}
100
101/// Every `wal.N.archive` in the directory, as `(n, len)`, sorted by `n`.
102pub fn local_archives(dir: &Path) -> io::Result<Vec<(u64, u64)>> {
103    let mut out = Vec::new();
104    for entry in std::fs::read_dir(dir)? {
105        let entry = entry?;
106        let name = entry.file_name();
107        if let Some(n) = crate::manifest::parse_archive_name(&name.to_string_lossy()) {
108            if let Some(m) = not_found_is_none(entry.metadata())? {
109                out.push((n, m.len()));
110            }
111        }
112    }
113    out.sort_unstable();
114    Ok(out)
115}
116
117/// A side file's identity, and its bytes when the identity differs from
118/// `known`. `Ok(None)` when the file does not exist.
119pub fn read_side(
120    dir: &Path,
121    name: &str,
122    known: Option<Ident>,
123) -> io::Result<Option<(Ident, Option<Vec<u8>>)>> {
124    let path = dir.join(name);
125    let Some(meta) = not_found_is_none(std::fs::metadata(&path))? else {
126        return Ok(None);
127    };
128    let ident = ident_of(&meta);
129    if known == Some(ident) {
130        return Ok(Some((ident, None)));
131    }
132    match not_found_is_none(std::fs::read(&path))? {
133        Some(bytes) => Ok(Some((ident, Some(bytes)))),
134        None => Ok(None),
135    }
136}
137
138/// The store file naming the backup lineage it belongs to.
139pub const LINEAGE_FILE: &str = "backup.lineage";
140
141/// The store's backup lineage, or `None` when it has never been backed up.
142pub fn read_lineage(dir: &Path) -> io::Result<Option<String>> {
143    Ok(
144        not_found_is_none(std::fs::read_to_string(dir.join(LINEAGE_FILE)))?
145            .map(|s| s.trim().to_string())
146            .filter(|s| !s.is_empty()),
147    )
148}
149
150/// Give the store the backup lineage `id`: written to a temporary file,
151/// fsynced, renamed into place, and the directory fsynced. The one file the
152/// backup worker ever writes into a store directory.
153pub fn create_lineage(dir: &Path, id: &str) -> io::Result<()> {
154    use std::io::Write as _;
155    let tmp = dir.join(format!("{LINEAGE_FILE}.tmp"));
156    {
157        let mut f = std::fs::File::create(&tmp)?;
158        f.write_all(id.as_bytes())?;
159        f.sync_all()?;
160    }
161    std::fs::rename(&tmp, dir.join(LINEAGE_FILE))?;
162    #[cfg(unix)]
163    std::fs::File::open(dir)?.sync_all()?;
164    Ok(())
165}
166
167/// The store file a restore-on-boot writes: which lineage and restore point
168/// the store was restored from. Never uploaded (it is not a side file).
169pub const PARENT_FILE: &str = "backup.parent";
170
171/// What [`PARENT_FILE`] holds.
172///
173/// A store restored at boot has a new lineage of its own, like every restore.
174/// This record is what lets its backup continue under the prefix it was
175/// restored from: the worker adopts that prefix only while `LATEST` still
176/// names exactly this manifest of exactly this lineage, and from then on
177/// writes only objects of its own lineage beside the parent's, which it never
178/// overwrites or deletes.
179#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
180pub struct ParentRecord {
181    /// The lineage of the backup the store was restored from.
182    pub lineage: String,
183    /// The key of the manifest it was restored from.
184    pub manifest: String,
185    /// The commits the restored store held.
186    pub commit_seq: u64,
187}
188
189/// The store's parent record, or `None` when it has none (or one that does
190/// not parse: it then adopts nothing).
191pub fn read_parent(dir: &Path) -> io::Result<Option<ParentRecord>> {
192    Ok(not_found_is_none(std::fs::read(dir.join(PARENT_FILE)))?
193        .and_then(|b| serde_json::from_slice(&b).ok()))
194}
195
196/// Write [`PARENT_FILE`] as [`create_lineage`] writes the lineage: a
197/// temporary file, fsynced, renamed into place, the directory fsynced.
198pub fn create_parent(dir: &Path, record: &ParentRecord) -> io::Result<()> {
199    use std::io::Write as _;
200    let bytes = serde_json::to_vec(record).map_err(io::Error::other)?;
201    let tmp = dir.join(format!("{PARENT_FILE}.tmp"));
202    {
203        let mut f = std::fs::File::create(&tmp)?;
204        f.write_all(&bytes)?;
205        f.sync_all()?;
206    }
207    std::fs::rename(&tmp, dir.join(PARENT_FILE))?;
208    #[cfg(unix)]
209    std::fs::File::open(dir)?.sync_all()?;
210    Ok(())
211}
212
213/// CRC-32 of a whole file, read from its start; the file is left at its
214/// start again.
215pub fn crc32_of(file: &mut std::fs::File) -> io::Result<String> {
216    use std::io::{Read as _, Seek as _, SeekFrom};
217    file.seek(SeekFrom::Start(0))?;
218    let mut h = crc32fast::Hasher::new();
219    let mut buf = vec![0u8; 1 << 20];
220    loop {
221        let n = file.read(&mut buf)?;
222        if n == 0 {
223            break;
224        }
225        h.update(&buf[..n]);
226    }
227    file.seek(SeekFrom::Start(0))?;
228    Ok(crate::manifest::format_crc32(h.finalize()))
229}
230
231/// The store's horizon floor (`wal.floor`), or 0.
232pub fn floor(dir: &Path) -> u64 {
233    std::fs::read(dir.join("wal.floor"))
234        .ok()
235        .and_then(|b| {
236            b.get(..8)
237                .map(|s| u64::from_le_bytes(s.try_into().unwrap()))
238        })
239        .unwrap_or(0)
240}
backup/disk.rs

backup/
disk.rs
