use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Durability {
Rename,
Fsync,
}
/// the mode is set at create time. nobody can read the file before it is private.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Access {
Shared,
Private,
}
impl Access {
pub(crate) fn create_dir_all(self, dir: &Path) -> std::io::Result<()> {
use std::os::unix::fs::DirBuilderExt as _;
let mode = match self {
Self::Shared => 0o777,
Self::Private => 0o700,
};
std::fs::DirBuilder::new()
.recursive(true)
.mode(mode)
.create(dir)
}
pub(crate) fn open_options(self) -> std::fs::OpenOptions {
use std::os::unix::fs::OpenOptionsExt as _;
let mut options = std::fs::OpenOptions::new();
options.mode(match self {
Self::Shared => 0o666,
Self::Private => 0o600,
});
options
}
}
static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
/// stage in the target's own directory, because a rename across filesystems fails.
fn staging(path: &Path) -> PathBuf {
let name = path
.file_name()
.map_or_else(|| "state".into(), |name| name.to_string_lossy());
path.with_file_name(format!(
".{name}.tmp.{}.{}",
std::process::id(),
TMP_COUNTER.fetch_add(1, Ordering::Relaxed)
))
}
pub(crate) fn publish(path: &Path, bytes: &[u8], durability: Durability) -> std::io::Result<()> {
publish_as(path, bytes, durability, Access::Shared)
}
pub(crate) fn publish_as(
path: &Path,
bytes: &[u8],
durability: Durability,
access: Access,
) -> std::io::Result<()> {
let (parent, tmp) = stage(path, bytes, durability, access)?;
std::fs::rename(&tmp, path).inspect_err(|_| {
let _ = std::fs::remove_file(&tmp);
})?;
if durability == Durability::Fsync {
fsync_dir(&parent)?;
}
Ok(())
}
fn stage(
path: &Path,
bytes: &[u8],
durability: Durability,
access: Access,
) -> std::io::Result<(PathBuf, PathBuf)> {
let parent = path.parent().unwrap_or(Path::new(".")).to_path_buf();
access.create_dir_all(&parent)?;
let tmp = staging(path);
let written = (|| -> std::io::Result<()> {
let mut file = access
.open_options()
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
file.write_all(bytes)?;
if durability == Durability::Fsync {
sync(&file)?;
}
Ok(())
})();
match written {
Ok(()) => Ok((parent, tmp)),
Err(error) => {
let _ = std::fs::remove_file(&tmp);
Err(error)
}
}
}
#[derive(Debug)]
pub(crate) enum Published {
Mine,
Existing(Vec<u8>),
}
pub(crate) fn publish_once(
path: &Path,
bytes: &[u8],
durability: Durability,
access: Access,
) -> std::io::Result<Published> {
let (parent, tmp) = stage(path, bytes, durability, access)?;
let linked = std::fs::hard_link(&tmp, path);
let _ = std::fs::remove_file(&tmp);
match linked {
Ok(()) => {
if durability == Durability::Fsync {
fsync_dir(&parent)?;
}
Ok(Published::Mine)
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
Ok(Published::Existing(std::fs::read(path)?))
}
Err(error) => Err(error),
}
}
#[cfg(test)]
thread_local! {
static SYNCS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
#[cfg(test)]
pub(crate) fn syncs() -> u64 {
SYNCS.get()
}
fn sync(file: &std::fs::File) -> std::io::Result<()> {
#[cfg(test)]
SYNCS.set(SYNCS.get() + 1);
file.sync_data()
}
/// one `syncfs` covers every file, cheaper than fsyncing tens of thousands of them.
pub(crate) fn sync_filesystem(dir: &Path) -> std::io::Result<()> {
let handle = std::fs::File::open(dir)?;
#[cfg(test)]
SYNCS.set(SYNCS.get() + 1);
#[cfg(target_os = "linux")]
{
use std::os::fd::AsRawFd as _;
// SAFETY: the fd is an open directory this call owns for the duration
if unsafe { libc::syncfs(handle.as_raw_fd()) } != 0 {
return Err(std::io::Error::last_os_error());
}
Ok(())
}
#[cfg(not(target_os = "linux"))]
handle.sync_all()
}
/// starts writeback early. a big file is mostly on disk when the generation syncs.
pub(crate) fn start_writeback(file: &std::fs::File) {
#[cfg(target_os = "linux")]
{
use std::os::fd::AsRawFd as _;
// SAFETY: the fd is open for the call, and a failed hint changes nothing
unsafe {
libc::sync_file_range(file.as_raw_fd(), 0, 0, libc::SYNC_FILE_RANGE_WRITE);
}
}
#[cfg(not(target_os = "linux"))]
let _ = file;
}
/// macos fsync does not reach the platter. this is best effort there.
pub(crate) fn sync_directory(dir: &Path, durability: Durability) -> std::io::Result<()> {
if durability == Durability::Fsync {
fsync_dir(dir)?;
}
Ok(())
}
fn fsync_dir(dir: &Path) -> std::io::Result<()> {
let handle = std::fs::File::open(dir)?;
#[cfg(test)]
SYNCS.set(SYNCS.get() + 1);
match handle.sync_all() {
// some filesystems fail a directory fsync with EINVAL, which counts as synced
Err(error) if error.kind() == std::io::ErrorKind::InvalidInput => Ok(()),
other => other,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Barrier};
#[test]
fn a_durable_write_syncs_the_file_and_its_directory_and_a_renamed_one_neither() {
let dir = tempfile::tempdir().unwrap();
let before = syncs();
publish(&dir.path().join("fast.json"), b"fast", Durability::Rename).unwrap();
assert_eq!(syncs(), before, "a renamed write syncs nothing");
publish(
&dir.path().join("durable.json"),
b"durable",
Durability::Fsync,
)
.unwrap();
assert_eq!(
syncs(),
before + 2,
"a durable write syncs the file and the directory"
);
assert_eq!(
std::fs::read(dir.path().join("durable.json")).unwrap(),
b"durable"
);
}
#[test]
fn a_private_write_makes_its_file_and_directories_the_users_alone() {
use std::os::unix::fs::PermissionsExt as _;
let dir = tempfile::tempdir().unwrap();
let mode = |path: &Path| std::fs::metadata(path).unwrap().permissions().mode() & 0o777;
let path = dir.path().join("a/b/secret");
publish_as(&path, b"secret", Durability::Rename, Access::Private).unwrap();
let once = dir.path().join("c/once");
publish_once(&once, b"once", Durability::Rename, Access::Private).unwrap();
for private in [
dir.path().join("a"),
dir.path().join("a/b"),
dir.path().join("c"),
] {
assert_eq!(mode(&private), 0o700, "{}", private.display());
}
assert_eq!(mode(&path), 0o600);
assert_eq!(mode(&once), 0o600);
}
#[test]
fn staging_names_are_unique_and_hidden() {
let path = Path::new("/state/repos/k/active.json");
let barrier = Arc::new(Barrier::new(4));
let threads: Vec<_> = (0..4)
.map(|_| {
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
(0..32).map(|_| staging(path)).collect::<Vec<_>>()
})
})
.collect();
let mut names = std::collections::BTreeSet::new();
let mut total = 0;
for thread in threads {
for name in thread.join().unwrap() {
total += 1;
assert_eq!(
name.parent(),
path.parent(),
"{name:?} must stage beside its target"
);
let file = name.file_name().unwrap().to_string_lossy().into_owned();
assert!(
file.starts_with(&format!(".active.json.tmp.{}", std::process::id())),
"{file} must be hidden and carry the pid"
);
assert!(names.insert(name), "duplicate staging name");
}
}
assert_eq!(names.len(), total);
}
#[test]
fn only_the_first_publisher_of_a_name_wins_and_the_rest_read_its_bytes() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("client_key.sealed");
let barrier = Arc::new(Barrier::new(4));
let threads: Vec<_> = (0..4u8)
.map(|index| {
let barrier = Arc::clone(&barrier);
let path = path.clone();
std::thread::spawn(move || {
barrier.wait();
publish_once(&path, &[index; 8], Durability::Fsync, Access::Shared).unwrap()
})
})
.collect();
let published: Vec<Published> = threads
.into_iter()
.map(|thread| thread.join().unwrap())
.collect();
let stored = std::fs::read(&path).unwrap();
let mine = published
.iter()
.filter(|published| matches!(published, Published::Mine))
.count();
assert_eq!(mine, 1, "exactly one publisher may own the name");
for published in &published {
if let Published::Existing(bytes) = published {
assert_eq!(bytes, &stored, "a loser must read the winner's bytes");
}
}
let residue: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name.contains(".tmp."))
.collect();
assert!(residue.is_empty(), "staging file left behind: {residue:?}");
}
#[test]
fn a_failed_write_leaves_no_staging_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("record.json");
std::fs::create_dir(&path).unwrap();
publish(&path, b"mine", Durability::Fsync).expect_err("the rename cannot succeed");
let residue: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name.contains(".tmp."))
.collect();
assert!(residue.is_empty(), "staging file left behind: {residue:?}");
}
}