durable.rs 93.4%

99 of 106 lines run, 31 of 37 functions · run #387 on main · run missed

//! Durable file updates (spec §13.6): every write that must survive a
//! crash goes through here.
//!
//! A file is replaced by writing a uniquely named temporary file in the
//! same directory, syncing it, renaming it over the target, and syncing
//! the directory so the rename itself is on disk. A crash at any point
//! leaves either the old file or the new one, never a torn one; at worst
//! a stray temporary file, which nothing reads.
//!
//! In tests, `crash_after(n)` makes the n-th durable step on the current
//! thread fail as if the process had died there, so a test can stop an
//! operation at every step and check what is left behind.

use rand::RngCore;
use std::fs;
use std::io::{self, Write};
use std::path::Path;

/// Replace `path` with `data`, atomically and durably.
pub fn write_atomic(path: &Path, data: &[u8]) -> io::Result<()> {
    write_atomic_mode(path, data, false)
}

/// [`write_atomic`] for a secret: on Unix the file is created readable by
/// its owner only, so the secret is never on disk with wider permissions.
pub fn write_atomic_private(path: &Path, data: &[u8]) -> io::Result<()> {
    write_atomic_mode(path, data, true)
}

fn write_atomic_mode(path: &Path, data: &[u8], private: bool) -> io::Result<()> {
    let dir = path.parent().unwrap_or(Path::new("."));
    let name = path
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_default();
    let mut nonce = [0u8; 6];
    rand::rngs::OsRng.fill_bytes(&mut nonce);
    let tmp = dir.join(format!(".{name}.tmp-{}", hex::encode(nonce)));
    step("create temporary file")?;
    let written = (|| {
        let mut open = fs::OpenOptions::new();
        open.write(true).create_new(true);
        #[cfg(unix)]
        if private {
            use std::os::unix::fs::OpenOptionsExt;
            open.mode(0o600);
        }
        #[cfg(not(unix))]
        let _ = private;
        let mut f = open.open(&tmp)?;
        f.write_all(data)?;
        f.sync_all()?;
        step("rename into place")?;
        fs::rename(&tmp, path)
    })();
    if let Err(e) = written {
        if !is_injected(&e) {
            let _ = fs::remove_file(&tmp);
        }
        return Err(at("write", path, e));
    }
    sync_dir(dir).map_err(|e| at("sync the directory of", path, e))
}

/// `e`, naming what failed and on which file, so a failure deep in a
/// command says where it happened (an injected test crash stays as is).
/// The kind is kept for callers that test it.
fn at(what: &str, path: &Path, e: io::Error) -> io::Error {
    if is_injected(&e) {
        return e;
    }
    io::Error::new(e.kind(), format!("{what} {}: {e}", path.display()))
}

/// Rename `from` to `to` and make the rename durable.
pub fn rename(from: &Path, to: &Path) -> io::Result<()> {
    step("rename")?;
    fs::rename(from, to).map_err(|e| at(&format!("rename {} to", from.display()), to, e))?;
    sync_dir(to.parent().unwrap_or(Path::new("."))).map_err(|e| at("sync the directory of", to, e))
}

/// Flush a directory's entries (new, renamed or removed files) to disk.
/// A no-op where directories cannot be opened for syncing.
pub fn sync_dir(dir: &Path) -> io::Result<()> {
    #[cfg(unix)]
    {
        fs::File::open(dir)?.sync_all()
    }
    #[cfg(not(unix))]
    {
        let _ = dir;
        Ok(())
    }
}

/// A point where a crash may happen. Outside tests it does nothing.
pub fn step(what: &str) -> io::Result<()> {
    #[cfg(test)]
    {
        failpoint::hit(what)
    }
    #[cfg(not(test))]
    {
        let _ = what;
        Ok(())
    }
}

const INJECTED: &str = "injected crash";

fn is_injected(e: &io::Error) -> bool {
    e.to_string().starts_with(INJECTED)
}

#[cfg(test)]
pub(crate) mod failpoint {
    use std::cell::Cell;
    use std::io;

    thread_local! {
        static COUNTDOWN: Cell<Option<usize>> = const { Cell::new(None) };
        static FIRED: Cell<bool> = const { Cell::new(false) };
    }

