chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! What workflows keep from run to run: for each path a workflow names
//! under `cache:`, a base, the directory its runs start from, which every
//! run sees through an overlay of its own, so runs of one workflow go side
//! by side rather than taking turns at it.
//!
//! A run's overlay has the base as its lower layer, read only, and a
//! directory of the run's own as its upper one, which takes all it writes.
//! Only a run of a push to the default branch that passes keeps what it
//! wrote: its upper layer goes into a new base, which the runs started
//! after it start from. Every other run's upper layer, a pull's among
//! them, goes when the run ends, so what a pull writes never reaches the
//! base, and a pull starts from what the default branch last left.
//!
//! A base is never changed while a run may read it: a new one is made
//! beside it, its files hard links to the old one's where the upper layer
//! left them as they were, and the old one goes once the last run reading
//! it ends. The overlays are mounted with `userxattr`, so what says a
//! directory in an upper layer hides the one below it is an extended
//! attribute the runner can read without privileges.
//!
//! A trusted pull's run that passes, which a push to the default branch
//! at the same tree may take as its own rather than run again, keeps its
//! upper layers instead, under the key of its check: the push that takes
//! the run takes its layers too, into new bases as its own run's would
//! go, but only over the base each was written over. Applied over another,
//! its whiteouts and what it left as it was would be of what that base
//! held, not what this one does, so a layer over a base that has moved on
//! is not kept, and the next run of the default branch warms it. Kept
//! layers go once taken, once their pull has moved on ([`Caches::forget`]),
//! and past [`KEPT_AGE`], [`KEPT_MAX`] of them or a budget of bytes,
//! the oldest first. Only a run the caller says to keep is kept: an
//! untrusted pull's never is, so its writes never reach a base.
//!
//! Where no push to the default branch runs, as on a runner that runs a
//! repository's pulls alone, a trusted pull's run that passes makes the
//! base instead ([`Lease::advance`]), as a push's would, so its pulls do
//! not all start cold: but only over the base each layer was written
//! over, still current, as a kept layer is taken, and only when the tip
//! of the branch the pull merged with is the one that base was made at or
//! newer, so the base never goes back to an older tip. Each base made so
//! is marked with that tip's generation, in [`MARKS`].
//!
//! The disk the caches are on is shared, with the site on the same host
//! among others, so the runner keeps a floor of it free: below it, before
//! a run starts, [`Caches::shed`] removes every kept run's layers and
//! every base no run reads, and the runs after start afresh.
//!
//! Under the root: `bases/<cache>/<generation>`, `runs/<run>/<n>/` with
//! each layer's `upper` and the `work` directory overlayfs needs beside
//! it, `kept/<run>/`, a kept run's [`KEPT`] and its `<n>/upper`s, and
//! [`MARKS`].
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeSet, HashMap, HashSet};
use std::fs::Metadata;
use std::io;
use std::os::unix::fs::{FileTypeExt, MetadataExt};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use crate::executor;
use knotless::keys;
/// The attribute overlayfs, mounted with `userxattr`, marks a directory in
/// an upper layer with when it hides the lower layer's: one deleted and
/// made again.
const OPAQUE: &str = "user.overlay.opaque";
/// What a kept run's directory says of it.
pub const KEPT: &str = "kept.json";
/// The generation of the branch's tip each cache's base a pull made was
/// made at, as `{cache: [base generation, tip generation]}`.
pub const MARKS: &str = "marks.json";
/// How long a passing pull run's layers are kept for a push to take, in
/// seconds: three days.
pub const KEPT_AGE: u64 = 3 * 24 * 60 * 60;
/// The most runs whose layers are kept at once.
pub const KEPT_MAX: usize = 16;
/// Says the bytes free on the disk holding a path.
pub type Space = Box<dyn Fn(&Path) -> io::Result<u64> + Send + Sync>;
pub struct Caches {
root: PathBuf,
/// The bytes free on the disk the caches are on: [`free`], but in tests.
space: Space,
/// This runner's start, so its runs' names are not an earlier one's.
boot: u128,
runs: AtomicU64,
bases: Mutex<Bases>,
/// Held while kept layers are made, taken or removed.
kept: Mutex<()>,
}
/// A kept run's layers, when it was kept, and the bytes they take.
#[derive(Serialize, Deserialize, Default)]
struct Kept {
at: u64,
size: u64,
layers: Vec<KeptLayer>,
}
/// A layer kept: its cache, where the run saw it, and the generation of
/// the base it was written over.
#[derive(Serialize, Deserialize)]
struct KeptLayer {
name: String,
path: String,
generation: u64,
}
/// A kept run's layers as a push takes them: a lease over the bases they
/// were written over, of those still current, for [`Lease::promote`]; and
/// where the run saw those over a base that has moved on, which are not.
pub struct Taken {
pub lease: Lease,
pub moved: Vec<String>,
}
impl Taken {
/// Where the run saw each layer that is taken.
pub fn kept(&self) -> Vec<String> {
self.lease.layers.iter().map(|l| l.path.clone()).collect()
}
}
/// The generations of each cache's base: the one runs start from, those
/// still on disk, and how many runs read each; and of the bases a pull
/// made, the generation of the branch's tip each was made at.
#[derive(Default)]
struct Bases {
current: HashMap<String, u64>,
kept: HashMap<String, BTreeSet<u64>>,
readers: HashMap<(String, u64), usize>,
next: u64,
marks: HashMap<String, (u64, u64)>,
}
impl Bases {
/// Whether a layer written over `name`'s base `generation` by a pull
/// merged with the branch's tip at generation `at` may make the next
/// base: that base is still current, and was made at that tip or an
/// older one, or by a push, which marks none.
fn ahead(&self, name: &str, generation: u64, at: u64) -> bool {
let mark = self.marks.get(name).filter(|(g, _)| *g == generation);
self.current.get(name) == Some(&generation) && mark.is_none_or(|(_, tip)| *tip <= at)
}
/// The generations of `name` no run reads and none will start from, no
/// longer counted as on disk.
fn unused(&mut self, name: &str) -> Vec<u64> {
let current = self.current.get(name).copied();
let readers = &self.readers;
let kept = self.kept.entry(name.to_string()).or_default();
let unused: Vec<u64> = kept
.iter()
.copied()
.filter(|g| Some(*g) != current && !readers.contains_key(&(name.to_string(), *g)))
.collect();
kept.retain(|g| !unused.contains(g));
unused
}
}
/// A cache as a run sees it: the base it starts from and the layer it
/// writes to.
#[derive(Clone, Debug)]
pub struct Layer {
/// The cache, as its bases are named.
pub name: String,
/// Where the run's container sees it.
pub path: String,
generation: u64,
pub lower: PathBuf,
pub upper: PathBuf,
pub work: PathBuf,
}
/// A run's layers over the bases it reads, which stay until it ends; its
/// own layers go then.
pub struct Lease {
caches: Arc<Caches>,
pub id: String,
dir: PathBuf,
pub layers: Vec<Layer>,
}
impl Caches {
/// The caches under `root`, with what runs of an earlier runner left
/// there gone, but for the layers of the runs `alive` names, whose
/// overlays are still mounted, and the bases those may read; and the
/// layers kept of the runs `kept` names, whose checks a push may yet
/// take, and of no other.
pub fn open(root: &Path, alive: &[String], kept: &[String]) -> Result<Arc<Caches>> {
Caches::open_with(root, alive, kept, Box::new(free))
}
/// [`Caches::open`], told the bytes free on its disk by `space`.
pub fn open_with(
root: &Path,
alive: &[String],
kept: &[String],
space: Space,
) -> Result<Arc<Caches>> {
let (bases_dir, runs_dir) = (root.join("bases"), root.join("runs"));
std::fs::create_dir_all(&bases_dir).context("making the caches' directory")?;
std::fs::create_dir_all(&runs_dir)?;
for run in std::fs::read_dir(&runs_dir)? {
let run = run?;
if !alive.iter().any(|a| run.file_name() == a.as_str()) {
remove(&run.path())?;
}
}
let kept_dir = root.join("kept");
std::fs::create_dir_all(&kept_dir)?;
let wanted: HashSet<String> = kept.iter().map(|r| keys::ref_hash(r)).collect();
for run in std::fs::read_dir(&kept_dir)? {
let run = run?.path();
let named = run.file_name().map(|n| n.to_string_lossy().into_owned());
if !named.is_some_and(|n| wanted.contains(&n)) || read_kept(&run).is_none() {
remove(&run)?;
}
}
let mut bases = Bases::default();
for cache in std::fs::read_dir(&bases_dir)? {
let cache = cache?;
let name = cache.file_name().to_string_lossy().into_owned();
// An earlier runner's, named so that its repository cannot be
// told, and so never dropped with it: its next run warms anew.
if name.starts_with(PREFIX) && repo_of(&name).is_none() {
remove(&cache.path())?;
continue;
}
let mut gens = BTreeSet::new();
for base in std::fs::read_dir(cache.path())? {
let base = base?;
match base.file_name().to_string_lossy().parse::<u64>() {
Ok(generation) => gens.insert(generation),
// One half made when the runner stopped.
Err(_) => remove(&base.path()).map(|_| false)?,
};
}
let Some(&newest) = gens.last() else {
continue;
};
bases.next = bases.next.max(newest + 1);
bases.current.insert(name.clone(), newest);
bases.kept.insert(name.clone(), gens);
if alive.is_empty() {
for old in bases.unused(&name) {
remove(&cache.path().join(old.to_string()))?;
}
}
}
// Each mark of a base still current, which a pull made.
let marks = std::fs::read(root.join(MARKS)).unwrap_or_default();
let marks: HashMap<String, (u64, u64)> = serde_json::from_slice(&marks).unwrap_or_default();
bases.marks = marks
.into_iter()
.filter(|(name, (g, _))| bases.current.get(name) == Some(g))
.collect();
let boot = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis();
Ok(Arc::new(Caches {
root: root.to_path_buf(),
space,
boot,
runs: AtomicU64::new(0),
bases: Mutex::new(bases),
kept: Mutex::new(()),
}))
}
/// Drops the caches of the repository `repo` ([`name`]), as of one
/// deleted: each base no run reads now goes, and those that runs do go
/// once they end, none being current any more.
pub fn drop_repo(&self, repo: &str) -> Result<()> {
let hash = keys::ref_hash(repo);
let mut bases = self.bases.lock().unwrap();
let names: Vec<String> = bases.kept.keys().cloned().collect();
let names = names.into_iter().filter(|n| repo_of(n) == Some(&hash));
for name in names.collect::<Vec<_>>() {
bases.current.remove(&name);
bases.marks.remove(&name);
let dir = self.root.join("bases").join(&name);
for generation in bases.unused(&name) {
remove(&dir.join(generation.to_string()))?;
}
// Empty, with no run reading it.
let _ = std::fs::remove_dir(&dir);
}
Ok(())
}
/// The bytes free on the disk the caches are on.
pub fn free(&self) -> io::Result<u64> {
(self.space)(&self.root)
}
/// Makes room on the disk, as it runs short: removes the layers kept
/// of every run, which a push would only have saved a run's work
/// with, and every cache's base no run reads, as [`Caches::drop_repo`]
/// drops them, for the next run of each to start afresh. Says how
/// many bases and kept runs went.
pub fn shed(&self) -> Result<(usize, usize)> {
let mut runs = 0;
{
let _kept = self.kept.lock().unwrap();
for run in std::fs::read_dir(self.root.join("kept"))? {
remove(&run?.path())?;
runs += 1;
}
}
let mut bases = self.bases.lock().unwrap();
let mut gone = 0;
let current: Vec<(String, u64)> = bases.current.clone().into_iter().collect();
for (name, generation) in current {
if bases.readers.contains_key(&(name.clone(), generation)) {
continue;
}
bases.current.remove(&name);
bases.marks.remove(&name);
for generation in bases.unused(&name) {
remove(&self.base(&name, generation))?;
gone += 1;
}
}
Ok((gone, runs))
}
/// Where the layers kept of the run `run` are.
fn kept_dir(&self, run: &str) -> PathBuf {
self.root.join("kept").join(keys::ref_hash(run))
}
/// Keeps the layers `lease` wrote to, those with anything in them, as
/// the run `run`'s, `now`, unless they would take more than `budget`
/// bytes; then keeps the kept layers within their bounds.
fn keeping(&self, lease: Lease, run: &str, budget: u64, now: u64) -> Result<()> {
let _kept = self.kept.lock().unwrap();
let dir = self.kept_dir(run);
let making = dir.with_extension("new");
remove(&dir)?;
remove(&making)?;
std::fs::create_dir_all(&making)?;
let mut layers = Vec::new();
for layer in &lease.layers {
if std::fs::read_dir(&layer.upper)?.next().is_none() {
continue;
}
let to = making.join(layers.len().to_string());
std::fs::create_dir(&to)?;
std::fs::rename(&layer.upper, to.join("upper"))?;
layers.push(KeptLayer {
name: layer.name.clone(),
path: layer.path.clone(),
generation: layer.generation,
});
}
let size = size(&making)?;
if layers.is_empty() || size > budget {
return Ok(remove(&making)?);
}
let kept = Kept {
at: now,
size,
layers,
};
std::fs::write(making.join(KEPT), serde_json::to_vec(&kept)?)?;
std::fs::rename(&making, &dir)?;
self.bound(budget, now)
}
/// Removes the kept layers past their bounds, `now`: older than
/// [`KEPT_AGE`], and, the newest first, past [`KEPT_MAX`] of them or
/// `budget` bytes in all. Called holding `kept`.
fn bound(&self, budget: u64, now: u64) -> Result<()> {
let mut all = Vec::new();
for run in std::fs::read_dir(self.root.join("kept"))? {
let run = run?.path();
let kept = read_kept(&run).unwrap_or_default();
all.push((kept.at, kept.size, run));
}
all.sort_by_key(|(at, ..)| std::cmp::Reverse(*at));
let (mut count, mut total) = (0, 0);
for (at, size, run) in all {
let fits = now.saturating_sub(at) <= KEPT_AGE && count < KEPT_MAX;
if fits && total + size <= budget {
(count, total) = (count + 1, total + size);
} else {
remove(&run)?;
}
}
Ok(())
}
/// Removes the layers kept of each of `runs`, as of runs a push can no
/// longer take, and those past their bounds, as [`Caches::bound`].
pub fn forget(&self, runs: &[String], budget: u64) -> Result<()> {
self.forgetting(runs, budget, now())
}
/// [`Caches::forget`], `now`.
fn forgetting(&self, runs: &[String], budget: u64, now: u64) -> Result<()> {
let _kept = self.kept.lock().unwrap();
for run in runs {
remove(&self.kept_dir(run))?;
}
self.bound(budget, now)
}
/// Takes the layers kept of the run `run`, for a push that takes its
/// check: each over a base still current, leased over it, and where
/// the run saw each over one that has moved on since, which is not.
/// They go once the lease does, promoted or not.
pub fn take(self: &Arc<Self>, run: &str) -> Taken {
let _kept = self.kept.lock().unwrap();
let dir = self.kept_dir(run);
let kept = read_kept(&dir).unwrap_or_default();
let mut bases = self.bases.lock().unwrap();
let (mut layers, mut moved) = (Vec::new(), Vec::new());
for (i, layer) in kept.layers.into_iter().enumerate() {
let KeptLayer {
name,
path,
generation,
} = layer;
if bases.current.get(&name) != Some(&generation) {
moved.push(path);
continue;
}
*bases.readers.entry((name.clone(), generation)).or_default() += 1;
let at = dir.join(i.to_string());
layers.push(Layer {
lower: self.base(&name, generation),
name,
path,
generation,
upper: at.join("upper"),
work: at.join("work"),
});
}
let id = keys::ref_hash(run);
let caches = self.clone();
let lease = Lease {
caches,
id,
dir,
layers,
};
Taken { lease, moved }
}
/// Writes down `marks`, the tips the bases pulls made were made at.
fn mark(&self, marks: &HashMap<String, (u64, u64)>) -> Result<()> {
let path = self.root.join(MARKS);
let making = path.with_extension("new");
std::fs::write(&making, serde_json::to_vec(marks)?)?;
Ok(std::fs::rename(&making, &path)?)
}
fn base(&self, name: &str, generation: u64) -> PathBuf {
self.root
.join("bases")
.join(name)
.join(generation.to_string())
}
/// Layers for a run over the current base of each of `caches`, as
/// `(name, path)`: an empty base for a cache that has none yet.
pub fn lease(self: &Arc<Self>, caches: &[(String, String)]) -> Result<Lease> {
let n = self.runs.fetch_add(1, Ordering::Relaxed);
let id = format!("{}-{n}", self.boot);
let dir = self.root.join("runs").join(&id);
let mut lease = Lease {
caches: self.clone(),
id,
dir,
layers: Vec::new(),
};
for (i, (name, path)) in caches.iter().enumerate() {
let generation = self.read(name)?;
let layer = lease.dir.join(i.to_string());
lease.layers.push(Layer {
name: name.clone(),
path: path.clone(),
generation,
lower: self.base(name, generation),
upper: layer.join("upper"),
work: layer.join("work"),
});
std::fs::create_dir_all(layer.join("upper"))?;
std::fs::create_dir_all(layer.join("work"))?;
}
Ok(lease)
}
/// The generation of `name`'s base a run starts from, counted as read
/// until it ends.
fn read(&self, name: &str) -> Result<u64> {
let mut bases = self.bases.lock().unwrap();
let generation = match bases.current.get(name) {
Some(g) => *g,
None => {
let g = bases.next;
std::fs::create_dir_all(self.base(name, g))?;
bases.next += 1;
bases.current.insert(name.to_string(), g);
bases.kept.entry(name.to_string()).or_default().insert(g);
g
}
};
*bases
.readers
.entry((name.to_string(), generation))
.or_default() += 1;
Ok(generation)
}
/// Removes the bases of `name` that are no longer used.
fn sweep(&self, name: &str) {
let unused = self.bases.lock().unwrap().unused(name);
for generation in unused {
let _ = remove(&self.base(name, generation));
}
}
}
impl Lease {
/// The overlays to mount, one docker volume each, named for the run.
pub fn mounts(&self) -> Vec<executor::Cache> {
self.layers
.iter()
.enumerate()
.map(|(i, l)| executor::Cache {
volume: format!("knotless-overlay-{}-{i}", self.id),
path: l.path.clone(),
run: self.id.clone(),
lower: l.lower.clone(),
upper: l.upper.clone(),
work: l.work.clone(),
})
.collect()
}
/// Keeps what the run wrote, once its overlays are unmounted, as the
/// run `run`'s, for a push that takes its check to take, within the
/// bounds of kept layers and `budget` bytes for them all.
pub async fn keep(self, run: &str, budget: u64) -> Result<()> {
let (caches, run) = (self.caches.clone(), run.to_string());
let kept = move || caches.keeping(self, &run, budget, now());
tokio::task::spawn_blocking(kept).await?
}
/// Makes what the run wrote, once its overlays are unmounted, the base
/// the runs after it start from: each layer it wrote to over its base,
/// as a new one. A base that would be over `max` bytes is dropped
/// instead, for the next run to start afresh.
pub async fn promote(self, max: u64) -> Result<()> {
self.promoting(max, None).await.map(|_| ())
}
/// Makes what a trusted pull's run wrote, once its overlays are
/// unmounted, the base the runs after it start from, as
/// [`Lease::promote`] does a push's, where no push runs to: each layer
/// only over the base it was written over, if that is still current
/// and was made at the branch's tip of generation `at`, the one the
/// pull merged with, or an older one. Says where the run saw the
/// layers made the base, and those that are not.
pub async fn advance(self, max: u64, at: u64) -> Result<(Vec<String>, Vec<String>)> {
self.promoting(max, Some(at)).await
}
/// [`Lease::promote`], or, given the tip a pull merged with, `at`,
/// [`Lease::advance`].
async fn promoting(self, max: u64, at: Option<u64>) -> Result<(Vec<String>, Vec<String>)> {
let (caches, layers) = (self.caches.clone(), self.layers.clone());
let building = caches.clone();
let made = tokio::task::spawn_blocking(move || -> Result<Vec<_>> {
let caches = building;
let mut made = Vec::new();
for layer in layers {
if std::fs::read_dir(&layer.upper)?.next().is_none() {
continue;
}
let generation = {
let mut bases = caches.bases.lock().unwrap();
bases.next += 1;
bases.next - 1
};
let base = caches.base(&layer.name, generation);
let making = base.with_extension("new");
clone(&layer.lower, &making)?;
apply(&layer.upper, &making)?;
let fits = size(&making)? <= max;
std::fs::rename(&making, &base)?;
made.push((layer, generation, fits));
}
Ok(made)
});
let made = made.await??;
let (mut kept, mut moved) = (Vec::new(), Vec::new());
for (layer, generation, fits) in made {
let mut bases = caches.bases.lock().unwrap();
let name = layer.name.clone();
bases
.kept
.entry(name.clone())
.or_default()
.insert(generation);
// Checked as it is made current, so of two pulls' runs over
// one base, the first to end makes the next, and the other's
// goes.
let ahead = at.is_none_or(|at| bases.ahead(&name, layer.generation, at));
if !ahead {
moved.push(layer.path.clone());
} else if fits {
bases.current.insert(name.clone(), generation);
kept.push(layer.path.clone());
if let Some(at) = at {
bases.marks.insert(name.clone(), (generation, at));
caches.mark(&bases.marks)?;
}
} else if bases.current.get(&name) == Some(&layer.generation) {
bases.current.remove(&name);
}
drop(bases);
caches.sweep(&name);
}
Ok((kept, moved))
}
}
impl Drop for Lease {
fn drop(&mut self) {
let _ = remove(&self.dir);
let mut bases = self.caches.bases.lock().unwrap();
for layer in &self.layers {
let key = (layer.name.clone(), layer.generation);
let n = bases.readers.entry(key.clone()).or_default();
*n = n.saturating_sub(1);
if *n == 0 {
bases.readers.remove(&key);
}
}
drop(bases);
for layer in &self.layers {
self.caches.sweep(&layer.name);
}
}
}
/// The bytes free to the runner on the disk holding `path`.
pub fn free(path: &Path) -> io::Result<u64> {
let stat = rustix::fs::statvfs(path)?;
Ok(stat.f_bavail.saturating_mul(stat.f_frsize))
}
/// The seconds since the epoch.
fn now() -> u64 {
let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
since.unwrap_or_default().as_secs()
}
/// What the kept run's directory `dir` says of it, if it says.
fn read_kept(dir: &Path) -> Option<Kept> {
let text = std::fs::read(dir.join(KEPT)).ok()?;
serde_json::from_slice(&text).ok()
}
/// Removes `path`, a directory and all in it or anything else, if it is
/// there.
/// What every cache's name starts with.
const PREFIX: &str = "knotless-cache-";
/// The name of a cache of the repository `repo`, `<owner did>/<key>`,
/// that `what` tells apart from its others: a hash of each, so that the
/// repository's caches can be told by name ([`Caches::drop_repo`]).
pub fn name(repo: &str, what: &str) -> String {
format!("{PREFIX}{}-{}", keys::ref_hash(repo), keys::ref_hash(what))
}
/// The hash of the repository the cache `name` is of, as [`name`] makes
/// one; none for a name it did not make.
fn repo_of(name: &str) -> Option<&str> {
let rest = name.strip_prefix(PREFIX)?;
Some(rest.split_once('-')?.0)
}
fn remove(path: &Path) -> io::Result<()> {
let removed = match std::fs::symlink_metadata(path) {
Ok(meta) if meta.is_dir() => std::fs::remove_dir_all(path),
Ok(_) => std::fs::remove_file(path),
Err(err) => Err(err),
};
match removed {
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
done => done,
}
}
/// Makes `to` as `from` is, its directories new and everything else hard
/// links to `from`'s.
fn clone(from: &Path, to: &Path) -> io::Result<()> {
std::fs::create_dir(to)?;
for entry in std::fs::read_dir(from)? {
let entry = entry?;
let dest = to.join(entry.file_name());
if entry.file_type()?.is_dir() {
clone(&entry.path(), &dest)?;
} else {
std::fs::hard_link(entry.path(), &dest)?;
}
}
like(from, to)
}
/// Gives the directory `to` the permissions and owner `from` has.
fn like(from: &Path, to: &Path) -> io::Result<()> {
let meta = std::fs::symlink_metadata(from)?;
std::fs::set_permissions(to, meta.permissions())?;
std::os::unix::fs::lchown(to, Some(meta.uid()), Some(meta.gid()))
}
/// Applies the upper layer `upper` to `base`, as overlayfs shows the two
/// together: each whiteout removes what it names, a directory not marked
/// opaque merges with the one below it, and anything else replaces what
/// is below it, moved rather than copied.
fn apply(upper: &Path, base: &Path) -> io::Result<()> {
for entry in std::fs::read_dir(upper)? {
let entry = entry?;
let (from, to) = (entry.path(), base.join(entry.file_name()));
let meta = entry.metadata()?;
let below = std::fs::symlink_metadata(&to).ok();
let merges =
meta.is_dir() && !opaque(&from) && below.as_ref().is_some_and(Metadata::is_dir);
if merges {
apply(&from, &to)?;
like(&from, &to)?;
continue;
}
remove(&to)?;
// A whiteout, once what it names is gone, has done its work.
let moved = (!whiteout(&meta)).then(|| std::fs::rename(&from, &to));
moved.transpose()?;
}
Ok(())
}
/// Whether `meta` is of a whiteout: a character device numbered 0, 0.
fn whiteout(meta: &Metadata) -> bool {
meta.file_type().is_char_device() && meta.rdev() == 0
}
/// Whether the directory `dir` hides the one below it.
fn opaque(dir: &Path) -> bool {
let mut value = [0u8; 1];
let read = rustix::fs::lgetxattr(dir, OPAQUE, &mut value[..]);
matches!(read, Ok(1)) && value[0] == b'y'
}
/// The bytes the files under `dir` take on disk, each file once however
/// many names it has, as `du` counts them: cargo links what it builds
/// under more than one.
fn size(dir: &Path) -> io::Result<u64> {
size_of(dir, &mut HashSet::new())
}
/// [`size`], not counting again the files `seen`, by inode.
fn size_of(dir: &Path, seen: &mut HashSet<u64>) -> io::Result<u64> {
let mut total = 0;
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let meta = entry.metadata()?;
if meta.nlink() > 1 && !meta.is_dir() && !seen.insert(meta.ino()) {
continue;
}
total += meta.blocks() * 512;
if meta.is_dir() {
total += size_of(&entry.path(), seen)?;
}
}
Ok(total)
}
#[cfg(test)]
mod tests {
use super::*;
fn write(path: &Path, text: &str) {
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(path, text).unwrap();
}
fn read(path: &Path) -> String {
std::fs::read_to_string(path).unwrap_or_default()
}
/// What `dir` holds, as `path=text` for each file, in order.
fn listing(dir: &Path) -> Vec<String> {
let mut all = Vec::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
for e in std::fs::read_dir(&d).unwrap() {
let p = e.unwrap().path();
if p.is_dir() {
stack.push(p);
} else {
let rel = p.strip_prefix(dir).unwrap().display().to_string();
all.push(format!("{rel}={}", read(&p)));
}
}
}
all.sort();
all
}
/// Makes a whiteout at `path`, as overlayfs does: false where that
/// takes privileges the tests do not have (macOS, not as root).
fn whiteout_at(path: &Path) -> bool {
let mut made = std::process::Command::new("mknod");
made.arg(path).args(["c", "0", "0"]);
made.stderr(std::process::Stdio::null())
.status()
.is_ok_and(|s| s.success())
}
fn names(caches: &[(&str, &str)]) -> Vec<(String, String)> {
caches
.iter()
.map(|(n, p)| (n.to_string(), p.to_string()))
.collect()
}
#[tokio::test]
async fn a_deleted_repositorys_caches_go() {
let root = tempfile::tempdir().unwrap();
let bases = root.path().join("bases");
// An earlier runner's, named for no repository, goes at the start.
let old = format!("{PREFIX}{}", keys::ref_hash("x"));
std::fs::create_dir_all(bases.join(&old).join("0")).unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
assert!(!bases.join(&old).exists());
let (gone, other) = (name("did:a/k", "t"), name("did:a/j", "t"));
assert_eq!(repo_of(&gone), Some(keys::ref_hash("did:a/k").as_str()));
assert_eq!(repo_of("t"), None);
// One run reads the deleted repository's cache as it goes.
let reading = caches.lease(&names(&[(gone.as_str(), "/a")])).unwrap();
let other_base = caches.lease(&names(&[(other.as_str(), "/a")])).unwrap();
let read = reading.layers[0].lower.clone();
caches.drop_repo("did:a/k").unwrap();
assert!(read.exists());
drop(reading);
assert!(!read.exists());
// Another repository's stays.
assert!(other_base.layers[0].lower.exists());
// One no run reads goes at once, its directory with it.
let unread = caches.lease(&names(&[(gone.as_str(), "/a")])).unwrap();
let base = unread.layers[0].lower.clone();
drop(unread);
caches.drop_repo("did:a/k").unwrap();
assert!(!base.exists());
assert!(!bases.join(&gone).exists());
}
#[tokio::test]
async fn runs_side_by_side_and_main_keeps() {
let root = tempfile::tempdir().unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
// Two runs at once, each over an empty base, with layers of its own.
let a = caches.lease(&wanted).unwrap();
let b = caches.lease(&wanted).unwrap();
assert_ne!(a.id, b.id);
assert_eq!(a.layers[0].lower, b.layers[0].lower);
assert_ne!(a.layers[0].upper, b.layers[0].upper);
let mounts = a.mounts();
assert_eq!(mounts[1].volume, format!("knotless-overlay-{}-1", a.id));
assert_eq!(mounts[1].path, "/usr/local/rustup");
assert_eq!(mounts[1].run, a.id);
// What a pull writes is gone with it, and its layers with it.
write(&b.layers[0].upper.join("pull"), "x");
let b_dir = b.dir.clone();
drop(b);
assert!(!b_dir.exists());
// A main run that wrote makes a new base; one it did not write to
// keeps its own.
write(&a.layers[0].upper.join("debug/dep"), "1");
write(&a.layers[0].upper.join("debug/.fingerprint/f"), "1");
let first = a.layers[0].lower.clone();
let rustup = a.layers[1].lower.clone();
a.promote(1 << 30).await.unwrap();
assert!(!first.exists(), "an old base no run reads goes");
let c = caches.lease(&wanted).unwrap();
assert_eq!(c.layers[1].lower, rustup);
let base = c.layers[0].lower.clone();
assert_eq!(listing(&base), ["debug/.fingerprint/f=1", "debug/dep=1"]);
assert!(listing(&root.path().join("runs")).is_empty());
// A pull reading that base keeps it while a later main run makes
// the next: overwritten, removed, and made again.
let pull = caches.lease(&wanted).unwrap();
let up = &c.layers[0].upper;
write(&up.join("debug/dep"), "2");
write(&up.join("new/file"), "n");
let whiteouts = whiteout_at(&up.join("debug/.fingerprint"));
c.promote(1 << 30).await.unwrap();
assert_eq!(
listing(&pull.layers[0].lower),
["debug/.fingerprint/f=1", "debug/dep=1"]
);
let next = caches.lease(&wanted).unwrap();
// Where no whiteout could be made, what it would remove stays.
let stays = ["debug/.fingerprint/f=1"]
.into_iter()
.filter(|_| !whiteouts);
let mut want: Vec<&str> = stays.collect();
want.extend(["debug/dep=2", "new/file=n"]);
assert_eq!(listing(&next.layers[0].lower), want);
// Once the pull ends, the base it read goes.
drop(pull);
assert!(!base.exists());
// A directory made again hides what was below it; a file replaces
// a directory, and a directory a file.
let up = &next.layers[0].upper;
write(&up.join("debug/fresh"), "f");
rustix::fs::lsetxattr(
up.join("debug"),
OPAQUE,
b"y",
rustix::fs::XattrFlags::empty(),
)
.unwrap();
write(&up.join("new"), "now a file");
write(&up.join("lone"), "l");
next.promote(1 << 30).await.unwrap();
let last = caches.lease(&wanted).unwrap();
assert_eq!(
listing(&last.layers[0].lower),
["debug/fresh=f", "lone=l", "new=now a file"]
);
let up = &last.layers[0].upper;
write(&up.join("lone/inside"), "i");
// Over the most a cache may hold, a base is dropped, and the next
// run starts afresh.
last.promote(1).await.unwrap();
let fresh = caches.lease(&wanted).unwrap();
assert!(listing(&fresh.layers[0].lower).is_empty());
let one = fresh.layers[1].lower.clone();
drop(fresh);
let bases = root.path().join("bases");
assert_eq!(std::fs::read_dir(bases.join("t")).unwrap().count(), 1);
// Opened again, the newest base of each is kept, and nothing else.
write(&bases.join("t/99.new/half"), "h");
std::fs::create_dir_all(bases.join("t/0")).unwrap();
std::fs::create_dir_all(bases.join("empty")).unwrap();
write(&root.path().join("runs/old/0/upper/x"), "x");
std::fs::create_dir_all(root.path().join("runs/alive")).unwrap();
let again = Caches::open(root.path(), &[], &[]).unwrap();
assert!(!bases.join("t/99.new").exists());
assert!(!bases.join("t/0").exists());
assert_eq!(
std::fs::read_dir(root.path().join("runs")).unwrap().count(),
0
);
let lease = again.lease(&wanted).unwrap();
assert_eq!(lease.layers[1].lower, one);
drop(lease);
// But for what runs whose overlays are still mounted read.
std::fs::create_dir_all(bases.join("t/0")).unwrap();
std::fs::create_dir_all(root.path().join("runs/alive")).unwrap();
Caches::open(root.path(), &["alive".into()], &[]).unwrap();
assert!(bases.join("t/0").exists());
assert!(root.path().join("runs/alive").exists());
// A root it cannot make is an error.
let file = root.path().join("file");
write(&file, "");
assert!(Caches::open(&file, &[], &[]).is_err());
assert!(remove(&root.path().join("gone")).is_ok());
// A file under two names counts once, as `du` counts it.
let linked = root.path().join("linked");
write(&linked.join("one"), &"x".repeat(1 << 16));
let once = size(&linked).unwrap();
std::fs::hard_link(linked.join("one"), linked.join("two")).unwrap();
assert_eq!(size(&linked).unwrap(), once);
}
#[tokio::test]
async fn pulls_advance_the_base_where_no_push_runs() {
let root = tempfile::tempdir().unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
let lower = |c: &Arc<Caches>| {
let lease = c.lease(&wanted).unwrap();
listing(&lease.layers[0].lower)
};
// Two pulls' runs over one base, merged with the branch's tip at
// generation 5: the first to end makes the next base, marked as of
// that tip; the other's, over a base that has moved on, goes.
let a = caches.lease(&wanted).unwrap();
let b = caches.lease(&wanted).unwrap();
write(&a.layers[0].upper.join("a"), "a");
write(&b.layers[0].upper.join("b"), "b");
let made = a.advance(1 << 30, 5).await.unwrap();
assert_eq!(made, (vec!["/work/target".to_string()], Vec::new()));
let made = b.advance(1 << 30, 5).await.unwrap();
assert_eq!(made, (Vec::new(), vec!["/work/target".to_string()]));
assert_eq!(lower(&caches), ["a=a"]);
assert_eq!(
std::fs::read_dir(root.path().join("bases/t"))
.unwrap()
.count(),
1
);
// One merged with an older tip never takes the base back to it,
// though it ran over that base; one merged with the same or a
// newer one goes on from it.
let old = caches.lease(&wanted).unwrap();
write(&old.layers[0].upper.join("old"), "o");
assert_eq!(
old.advance(1 << 30, 4).await.unwrap().0,
Vec::<String>::new()
);
let same = caches.lease(&wanted).unwrap();
write(&same.layers[0].upper.join("same"), "s");
assert_eq!(
same.advance(1 << 30, 5).await.unwrap().1,
Vec::<String>::new()
);
assert_eq!(lower(&caches), ["a=a", "same=s"]);
// The mark outlasts the runner: opened again, an older tip still
// does not take it back, and a newer one goes on.
let again = Caches::open(root.path(), &[], &[]).unwrap();
let old = again.lease(&wanted).unwrap();
write(&old.layers[0].upper.join("old"), "o");
assert!(old.advance(1 << 30, 4).await.unwrap().0.is_empty());
let newer = again.lease(&wanted).unwrap();
write(&newer.layers[0].upper.join("new"), "n");
newer.advance(1 << 30, 6).await.unwrap();
assert_eq!(lower(&again), ["a=a", "new=n", "same=s"]);
// A push's base is marked with no tip, so any pull goes on from
// it; and a mark of a base no longer current is let go on opening.
let push = again.lease(&wanted).unwrap();
write(&push.layers[0].upper.join("push"), "p");
push.promote(1 << 30).await.unwrap();
let again = Caches::open(root.path(), &[], &[]).unwrap();
assert!(again.bases.lock().unwrap().marks.is_empty());
let after = again.lease(&wanted).unwrap();
write(&after.layers[0].upper.join("after"), "f");
after.advance(1 << 30, 1).await.unwrap();
assert_eq!(
lower(&again),
["a=a", "after=f", "new=n", "push=p", "same=s"]
);
// Over the most a cache may hold, the base is dropped, as a push's.
let big = again.lease(&wanted).unwrap();
write(&big.layers[0].upper.join("big"), "b");
assert_eq!(big.advance(1, 9).await.unwrap(), (Vec::new(), Vec::new()));
assert!(lower(&again).is_empty());
}
/// The runs whose layers are kept, by the names their directories have.
fn kept_of(root: &Path) -> Vec<String> {
let all = std::fs::read_dir(root.join("kept")).unwrap();
let mut all: Vec<String> = all
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
all.sort();
all
}
#[tokio::test]
async fn pulls_keep_what_a_push_takes() {
let root = tempfile::tempdir().unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
let main = caches.lease(&wanted).unwrap();
write(&main.layers[0].upper.join("dep"), "1");
main.promote(1 << 30).await.unwrap();
// A pull's run that passed keeps the layers it wrote to, and only
// those, its own run's layers gone.
let pull = caches.lease(&wanted).unwrap();
let (over, dir) = (pull.layers[0].lower.clone(), pull.dir.clone());
write(&pull.layers[0].upper.join("dep"), "2");
write(&pull.layers[0].upper.join("new"), "n");
pull.keep("pull-run", 1 << 30).await.unwrap();
assert!(!dir.exists());
assert_eq!(kept_of(root.path()), [keys::ref_hash("pull-run")]);
// The push that takes its check takes them, into a new base over
// the one it was written over, as its own run's would go.
let taken = caches.take("pull-run");
assert_eq!(taken.kept(), ["/work/target"]);
assert!(taken.moved.is_empty());
taken.lease.promote(1 << 30).await.unwrap();
assert!(kept_of(root.path()).is_empty());
assert!(!over.exists(), "the base it was over goes");
let next = caches.lease(&wanted).unwrap();
assert_eq!(listing(&next.layers[0].lower), ["dep=2", "new=n"]);
drop(next);
// A run nothing was kept of has none to take.
let none = caches.take("never");
assert!(none.kept().is_empty() && none.moved.is_empty());
none.lease.promote(1 << 30).await.unwrap();
// One written over a base that has moved on since is not kept: its
// writes and whiteouts are of what that base held.
let stale = caches.lease(&wanted).unwrap();
write(&stale.layers[0].upper.join("stale"), "s");
write(&stale.layers[1].upper.join("toolchain"), "t");
let main = caches.lease(&wanted).unwrap();
write(&main.layers[0].upper.join("main"), "m");
stale.keep("stale-run", 1 << 30).await.unwrap();
main.promote(1 << 30).await.unwrap();
let taken = caches.take("stale-run");
assert_eq!(taken.kept(), ["/usr/local/rustup"]);
assert_eq!(taken.moved, ["/work/target"]);
taken.lease.promote(1 << 30).await.unwrap();
assert!(kept_of(root.path()).is_empty());
let next = caches.lease(&wanted).unwrap();
assert_eq!(listing(&next.layers[0].lower), ["dep=2", "main=m", "new=n"]);
assert_eq!(listing(&next.layers[1].lower), ["toolchain=t"]);
drop(next);
// A run that wrote nothing keeps nothing, and nor does one over
// the budget.
let bare = caches.lease(&wanted).unwrap();
bare.keep("bare", 1 << 30).await.unwrap();
let big = caches.lease(&wanted).unwrap();
write(&big.layers[0].upper.join("big"), "b");
big.keep("big", 1).await.unwrap();
assert!(kept_of(root.path()).is_empty());
// Kept, they are bounded: in number, the oldest going first...
let t0 = 1_000_000_000;
let keep = |run: &str, at: u64, budget: u64| {
let lease = caches.lease(&wanted).unwrap();
write(&lease.layers[0].upper.join(run), run);
caches.keeping(lease, run, budget, at).unwrap();
};
for n in 0..=KEPT_MAX as u64 {
keep(&format!("k{n}"), t0 + n, 1 << 30);
}
let all = kept_of(root.path());
assert_eq!(all.len(), KEPT_MAX);
assert!(!all.contains(&keys::ref_hash("k0")));
// ...in age...
let k1 = keys::ref_hash("k1");
let late = t0 + 2 + KEPT_AGE;
caches.forgetting(&[], 1 << 30, late).unwrap();
assert!(!kept_of(root.path()).contains(&k1));
assert_eq!(kept_of(root.path()).len(), KEPT_MAX - 1);
// ...and in bytes, the newest kept.
let newest = format!("k{KEPT_MAX}");
let one = read_kept(&caches.kept_dir(&newest)).unwrap().size;
caches.forgetting(&[], one, late).unwrap();
assert_eq!(kept_of(root.path()), [keys::ref_hash(&newest)]);
// A run kept again replaces what it kept; one forgotten goes.
keep(&newest, late, 1 << 30);
assert_eq!(kept_of(root.path()), [keys::ref_hash(&newest)]);
caches.forgetting(&[newest], 1 << 30, late).unwrap();
assert!(kept_of(root.path()).is_empty());
// Opened again, it keeps the layers of the runs it is told a push
// may yet take, and nothing else.
keep("wanted", late, 1 << 30);
keep("unwanted", late, 1 << 30);
std::fs::create_dir_all(caches.kept_dir("broken")).unwrap();
let half = caches.kept_dir("half").with_extension("new");
std::fs::create_dir_all(half).unwrap();
let named = ["wanted".into(), "broken".into()];
Caches::open(root.path(), &[], &named).unwrap();
assert_eq!(kept_of(root.path()), [keys::ref_hash("wanted")]);
// A kept run's directory that says nothing of it counts for none.
write(&root.path().join("kept/junk/0/upper/x"), "x");
caches.forgetting(&[], 1 << 30, late).unwrap();
assert_eq!(kept_of(root.path()), [keys::ref_hash("wanted")]);
}
#[tokio::test]
async fn short_of_disk_it_sheds_what_no_run_reads() {
let root = tempfile::tempdir().unwrap();
// The disk as statvfs tells it, of somewhere real.
assert!(free(root.path()).unwrap() > 0);
assert!(free(&root.path().join("gone")).is_err());
let space: Space = Box::new(|_| Ok(7));
let caches = Caches::open_with(root.path(), &[], &[], space).unwrap();
assert_eq!(caches.free().unwrap(), 7);
let wanted = names(&[("read", "/a"), ("unread", "/b")]);
// Bases for both caches, and a pull's run kept.
let main = caches.lease(&wanted).unwrap();
write(&main.layers[0].upper.join("x"), "x");
write(&main.layers[1].upper.join("y"), "y");
main.promote(1 << 30).await.unwrap();
let pull = caches.lease(&wanted).unwrap();
write(&pull.layers[0].upper.join("z"), "z");
pull.keep("pull-run", 1 << 30).await.unwrap();
assert!(caches.kept_dir("pull-run").exists());
// A run reads one cache's base as the disk runs short.
let reading = caches.lease(&names(&[("read", "/a")])).unwrap();
let (read, unread) = {
let all = caches.lease(&wanted).unwrap();
(all.layers[0].lower.clone(), all.layers[1].lower.clone())
};
assert_eq!(caches.shed().unwrap(), (1, 1));
assert!(!caches.kept_dir("pull-run").exists());
assert!(!unread.exists());
assert!(read.exists());
// The next run of the one shed starts afresh; the one read goes
// on from its base.
let next = caches.lease(&wanted).unwrap();
assert_eq!(next.layers[0].lower, read);
assert!(listing(&next.layers[1].lower).is_empty());
drop((next, reading));
// Nothing left to shed but the bases no run reads now.
assert_eq!(caches.shed().unwrap(), (2, 0));
}
}