cache.rs
raw
//! A local record of chunks the repository is known to hold.
//!
//! Storing a chunk that already exists is harmless — writes are
//! idempotent and deduplication is the whole point — but *finding out*
//! costs a round trip to the backend for every chunk considered. Over a
//! remote link that dominates: re-reading an interrupted twenty-gigabyte
//! file asks about twenty thousand chunks before it gets back to where
//! it stopped, and a second backup of an unchanged tree asks about every
//! chunk in it.
//!
//! The cache answers those questions locally. It is advisory in one
//! direction only: not knowing about a chunk simply means asking the
//! backend as before, so a truncated, half-written, or deleted cache
//! costs speed and nothing else. Claiming a chunk the repository does
//! not have would be another matter, which is why entries are only ever
//! added after the backend has confirmed the chunk — by storing it, or
//! by saying it already had it.
use std::{
collections::HashSet,
fs::{File, OpenOptions},
io::{BufReader, BufWriter, Write as _},
path::{Path, PathBuf},
sync::Mutex,
};
use crate::{ChunkId, Error};
/// Chunk ids are 32 bytes and the body is a plain concatenation of them,
/// so a torn write can only ever leave a trailing partial id, which
/// loading discards.
const ID_BYTES: usize = 32;
const MAGIC: &[u8; 8] = b"beepchnk";
const FORMAT_VERSION: u8 = 1;
/// Magic, version, repository fingerprint, epoch.
const HEADER_BYTES: usize = 8 + 1 + 16 + 16;
/// What a cache file must agree with before a single id in it is
/// trusted: which repository it belongs to, and what that repository had
/// removed when it was written.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CacheIdentity {
/// [`crate::Repository::fingerprint`] — the repository *and*
/// passphrase this cache describes.
pub fingerprint: [u8; 16],
/// [`crate::Repository::epoch`] — a digest of everything the
/// repository has ever had pruned out of it.
pub epoch: [u8; 16],
}
impl CacheIdentity {
fn encode(&self) -> Vec<u8> {
let mut header = Vec::with_capacity(HEADER_BYTES);
header.extend_from_slice(MAGIC);
header.push(FORMAT_VERSION);
header.extend_from_slice(&self.fingerprint);
header.extend_from_slice(&self.epoch);
header
}
/// Reads a header, or `None` if this is not a cache file of ours.
fn decode(bytes: &[u8]) -> Option<CacheIdentity> {
if bytes.len() < HEADER_BYTES || &bytes[..8] != MAGIC || bytes[8] != FORMAT_VERSION {
return None;
}
Some(CacheIdentity {
fingerprint: bytes[9..25].try_into().ok()?,
epoch: bytes[25..41].try_into().ok()?,
})
}
}
pub struct ChunkCache {
known: Mutex<HashSet<ChunkId>>,
/// Appended to as chunks are confirmed. Never fsynced: losing recent
/// additions costs a few round trips next time, and nothing else.
log: Mutex<BufWriter<File>>,
path: PathBuf,
/// Which repository this cache describes. Fixed for the life of the
/// file, unlike the epoch, which every prune moves on.
fingerprint: [u8; 16],
}
impl ChunkCache {
/// Opens the cache at `path` for the repository described by
/// `identity`, keeping what it holds only if it agrees.
///
/// A file belonging to another repository, written by another format
/// version, or written before a prune is discarded and started
/// again — never read. Starting empty is always correct here: it
/// costs round trips and nothing else, whereas trusting the wrong
/// file would have a backup skip storing chunks the repository does
/// not have.
pub fn open(path: impl Into<PathBuf>, identity: CacheIdentity) -> Result<ChunkCache, Error> {
let path = path.into();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let known = match File::open(&path) {
Ok(file) => Self::load(file, identity)?,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => None,
Err(err) => return Err(err.into()),
};
let known = match known {
Some(known) => known,
None => {
// Start the file again under this identity
let mut fresh = File::create(&path)?;
fresh.write_all(&identity.encode())?;
HashSet::new()
}
};
let log = OpenOptions::new().append(true).open(&path)?;
Ok(ChunkCache {
known: Mutex::new(known),
log: Mutex::new(BufWriter::new(log)),
path,
fingerprint: identity.fingerprint,
})
}
/// Reads a cache file, returning its contents if its header agrees
/// with `identity` and `None` if the file is not this repository's
/// to trust. A trailing partial id — all a crash can leave — is
/// dropped.
fn load(file: File, identity: CacheIdentity) -> Result<Option<HashSet<ChunkId>>, Error> {
let mut reader = BufReader::new(file);
let mut header = [0u8; HEADER_BYTES];
if read_full(&mut reader, &mut header)? != HEADER_BYTES {
return Ok(None);
}
if CacheIdentity::decode(&header) != Some(identity) {
return Ok(None);
}
let mut known = HashSet::new();
let mut buffer = [0u8; ID_BYTES];
while read_full(&mut reader, &mut buffer)? == ID_BYTES {
known.insert(ChunkId::from_bytes(buffer));
}
Ok(Some(known))
}
/// Drops the given chunks and adopts a new epoch: what a prune does
/// to the cache on the machine it ran from, so that the rest of the
/// cache stays useful instead of being thrown away wholesale.
///
/// Rewritten in place through a temporary file, so an interrupted
/// rewrite leaves either the old cache (whose epoch no longer
/// matches, and which is therefore discarded on the next open) or
/// the new one.
pub fn forget(&self, swept: &HashSet<ChunkId>, epoch: [u8; 16]) -> Result<(), Error> {
let mut known = self.known.lock().unwrap();
let mut log = self.log.lock().unwrap();
known.retain(|id| !swept.contains(id));
let identity = CacheIdentity {
fingerprint: self.fingerprint,
epoch,
};
let temporary = self.path.with_extension("rewriting");
let mut fresh = BufWriter::new(File::create(&temporary)?);
fresh.write_all(&identity.encode())?;
for id in known.iter() {
fresh.write_all(id.as_bytes())?;
}
fresh.flush()?;
drop(fresh);
std::fs::rename(&temporary, &self.path)?;
*log = BufWriter::new(OpenOptions::new().append(true).open(&self.path)?);
Ok(())
}
/// Where the cache lives, for diagnostics.
pub fn path(&self) -> &Path {
&self.path
}
/// How many chunks it knows about.
pub fn len(&self) -> usize {
self.known.lock().unwrap().len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// Whether the repository is known to hold this chunk. `false` means
/// "ask the backend", never "the chunk is absent".
pub fn holds(&self, id: ChunkId) -> bool {
self.known.lock().unwrap().contains(&id)
}
/// Records a chunk the backend has confirmed — because it was just
/// stored, or because it said it already had it.
///
/// Write failures are deliberately ignored: this is a cache, and a
/// backup must not fail because one could not be written.
pub fn remember(&self, id: ChunkId) {
if !self.known.lock().unwrap().insert(id) {
return;
}
let mut log = self.log.lock().unwrap();
let _ = log.write_all(id.as_bytes());
}
/// Pushes buffered additions to the file. Called at the points a
/// pipeline makes its own state durable; missing them loses only
/// speed.
pub fn flush(&self) {
let _ = self.log.lock().unwrap().flush();
}
}
impl Drop for ChunkCache {
fn drop(&mut self) {
self.flush();
}
}
/// Reads until the buffer is full or the file ends, returning how much
/// was read. `Read::read` is free to return less than asked for, and a
/// short read at the end is exactly what a torn write looks like.
fn read_full(reader: &mut impl std::io::Read, buffer: &mut [u8]) -> Result<usize, Error> {
let mut filled = 0;
while filled < buffer.len() {
match reader.read(&mut buffer[filled..])? {
0 => break,
count => filled += count,
}
}
Ok(filled)
}
#[cfg(test)]
mod tests {
use super::*;
fn id(seed: u8) -> ChunkId {
ChunkId::from_bytes([seed; ID_BYTES])
}
fn identity(fingerprint: u8, epoch: u8) -> CacheIdentity {
CacheIdentity {
fingerprint: [fingerprint; 16],
epoch: [epoch; 16],
}
}
#[test]
fn remembers_across_reopening() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("chunks");
{
let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
assert!(!cache.holds(id(1)));
cache.remember(id(1));
cache.remember(id(2));
cache.remember(id(1)); // already known, not written twice
assert!(cache.holds(id(1)));
assert_eq!(cache.len(), 2);
}
let reopened = ChunkCache::open(&path, identity(1, 0)).unwrap();
assert_eq!(reopened.len(), 2);
assert!(reopened.holds(id(1)));
assert!(reopened.holds(id(2)));
assert!(!reopened.holds(id(3)));
assert_eq!(
std::fs::metadata(&path).unwrap().len() as usize,
HEADER_BYTES + 2 * ID_BYTES,
"a repeated chunk is not appended again"
);
}
#[test]
fn a_torn_write_costs_only_its_own_record() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("chunks");
{
let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
cache.remember(id(1));
cache.remember(id(2));
}
// Half of the second id survives the crash
let mut truncated = std::fs::read(&path).unwrap();
truncated.truncate(HEADER_BYTES + ID_BYTES + ID_BYTES / 2);
std::fs::write(&path, truncated).unwrap();
let reopened = ChunkCache::open(&path, identity(1, 0)).unwrap();
assert!(reopened.holds(id(1)), "whole records still load");
assert!(!reopened.holds(id(2)), "the partial one is dropped");
// And it goes on being usable
reopened.remember(id(3));
assert!(reopened.holds(id(3)));
}
#[test]
fn another_repositorys_cache_is_never_read() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("chunks");
{
let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
cache.remember(id(1));
}
let foreign = ChunkCache::open(&path, identity(2, 0)).unwrap();
assert!(
!foreign.holds(id(1)),
"a cache belonging to another repository must not be trusted"
);
// ...and it is now this repository's cache, not the other's
foreign.remember(id(7));
drop(foreign);
let reopened = ChunkCache::open(&path, identity(2, 0)).unwrap();
assert!(reopened.holds(id(7)));
assert!(!reopened.holds(id(1)));
}
#[test]
fn a_prune_since_the_cache_was_written_discards_it() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("chunks");
{
let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
cache.remember(id(1));
cache.remember(id(2));
}
// Something was pruned elsewhere: every id in the file may name
// a chunk that is gone
let after = ChunkCache::open(&path, identity(1, 9)).unwrap();
assert!(after.is_empty());
}
#[test]
fn forgetting_keeps_the_rest_of_the_cache_warm() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("chunks");
let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
for seed in 1..=5 {
cache.remember(id(seed));
}
let swept: HashSet<ChunkId> = [id(2), id(4)].into_iter().collect();
cache.forget(&swept, [9u8; 16]).unwrap();
assert!(cache.holds(id(1)) && cache.holds(id(3)) && cache.holds(id(5)));
assert!(!cache.holds(id(2)) && !cache.holds(id(4)));
// It goes on being appendable, and reopens under the new epoch
cache.remember(id(6));
drop(cache);
let reopened = ChunkCache::open(&path, identity(1, 9)).unwrap();
assert_eq!(reopened.len(), 4);
assert!(reopened.holds(id(6)) && !reopened.holds(id(2)));
}
#[test]
fn a_missing_cache_simply_starts_empty() {
let dir = tempfile::tempdir().unwrap();
let cache =
ChunkCache::open(dir.path().join("nested/deeper/chunks"), identity(1, 0)).unwrap();
assert!(cache.is_empty());
cache.remember(id(9));
assert!(cache.holds(id(9)));
}
}