//! a whole car kept in a file instead of memory, for a repo too big to hold while it's written.
//! the file is unlinked as soon as it's made, so a crash leaves nothing behind. every block is
//! hashed against its cid as it lands and again as it's read back, so a byte that rots on disk
//! in between fails the backfill instead of being stored
use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Write};
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use bytes::Bytes;
use cid::Cid as IpldCid;
use miette::{IntoDiagnostic, Result};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tracing::warn;
use crate::backfill::error::BackfillError;
use crate::car::{CarBlock, CarStream, WholeCar, validate_block_cid};
use crate::mst::BlockSource;
/// disk a spill always leaves free, so the database never runs out of room because of a
/// backfill
const DISK_HEADROOM: u64 = 256 * 1024 * 1024;
/// how much every spill together writes between checks that the headroom is still there.
/// checking as they go sizes the check to the cars that actually arrive, whatever their encoding
/// said, and counting them together keeps several spills from each writing this much past it
const DISK_CHECK_EVERY: u64 = 64 * 1024 * 1024;
#[cfg(test)]
type AfterSpill = Arc<dyn Fn(&SpilledCar) + Send + Sync>;
/// where backfills spill their cars, and how many may at once
#[derive(Clone)]
pub(crate) struct Spills(Arc<Inner>);
struct Inner {
dir: PathBuf,
threshold: usize,
#[cfg(test)]
at_once: usize,
permits: Arc<Semaphore>,
/// names each file apart from the others this process made
next: AtomicU64,
/// what every spill wrote since the last headroom check
unchecked: AtomicU64,
/// stands in for the disk's free space before anything is spilled, so a test can run
/// out of it
#[cfg(test)]
free_bytes: parking_lot::Mutex<Option<u64>>,
/// every byte spilled, which the stand-in free space goes down by
#[cfg(test)]
spilled: AtomicU64,
/// stands in for [`DISK_CHECK_EVERY`], so a test can check after every byte
#[cfg(test)]
check_every: AtomicU64,
/// runs once a spilled car has every block on disk, before anything reads it
#[cfg(test)]
after_spill: parking_lot::Mutex<Option<AfterSpill>>,
}
impl Spills {
/// spills go to `dir`, made if it's missing and emptied of anything a crash left between
/// making a file and unlinking it
pub(crate) fn open(dir: PathBuf, threshold: usize, at_once: usize) -> Result<Self> {
std::fs::create_dir_all(&dir).into_diagnostic()?;
for entry in std::fs::read_dir(&dir).into_diagnostic()? {
let path = entry.into_diagnostic()?.path();
// only ever a car of ours, in case the dir is shared with something else
if !is_spill_name(&path) {
continue;
}
if let Err(e) = std::fs::remove_file(&path) {
warn!(path = %path.display(), err = %e, "failed to remove a leftover spill");
}
}
Ok(Self::new(dir, threshold, at_once))
}
/// spills that never happen, for a fetch that never gets past `usize::MAX` bytes
#[cfg(test)]
pub(crate) fn never() -> Self {
Self::new(PathBuf::new(), usize::MAX, 1)
}
fn new(dir: PathBuf, threshold: usize, at_once: usize) -> Self {
let at_once = at_once.max(1);
Self(Arc::new(Inner {
dir,
threshold,
#[cfg(test)]
at_once,
permits: Arc::new(Semaphore::new(at_once)),
next: AtomicU64::new(0),
unchecked: AtomicU64::new(0),
#[cfg(test)]
free_bytes: parking_lot::Mutex::new(None),
#[cfg(test)]
spilled: AtomicU64::new(0),
#[cfg(test)]
check_every: AtomicU64::new(DISK_CHECK_EVERY),
#[cfg(test)]
after_spill: parking_lot::Mutex::new(None),
}))
}
/// how big a car may get in memory before it spills
pub(crate) fn threshold(&self) -> usize {
self.0.threshold
}
/// a file for a car, once fewer than the cap are spilling. a disk without the headroom
/// left, now or as the car lands, is an error the backfill retries like a failed fetch,
/// with nothing written
pub(crate) async fn start(&self) -> Result<SpillFile, BackfillError> {
let permit = self
.0
.permits
.clone()
.acquire_owned()
.await
.into_diagnostic()?;
self.check_headroom()?;
let path = self.0.dir.join(format!(
"{}-{}.car",
std::process::id(),
self.0.next.fetch_add(1, Ordering::Relaxed)
));
let file = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&path)
.into_diagnostic()?;
// the open handle keeps it, and nothing is left on disk if hydrant dies
std::fs::remove_file(&path).into_diagnostic()?;
Ok(SpillFile {
file: BufWriter::new(file),
written: 0,
index: Vec::new(),
_permit: permit,
spills: self.clone(),
})
}
/// counts `bytes` a spill wrote, and looks at the headroom again once every spill together
/// wrote [`DISK_CHECK_EVERY`] since the last look
fn wrote(&self, bytes: u64) -> Result<()> {
#[cfg(test)]
self.0.spilled.fetch_add(bytes, Ordering::Relaxed);
let unchecked = self.0.unchecked.fetch_add(bytes, Ordering::Relaxed) + bytes;
if unchecked >= self.check_every() {
self.0.unchecked.store(0, Ordering::Relaxed);
self.check_headroom()?;
}
Ok(())
}
#[cfg(not(test))]
fn check_every(&self) -> u64 {
DISK_CHECK_EVERY
}
#[cfg(test)]
fn check_every(&self) -> u64 {
self.0.check_every.load(Ordering::Relaxed)
}
/// fails once less than the headroom is free
fn check_headroom(&self) -> Result<()> {
let free = self.free_bytes()?;
if free < DISK_HEADROOM {
miette::bail!(
"{free} bytes free in {}, under the {DISK_HEADROOM} a spill leaves",
self.0.dir.display()
);
}
Ok(())
}
fn free_bytes(&self) -> Result<u64> {
#[cfg(test)]
if let Some(free) = *self.0.free_bytes.lock() {
return Ok(free.saturating_sub(self.0.spilled.load(Ordering::Relaxed)));
}
free_bytes(&self.0.dir)
}
/// cars spilling right now
#[cfg(test)]
pub(crate) fn spilling(&self) -> usize {
self.0.at_once - self.0.permits.available_permits()
}
/// the disk's free space before anything is spilled, which each spilled byte takes from
#[cfg(test)]
pub(crate) fn set_free_bytes(&self, free: u64) {
*self.0.free_bytes.lock() = Some(free);
}
/// how much every spill together writes between headroom checks
#[cfg(test)]
pub(crate) fn set_check_every(&self, bytes: u64) {
self.0.check_every.store(bytes, Ordering::Relaxed);
}
#[cfg(test)]
pub(crate) const HEADROOM: u64 = DISK_HEADROOM;
#[cfg(test)]
pub(crate) fn after_spill(&self, hook: impl Fn(&SpilledCar) + Send + Sync + 'static) {
*self.0.after_spill.lock() = Some(Arc::new(hook));
}
#[cfg(test)]
pub(crate) fn dir(&self) -> &Path {
&self.0.dir
}
}
/// whether `path` is named like the files [`Spills::start`] makes
fn is_spill_name(path: &Path) -> bool {
let digits = |part: &str| !part.is_empty() && part.bytes().all(|b| b.is_ascii_digit());
path.extension().is_some_and(|ext| ext == "car")
&& path
.file_stem()
.and_then(|stem| stem.to_str())
.and_then(|stem| stem.split_once('-'))
.is_some_and(|(pid, n)| digits(pid) && digits(n))
}
fn free_bytes(dir: &Path) -> Result<u64> {
use std::os::unix::ffi::OsStrExt;
let path = std::ffi::CString::new(dir.as_os_str().as_bytes()).into_diagnostic()?;
// SAFETY: statvfs only fills in the plain struct it's given, from a nul terminated path
let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
if unsafe { libc::statvfs(path.as_ptr(), &mut stat) } != 0 {
return Err(std::io::Error::last_os_error()).into_diagnostic();
}
#[allow(clippy::useless_conversion)]
Ok(u64::from(stat.f_bavail).saturating_mul(u64::from(stat.f_frsize)))
}
/// where a block sits in the file. 8 bytes of its digest are enough to find it, and the read
/// hashes the whole block, which tells it apart from another under the same prefix
#[derive(Debug, Clone, Copy)]
struct Entry {
prefix: u64,
offset: u64,
len: u32,
}
fn prefix(cid: &IpldCid) -> u64 {
let mut prefix = [0; 8];
let digest = cid.hash().digest();
let n = digest.len().min(8);
prefix[..n].copy_from_slice(&digest[..n]);
u64::from_be_bytes(prefix)
}
/// a spill being written, one block after another as the car streams
pub(crate) struct SpillFile {
file: BufWriter<File>,
written: u64,
index: Vec<Entry>,
_permit: OwnedSemaphorePermit,
spills: Spills,
}
impl SpillFile {
fn append(&mut self, cid: &IpldCid, payload: &[u8]) -> Result<()> {
let len = u32::try_from(payload.len()).into_diagnostic()?;
self.file.write_all(payload).into_diagnostic()?;
self.index.push(Entry {
prefix: prefix(cid),
offset: self.written,
len,
});
self.written += u64::from(len);
self.spills.wrote(u64::from(len))
}
fn finish(self) -> Result<SpilledCar> {
let file = self
.file
.into_inner()
.map_err(|e| e.into_error())
.into_diagnostic()?;
let mut index = self.index;
index.sort_unstable_by_key(|entry| entry.prefix);
let spilled = SpilledCar {
file,
index,
_permit: self._permit,
};
#[cfg(test)]
if let Some(after_spill) = self.spills.0.after_spill.lock().clone() {
after_spill(&spilled);
}
Ok(spilled)
}
}
/// a whole car streaming into its spill file. every block is hashed as it lands, whether or
/// not cids are checked otherwise, since it's read back from a disk that can rot
pub(crate) struct SpillingCar {
stream: CarStream,
file: SpillFile,
}
impl SpillingCar {
/// carries on with `car`, writing what it read so far to `file` first
pub(crate) fn from_memory(car: WholeCar, mut file: SpillFile) -> Result<Self> {
let (mut stream, blocks) = car.into_parts();
stream.check_cids();
for (cid, bytes) in blocks {
validate_block_cid(&cid, &bytes)?;
file.append(&cid, &bytes)?;
}
Ok(Self { stream, file })
}
pub(crate) fn push(&mut self, chunk: &[u8]) -> Result<()> {
let file = &mut self.file;
self.stream
.push(chunk, |_, cid, payload| file.append(&cid, payload))
}
/// the car's root and its blocks, ready to read
pub(crate) fn finish(self) -> Result<(IpldCid, SpilledCar)> {
let root = self.stream.finish_with(())?.root;
Ok((root, self.file.finish()?))
}
}
/// a spilled car's blocks, read from disk as they're asked for
pub(crate) struct SpilledCar {
file: File,
index: Vec<Entry>,
_permit: OwnedSemaphorePermit,
}
impl SpilledCar {
/// whether the car has a block for `cid`, without reading it
pub(crate) fn contains(&self, cid: &IpldCid) -> bool {
self.entries(cid).next().is_some()
}
pub(crate) fn len(&self) -> usize {
self.index.len()
}
/// the block under `cid`, read and hashed now
pub(crate) fn block(&self, cid: &IpldCid) -> Result<Option<CarBlock>> {
Ok(self.read(cid)?.map(|bytes| CarBlock::hashed(*cid, bytes)))
}
fn entries(&self, cid: &IpldCid) -> impl Iterator<Item = &Entry> {
let prefix = prefix(cid);
let start = self.index.partition_point(|entry| entry.prefix < prefix);
self.index[start..]
.iter()
.take_while(move |entry| entry.prefix == prefix)
}
fn read_entry(&self, entry: &Entry) -> Result<Vec<u8>> {
let mut bytes = vec![0; entry.len as usize];
self.file
.read_exact_at(&mut bytes, entry.offset)
.into_diagnostic()?;
Ok(bytes)
}
/// flips a byte of the block under `cid` on disk, the way a bad disk would
#[cfg(test)]
pub(crate) fn corrupt(&self, cid: &IpldCid) -> Result<()> {
for entry in self.entries(cid) {
let mut bytes = self.read_entry(entry)?;
if crate::car::compute_cid(&bytes) == *cid {
bytes[0] ^= 0xff;
return self
.file
.write_all_at(&bytes[..1], entry.offset)
.into_diagnostic();
}
}
Err(miette::miette!("{cid} isn't in the spill"))
}
}
impl BlockSource for SpilledCar {
fn read(&self, cid: &IpldCid) -> Result<Option<Bytes>> {
let mut found = false;
for entry in self.entries(cid) {
let bytes = self.read_entry(entry)?;
if crate::car::compute_cid(&bytes) == *cid {
return Ok(Some(bytes.into()));
}
found = true;
}
// another block sharing the prefix is one chance in 2^64, a rotten disk isn't
if found {
return Err(miette::miette!(
"spilled block {cid} no longer hashes to its cid"
));
}
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::FutureExt;
#[tokio::test]
async fn no_more_cars_spill_at_once_than_the_cap() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let spills = Spills::open(tmp.path().join("spill"), 0, 2)?;
let first = spills.start().await?;
let _second = spills.start().await?;
assert_eq!(spills.spilling(), 2);
let mut third = Box::pin(spills.start());
assert!((&mut third).now_or_never().is_none());
drop(first);
let _third = third
.now_or_never()
.expect("a freed slot goes to the next car")?;
assert_eq!(spills.spilling(), 2);
// every file was unlinked as soon as it was made
assert!(
std::fs::read_dir(spills.dir())
.into_diagnostic()?
.next()
.is_none()
);
Ok(())
}
#[tokio::test]
async fn spills_running_together_check_the_headroom_together() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let spills = Spills::open(tmp.path().join("spill"), 0, 2)?;
spills.set_free_bytes(Spills::HEADROOM + 100);
spills.set_check_every(100);
let mut first = spills.start().await?;
let mut second = spills.start().await?;
let cid = crate::mst::tests::cid(1);
// neither writes a check's worth on its own, but both together leave less than the
// headroom
first.append(&cid, &[0; 60])?;
let err = second.append(&cid, &[0; 60]).unwrap_err();
assert!(err.to_string().contains("bytes free"), "{err}");
Ok(())
}
#[test]
fn a_start_empties_what_a_crash_left() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let dir = tmp.path().join("spill");
std::fs::create_dir_all(&dir).into_diagnostic()?;
std::fs::write(dir.join("1-0.car"), b"left behind").into_diagnostic()?;
std::fs::write(dir.join("export.car"), b"someone else's").into_diagnostic()?;
Spills::open(dir.clone(), 0, 1)?;
let left = std::fs::read_dir(&dir)
.into_diagnostic()?
.map(|entry| Ok(entry.into_diagnostic()?.file_name()))
.collect::<Result<Vec<_>>>()?;
assert_eq!(left, ["export.car"]);
Ok(())
}
}