use std::collections::BTreeMap;
use std::io::Read as _;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use sha2::{Digest as _, Sha256};
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct Digest([u8; 32]);
impl Digest {
pub fn of_bytes(bytes: &[u8]) -> Self {
Self(Sha256::digest(bytes).into())
}
pub fn of_file(path: &Path) -> std::io::Result<Self> {
let mut file = std::fs::File::open(path)?;
let mut hasher = Sha256::new();
let mut buffer = [0u8; 64 * 1024];
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
Ok(Self(hasher.finalize().into()))
}
pub fn from_raw(bytes: [u8; 32]) -> Self {
Self(bytes)
}
pub fn hex(&self) -> String {
knot_types::lowercase_hex(&self.0)
}
}
/// digests of files this process wrote, which publication reads instead of rehashing.
static WRITTEN_DIGESTS: OnceLock<Mutex<BTreeMap<PathBuf, String>>> = OnceLock::new();
pub(crate) fn note_written_digest(path: &Path, digest: &Digest) {
let map = WRITTEN_DIGESTS.get_or_init(|| Mutex::new(Default::default()));
if let Ok(mut guard) = map.lock() {
guard.insert(path.to_path_buf(), digest.hex());
}
}
pub(crate) fn written_digest(path: &Path) -> Option<String> {
WRITTEN_DIGESTS
.get()
.and_then(|map| map.lock().ok())
.and_then(|guard| guard.get(path).cloned())
}
/// no fsync here, a crash can lose the file and the next build rebuilds it.
pub fn atomic_write(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
crate::datastore::write::publish(path, bytes, crate::datastore::Durability::Rename)?;
note_written_digest(path, &Digest::of_bytes(bytes));
Ok(())
}
/// hardlinked trees share inodes, and a write through one path would change them all.
/// read-only makes that write fail, a rename still replaces the file.
pub fn seal_read_only(root: &Path) -> std::io::Result<()> {
for entry in std::fs::read_dir(root)? {
let entry = entry?;
let file_type = entry.file_type()?;
if file_type.is_dir() {
seal_read_only(&entry.path())?;
} else if file_type.is_file() {
seal_file(&entry.path())?;
}
}
Ok(())
}
/// an existing `dst` is fine, names are content addressed.
pub fn link_or_reflink(src: &Path, dst: &Path) -> std::io::Result<()> {
if let Some(parent) = dst.parent() {
std::fs::create_dir_all(parent)?;
}
match std::fs::hard_link(src, dst) {
Err(error) if error.kind() != std::io::ErrorKind::AlreadyExists => {}
_ => return Ok(()),
}
match reflink(src, dst) {
Err(error) if error.kind() != std::io::ErrorKind::AlreadyExists => {
atomic_write(dst, &std::fs::read(src)?)
}
_ => Ok(()),
}
}
fn c_path(path: &Path) -> std::io::Result<std::ffi::CString> {
use std::os::unix::ffi::OsStrExt as _;
std::ffi::CString::new(path.as_os_str().as_bytes())
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))
}
#[cfg(target_os = "macos")]
fn reflink(src: &Path, dst: &Path) -> std::io::Result<()> {
let (from, to) = (c_path(src)?, c_path(dst)?);
// SAFETY: both are nul-terminated paths that outlive the call.
match unsafe { libc::clonefile(from.as_ptr(), to.as_ptr(), 0) } {
0 => Ok(()),
_ => Err(std::io::Error::last_os_error()),
}
}
/// clone into a staging file first, readers of `dst` must never see half a clone.
#[cfg(target_os = "linux")]
fn reflink(src: &Path, dst: &Path) -> std::io::Result<()> {
use std::os::fd::AsRawFd as _;
const FICLONE: u64 = 0x4004_9409;
let source = std::fs::File::open(src)?;
let staging = dst.with_file_name(format!(
".{}.clone.{}",
dst.file_name()
.and_then(|name| name.to_str())
.unwrap_or("file"),
std::process::id()
));
let path = c_path(&staging)?;
// SAFETY: `path` is a nul-terminated CString.
let fd = unsafe {
libc::open(
path.as_ptr(),
libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC,
0o644,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error());
}
// SAFETY: both descriptors are open for the duration of the call.
let cloned = unsafe { libc::ioctl(fd, FICLONE as _, source.as_raw_fd()) };
let error = std::io::Error::last_os_error();
// SAFETY: fd came from the open above.
unsafe { libc::close(fd) };
if cloned != 0 {
let _ = std::fs::remove_file(&staging);
return Err(error);
}
std::fs::rename(&staging, dst).inspect_err(|_| {
let _ = std::fs::remove_file(&staging);
})
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
fn reflink(_src: &Path, _dst: &Path) -> std::io::Result<()> {
Err(std::io::ErrorKind::Unsupported.into())
}
pub fn seal_file(path: &Path) -> std::io::Result<()> {
use std::os::unix::fs::PermissionsExt as _;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o444))
}
#[cfg(test)]
mod atomic_writes {
use super::*;
#[test]
fn a_failed_rename_is_an_error_and_leaves_no_digest() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("record.json");
std::fs::create_dir(&path).unwrap();
atomic_write(&path, b"mine").expect_err("a failed rename is not a write");
assert_eq!(written_digest(&path), None, "nothing was published here");
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:?}");
}
}
#[cfg(test)]
mod linked_copies {
use super::link_or_reflink;
#[test]
fn a_linked_copy_holds_the_same_bytes_and_tolerates_an_existing_name() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("src");
super::atomic_write(&src, b"shared").unwrap();
let dst = dir.path().join("nested/dst");
link_or_reflink(&src, &dst).unwrap();
assert_eq!(std::fs::read(&dst).unwrap(), b"shared");
link_or_reflink(&src, &dst).unwrap();
assert_eq!(std::fs::read(&dst).unwrap(), b"shared");
}
#[test]
fn a_reflink_or_copy_stands_in_when_no_hardlink_can() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("src");
super::atomic_write(&src, b"cloned").unwrap();
let dst = dir.path().join("dst");
match super::reflink(&src, &dst) {
Ok(()) => assert_eq!(std::fs::read(&dst).unwrap(), b"cloned"),
Err(error) => eprintln!("no reflink on this filesystem: {error}"),
}
}
}
#[cfg(test)]
mod hex_regressions {
use super::Digest;
#[test]
fn hex_matches_formatting_for_every_byte() {
for byte in 0..=u8::MAX {
let digest = Digest::from_raw([byte; 32]);
let hex = digest.hex();
assert_eq!(hex, format!("{byte:02x}").repeat(32));
}
}
}
#[cfg(test)]
pub(crate) mod sealed {
use std::path::Path;
/// root can write read-only files, which is why the open check only runs for non-root.
pub(crate) fn assert_sealed(root: &Path) {
use std::os::unix::fs::PermissionsExt as _;
let probe = tempfile::NamedTempFile::new().unwrap();
std::fs::set_permissions(probe.path(), std::fs::Permissions::from_mode(0o444)).unwrap();
let writes_refused = std::fs::OpenOptions::new()
.write(true)
.open(probe.path())
.is_err();
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
for entry in std::fs::read_dir(&dir).unwrap() {
let entry = entry.unwrap();
let file_type = entry.file_type().unwrap();
if file_type.is_dir() {
stack.push(entry.path());
} else if file_type.is_file() {
let path = entry.path();
let mode = std::fs::metadata(&path).unwrap().permissions().mode();
assert_eq!(mode & 0o222, 0, "{} is writable", path.display());
if writes_refused {
let error = std::fs::OpenOptions::new()
.write(true)
.open(&path)
.unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
}
}
}
}
}
}
#[cfg(test)]
mod raw_write_lint {
use std::path::Path;
const RAW_WRITES: &[&str] = &[
"fs::write(",
"fs::copy(",
"File::create(",
".write(true)",
".append(true)",
];
const ALLOWED: &[(&str, &str, usize)] = &[
("write.rs", ".write(true)", 1),
("record.rs", ".write(true)", 1),
("marks.rs", ".write(true)", 1),
// lock files, opened writable only for flock.
("output.rs", ".write(true)", 1),
("repolock.rs", ".write(true)", 1),
("gc.rs", ".write(true)", 1),
("mod.rs", ".write(true)", 1),
// copied because the export can't hardlink the source pack.
("reach.rs", "let target = std::fs::File::create(&to)?;", 1),
(
"packfile.rs",
"File::create(&path).with_context(|| format!(\"create {}\", path.display()))",
1,
),
("flow.rs", ".write(true)", 1),
(
"fork.rs",
"let mut out = std::fs::File::create(pack.path()).map_err(internal)?;",
2,
),
("git_ssh.rs", ".append(true)", 1),
("git_ssh.rs", ".write(true)", 1),
("provision.rs", "std::fs::copy(entry.path(), &target)", 1),
(
"failpoint.rs",
"let _ = std::fs::write(path, format!(\"{name}\\n\"));",
1,
),
("atomic.rs", ".write(true)", 1),
(
"atomic.rs",
".or_else(|_| std::fs::copy(&path, &target).map(|_| ()))?;",
1,
),
(
"store.rs",
".or_else(|_| std::fs::copy(&source, &target).map(|_| ()))?;",
1,
),
// blobs aren't a published tree, and their mtime is the grace clock.
("blobs.rs", ".write(true)", 1),
// appends only, and no shared bytes change under an append.
("firehose.rs", ".append(true)", 1),
];
fn shipped_lines(path: &Path) -> Vec<String> {
let text = std::fs::read_to_string(path).unwrap();
let lines: Vec<&str> = text.lines().collect();
let end = lines
.windows(2)
.position(|pair| {
pair[0].trim() == "#[cfg(test)]"
&& pair[1].contains("mod ")
&& pair[1].contains('{')
})
.unwrap_or(lines.len());
lines[..end]
.iter()
.map(|line| line.trim().to_string())
.collect()
}
fn sources(path: &Path, out: &mut Vec<std::path::PathBuf>) {
if path.is_dir() {
for entry in std::fs::read_dir(path).unwrap() {
sources(&entry.unwrap().path(), out);
}
} else if path.extension().is_some_and(|ext| ext == "rs")
&& path
.file_name()
.is_some_and(|name| name != "tests.rs" && name != "fuzz.rs")
&& !path
.parent()
.is_some_and(|dir| dir.file_name().is_some_and(|name| name == "tests"))
{
out.push(path.to_path_buf());
}
}
#[test]
fn published_trees_are_written_by_rename_only() {
let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
let mut files = Vec::new();
sources(&src, &mut files);
let mut found = Vec::new();
// the count stops a second write of the same shape in one file from passing.
let mut seen = vec![0usize; ALLOWED.len()];
for file in &files {
let name = file.file_name().unwrap().to_str().unwrap();
for (number, line) in shipped_lines(file).iter().enumerate() {
if !RAW_WRITES.iter().any(|write| line.contains(write)) {
continue;
}
match ALLOWED
.iter()
.position(|(file, site, _)| *file == name && line == site)
{
Some(index) => seen[index] += 1,
None => found.push(format!("{}:{}: {line}", file.display(), number + 1)),
}
}
}
let miscounted: Vec<String> = ALLOWED
.iter()
.zip(&seen)
.filter(|((_, _, allowed), seen)| *seen != allowed)
.map(|((file, site, allowed), seen)| format!("{file}: {site} ({seen} of {allowed})"))
.collect();
assert!(
miscounted.is_empty(),
"allowed writes that no longer appear as often as they may:\n{}",
miscounted.join("\n")
);
assert!(
found.is_empty(),
"write these with cas::atomic_write (temp file + rename):\n{}",
found.join("\n")
);
}
}