    /// Fail the `n`-th durable step from now (1-based) on this thread,
    /// and every step after it, as a dead process would.
    pub fn crash_after(n: usize) {
        COUNTDOWN.with(|c| c.set(Some(n)));
        FIRED.with(|f| f.set(false));
    }

    /// Stop injecting; returns whether a crash was injected.
    pub fn disarm() -> bool {
        COUNTDOWN.with(|c| c.set(None));
        FIRED.with(|f| f.replace(false))
    }

    pub(super) fn hit(what: &str) -> io::Result<()> {
        if FIRED.with(|f| f.get()) {
            return Err(io::Error::other(format!(
                "{} (after) at {what}",
                super::INJECTED
            )));
        }
        COUNTDOWN.with(|c| match c.get() {
            Some(1) => {
                c.set(None);
                FIRED.with(|f| f.set(true));
                Err(io::Error::other(format!("{} at {what}", super::INJECTED)))
            }
            Some(n) => {
                c.set(Some(n - 1));
                Ok(())
            }
            None => Ok(()),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_failed_write_names_the_file_and_keeps_its_kind() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("attest").join("feature").join("x");
        let e = write_atomic(&path, b"state").unwrap_err();
        assert_eq!(e.kind(), io::ErrorKind::NotFound);
        let msg = e.to_string();
        assert!(
            msg.starts_with("write ") && msg.contains("attest/feature/x"),
            "{msg}"
        );
        write_atomic(&dir.path().join("ok"), b"state").unwrap();
        let e = rename(&dir.path().join("missing"), &dir.path().join("to")).unwrap_err();
        assert!(
            e.to_string().contains("missing") && e.to_string().contains("to"),
            "{e}"
        );
    }
}

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	8682×	pub fn write_atomic(path: &Path, data: &[u8]) -> io::Result<()> {
21	8682×	write_atomic_mode(path, data, false)
22	8682×	}
23
24		/// [`write_atomic`] for a secret: on Unix the file is created readable by
25		/// its owner only, so the secret is never on disk with wider permissions.
26	104×	pub fn write_atomic_private(path: &Path, data: &[u8]) -> io::Result<()> {
27	104×	write_atomic_mode(path, data, true)
28	104×	}
29
30	8786×	fn write_atomic_mode(path: &Path, data: &[u8], private: bool) -> io::Result<()> {
31	8786×	let dir = path.parent().unwrap_or(Path::new("."));
32	8786×	let name = path
33	8786×	.file_name()
34	8786×	.map(\|n\| n.to_string_lossy().into_owned())
35	8786×	.unwrap_or_default();
36	8786×	let mut nonce = [0u8; 6];
37	8786×	rand::rngs::OsRng.fill_bytes(&mut nonce);
38	8786×	let tmp = dir.join(format!(".{name}.tmp-{}", hex::encode(nonce)));
39	8786×	step("create temporary file")?;
40	8781×	let written = (\|\| {
41	8781×	let mut open = fs::OpenOptions::new();
42	8781×	open.write(true).create_new(true);
43		#[cfg(unix)]
44	8781×	if private {
45		use std::os::unix::fs::OpenOptionsExt;
46	104×	open.mode(0o600);
47	8677×	}
48		#[cfg(not(unix))]
49		let _ = private;
50	8781×	let mut f = open.open(&tmp)?;
51	8779×	f.write_all(data)?;
52	8779×	f.sync_all()?;
53	8779×	step("rename into place")?;
54	8774×	fs::rename(&tmp, path)
55		})();
56	8781×	if let Err(e) = written {
57	7×	if !is_injected(&e) {
58	2×	let _ = fs::remove_file(&tmp);
59	5×	}
60	7×	return Err(at("write", path, e));
61	8774×	}
62	8774×	sync_dir(dir).map_err(\|e\| at("sync the directory of", path, e))
63	8786×	}
64
65		/// `e`, naming what failed and on which file, so a failure deep in a
66		/// command says where it happened (an injected test crash stays as is).
67		/// The kind is kept for callers that test it.
68	8×	fn at(what: &str, path: &Path, e: io::Error) -> io::Error {
69	8×	if is_injected(&e) {
70	5×	return e;
71	3×	}
72	3×	io::Error::new(e.kind(), format!("{what} {}: {e}", path.display()))
73	8×	}
74
75		/// Rename `from` to `to` and make the rename durable.
76	1×	pub fn rename(from: &Path, to: &Path) -> io::Result<()> {
77	1×	step("rename")?;
78	1×	fs::rename(from, to).map_err(\|e\| at(&format!("rename {} to", from.display()), to, e))?;
79	0×	sync_dir(to.parent().unwrap_or(Path::new("."))).map_err(\|e\| at("sync the directory of", to, e))
80	1×	}
81
82		/// Flush a directory's entries (new, renamed or removed files) to disk.
83		/// A no-op where directories cannot be opened for syncing.
84	36663×	pub fn sync_dir(dir: &Path) -> io::Result<()> {
85		#[cfg(unix)]
86		{
87	36663×	fs::File::open(dir)?.sync_all()
88		}
89		#[cfg(not(unix))]
90		{
91		let _ = dir;
92		Ok(())
93		}
94	36663×	}
95
96		/// A point where a crash may happen. Outside tests it does nothing.
97	29060×	pub fn step(what: &str) -> io::Result<()> {
98		#[cfg(test)]
99		{
100	18428×	failpoint::hit(what)
101		}
102		#[cfg(not(test))]
103		{
104	10632×	let _ = what;
105	10632×	Ok(())
106		}
107	29060×	}
108
109		const INJECTED: &str = "injected crash";
110
111	15×	fn is_injected(e: &io::Error) -> bool {
112	15×	e.to_string().starts_with(INJECTED)
113	15×	}
114
115		#[cfg(test)]
116		pub(crate) mod failpoint {
117		use std::cell::Cell;
118		use std::io;
119
120		thread_local! {
121		static COUNTDOWN: Cell<Option<usize>> = const { Cell::new(None) };
122		static FIRED: Cell<bool> = const { Cell::new(false) };
123		}
124
125		/// Fail the `n`-th durable step from now (1-based) on this thread,
126		/// and every step after it, as a dead process would.
127	21×	pub fn crash_after(n: usize) {
128	21×	COUNTDOWN.with(\|c\| c.set(Some(n)));
129	21×	FIRED.with(\|f\| f.set(false));
130	21×	}
131
132		/// Stop injecting; returns whether a crash was injected.
133	21×	pub fn disarm() -> bool {
134	21×	COUNTDOWN.with(\|c\| c.set(None));
135	21×	FIRED.with(\|f\| f.replace(false))
136	21×	}
137
138	18428×	pub(super) fn hit(what: &str) -> io::Result<()> {
139	18428×	if FIRED.with(\|f\| f.get()) {
140	0×	return Err(io::Error::other(format!(
141	0×	"{} (after) at {what}",
142	0×	super::INJECTED
143	0×	)));
144	18428×	}
145	18428×	COUNTDOWN.with(\|c\| match c.get() {
146		Some(1) => {
147	19×	c.set(None);
148	19×	FIRED.with(\|f\| f.set(true));
149	19×	Err(io::Error::other(format!("{} at {what}", super::INJECTED)))
150		}
151	130×	Some(n) => {
152	130×	c.set(Some(n - 1));
153	130×	Ok(())
154		}
155	18279×	None => Ok(()),
156	18428×	})
157	18428×	}
158		}
159
160		#[cfg(test)]
161		mod tests {
162		use super::*;
163
164		#[test]
165	1×	fn a_failed_write_names_the_file_and_keeps_its_kind() {
166	1×	let dir = tempfile::tempdir().unwrap();
167	1×	let path = dir.path().join("attest").join("feature").join("x");
168	1×	let e = write_atomic(&path, b"state").unwrap_err();
169	1×	assert_eq!(e.kind(), io::ErrorKind::NotFound);
170	1×	let msg = e.to_string();
171	1×	assert!(
172	1×	msg.starts_with("write ") && msg.contains("attest/feature/x"),
173		"{msg}"
174		);
175	1×	write_atomic(&dir.path().join("ok"), b"state").unwrap();
176	1×	let e = rename(&dir.path().join("missing"), &dir.path().join("to")).unwrap_err();
177	1×	assert!(
178	1×	e.to_string().contains("missing") && e.to_string().contains("to"),
179		"{e}"
180		);
181	1×	}
182		}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/oxy-core/src/durable.rs 93.4%

Keyboard