use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use anyhow::{Context, Result};
pub fn backoff(attempt: u32, base: Duration, max: Duration) -> Duration {
if attempt == 0 {
return Duration::ZERO;
}
let factor = 1u32.checked_shl((attempt - 1).min(30)).unwrap_or(u32::MAX);
base.saturating_mul(factor).min(max)
}
pub fn write_atomic(root: &Path, filename: &str, bytes: &[u8]) -> Result<()> {
let _framed = gunnar_core::txlog::sink::replace_file(&root.join(filename), bytes, false)
.with_context(|| format!("framing {filename} in the transaction log"))?;
write_atomic_unframed(root, filename, bytes)
}
pub fn write_atomic_unframed(root: &Path, filename: &str, bytes: &[u8]) -> Result<()> {
static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
let path = root.join(filename);
let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
let temp = root.join(format!("{filename}.{}-{seq}.tmp", std::process::id()));
let written = (|| -> Result<()> {
let mut f = std::fs::File::create(&temp)
.with_context(|| format!("writing {}", temp.display()))?;
std::io::Write::write_all(&mut f, bytes)
.with_context(|| format!("writing {}", temp.display()))?;
f.sync_all()
.with_context(|| format!("flushing {} to disk", temp.display()))?;
std::fs::rename(&temp, &path).with_context(|| format!("replacing {}", path.display()))
})();
if let Err(e) = written {
let _ = std::fs::remove_file(&temp);
return Err(e);
}
sync_dir(root).with_context(|| format!("flushing the save of {filename} in {}", root.display()))
}
pub fn sync_dir(dir: &Path) -> Result<()> {
#[cfg(test)]
if let Some(e) = dir_sync_seam::injected() {
return Err(e).with_context(|| format!("flushing {}", dir.display()));
}
let d = std::fs::File::open(dir).with_context(|| format!("opening {} to flush it", dir.display()))?;
d.sync_all().with_context(|| format!("flushing {}", dir.display()))
}
pub fn remove_durable(root: &Path, filename: &str) -> Result<()> {
match std::fs::remove_file(root.join(filename)) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(e) => return Err(e).with_context(|| format!("removing {}", root.join(filename).display())),
}
sync_dir(root).with_context(|| format!("flushing the removal of {filename} from {}", root.display()))
}
pub fn write_atomic_mirrored(
root: &Path,
slow: &crate::slow::SlowRoot,
filename: &str,
bytes: &[u8],
) -> Result<()> {
write_atomic(root, filename, bytes)?;
let mirrored = slow
.meta_dir()
.and_then(|dir: std::path::PathBuf| write_atomic_unframed(&dir, filename, bytes));
if let Err(err) = mirrored {
tracing::warn!(
target: crate::slow::TARGET_SLOW,
file = filename,
mirror_dir = %slow.path_of(crate::slow::META_DIR).display(),
error = %format!("{err:#}"),
"the metadata mirror on the slow disk could not be written; the file itself is \
saved on /data. This box now has ONE copy of it."
);
}
Ok(())
}
#[cfg(test)]
pub(crate) mod dir_sync_seam {
use std::cell::Cell;
thread_local! {
static FAIL: Cell<bool> = const { Cell::new(false) };
}
pub(crate) fn fail_from_now() {
FAIL.with(|f| f.set(true));
}
pub(crate) fn injected() -> Option<std::io::Error> {
FAIL.with(|f| f.get()).then(|| std::io::Error::other("injected directory fsync failure"))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn concurrent_writers_of_one_file_never_trip_over_each_others_temp() {
let dir = tempfile::tempdir().unwrap();
let root = std::sync::Arc::new(dir.path().to_path_buf());
let writers: Vec<_> = (0..8u8)
.map(|w| {
let root = std::sync::Arc::clone(&root);
std::thread::spawn(move || {
let body = vec![b'a' + w; 64 * 1024];
(0..40).map(|_| write_atomic_unframed(&root, "doc.json", &body).map_err(|e| format!("{e:#}"))).collect::<Vec<_>>()
})
})
.collect();
let failures: Vec<String> = writers.into_iter().flat_map(|w| w.join().unwrap()).filter_map(|r| r.err()).collect();
assert!(failures.is_empty(), "{} of 320 saves failed, first: {:?}", failures.len(), failures.first());
let landed = std::fs::read(dir.path().join("doc.json")).unwrap();
assert_eq!(landed.len(), 64 * 1024, "a save landed torn");
assert!(landed.iter().all(|b| *b == landed[0]), "a save landed interleaved");
}
#[test]
fn a_leftover_temp_does_not_wedge_the_next_save() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("doc.json.tmp")).unwrap();
write_atomic_unframed(dir.path(), "doc.json", b"{}").unwrap();
assert_eq!(std::fs::read(dir.path().join("doc.json")).unwrap(), b"{}");
}
#[test]
fn a_failed_directory_fsync_fails_the_save() {
let dir = tempfile::tempdir().unwrap();
dir_sync_seam::fail_from_now();
assert!(write_atomic_unframed(dir.path(), "doc.json", b"{}").is_err(), "a save whose directory fsync failed reported success");
}
}