OXYGENHUB
# 696bbf29d3eb oxygen / crates / oxy-core / src / durable.rs
3.7 KiB · Rust · last changed Crash safety: durable writes and the op head as commit point 1 month ago source blame history raw
1
//! Durable file updates (spec §13.6): every write that must survive a
2
//! crash goes through here.
3
//!
4
//! A file is replaced by writing a uniquely named temporary file in the
5
//! same directory, syncing it, renaming it over the target, and syncing
6
//! the directory so the rename itself is on disk. A crash at any point
7
//! leaves either the old file or the new one, never a torn one; at worst
8
//! a stray temporary file, which nothing reads.
9
//!
10
//! In tests, `crash_after(n)` makes the n-th durable step on the current
11
//! thread fail as if the process had died there, so a test can stop an
12
//! operation at every step and check what is left behind.
13
14
use rand::RngCore;
15
use std::fs;
16
use std::io::{self, Write};
17
use std::path::Path;
18
19
/// Replace `path` with `data`, atomically and durably.
20
pub fn write_atomic(path: &Path, data: &[u8]) -> io::Result<()> {
21
    let dir = path.parent().unwrap_or(Path::new("."));
22
    let name = path
23
        .file_name()
24
        .map(|n| n.to_string_lossy().into_owned())
25
        .unwrap_or_default();
26
    let mut nonce = [0u8; 6];
27
    rand::rngs::OsRng.fill_bytes(&mut nonce);
28
    let tmp = dir.join(format!(".{name}.tmp-{}", hex::encode(nonce)));
29
    step("create temporary file")?;
30
    let written = (|| {
31
        let mut f = fs::File::create(&tmp)?;
32
        f.write_all(data)?;
33
        f.sync_all()?;
34
        step("rename into place")?;
35
        fs::rename(&tmp, path)
36
    })();
37
    if let Err(e) = written {
38
        if !is_injected(&e) {
39
            let _ = fs::remove_file(&tmp);
40
        }
41
        return Err(e);
42
    }
43
    sync_dir(dir)
44
}
45
46
/// Rename `from` to `to` and make the rename durable.
47
pub fn rename(from: &Path, to: &Path) -> io::Result<()> {
48
    step("rename")?;
49
    fs::rename(from, to)?;
50
    sync_dir(to.parent().unwrap_or(Path::new(".")))
51
}
52
53
/// Flush a directory's entries (new, renamed or removed files) to disk.
54
/// A no-op where directories cannot be opened for syncing.
55
pub fn sync_dir(dir: &Path) -> io::Result<()> {
56
    #[cfg(unix)]
57
    {
58
        fs::File::open(dir)?.sync_all()
59
    }
60
    #[cfg(not(unix))]
61
    {
62
        let _ = dir;
63
        Ok(())
64
    }
65
}
66
67
/// A point where a crash may happen. Outside tests it does nothing.
68
pub fn step(what: &str) -> io::Result<()> {
69
    #[cfg(test)]
70
    {
71
        failpoint::hit(what)
72
    }
73
    #[cfg(not(test))]
74
    {
75
        let _ = what;
76
        Ok(())
77
    }
78
}
79
80
const INJECTED: &str = "injected crash";
81
82
fn is_injected(e: &io::Error) -> bool {
83
    e.to_string().starts_with(INJECTED)
84
}
85
86
#[cfg(test)]
87
pub(crate) mod failpoint {
88
    use std::cell::Cell;
89
    use std::io;
90
91
    thread_local! {
92
        static COUNTDOWN: Cell<Option<usize>> = const { Cell::new(None) };
93
        static FIRED: Cell<bool> = const { Cell::new(false) };
94
    }
95
96
    /// Fail the `n`-th durable step from now (1-based) on this thread,
97
    /// and every step after it, as a dead process would.
98
    pub fn crash_after(n: usize) {
99
        COUNTDOWN.with(|c| c.set(Some(n)));
100
        FIRED.with(|f| f.set(false));
101
    }
102
103
    /// Stop injecting; returns whether a crash was injected.
104
    pub fn disarm() -> bool {
105
        COUNTDOWN.with(|c| c.set(None));
106
        FIRED.with(|f| f.replace(false))
107
    }
108
109
    pub(super) fn hit(what: &str) -> io::Result<()> {
110
        if FIRED.with(|f| f.get()) {
111
            return Err(io::Error::other(format!(
112
                "{} (after) at {what}",
113
                super::INJECTED
114
            )));
115
        }
116
        COUNTDOWN.with(|c| match c.get() {
117
            Some(1) => {
118
                c.set(None);
119
                FIRED.with(|f| f.set(true));
120
                Err(io::Error::other(format!("{} at {what}", super::INJECTED)))
121
            }
122
            Some(n) => {
123
                c.set(Some(n - 1));
124
                Ok(())
125
            }
126
            None => Ok(()),
127
        })
128
    }
129
}