~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

knotless-run: a feature branch's pulls start from its own warm caches

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:run/branch-caches
.changeset/branch-caches.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: patch
---
# A feature branch's pulls start from its own warm caches
A branch whose pushes and pulls both run a workflow, as `on.push.refs` and `on.pull.base` both naming `refs/heads/feature/*` do, keeps that workflow's caches for itself: each passing push to it leaves what the pulls into it start from, and a pull landed there at the tree it was checked at is taken as the push's run, caches and all. A branch with none yet starts from the default branch's, and its caches go when it is deleted.
crates/knotless-run/src/caches.rs+259−36
View file
@@ -5,11 +5,19 @@
//!
//! 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.
//! Only a run of a push 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 its base branch's pushes last left.
//!
//! A cache is the default branch's, or, of a workflow whose pushes run on
//! another branch, that branch's own ([`on_branch`]): what pushes there
//! leave, which the pulls into it start from. A branch's cache with no
//! base yet reads the default branch's instead, so a new branch starts
//! as warm as the default one, and its first push that passes makes a
//! base of its own from that one and what it wrote. A branch's caches go
//! once the branch does ([`Caches::drop_branches`]).
//!
//! 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
@@ -45,7 +53,9 @@
//! 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
//! Under the root: `bases/<cache>/<generation>`, a branch's caches named
//! as the default branch's with a hash of the branch after them,
//! `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`].
@@ -106,12 +116,15 @@ struct Kept {
}
/// A layer kept: its cache, where the run saw it, and the generation of
/// the base it was written over.
/// the base it was written over, of the cache `read` when that was the
/// default branch's, for a branch with no base of its own yet.
#[derive(Serialize, Deserialize)]
struct KeptLayer {
name: String,
path: String,
generation: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
read: Option<String>,
}
/// A kept run's layers as a push takes them: a lease over the bases they
@@ -142,13 +155,28 @@ struct Bases {
}
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 {
/// The base a run of `name` starts from, as the cache it is of and its
/// generation: `name`'s current one, or, for a branch's cache with
/// none, the default branch's.
fn start(&self, name: &str) -> Option<(String, u64)> {
let own = self.current.get(name).map(|g| (name.to_string(), *g));
let default = || {
let default = fallback(name)?;
Some((default.to_string(), *self.current.get(default)?))
};
own.or_else(default)
}
/// Whether a layer of `name` written over the base `generation` of
/// `read` by a pull merged with the branch's tip at generation `at`
/// may make the next base: that base is still the one runs of `name`
/// start from, and was made at that tip or an older one, or by a
/// push, which marks none.
fn ahead(&self, name: &str, read: &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)
let start = self.start(name);
let over = start.as_ref().map(|(r, g)| (r.as_str(), *g));
over == Some((read, generation)) && mark.is_none_or(|(_, tip)| *tip <= at)
}
/// The generations of `name` no run reads and none will start from, no
@@ -175,6 +203,9 @@ pub struct Layer {
pub name: String,
/// Where the run's container sees it.
pub path: String,
/// The cache whose base it reads: `name`, or the default branch's for
/// a branch's cache with no base of its own yet.
read: String,
generation: u64,
pub lower: PathBuf,
pub upper: PathBuf,
@@ -282,10 +313,27 @@ impl Caches {
/// 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<()> {
self.dropping(repo, |_| true)
}
/// Drops the caches of the branches of the repository `repo` that are
/// not among `live`, its branches now, in full, as [`Caches::drop_repo`]
/// drops a deleted repository's.
pub fn drop_branches(&self, repo: &str, live: &[String]) -> Result<()> {
let live: HashSet<String> = live.iter().map(|b| keys::ref_hash(b)).collect();
self.dropping(repo, |name| {
branch_of(name).is_some_and(|b| !live.contains(b))
})
}
/// Drops the caches of `repo` whose names `of` says.
fn dropping(&self, repo: &str, of: impl Fn(&str) -> bool) -> 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));
let names = names
.into_iter()
.filter(|n| repo_of(n) == Some(&hash) && of(n));
for name in names.collect::<Vec<_>>() {
bases.current.remove(&name);
bases.marks.remove(&name);
@@ -358,10 +406,12 @@ impl Caches {
let to = making.join(layers.len().to_string());
std::fs::create_dir(&to)?;
std::fs::rename(&layer.upper, to.join("upper"))?;
let read = (layer.read != layer.name).then(|| layer.read.clone());
layers.push(KeptLayer {
name: layer.name.clone(),
path: layer.path.clone(),
generation: layer.generation,
read,
});
}
let size = size(&making)?;
@@ -431,17 +481,20 @@ impl Caches {
name,
path,
generation,
read,
} = layer;
if bases.current.get(&name) != Some(&generation) {
let read = read.unwrap_or_else(|| name.clone());
if bases.start(&name) != Some((read.clone(), generation)) {
moved.push(path);
continue;
}
*bases.readers.entry((name.clone(), generation)).or_default() += 1;
*bases.readers.entry((read.clone(), generation)).or_default() += 1;
let at = dir.join(i.to_string());
layers.push(Layer {
lower: self.base(&name, generation),
lower: self.base(&read, generation),
name,
path,
read,
generation,
upper: at.join("upper"),
work: at.join("work"),
@@ -486,13 +539,14 @@ impl Caches {
layers: Vec::new(),
};
for (i, (name, path)) in caches.iter().enumerate() {
let generation = self.read(name)?;
let (read, generation) = self.read(name)?;
let layer = lease.dir.join(i.to_string());
lease.layers.push(Layer {
name: name.clone(),
path: path.clone(),
lower: self.base(&read, generation),
read,
generation,
lower: self.base(name, generation),
upper: layer.join("upper"),
work: layer.join("work"),
});
@@ -502,26 +556,24 @@ impl Caches {
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> {
/// The base a run of `name` starts from, as [`Bases::start`] says,
/// counted as read until it ends: an empty one of `name`'s own where
/// there is none.
fn read(&self, name: &str) -> Result<(String, u64)> {
let mut bases = self.bases.lock().unwrap();
let generation = match bases.current.get(name) {
Some(g) => *g,
let start = match bases.start(name) {
Some(start) => start,
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
(name.to_string(), g)
}
};
*bases
.readers
.entry((name.to_string(), generation))
.or_default() += 1;
Ok(generation)
*bases.readers.entry(start.clone()).or_default() += 1;
Ok(start)
}
/// Removes the bases of `name` that are no longer used.
@@ -597,6 +649,8 @@ impl Lease {
};
let base = caches.base(&layer.name, generation);
let making = base.with_extension("new");
// A branch's first base is beside none of its own.
std::fs::create_dir_all(caches.root.join("bases").join(&layer.name))?;
clone(&layer.lower, &making)?;
apply(&layer.upper, &making)?;
let fits = size(&making)? <= max;
@@ -618,7 +672,7 @@ impl Lease {
// 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));
let ahead = at.is_none_or(|at| bases.ahead(&name, &layer.read, layer.generation, at));
if !ahead {
moved.push(layer.path.clone());
} else if fits {
@@ -628,11 +682,12 @@ impl Lease {
bases.marks.insert(name.clone(), (generation, at));
caches.mark(&bases.marks)?;
}
} else if bases.current.get(&name) == Some(&layer.generation) {
} else if layer.read == name && bases.current.get(&name) == Some(&layer.generation) {
bases.current.remove(&name);
}
drop(bases);
caches.sweep(&name);
caches.sweep(&layer.read);
}
Ok((kept, moved))
}
@@ -643,7 +698,7 @@ impl Drop for Lease {
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 key = (layer.read.clone(), layer.generation);
let n = bases.readers.entry(key.clone()).or_default();
*n = n.saturating_sub(1);
if *n == 0 {
@@ -652,7 +707,7 @@ impl Drop for Lease {
}
drop(bases);
for layer in &self.layers {
self.caches.sweep(&layer.name);
self.caches.sweep(&layer.read);
}
}
}
@@ -675,8 +730,6 @@ fn read_kept(dir: &Path) -> Option<Kept> {
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-";
@@ -687,6 +740,13 @@ pub fn name(repo: &str, what: &str) -> String {
format!("{PREFIX}{}-{}", keys::ref_hash(repo), keys::ref_hash(what))
}
/// The name of the cache `name`, the default branch's, of the branch
/// `branch`, in full: a hash of the branch after it, so that the branch's
/// caches can be told by name ([`Caches::drop_branches`]).
pub fn on_branch(name: &str, branch: &str) -> String {
format!("{name}-{}", keys::ref_hash(branch))
}
/// 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> {
@@ -694,6 +754,23 @@ fn repo_of(name: &str) -> Option<&str> {
Some(rest.split_once('-')?.0)
}
/// The hash of the branch the cache `name` is of, as [`on_branch`] makes
/// one; none for the default branch's.
fn branch_of(name: &str) -> Option<&str> {
let rest = name.strip_prefix(PREFIX)?;
let mut parts = rest.splitn(3, '-');
parts.nth(2)
}
/// The default branch's cache that the branch's cache `name` reads until
/// it has a base of its own; none for the default branch's.
fn fallback(name: &str) -> Option<&str> {
let branch = branch_of(name)?;
name.strip_suffix(branch)?.strip_suffix('-')
}
/// Removes `path`, a directory and all in it or anything else, if it is
/// there.
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),
@@ -1068,6 +1145,152 @@ mod tests {
assert!(lower(&again).is_empty());
}
#[tokio::test]
async fn a_branch_starts_from_the_default_and_keeps_its_own() {
let root = tempfile::tempdir().unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
let main = name("did:a/k", "t");
let (f, g) = ("refs/heads/feature/f", "refs/heads/feature/g");
let on = |branch: &str| names(&[(on_branch(&main, branch).as_str(), "/t")]);
let of = |wanted: &[(String, String)]| {
let lease = caches.lease(wanted).unwrap();
listing(&lease.layers[0].lower)
};
let push = caches.lease(&names(&[(main.as_str(), "/t")])).unwrap();
write(&push.layers[0].upper.join("m"), "1");
push.promote(1 << 30).await.unwrap();
// A branch with no base of its own reads the default branch's.
let first = caches.lease(&on(f)).unwrap();
let mains = caches.lease(&names(&[(main.as_str(), "/t")])).unwrap();
assert_eq!(first.layers[0].lower, mains.layers[0].lower);
drop(mains);
// Its push that passes makes one of its own from it, and leaves the
// default branch's as it was.
write(&first.layers[0].upper.join("f"), "1");
first.promote(1 << 30).await.unwrap();
assert_eq!(of(&on(f)), ["f=1", "m=1"]);
assert_eq!(of(&names(&[(main.as_str(), "/t")])), ["m=1"]);
// The default branch moving on leaves it, and a branch with none
// of its own yet starts from where the default is now.
let push = caches.lease(&names(&[(main.as_str(), "/t")])).unwrap();
write(&push.layers[0].upper.join("m"), "2");
push.promote(1 << 30).await.unwrap();
assert_eq!(of(&on(f)), ["f=1", "m=1"]);
assert_eq!(of(&on(g)), ["m=2"]);
// A pull into a branch with none, kept over the default's base,
// is taken by the push to the branch into a base of its own.
let pull = caches.lease(&on(g)).unwrap();
write(&pull.layers[0].upper.join("g"), "1");
pull.keep("g-pull", 1 << 30).await.unwrap();
let taken = caches.take("g-pull");
assert_eq!(taken.kept(), ["/t"]);
taken.lease.promote(1 << 30).await.unwrap();
assert_eq!(of(&on(g)), ["g=1", "m=2"]);
// But not once the branch has a base of its own meanwhile, nor the
// default's has moved on: either is a base the layer was not over.
let h = "refs/heads/feature/h";
let pull = caches.lease(&on(h)).unwrap();
write(&pull.layers[0].upper.join("h"), "1");
pull.keep("h-pull", 1 << 30).await.unwrap();
let push = caches.lease(&on(h)).unwrap();
write(&push.layers[0].upper.join("p"), "1");
push.promote(1 << 30).await.unwrap();
assert_eq!(caches.take("h-pull").moved, ["/t"]);
let i = "refs/heads/feature/i";
let pull = caches.lease(&on(i)).unwrap();
write(&pull.layers[0].upper.join("i"), "1");
pull.keep("i-pull", 1 << 30).await.unwrap();
let push = caches.lease(&names(&[(main.as_str(), "/t")])).unwrap();
write(&push.layers[0].upper.join("m"), "3");
push.promote(1 << 30).await.unwrap();
assert_eq!(caches.take("i-pull").moved, ["/t"]);
// A run of a branch reading the default's base keeps it while the
// default moves on, and it goes once that run ends.
let reading = caches.lease(&on(i)).unwrap();
let read = reading.layers[0].lower.clone();
let push = caches.lease(&names(&[(main.as_str(), "/t")])).unwrap();
write(&push.layers[0].upper.join("m"), "4");
push.promote(1 << 30).await.unwrap();
assert_eq!(listing(&read), ["m=3"]);
drop(reading);
assert!(!read.exists());
// Where pushes run elsewhere, a pull into a branch with none makes
// its first over the default's, which a pull over a base that has
// moved on since does not.
let j = "refs/heads/feature/j";
let (a, b) = (caches.lease(&on(j)).unwrap(), caches.lease(&on(j)).unwrap());
write(&a.layers[0].upper.join("a"), "a");
write(&b.layers[0].upper.join("b"), "b");
assert_eq!(a.advance(1 << 30, 1).await.unwrap().0, ["/t"]);
assert_eq!(b.advance(1 << 30, 1).await.unwrap().1, ["/t"]);
assert_eq!(of(&on(j)), ["a=a", "m=4"]);
// Opened again, each branch goes on from its own.
let again = Caches::open(root.path(), &[], &[]).unwrap();
let lease = again.lease(&on(f)).unwrap();
assert_eq!(listing(&lease.layers[0].lower), ["f=1", "m=1"]);
let lease = again.lease(&on("refs/heads/new")).unwrap();
assert_eq!(listing(&lease.layers[0].lower), ["m=4"]);
// A branch's cache whose default has no base either starts empty,
// of its own.
let lone = name("did:a/k", "lone");
let lease = again
.lease(&names(&[(on_branch(&lone, f).as_str(), "/t")]))
.unwrap();
assert!(listing(&lease.layers[0].lower).is_empty());
assert_eq!(lease.layers[0].read, on_branch(&lone, f));
}
#[tokio::test]
async fn a_deleted_branchs_caches_go() {
let root = tempfile::tempdir().unwrap();
let caches = Caches::open(root.path(), &[], &[]).unwrap();
let (main, other) = (name("did:a/k", "t"), name("did:a/j", "t"));
let (f, g) = ("refs/heads/feature/f", "refs/heads/feature/g");
assert_eq!(branch_of(&main), None);
assert_eq!(fallback(&main), None);
assert_eq!(fallback("t"), None);
let on_f = on_branch(&main, f);
assert_eq!(branch_of(&on_f), Some(keys::ref_hash(f).as_str()));
assert_eq!(fallback(&on_f), Some(main.as_str()));
// Bases of the default branch, of two branches, and of another
// repository's branch of the same name.
let wanted = |name: &str| names(&[(name, "/t")]);
for cache in [
main.clone(),
on_f.clone(),
on_branch(&main, g),
on_branch(&other, f),
] {
let lease = caches.lease(&wanted(&cache)).unwrap();
write(&lease.layers[0].upper.join("x"), "x");
lease.promote(1 << 30).await.unwrap();
}
let bases = root.path().join("bases");
// One run reads the deleted branch's base as it goes.
let reading = caches.lease(&wanted(&on_f)).unwrap();
let read = reading.layers[0].lower.clone();
let live = ["refs/heads/main".to_string(), g.to_string()];
caches.drop_branches("did:a/k", &live).unwrap();
assert!(read.exists());
drop(reading);
assert!(!read.exists());
// The branches still there keep theirs, and so does the default
// branch, and every other repository.
for cache in [&main, &on_branch(&main, g), &on_branch(&other, f)] {
assert_eq!(std::fs::read_dir(bases.join(cache)).unwrap().count(), 1);
}
// Its next run, as of a branch made again, starts from the
// default's.
let lease = caches.lease(&wanted(&on_f)).unwrap();
assert_eq!(lease.layers[0].read, main);
drop(lease);
// With no branch left, the default branch's stay all the same.
caches.drop_branches("did:a/k", &[]).unwrap();
assert!(!bases.join(on_branch(&main, g)).exists());
assert!(bases.join(&main).exists());
assert!(bases.join(on_branch(&other, f)).exists());
}
/// 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();
crates/knotless-run/src/lib.rs+185−43
View file
@@ -740,9 +740,8 @@ struct Due {
branch: String,
base: String,
base_sha: String,
/// The generation of `base` at `base_sha`, when it is the default
/// branch.
main_generation: Option<u64>,
/// The generation of `base` at `base_sha`.
base_generation: u64,
uri: String,
/// Whether its author and its head's owner may both push here.
trusted: bool,
@@ -761,6 +760,9 @@ struct Subject {
/// Whether it is a push to the repository's default branch, whose runs
/// go first.
main: bool,
/// The repository's default branch, in full, whose caches the others'
/// start from until they have their own.
default: String,
pull: Option<Pull>,
/// The tree of what is checked out, for a push to the default branch
/// and a trusted pull that merges: what runs that passed are found by.
@@ -788,9 +790,10 @@ struct Pull {
base: String,
/// That branch's tip, which the head is merged with.
base_commit: String,
/// For a pull into the default branch, its generation at that tip:
/// how many times it had moved, so which of two tips is the newer.
main_generation: Option<u64>,
/// That branch's generation at that tip: how many times it had moved,
/// so which of two tips is the newer. None for a pull's pre-flight,
/// which the merge queue runs.
base_generation: Option<u64>,
/// The commit the head merged into the base makes, or the paths that
/// conflict.
merged: Result<String, Vec<String>>,
@@ -844,16 +847,21 @@ impl Subject {
self.pull.is_some() || self.batch.is_some()
}
/// Whether runs that passed are found by its tree: a push to the
/// default branch, which may reuse them, or a trusted pull that merges,
/// whose runs that pass are kept for it.
/// Whether runs that passed are found by its tree: a push to a
/// branch, which may reuse those of the pulls into it, or a trusted
/// pull that merges, whose runs that pass are kept for it.
fn by_tree(&self) -> bool {
match &self.pull {
Some(pull) => pull.trusted && pull.merged.is_ok(),
None => self.main,
None => self.ref_name.starts_with("refs/heads/"),
}
}
/// The branch it is of, in full: a pull's base, or the ref.
fn branch(&self) -> &str {
self.pull.as_ref().map_or(&self.ref_name, |p| &p.base)
}
/// What its checks are keyed by: its pull, its branch's queue for a
/// batch, or its ref.
fn keyed(&self) -> String {
@@ -1228,6 +1236,9 @@ impl Runner {
let repo = snap
.reference(watch.remote.did())
.context("a snapshot with a key has a repository")?;
// A deleted branch's caches go with it.
let live: Vec<String> = snap.live_refs().into_iter().map(|(name, _)| name).collect();
self.caches.drop_branches(&watch.id(), &live)?;
watch.at("reading reruns");
self.reruns(watch).await?;
let moved = self.moved(watch, &snap)?;
@@ -1346,6 +1357,7 @@ impl Runner {
commit,
generation: moved.generation,
main: moved.name == main,
default: main.clone(),
pull: None,
tree: None,
changed,
@@ -1483,7 +1495,6 @@ impl Runner {
let runs = runs || asked || dropped;
let kept = last.filter(|_| lands && !head_moved && !runs);
if runs {
let main_generation = (base == main).then_some(base_generation);
due.push(Due {
remote: remote.clone(),
packs: theirs.packs.clone(),
@@ -1493,7 +1504,7 @@ impl Runner {
branch,
base,
base_sha,
main_generation,
base_generation,
uri: uri.clone(),
trusted,
later: lands,
@@ -1528,11 +1539,12 @@ impl Runner {
commit: d.sha,
generation: d.generation,
main: false,
default: main.clone(),
pull: Some(Pull {
uri: d.uri,
base: d.base,
base_commit: d.base_sha,
main_generation: d.main_generation,
base_generation: Some(d.base_generation),
merged,
trusted: d.trusted && strangers.is_empty(),
later: d.later,
@@ -2254,7 +2266,7 @@ impl Runner {
}
/// The pull runs that passed at `subject`'s tree, into its ref, for a
/// push to the default branch.
/// push.
fn passed_at(&self, watch: &Watch, subject: &Subject) -> Vec<Passed> {
let Some(tree) = subject.tree.as_ref().filter(|_| subject.pull.is_none()) else {
return Vec::new();
@@ -2270,9 +2282,11 @@ impl Runner {
/// `passed` at its tree did, as that run's check still says: filled in
/// from it if so, its log saying whose it is, and what that run wrote
/// to its caches taken as the push's own run's would be, where it can
/// be. Never one that asks for secrets, which the pull's run did not
/// get, nor one whose runs finish, which a push runs for what it does,
/// as a deploy.
/// be: a push that runs `wf`, as this one does, keeps what its run
/// writes ([`promotes`]), to the caches of the branch the pulls it
/// takes from ran into. Never one that asks for secrets, which the
/// pull's run did not get, nor one whose runs finish, which a push
/// runs for what it does, as a deploy.
async fn reused(
&self,
watch: &Watch,
@@ -2791,7 +2805,8 @@ impl Runner {
});
let outputs = dir.path().join("outputs");
std::fs::create_dir(&outputs)?;
let wanted = caches(&watch.id(), kept_apart(subject, wf), wf);
let branch = own_branch(subject, wf);
let wanted = caches(&watch.id(), kept_apart(subject, wf), wf, branch);
let lease = self.caches.lease(&wanted)?;
let caches = lease.mounts();
let trusts = self.opts.trust.contains(&watch.remote.did().to_string());
@@ -2840,15 +2855,15 @@ impl Runner {
status::SUCCESS
}
.into();
// What a push to the default branch that passed wrote is what the
// runs after it start from; a trusted pull's or batch's that passed
// is kept, for the push that takes its check to take, or, where no
// push runs, is what the runs after it start from, as a push's
// would be; any other run's goes with it. A batch, not landed yet,
// never writes there.
// What a push that passed wrote, to the default branch or to one
// whose caches are its own, is what the runs after it start from;
// a trusted pull's or batch's that passed is kept, for the push
// that takes its check to take, or, where no push runs, is what the
// runs after it start from, as a push's would be; any other run's
// goes with it. A batch, not landed yet, never writes there.
let passed = check.status == status::SUCCESS;
let mut warmed = String::new();
if passed && subject.main && subject.batch.is_none() {
if passed && promotes(subject, wf) {
lease.promote(self.opts.cache_max).await?;
} else if let Some(at) = warms(watch.pulls_only, subject, wf).filter(|_| passed) {
let (kept, moved) = lease.advance(self.opts.cache_max, at).await?;
@@ -3816,20 +3831,48 @@ async fn until_cancelled<T>(cancelled: &mut Cancelled, f: impl Future<Output = T
/// The caches `wf` keeps, as `(name, path)`: one for each path it names,
/// for the repository `repo`, which every run of it reads and runs of
/// pushes to the default branch alone write; or, `apart`, ones no run
/// writes, and so always empty.
fn caches(repo: &str, apart: bool, wf: &Workflow) -> Vec<(String, String)> {
/// pushes alone write: the default branch's, or `branch`'s own; or,
/// `apart`, ones no run writes, and so always empty.
fn caches(repo: &str, apart: bool, wf: &Workflow, branch: Option<&str>) -> Vec<(String, String)> {
let scope = if apart { "\napart" } else { "" };
wf.cache
.iter()
.filter_map(|path| {
let path = workflow::mount_path(path)?;
let what = format!("{}\n{path}{scope}", wf.name);
Some((caches::name(repo, &what), path))
let name = caches::name(repo, &what);
let name = match branch.filter(|_| !apart) {
Some(branch) => caches::on_branch(&name, branch),
None => name,
};
Some((name, path))
})
.collect()
}
/// The branch whose own caches `subject`'s run of `wf` reads: its branch
/// ([`Subject::branch`]) when that is not the default branch, and both a
/// push to it and a pull into it run `wf`, so that its pushes leave what
/// the pulls into it start from, as the default branch's do. `None` for
/// the default branch's caches, which every other branch reads: one only
/// pushed to, as a pull's head, is not worth a base of its own.
fn own_branch<'a>(subject: &'a Subject, wf: &Workflow) -> Option<&'a str> {
let branch = subject.branch();
let own = branch != subject.default && wf.runs_on_push(branch) && wf.runs_on_pull(branch);
own.then_some(branch)
}
/// Whether `subject`'s run of `wf`, once it passes, makes what it wrote
/// the base the runs after it start from: a push's, not a pull's or a
/// batch's, to the default branch or to a branch whose caches are its own
/// ([`own_branch`]). Not a push's to any other branch, which reads the
/// default branch's caches, or an asked-for run's at a ref no push runs
/// `wf` at.
fn promotes(subject: &Subject, wf: &Workflow) -> bool {
let push = subject.pull.is_none() && subject.batch.is_none();
push && (subject.main || own_branch(subject, wf).is_some())
}
/// Whether a push's run of `wf` may be taken from a pull's that passed at
/// the same tree: not if it asks for secrets, which the pull's run did not
/// get, nor if its runs finish, as a deploy's, which a push runs for what
@@ -3851,11 +3894,16 @@ fn keeps(subject: &Subject, wf: &Workflow) -> bool {
/// Whether `subject`'s run of `wf`, once it passes, makes what it wrote
/// the base the runs after it start from, as a push's run's would, on a
/// repository that is `pulls_only`, whose pushes run elsewhere and so never
/// take kept layers: a trusted pull's into the default branch, of a
/// workflow a push may take, as [`keeps`] says. Never an untrusted pull's.
/// If so, the default branch's generation at the tip it merged with.
/// take kept layers: a trusted pull's into the default branch, or into a
/// branch whose caches are its own ([`own_branch`]), of a workflow a push
/// may take, as [`keeps`] says. Never an untrusted pull's, nor one into a
/// branch that reads the default branch's caches, whose tips are not the
/// default branch's to put in order. If so, the base branch's generation
/// at the tip it merged with.
fn warms(pulls_only: bool, subject: &Subject, wf: &Workflow) -> Option<u64> {
let tip = subject.pull.as_ref().and_then(|p| p.main_generation);
let pull = subject.pull.as_ref()?;
let ours = pull.base == subject.default || own_branch(subject, wf).is_some();
let tip = pull.base_generation.filter(|_| ours);
tip.filter(|_| pulls_only && keeps(subject, wf))
}
@@ -4055,7 +4103,7 @@ mod tests {
uri: "at://p".into(),
base: "refs/heads/main".into(),
base_commit: "b".into(),
main_generation: Some(1),
base_generation: Some(1),
merged: Ok("m".into()),
strangers: Vec::new(),
trusted,
@@ -4067,6 +4115,7 @@ mod tests {
commit: "h".into(),
generation: 1,
main: false,
default: "refs/heads/main".into(),
pull,
tree: None,
changed: None,
@@ -4106,7 +4155,7 @@ mod tests {
uri: "at://p".into(),
base: "refs/heads/main".into(),
base_commit: "b".into(),
main_generation: Some(1),
base_generation: Some(1),
merged,
strangers: Vec::new(),
trusted,
@@ -4118,6 +4167,7 @@ mod tests {
commit: "h".into(),
generation: 1,
main,
default: "refs/heads/main".into(),
pull,
tree: None,
changed: None,
@@ -4172,13 +4222,105 @@ mod tests {
let untrusted = subject(Some(pull(false, merges())), false);
assert_eq!(warms(true, &untrusted, &plain), None);
assert_eq!(warms(true, &batch(true), &plain), None);
// Nor a pull into another branch, whose tips are not the default
// branch's to put in order.
// A pull into another branch whose pushes run the workflow warms
// that branch's own caches, its tips put in order as the default
// branch's are; one into a branch that reads the default branch's
// caches does not, nor a pre-flight, with no tip to order by.
let side = Pull {
main_generation: None,
base: "refs/heads/feature/x".into(),
..pull(true, merges())
};
assert_eq!(warms(true, &subject(Some(side), false), &plain), None);
let into_side = subject(Some(side.clone()), false);
let both = parse(
"on: {push: {refs: [refs/heads/*]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n",
);
assert_eq!(warms(true, &into_side, &both), Some(1));
assert_eq!(warms(true, &into_side, &plain), None);
let main_only = parse(
"on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n",
);
assert_eq!(warms(true, &into_side, &main_only), None);
let preflight = Pull {
base_generation: None,
..pull(true, merges())
};
assert_eq!(warms(true, &subject(Some(preflight), false), &plain), None);
}
#[test]
fn whose_branch_the_caches_are() {
let parse = |text: &str| workflow::parse("ci.yml", text.as_bytes()).unwrap();
let feature = parse(
"on: {push: {refs: [refs/heads/main, refs/heads/feature/*]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n",
);
let main_only = parse(
"on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n",
);
let push = |ref_name: &str| Subject {
ref_name: ref_name.into(),
commit: "h".into(),
generation: 1,
main: ref_name == "refs/heads/main",
default: "refs/heads/main".into(),
pull: None,
tree: None,
changed: None,
batch: None,
pushed: true,
asked: Asks::new(),
only: None,
};
let pull = |base: &str| Subject {
pull: Some(Pull {
uri: "at://p".into(),
base: base.into(),
base_commit: "b".into(),
base_generation: Some(1),
merged: Ok("m".into()),
strangers: Vec::new(),
trusted: true,
later: false,
others: Vec::new(),
}),
..push("refs/heads/topic")
};
let batch = |base: &str| Subject {
batch: Some(queue::Batch::new(base, "b", Vec::new(), true)),
main: true,
..push(base)
};
let f = "refs/heads/feature/x";
// A push to a branch its pushes run on, a pull into it and a
// batch on it read that branch's own caches; the default branch's
// runs, and those of a branch only asked for, read the default's.
assert_eq!(own_branch(&push(f), &feature), Some(f));
assert_eq!(own_branch(&pull(f), &feature), Some(f));
assert_eq!(own_branch(&batch(f), &feature), Some(f));
assert_eq!(own_branch(&push("refs/heads/main"), &feature), None);
assert_eq!(own_branch(&pull("refs/heads/main"), &feature), None);
assert_eq!(own_branch(&push(f), &main_only), None);
assert_eq!(own_branch(&pull(f), &main_only), None);
// Nor does a branch only pushed to, which no pull goes into.
let pushes = parse("cache: [target]\nsteps: []\n");
assert_eq!(own_branch(&push(f), &pushes), None);
assert!(!promotes(&push(f), &pushes));
// Only a push writes them: to the default branch, or one whose
// caches are its own.
assert!(promotes(&push(f), &feature));
assert!(promotes(&push("refs/heads/main"), &main_only));
assert!(!promotes(&push(f), &main_only));
assert!(!promotes(&pull(f), &feature));
assert!(!promotes(&batch(f), &feature));
assert!(!promotes(&batch("refs/heads/main"), &feature));
// A branch's caches are the default's named for the branch too; a
// cache read apart is never a branch's.
let main = caches("did:a/k", false, &feature, None);
let on_f = caches("did:a/k", false, &feature, Some(f));
assert_eq!(on_f[0].0, caches::on_branch(&main[0].0, f));
assert_eq!(on_f[0].1, main[0].1);
let apart = caches("did:a/k", true, &feature, Some(f));
assert_eq!(apart, caches("did:a/k", true, &feature, None));
assert_ne!(apart[0].0, main[0].0);
}
#[test]
@@ -4209,16 +4351,16 @@ mod tests {
fn caches_apart() {
let wf =
workflow::parse("ci.yml", b"cache: [target, /usr/local/cargo/]\nsteps: []\n").unwrap();
let all = caches("did:plc:a/k", false, &wf);
let all = caches("did:plc:a/k", false, &wf, None);
let paths: Vec<&str> = all.iter().map(|(_, p)| p.as_str()).collect();
assert_eq!(paths, ["/work/target", "/usr/local/cargo"]);
assert!(all[0].0.starts_with("knotless-cache-"));
assert_ne!(all[0].0, all[1].0);
// The same for every run, and apart for each repository.
assert_eq!(caches("did:plc:a/k", false, &wf), all);
assert_ne!(caches("did:plc:b/k", false, &wf)[0], all[0]);
assert_eq!(caches("did:plc:a/k", false, &wf, None), all);
assert_ne!(caches("did:plc:b/k", false, &wf, None)[0], all[0]);
// Kept apart, others, whose paths are the same.
let apart = caches("did:plc:a/k", true, &wf);
let apart = caches("did:plc:a/k", true, &wf, None);
assert_ne!(apart[0].0, all[0].0);
assert_eq!(apart[0].1, all[0].1);
}
crates/knotless-run/src/queue.rs+3−1
View file
@@ -667,13 +667,14 @@ impl Runner {
commit: head.to_string(),
generation,
main: false,
default: crate::main_ref(on.snap),
pull: Some(Pull {
uri: item.uri.clone(),
base: format!("refs/heads/{}", on.branch),
base_commit: tip.to_string(),
// Only a --pulls-only repository's pulls warm its caches,
// and its queue is landed elsewhere.
main_generation: None,
base_generation: None,
merged,
trusted,
later: false,
@@ -723,6 +724,7 @@ impl Runner {
commit: stack.tip().to_string(),
generation,
main: true,
default: crate::main_ref(on.snap),
pull: None,
tree: None,
changed,
docs/design/runner.md+45−7
View file
@@ -677,7 +677,8 @@ flowchart BT
u -->|removed when the run ends| gone[nothing kept]
```
Only a passing push to the default branch promotes, or, on a repository
Only a passing push promotes, to the default branch or to a branch with
caches of its own ([below](#a-branchs-own-caches)), or, on a repository
whose pushes run on another runner, a trusted pull's passing run
([below](#where-pushes-run-elsewhere)). Each non-empty upper
layer is applied over a hard-link clone of its base, as the base's next
@@ -713,6 +714,40 @@ warms the rest.
On start, the runner removes its leftover overlays and half-made bases,
and sweeps no old base while an old overlay is still mounted.
### A branch's own caches
Work kept off the default branch for a while, on a `feature/<name>`
branch with pulls into it, has workflows of its own, or builds that
drift from the default branch's. Read from the default branch's base,
every pull into it would start cold, or warm only for what the two
branches share. So a branch has caches of its own when, of a workflow,
both a push to it and a pull into it run it (`on.push.refs` and
`on.pull.base` both name it), and it is not the default branch. Then,
of that workflow:
- a passing push to the branch promotes, as one to the default branch
does, into the branch's own bases;
- pulls into it, and the merge queue's batches on it, start from those,
and a trusted pull's passing run is kept for the push to the branch
that [takes its pass](#reusing-a-pulls-pass-at-the-push), so a pull
landed by fast-forward warms the branch with no run again;
- under `--pulls-only`, a trusted pull into it promotes into the
branch's bases, ordered by the branch's own tips.
A branch with no base of its own yet reads the default branch's, so a
new branch starts as warm as the default one; its first passing push
makes its first base, over the default's and what it wrote. A layer kept
over the default's base is taken only while the branch still has none
and the default's has not moved on. Any other branch, one only pushed to
as a pull's head among them, reads the default branch's caches and
keeps nothing, as before. Requiring a pull trigger keeps every
pushed branch from growing a base under a workflow with no `on`.
A branch's caches go as the branch does: each pass over the repository
drops those of branches it no longer has, as a deleted repository's
are, a base a run still reads once that run ends. The disk floor below
counts them as any others.
### Room on the disk
The caches share their disk with whatever else the host runs. On
@@ -740,7 +775,8 @@ the disk.
A repository named with `--pulls-only` has its pushes run by another
runner, with caches of its own, so here no base would ever be written
and no kept layer taken: every pull would start cold. Instead a trusted
pull's passing run into the default branch, of a workflow that could be
pull's passing run into the default branch, or into a branch with
[caches of its own](#a-branchs-own-caches), of a workflow that could be
[reused](#reusing-a-pulls-pass-at-the-push), promotes its layers itself,
as a push's run would, and keeps none. Two rules keep the base from
going backwards:
@@ -749,13 +785,15 @@ going backwards:
still current, checked as the new base is made current. Of two pulls'
runs over one base, the first to pass makes the next one, and the
other's layers go.
- A base a pull made is marked with the default branch's generation at
- A base a pull made is marked with its base branch's generation at
the tip the pull merged with (`marks.json` under the caches). A run
merged with an older tip than the mark does not promote, so the base
never goes back to an older `main`. A base a push made has no mark.
A branch's first base, made over the default branch's, has nothing
of the branch's to be ordered against.
The pull's log says which of its paths it made the base and which it
did not. Pulls into another branch, untrusted pulls (an agent's among
did not. Pulls into a branch that reads the default branch's caches, untrusted pulls (an agent's among
them) and workflows that ask for secrets or are `superseded: finish`
never promote.
@@ -773,9 +811,9 @@ main's own checks there.
When a trusted pull merges, the merge commit often has exactly the tree
the pull was checked at, and running the same workflows again would test
nothing new. So a push to the default branch whose tree equals the tree
of a trusted pull's passing run into that branch, or a trusted batch's,
takes that run's result. The check is filed `success` with the earlier
nothing new. So a push to a branch whose tree equals the tree of a
trusted pull's passing run into that branch, or a trusted batch's, takes
that run's result. The check is filed `success` with the earlier
steps and outputs, and its log names the run it took, then copies that
run's log, so it still reads whole once the earlier check is pruned. The tree holds the
workflows, so they were the same workflows.
docs/reference/runner-internals.md+3−3
View file
@@ -27,10 +27,10 @@ Under `--caches`:
| Path | |
| --- | --- |
| `bases/<cache>/<generation>` | each generation of a cache's base; `<cache>` is `knotless-cache-<h1>-<h2>`, `<h1>` a hash of the repository and `<h2>` of the workflow, mount path and, for a cache read apart, `apart`; a deleted repository's go |
| `bases/<cache>/<generation>` | each generation of a cache's base; `<cache>` is `knotless-cache-<h1>-<h2>`, `<h1>` a hash of the repository and `<h2>` of the workflow, mount path and, for a cache read apart, `apart`; a branch's own caches are `knotless-cache-<h1>-<h2>-<h3>`, `<h3>` a hash of the branch's full ref, and read the default branch's `knotless-cache-<h1>-<h2>` until they have a base; a deleted repository's go, and a deleted branch's |
| `runs/<run>/<i>/{upper,work}` | a run's overlay layers; `<run>` is the runner's boot time in milliseconds and a counter |
| `kept/<hash of check key>/` | a passing trusted run's kept layers: `kept.json`, and `<i>/upper` per cache |
| `marks.json` | of each cache whose current base a trusted pull made under `--pulls-only`, that base's generation and the default branch's generation at the tip the pull merged with: `{"<cache>": [<base>, <tip>]}` |
| `kept/<hash of check key>/` | a passing trusted run's kept layers: `kept.json`, and `<i>/upper` per cache; each layer in `kept.json` names its cache, path and base generation, and `read`, the default branch's cache, when it was over that one's base |
| `marks.json` | of each cache whose current base a trusted pull made under `--pulls-only`, that base's generation and the base branch's generation at the tip the pull merged with: `{"<cache>": [<base>, <tip>]}` |
## Containers and volumes
AGENTS.md+4−2
View file
@@ -296,8 +296,10 @@ author and its head repository's owner are the owner or a collaborator;
anyone else's is not checked until the owner runs it with `knotless rerun
<repo> --pull <pull uri>`. Every run starts warm from the caches `main`'s
last passing run left, each over a copy-on-write layer of its own, so runs
of one workflow go side by side, and only `main`'s keep what they wrote:
a pull's writes go when it ends. (Anyone else's pull of a workflow that
of one workflow go side by side, and only pushes keep what they wrote:
a pull's writes go when it ends. A branch whose pushes and pulls both run
a workflow, as a feature branch's can, keeps caches of its own, from its
pushes, which the pulls into it start from (`docs/design/runner.md`). (Anyone else's pull of a workflow that
asks for secrets starts from empty caches instead.)
While the runner lands the queue, a pull not in it is checked again only
as its head moves, not as `main` does, and after the queue's batches: