chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! A runner watching a repository on the reference PDS through a stand-in
//! Jetstream: a push rings, the runner reads the refs back from the PDS,
//! runs the workflows at the new commit and files a check for each in its
//! own account.
#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::{Sandbox, net, wait_until};
use futures::{SinkExt, StreamExt};
use knotless::keys::{ref_hash, ref_rkey};
use knotless_run::check;
use knotless_run::executor::Executor;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
const KIND: &str = "network.bsky.jetstream.subscribeEvents#";
/// A Jetstream that sends what the test hands it, to whoever is connected.
struct Jetstream {
url: String,
frames: tokio::sync::mpsc::UnboundedSender<Message>,
}
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
fn jetstream(rt: &tokio::runtime::Runtime) -> Jetstream {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let url = format!("ws://{}", listener.local_addr().unwrap());
let (frames, rx) = tokio::sync::mpsc::unbounded_channel::<Message>();
let rx = Arc::new(Mutex::new(rx));
rt.spawn(async move {
loop {
let (tcp, _) = listener.accept().await.unwrap();
let rx = rx.clone();
tokio::spawn(async move {
// Jetstream answers only a client that offers its framing.
let offered = |req: &Request, mut res: Response| {
let protocol = req.headers().get("sec-websocket-protocol").cloned();
assert_eq!(protocol.as_ref().unwrap(), "xrpc.v1.json");
res.headers_mut()
.insert("sec-websocket-protocol", protocol.unwrap());
Ok(res)
};
let Ok(mut ws) = tokio_tungstenite::accept_hdr_async(tcp, offered).await else {
return;
};
// Reading answers the client's pings, as Jetstream does: one
// left unanswered gives the connection up, and takes the
// next frame with it.
let mut rx = rx.lock().await;
loop {
tokio::select! {
frame = rx.recv() => {
let Some(frame) = frame else { return };
if ws.send(frame).await.is_err() {
return;
}
}
read = ws.next() => {
if !matches!(read, Some(Ok(_))) {
return;
}
}
}
}
});
}
});
Jetstream { url, frames }
}
impl Jetstream {
fn ring(&self, did: &str, rkey: &str) {
self.send(did, "dev.knotless.ref", rkey, None);
}
fn send(&self, did: &str, collection: &str, rkey: &str, record: Option<serde_json::Value>) {
let mut payload = serde_json::json!({
"$type": format!("{KIND}commit"), "seq": 1, "did": did, "time": "2026-10-21T00:00:00.000001Z",
"rev": "3l3qo2vutsw2b", "operation": "update", "collection": collection, "rkey": rkey,
});
if let Some(record) = record {
payload["record"] = record;
}
let frame = serde_json::json!({"$type": "message", "payload": payload});
self.frames
.send(Message::Text(frame.to_string().into()))
.unwrap();
}
}
const CI: &str = r#"
on:
push:
refs: [refs/heads/main]
secrets: [TOKEN, UNGRANTED, MAIN_ONLY, TAGS_ONLY]
env:
GREETING: hello
steps:
- name: look around
run: |
set -e
echo "$GREETING from $KNOTLESS_RUN_REF at $KNOTLESS_RUN_SHA of $KNOTLESS_RUN_REPO"
echo "token=$TOKEN ungranted=${UNGRANTED:-none} other=${OTHER:-none} password=${KNOTLESS_PASSWORD:-none}"
echo "main=$MAIN_ONLY tags=${TAGS_ONLY:-none}"
test "$(cat f)" = "$(git rev-parse HEAD~1)"
- run: test -f f
"#;
/// Fails, leaving pictures and what is not kept of them as outputs.
const FAIL: &str = r#"
steps:
- name: break
run: |
out=$KNOTLESS_RUN_OUTPUTS
echo iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg== | base64 -d > "$out/page 00.png"
for i in $(seq -w 1 32); do cp "$out/page 00.png" "$out/page $i.png"; done
printf x >> "$out/page 31.png"
head -c 1000001 /dev/zero > "$out/big.png"
echo notes > "$out/notes.txt"
mkdir "$out/nested"
echo broken; exit 4
- name: after
run: echo never
"#;
const TAGS: &str = "on: {push: {refs: [refs/tags/v*]}}\nsteps: [{run: 'true'}]\n";
fn runner_opts(runner: &Sandbox, owner: &Sandbox, js: &Jetstream) -> knotless_run::Options {
let secrets = runner.path("secrets.json");
let label = format!("{}/proj", owner.account.handle);
std::fs::write(
&secrets,
serde_json::json!({ label.clone(): {
"TOKEN": "s3cret",
"OTHER": "nope",
"MAIN_ONLY": {"value": "m4inonly", "refs": ["refs/heads/main"]},
"TAGS_ONLY": {"value": "t4gsonly", "refs": ["refs/tags/*"]},
} })
.to_string(),
)
.unwrap();
knotless_run::Options {
jetstream: js.url.clone(),
repos: vec![label],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: Some(secrets),
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 4,
pull_jobs: 4,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: Some(runner.path("paused.json")),
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
}
}
fn check_of(runner: &Sandbox, owner: &Sandbox, key: &str, workflow: &str) -> serde_json::Value {
check_on(runner, owner, key, "refs/heads/main", workflow)
}
fn blob(did: &str, cid: &str) -> String {
reqwest::blocking::get(format!(
"{}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}",
net().pds
))
.unwrap()
.text()
.unwrap()
}
#[test]
fn a_push_runs_its_workflows() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("proj", 1);
std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
for (file, text) in [
("ci.yml", CI),
("fail.yaml", FAIL),
("tags.yml", TAGS),
("bad.yml", "stpes: []\n"),
("notes.txt", "not a workflow"),
] {
std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
}
// Those naming runners run only on those, by handle or DID.
let did = runner.account.did.clone();
for (file, runners) in [
("by-handle.yml", format!("[nobody.invalid, {handle}]")),
("by-did.yml", format!("['{did}']")),
(
"elsewhere.yml",
"['did:plc:elsewhere', nobody.invalid]".to_string(),
),
] {
let text = format!(
"on: {{push: {{refs: [refs/heads/main]}}}}\nrunners: {runners}\nsteps: [{{run: 'true'}}]\n"
);
std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
}
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let ref_key = ref_rkey(&key, "refs/heads/main");
let rt = tokio::runtime::Runtime::new().unwrap();
let js = jetstream(&rt);
let opts = knotless_run::Options {
allowance: Some(1),
unmetered: vec![owner.account.did.clone()],
trust: Vec::new(),
..runner_opts(&runner, &owner, &js)
};
rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));
// Watching starts where the refs are: nothing runs for what was there.
wait_until(|| runner.path("state.json").exists());
// It made a key, kept beside its state, and published its public half
// for owners to seal secrets to.
let keys = knotless::sealed::RunnerKeys::open_or_make(&runner.path("secrets-key.json"));
let sealing = keys.unwrap().public().to_did_key();
let published = runner.record(knotless::sealed::KEY_NSID, knotless::sealed::KEY_RKEY);
assert_eq!(published["key"], sealing.as_str(), "{published}");
js.ring(&owner.account.did, &ref_key);
std::thread::sleep(Duration::from_millis(500));
assert!(check_of(&runner, &owner, &key, "ci").is_null());
// A push, and the ring that follows it, run the workflows at its commit.
let sha = owner.commit_file(&work, "f", "");
let sha = {
std::fs::write(work.join("f"), &sha).unwrap();
owner.git(&work, &["commit", "-q", "-am", "f names its parent"]);
owner.git(&work, &["rev-parse", "HEAD"])
};
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
// Side by side, each finishing in its own time.
wait_until(|| {
["ci", "fail"]
.iter()
.all(|wf| check_of(&runner, &owner, &key, wf)["finishedAt"].is_string())
});
let ci = check_of(&runner, &owner, &key, "ci");
assert_eq!(ci["status"], "success", "{ci}");
assert_eq!(ci["commit"], sha.as_str());
wait_until(|| {
["by-handle", "by-did"]
.iter()
.all(|wf| check_of(&runner, &owner, &key, wf)["status"] == "success")
});
assert!(check_of(&runner, &owner, &key, "elsewhere").is_null());
// The owner's minutes are filed in the runner's account, under its DID.
let minutes = runner.record("dev.knotless.minutes", &owner.account.did);
assert_eq!(minutes["owner"], owner.account.did.as_str(), "{minutes}");
assert!(minutes["minutes"].is_u64(), "{minutes}");
assert!(minutes["updatedAt"].is_string(), "{minutes}");
assert_eq!(ci["ref"], "refs/heads/main");
assert_eq!(ci["repo"]["key"], key.as_str());
assert_eq!(ci["steps"][0]["name"], "look around");
assert_eq!(ci["steps"][1]["name"], "step 2");
assert_eq!(ci["steps"][1]["exitCode"], 0);
let log = blob(
&runner.account.did,
ci["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.contains(&format!(
"hello from refs/heads/main at {sha} of {}/proj",
owner.account.handle
)),
"{log}"
);
// Only what the workflow asks for and the runner grants, and never the
// runner's own credentials; a granted secret is masked in the log.
assert!(
log.contains("token=*** ungranted=none other=none password=none"),
"{log}"
);
assert!(!log.contains("s3cret"), "{log}");
// A secret granted to some refs alone reaches only their pushes.
assert!(log.contains("main=*** tags=none"), "{log}");
// The log says which it asked for and was not given.
let label = format!("{}/proj", owner.account.handle);
for name in ["UNGRANTED", "TAGS_ONLY"] {
let said = format!("==> {name} is not granted to {label} at refs/heads/main\n");
assert!(log.contains(&said), "{log}");
}
assert!(!log.contains("TOKEN is not granted"), "{log}");
let fail = check_of(&runner, &owner, &key, "fail");
assert_eq!(fail["status"], "failure");
assert_eq!(fail["steps"][0]["exitCode"], 4);
assert_eq!(fail["steps"][1]["status"], "skipped");
// What it left to be looked at, failing: the first 32 pictures, each
// named for its file; the rest of what it left, in its log, and why.
let outputs = fail["outputs"].as_array().unwrap();
assert_eq!(outputs.len(), 32);
assert_eq!(outputs[0]["$type"], "dev.knotless.check#image");
assert_eq!(outputs[0]["alt"], "page 00");
assert_eq!(outputs[31]["alt"], "page 31");
assert_eq!(outputs[0]["image"]["mimeType"], "image/png");
// Files alike share one blob, uploaded once; each keeps its place.
let cid = |i: usize| outputs[i]["image"]["ref"]["$link"].as_str().unwrap();
assert_eq!(cid(0), cid(30));
assert_ne!(cid(30), cid(31));
let log = blob(
&runner.account.did,
fail["log"]["ref"]["$link"].as_str().unwrap(),
);
for left in [
"big.png: not kept, being over 1000000 bytes",
"nested: not kept, as only images are outputs",
"notes.txt: not kept, as only images are outputs",
"page 32.png: not kept, past 32 outputs",
] {
assert!(log.contains(&format!("==> {left}\n")), "{log}");
}
let bad = check_of(&runner, &owner, &key, "bad");
assert_eq!(bad["status"], "error");
assert!(
bad["error"]
.as_str()
.unwrap()
.contains("unknown field `stpes`"),
"{bad}"
);
assert!(check_of(&runner, &owner, &key, "tags").is_null());
// Named as a runner, what it wrote is what a reader finds.
let me = runner.account.handle.clone();
assert_eq!(
owner
.cli(&["runners", &owner.url("proj"), "add", &me], b"")
.0,
0
);
let (code, out) = owner.cli(
&["checks", &owner.url("proj"), "--ref", "refs/heads/main"],
b"",
);
assert_eq!(code, 0, "{out}");
let lines: Vec<&str> = out.lines().collect();
assert_eq!(lines.len(), 5, "{out}");
// And the minutes it counts for the owner, with no allowance for an
// owner unmetered.
let (code, out) = owner.cli(&["minutes", &owner.url("proj")], b"");
assert_eq!(code, 0, "{out}");
assert!(out.starts_with(&format!("{me}\t")), "{out}");
assert!(out.ends_with(" min in 30 days, no allowance"), "{out}");
// Its pictures, saved to open, each named for its workflow and what it
// shows: after the listing, or beside the JSON.
let saved = owner.path("pictures");
let dir = saved.to_str().unwrap();
let (code, out) = owner.cli(
&[
"checks",
&owner.url("proj"),
"--workflow",
"fail",
"--images",
dir,
],
b"",
);
assert_eq!(code, 0, "{out}");
let shown: Vec<&str> = out.lines().filter(|l| l.starts_with("image\t")).collect();
assert_eq!(shown.len(), 32, "{out}");
assert_eq!(shown[0], format!("image\t{dir}/fail page 00.png"));
let first = std::fs::read(saved.join("fail page 00.png")).unwrap();
assert!(first.starts_with(b"\x89PNG"));
let (code, out) = owner.cli(
&[
"checks",
&owner.url("proj"),
"--workflow",
"fail",
"--images",
dir,
"--json",
],
b"",
);
assert_eq!(code, 0, "{out}");
assert!(
serde_json::from_str::<serde_json::Value>(&out).is_ok(),
"{out}"
);
// With their logs, each after its line.
let (_, out) = owner.cli(
&[
"checks",
&owner.url("proj"),
"--ref",
"refs/heads/main",
"--log",
],
b"",
);
assert!(out.contains("==> break\nbroken\n"), "{out}");
assert!(
lines
.iter()
.any(|l| l.starts_with("success\tci\trefs/heads/main\t")),
"{out}"
);
// A ring for a ref that has not moved runs nothing again.
let started = ci["startedAt"].clone();
js.ring(&owner.account.did, &ref_key);
std::thread::sleep(Duration::from_millis(500));
assert_eq!(check_of(&runner, &owner, &key, "ci")["startedAt"], started);
// Unless its owner asks: requests that are not one are passed over.
let junk = serde_json::json!({"$type": "dev.knotless.rerun"});
owner.put_record("dev.knotless.rerun", &format!("{key}.zzzz"), junk.clone());
owner.put_record(
"dev.knotless.rerun",
&format!("{key}.{}", knotless::keys::tid()),
junk,
);
let (code, asked) = owner.cli(
&["rerun", &owner.url("proj"), "--ref", "refs/heads/main"],
b"",
);
assert_eq!(code, 0, "{asked}");
let asked = asked.rsplit('/').next().unwrap().to_string();
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
wait_until(|| {
let ci = check_of(&runner, &owner, &key, "ci");
ci["startedAt"] != started && ci["finishedAt"].is_string()
});
assert_eq!(
check_of(&runner, &owner, &key, "ci")["commit"],
sha.as_str()
);
// Once.
let started = check_of(&runner, &owner, &key, "ci")["startedAt"].clone();
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
std::thread::sleep(Duration::from_millis(500));
assert_eq!(check_of(&runner, &owner, &key, "ci")["startedAt"], started);
// A push while the runner is down runs when it is back, rung or not.
drop(rt);
let next = owner.commit_file(&work, "f", "elsewhere\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
let rt = tokio::runtime::Runtime::new().unwrap();
let js = jetstream(&rt);
let opts = knotless_run::Options {
jetstream: js.url.clone(),
..opts
};
rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));
wait_until(|| check_of(&runner, &owner, &key, "ci")["commit"] == next.as_str());
wait_until(|| check_of(&runner, &owner, &key, "ci")["finishedAt"].is_string());
// The checkout is that commit, whose f no longer names its parent.
let again = check_of(&runner, &owner, &key, "ci");
assert_eq!(again["status"], "failure", "{again}");
// The commit before keeps its own check.
let before = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
&sha,
None,
"ci",
);
let before = runner.record(check::NSID, &before);
assert_eq!(before["commit"], sha.as_str());
assert_eq!(before["status"], "success", "{before}");
// A newer commit's run of a workflow on the ref cancels the older
// one's still going: its step is killed, the rest skipped.
let long = "steps:\n - name: wait\n run: test -f quick || sleep 30\n - run: 'true'\n";
std::fs::write(work.join(".knotless/workflows/long.yml"), long).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "long"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let slow = owner.git(&work, &["rev-parse", "HEAD"]);
js.ring(&owner.account.did, &ref_key);
let slow_key = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
&slow,
None,
"long",
);
wait_until(|| runner.record(check::NSID, &slow_key)["status"] == "running");
std::thread::sleep(Duration::from_millis(500));
let quick = owner.commit_file(&work, "quick", "");
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
let quick_key = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
&quick,
None,
"long",
);
wait_until(|| runner.record(check::NSID, &quick_key)["status"] == "success");
wait_until(|| runner.record(check::NSID, &slow_key)["finishedAt"].is_string());
let cancelled = runner.record(check::NSID, &slow_key);
assert_eq!(cancelled["status"], "cancelled", "{cancelled}");
assert_eq!(cancelled["steps"][0]["status"], "cancelled");
assert_eq!(cancelled["steps"][1]["status"], "skipped");
let log = blob(
&runner.account.did,
cancelled["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.ends_with("==> cancelled for a newer commit's run\n"),
"{log}"
);
// One whose runs finish keeps going, and the newer commit's waits its
// turn; one waiting is cancelled for a newer still, so one waits.
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let finish = format!(
"superseded: finish\nsteps:\n - name: hold\n run: until [ -f {}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
go.display()
);
std::fs::write(work.join(".knotless/workflows/finish.yml"), finish).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "finish"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let first = owner.git(&work, &["rev-parse", "HEAD"]);
js.ring(&owner.account.did, &ref_key);
let at = |sha: &str| {
let rkey = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
sha,
None,
"finish",
);
runner.record(check::NSID, &rkey)
};
wait_until(|| at(&first)["status"] == "running");
std::thread::sleep(Duration::from_millis(500));
let second = owner.commit_file(&work, "f", "second\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
// The second, waiting for the first to end, is filed queued.
wait_until(|| at(&second)["status"] == "queued");
std::thread::sleep(Duration::from_millis(500));
assert!(at(&first)["finishedAt"].is_null(), "{}", at(&first));
let third = owner.commit_file(&work, "f", "third\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
wait_until(|| at(&second)["finishedAt"].is_string());
let never = at(&second);
assert_eq!(never["status"], "cancelled", "{never}");
assert!(never.get("steps").is_none(), "{never}");
let log = blob(
&runner.account.did,
never["log"]["ref"]["$link"].as_str().unwrap(),
);
assert_eq!(
log,
"==> cancelled for a newer commit's run before it started\n"
);
// The third waits for the first, even free to end at once.
std::fs::write(go.join(&third), "").unwrap();
std::thread::sleep(Duration::from_millis(500));
assert_eq!(at(&third)["status"], "queued", "{}", at(&third));
assert!(at(&first)["finishedAt"].is_null(), "{}", at(&first));
std::fs::write(go.join(&first), "").unwrap();
wait_until(|| at(&third)["finishedAt"].is_string());
let (first, third) = (at(&first), at(&third));
assert_eq!(first["status"], "success", "{first}");
assert_eq!(third["status"], "success", "{third}");
assert!(
first["finishedAt"].as_str() <= third["finishedAt"].as_str(),
"{first} {third}"
);
// Runs of a workflow with a cache, on two branches, go side by side,
// each over its own layers, which go when they end. Each writes to
// every run's upper layer going, as the shell executor mounts none.
std::fs::remove_file(work.join(".knotless/workflows/finish.yml")).unwrap();
let caches = runner.path("caches");
let hold = format!(
"cache: [target]\nsteps:\n - name: hold\n run: |\n b=${{KNOTLESS_RUN_REF##*/}}\n for u in {caches}/runs/*/*/upper; do echo $b > $u/$b; done\n touch {go}/started-vol-$b\n until [ -f {go}/vol-$b ]; do sleep 0.1; done\n",
go = go.display(),
caches = caches.display()
);
std::fs::write(work.join(".knotless/workflows/vol.yml"), hold).unwrap();
owner.git(&work, &["add", "-A"]);
owner.git(&work, &["commit", "-q", "-m", "volumes"]);
owner.git(
&work,
&[
"push",
"-q",
"origin",
"HEAD:refs/heads/a",
"HEAD:refs/heads/b",
],
);
js.ring(&owner.account.did, &ref_rkey(&key, "refs/heads/a"));
let started = || {
["a", "b"]
.into_iter()
.filter(|b| go.join(format!("started-vol-{b}")).exists())
.count()
};
wait_until(|| started() == 2);
let runs = caches.join("runs");
assert_eq!(std::fs::read_dir(&runs).unwrap().count(), 2);
for name in ["vol-a", "vol-b", "vol-main"] {
std::fs::write(go.join(name), "").unwrap();
}
let vol_done = |b: &str| {
let r = format!("refs/heads/{b}");
wait_until(|| check_on(&runner, &owner, &key, &r, "vol")["finishedAt"].is_string());
let done = check_on(&runner, &owner, &key, &r, "vol");
assert_eq!(done["status"], "success", "{done}");
};
vol_done("a");
vol_done("b");
wait_until(|| std::fs::read_dir(&runs).unwrap().count() == 0);
// What they wrote went with them: the base they started from is as
// empty as it was.
let base = || {
let cache = std::fs::read_dir(caches.join("bases")).unwrap();
let cache = cache.map(|c| c.unwrap().path()).next().unwrap();
let newest = std::fs::read_dir(cache).unwrap().map(|g| g.unwrap().path());
let newest = newest.max().unwrap();
let mut held: Vec<String> = std::fs::read_dir(&newest)
.unwrap()
.map(|f| f.unwrap().file_name().to_string_lossy().into_owned())
.collect();
held.sort();
held
};
assert!(base().is_empty(), "{:?}", base());
// A push to main that passes keeps what it wrote, for the runs after.
owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
js.ring(&owner.account.did, &ref_key);
vol_done("main");
wait_until(|| std::fs::read_dir(&runs).unwrap().count() == 0);
assert_eq!(base(), ["main"]);
// An owner its operator pauses has its runs filed as errors, saying
// why, until they are taken off the list.
let paused = runner.path("paused.json");
let list = serde_json::json!({ owner.account.did.as_str(): "mining, by the looks of it" });
std::fs::write(&paused, list.to_string()).unwrap();
let sha = owner.commit_file(&work, "f", "paused\n");
owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
js.ring(&owner.account.did, &ref_key);
wait_until(|| check_of(&runner, &owner, &key, "ci")["commit"] == sha.as_str());
let ci = check_of(&runner, &owner, &key, "ci");
assert_eq!(ci["status"], "error", "{ci}");
let why = "paused by this runner's operator: mining, by the looks of it";
assert_eq!(ci["error"], why, "{ci}");
std::fs::remove_file(&paused).unwrap();
// A repository whose name now holds another one is not run: the one
// it was is gone, and its checks with it.
let mut repo = owner.record("dev.knotless.repo", "proj");
repo["key"] = "3l3qo2vuowo2z".into();
owner.put_record("dev.knotless.repo", "proj", repo);
owner.commit_file(&work, "f", "again\n");
js.ring(&owner.account.did, &ref_key);
wait_until(|| check_of(&runner, &owner, &key, "ci").is_null());
std::thread::sleep(Duration::from_millis(500));
assert!(check_of(&runner, &owner, "3l3qo2vuowo2z", "ci").is_null());
// One it cannot find beside one it can is passed over.
let nope = format!("{}/nope", owner.account.handle);
let both = knotless_run::Options {
repos: vec![nope, opts.repos[0].clone()],
..opts.clone()
};
let started = knotless_run::Runner::start(both, common::ctx(&runner));
let (_, watches) = rt.block_on(started).unwrap();
assert_eq!(watches.len(), 1);
assert_eq!(watches[0].label, opts.repos[0]);
// Alone, it stops it before it watches anything.
let missing = knotless_run::Options {
repos: vec![format!("{}/nope", owner.account.handle)],
..opts
};
let err = rt
.block_on(knotless_run::run(missing, common::ctx(&runner)))
.unwrap_err();
assert!(
err.to_string()
.ends_with("/nope: no repository by that name this runner can read"),
"{err}"
);
}
/// The newest of `runner`'s checks of `wf` on `r`, of any commit, for the
/// repository keyed `key` of `owner`'s: null if there is none.
fn check_on(runner: &Sandbox, owner: &Sandbox, key: &str, r: &str, wf: &str) -> serde_json::Value {
let prefix = knotless::keys::check_prefix(&owner.account.did, key);
let (of, named) = (format!("{prefix}.{}.", ref_hash(r)), format!(".{wf}"));
runner
.records(check::NSID)
.into_iter()
.filter(|(k, _)| k.starts_with(&of) && k.ends_with(&named))
.map(|(_, v)| v)
.max_by(|a, b| a["startedAt"].as_str().cmp(&b["startedAt"].as_str()))
.unwrap_or_default()
}
#[test]
fn what_goes_wrong_is_reported() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("tools", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("tools")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let label = format!("{}/tools", owner.account.handle);
std::fs::write(runner.path("secrets.json"), "{ not json").unwrap();
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![label],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: Some(runner.path("secrets.json")),
secrets_key: None,
state: Some(runner.path("state.json")),
work: None,
step_timeout: Duration::from_secs(1),
// Past the one step that is out of its own time first, and short
// of the two that run out of the run's.
run_timeout: Some(Duration::from_millis(1200)),
jobs: 2,
pull_jobs: 2,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: None,
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let rt = tokio::runtime::Runtime::new().unwrap();
let (run, watches) = rt
.block_on(knotless_run::Runner::start(
opts.clone(),
common::ctx(&runner),
))
.unwrap();
let watch = &watches[0];
rt.block_on(run.reconcile(watch)).unwrap();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
for (file, text) in [
(
"slow.yml",
"steps: [{run: 'echo going; sleep 30'}, {run: 'true'}]\n",
),
(
"long.yml",
"steps: [{run: 'sleep 0.5'}, {run: 'echo late; sleep 30'}, {run: 'true'}]\n",
),
("needs.yml", "secrets: [X]\nsteps: [{run: 'true'}]\n"),
("twin-a.yml", "name: twin\nsteps: []\n"),
("twin-b.yml", "name: twin\nsteps: []\n"),
("bad name!.yml", "{"),
] {
std::fs::write(dir.join(file), text).unwrap();
}
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
rt.block_on(run.reconcile(watch)).unwrap();
rt.block_on(run.idle());
let main = "refs/heads/main";
let key = watch.key.clone();
// Out of time: killed, with no exit code, and said so in the log.
let slow = check_on(&runner, &owner, &key, main, "slow");
assert_eq!(slow["status"], "failure", "{slow}");
assert!(slow["steps"][0].get("exitCode").is_none(), "{slow}");
assert_eq!(slow["steps"][1]["status"], "skipped");
let log = blob(
&runner.account.did,
slow["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(log.ends_with("going\n==> killed after 1s\n"), "{log}");
// Out of the run's time, each step in its own: the step going is
// killed, and the rest skipped.
let long = check_on(&runner, &owner, &key, main, "long");
assert_eq!(long["status"], "failure", "{long}");
assert_eq!(long["steps"][0]["status"], "success", "{long}");
assert!(long["steps"][1].get("exitCode").is_none(), "{long}");
assert_eq!(long["steps"][2]["status"], "skipped");
let log = blob(
&runner.account.did,
long["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.ends_with("late\n==> killed as the run's 1s were up\n"),
"{log}"
);
// A workflow that asks for secrets the runner cannot read does not run.
let needs = check_on(&runner, &owner, &key, main, "needs");
assert_eq!(needs["status"], "error");
assert!(
needs["error"]
.as_str()
.unwrap()
.starts_with("reading the runner's secrets"),
"{needs}"
);
let twin = check_on(&runner, &owner, &key, main, "twin");
assert_eq!(
twin["error"],
"more than one workflow is named twin, twin-b.yml among them"
);
let invalid = check_on(&runner, &owner, &key, main, "invalid");
assert!(
invalid["error"]
.as_str()
.unwrap()
.starts_with("bad name!.yml: "),
"{invalid}"
);
// A branch without the directory has no workflows to run.
owner.git(&work, &["checkout", "-q", "--orphan", "bare"]);
owner.git(&work, &["rm", "-rqf", "."]);
owner.commit_file(&work, "g", "g\n");
owner.git(&work, &["push", "-q", "origin", "bare"]);
rt.block_on(run.reconcile(watch)).unwrap();
assert!(check_on(&runner, &owner, &key, "refs/heads/bare", "slow").is_null());
// A name that comes to hold another repository, the one it held kept
// under another name, is no longer watched by that name, its checks
// kept.
let renamed = ["rename", &owner.url("tools"), "elsewhere"];
assert_eq!(owner.cli(&renamed, b"").0, 0);
let mut repo = owner.record("dev.knotless.repo", "elsewhere");
repo["key"] = "3l3qo2vuowo2z".into();
owner.put_record("dev.knotless.repo", "tools", repo);
let err = rt.block_on(run.reconcile(watch)).unwrap_err();
assert!(
err.to_string()
.ends_with("/tools no longer names the repository it did"),
"{err}"
);
assert!(!check_on(&runner, &owner, &key, main, "slow").is_null());
// The one it held gone, the checks of it go with it.
owner.delete_record("dev.knotless.repo", "elsewhere");
rt.block_on(run.reconcile(watch)).unwrap();
let of_it = knotless::keys::check_prefix(&owner.account.did, &key);
let filed = runner.records("dev.knotless.check");
assert!(filed.iter().all(|(rkey, _)| !rkey.starts_with(&of_it)));
// A check it left running, as a runner stopped mid-run leaves one, is
// ended when it starts again; one it finished is left.
let prefix = knotless::keys::check_prefix(&owner.account.did, "3l3qo2vuowo2z");
let mut stale = slow.clone();
// Its log went with the checks it was of.
let fields = stale.as_object_mut().unwrap();
fields.retain(|_, v| v["$type"] != "blob");
stale["repo"]["key"] = "3l3qo2vuowo2z".into();
stale["status"] = "running".into();
runner.put_record(
"dev.knotless.check",
&format!("{prefix}.a.b.stale"),
stale.clone(),
);
stale["status"] = "queued".into();
runner.put_record(
"dev.knotless.check",
&format!("{prefix}.a.b.waiting"),
stale.clone(),
);
stale["status"] = "success".into();
runner.put_record("dev.knotless.check", &format!("{prefix}.a.b.done"), stale);
// A Jetstream it cannot reach: it reads the repositories anyway, and
// tries again later.
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
let ended = || runner.record("dev.knotless.check", &format!("{prefix}.a.b.stale"));
wait_until(|| ended()["status"] == "error");
assert_eq!(
ended()["error"],
"the runner stopped before this run finished"
);
let waiting = runner.record("dev.knotless.check", &format!("{prefix}.a.b.waiting"));
assert_eq!(waiting["status"], "error");
assert_eq!(
waiting["error"],
"the runner stopped before this run started"
);
let done = runner.record("dev.knotless.check", &format!("{prefix}.a.b.done"));
assert_eq!(done["status"], "success");
}
/// A run of a workflow whose runs finish is never cut off by a newer
/// commit's, even while it waits for a job: with the one job held, the
/// first deploy has its turn on main and waits; the second waits behind it
/// and is cancelled for the third, which runs once the first has finished.
#[test]
fn a_deploy_is_never_cut_off() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("ship", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("ship")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![format!("{}/ship", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: None,
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 1,
pull_jobs: 1,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let rt = tokio::runtime::Runtime::new().unwrap();
let (run, watches) = rt
.block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
.unwrap();
let watch = &watches[0];
let key = watch.key.clone();
rt.block_on(run.reconcile(watch)).unwrap();
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let block = format!(
"on: {{push: {{refs: [refs/heads/block]}}}}\nsteps:\n - run: |\n touch {go}/started-block\n until [ -f {go}/block ]; do sleep 0.1; done\n",
go = go.display()
);
let deploy = format!(
"superseded: finish\non: {{push: {{refs: [refs/heads/main]}}}}\nsteps:\n - run: until [ -f {}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
go.display()
);
std::fs::write(dir.join("block.yml"), block).unwrap();
std::fs::write(dir.join("deploy.yml"), deploy).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
// The one job, held.
owner.git(&work, &["push", "-q", "origin", "HEAD:block"]);
rt.block_on(run.reconcile(watch)).unwrap();
wait_until(|| go.join("started-block").exists());
let mut shas = Vec::new();
for n in 0..3 {
if n > 0 {
owner.commit_file(&work, "f", &format!("{n}\n"));
}
owner.git(&work, &["push", "-q", "origin", "HEAD:main"]);
shas.push(owner.git(&work, &["rev-parse", "HEAD"]));
rt.block_on(run.reconcile(watch)).unwrap();
}
let at = |sha: &str| {
let rkey = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
sha,
None,
"deploy",
);
runner.record(check::NSID, &rkey)
};
let (first, second, third) = (&shas[0], &shas[1], &shas[2]);
wait_until(|| at(second)["finishedAt"].is_string());
let never = at(second);
assert_eq!(never["status"], "cancelled", "{never}");
let log = blob(
&runner.account.did,
never["log"]["ref"]["$link"].as_str().unwrap(),
);
assert_eq!(
log,
"==> cancelled for a newer commit's run before it started\n"
);
// The first, still waiting for the job, is not: it is filed queued,
// and filed running once it has the job, started then.
let waiting = at(first);
assert_eq!(waiting["status"], "queued", "{waiting}");
std::fs::write(go.join("block"), "").unwrap();
wait_until(|| at(first)["status"] == "running");
let started = at(first)["startedAt"].clone();
assert!(
started.as_str() > waiting["startedAt"].as_str(),
"{started}"
);
for name in [first.as_str(), third.as_str()] {
std::fs::write(go.join(name), "").unwrap();
}
rt.block_on(run.idle());
let (first, third) = (at(first), at(third));
assert_eq!(first["status"], "success", "{first}");
assert_eq!(first["startedAt"], started, "{first}");
assert_eq!(third["status"], "success", "{third}");
assert!(
first["finishedAt"].as_str() <= third["finishedAt"].as_str(),
"{first} {third}"
);
}
/// A stop waits only for the runs whose steps have started, a deploy's
/// among them: those waiting their turn are dropped, their checks filed
/// as cancelled, not deleted, and run when the runner starts again, and
/// only they. Those
/// of a going deploy's commit are kept until it ends.
#[test]
fn a_stop_drops_what_has_not_started() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let base = appview(&rt, &Sandbox::new());
owner.set("KNOTLESS_APPVIEW", &base);
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
runner.set("KNOTLESS_APPVIEW", &base);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("ship", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("ship")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![format!("{}/ship", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: None,
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 1,
pull_jobs: 1,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: true,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let start = || {
let started = knotless_run::Runner::start(opts.clone(), common::ctx(&runner));
let (run, mut watches) = rt.block_on(started).unwrap();
(run, watches.remove(0))
};
let (run, watch) = start();
let key = watch.key.clone();
rt.block_on(run.reconcile(&watch)).unwrap();
// `block` holds the one job until `go/block` is there, so the runs
// after it wait in line, where the first ranked goes first: the deploy,
// which then holds the job until its commit's file is there. `then`
// waits for `go/then`, and notes each run of it.
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let block = format!(
"on: {{push: {{refs: [refs/heads/block]}}}}\nsteps:\n - run: |\n touch {go}/started-block\n until [ -f {go}/block ]; do sleep 0.1; done\n",
go = go.display()
);
std::fs::write(dir.join("block.yml"), block).unwrap();
let deploy = format!(
"superseded: finish\non: {{push: {{refs: [refs/heads/main]}}}}\nsteps:\n - run: |\n echo $KNOTLESS_RUN_SHA >> {go}/deploys\n until [ -f {go}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n",
go = go.display()
);
let then = format!(
"on: {{push: {{refs: [refs/heads/main]}}, pull: {{base: [refs/heads/main]}}}}\nsteps:\n - run: |\n echo \"$KNOTLESS_RUN_SHA ${{KNOTLESS_RUN_PULL:-push}}\" >> {go}/thens\n until [ -f {go}/then ]; do sleep 0.1; done\n",
go = go.display()
);
std::fs::write(dir.join("deploy.yml"), deploy).unwrap();
std::fs::write(dir.join("then.yml"), then).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["push", "-q", "origin", "HEAD:block"]);
rt.block_on(run.reconcile(&watch)).unwrap();
wait_until(|| go.join("started-block").exists());
owner.git(&work, &["push", "-q", "origin", "main"]);
let main = owner.git(&work, &["rev-parse", "HEAD"]);
rt.block_on(run.reconcile(&watch)).unwrap();
// Both of main's runs are in line by the time the job is let go: a
// run is in line once its task first runs, a moment after it is filed.
std::thread::sleep(Duration::from_millis(500));
std::fs::write(go.join("block"), "").unwrap();
let ran = |name: &str| std::fs::read_to_string(go.join(name)).unwrap_or_default();
wait_until(|| ran("deploys") == format!("{main}\n"));
// A pull, its run waiting behind the deploy too.
let topic = owner.commit_file(&work, "t", "t\n");
owner.git(&work, &["push", "-q", "origin", "HEAD:topic"]);
let (code, pull) = owner.cli(
&[
"pull",
&owner.url("ship"),
"main",
&owner.url("ship"),
"topic",
"t",
],
b"",
);
assert_eq!(code, 0, "{pull}");
let pull = pull.trim().to_string();
let of = |r: &str, wf: &str| check_on(&runner, &owner, &key, r, wf);
wait_until(|| {
rt.block_on(run.reconcile(&watch)).unwrap();
of(&pull, "then")["status"] == "queued"
});
assert_eq!(of("refs/heads/main", "then")["status"], "queued");
// Stopped, it cancels the pull's run and reads no more, but waits
// for the deploy going, and keeps the push's run of its commit, which
// a deploy may wait for, until the deploy ends.
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
});
wait_until(|| of(&pull, "then")["status"] == "cancelled");
rt.block_on(run.reconcile(&watch)).unwrap();
std::thread::sleep(Duration::from_millis(500));
assert!(!stopping.is_finished());
assert_eq!(of("refs/heads/main", "then")["status"], "queued");
std::fs::write(go.join(&main), "").unwrap();
promptly(&rt, stopping);
assert_eq!(of("refs/heads/main", "deploy")["status"], "success");
assert_eq!(of("refs/heads/main", "then")["status"], "cancelled");
assert_eq!(ran("thens"), "");
// Started again, it runs what it dropped, and not the deploy that
// finished: one of the two takes the job. Stopped while that one goes,
// it waits for it, an ordinary run, and drops the other again.
let (run, watch) = start();
rt.block_on(run.reconcile(&watch)).unwrap();
let (push, pulled) = (format!("{main} push\n"), format!("{topic} {pull}\n"));
wait_until(|| [&push, &pulled].contains(&&ran("thens")));
let (first, second) = match ran("thens") == push {
true => ("refs/heads/main", pull.as_str()),
false => (pull.as_str(), "refs/heads/main"),
};
wait_until(|| of(second, "then")["status"] == "queued");
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
});
wait_until(|| of(second, "then")["status"] == "cancelled");
assert!(!stopping.is_finished());
std::fs::write(go.join("then"), "").unwrap();
promptly(&rt, stopping);
assert_eq!(of(first, "then")["status"], "success");
// And the other on the start after.
let (run, watch) = start();
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
let ran_second = of(second, "then");
assert_eq!(ran_second["status"], "success", "{ran_second}");
let mut thens: Vec<String> = ran("thens").lines().map(str::to_string).collect();
thens.sort();
let mut both = vec![push.trim().to_string(), pulled.trim().to_string()];
both.sort();
assert_eq!(thens, both);
assert_eq!(ran("deploys"), format!("{main}\n"));
}
/// A stand-in for the runner's PDS, in front of the reference one: it fails
/// the next `fail` writes, holds the next `hang` past any wait for an
/// answer, as the PDS did one night for a few minutes, and refuses the next
/// `limit` as a rate limit does, saying it lets writes in a second on.
struct Flaky {
url: String,
fail: Arc<AtomicUsize>,
hang: Arc<AtomicUsize>,
limit: Arc<AtomicUsize>,
}
fn flaky(rt: &tokio::runtime::Runtime) -> Flaky {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let (fail, hang) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
let limit = Arc::new(AtomicUsize::new(0));
let (failing, hanging, limiting) = (fail.clone(), hang.clone(), limit.clone());
let pass = move |req: axum::extract::Request| {
let (fail, hang, limit) = (failing.clone(), hanging.clone(), limiting.clone());
async move {
let path = req.uri().path_and_query().unwrap().as_str().to_string();
let write = ["putRecord", "deleteRecord", "uploadBlob"]
.iter()
.any(|w| path.ends_with(w));
let take = |n: &AtomicUsize| {
write && n.fetch_update(SeqCst, SeqCst, |n| n.checked_sub(1)).is_ok()
};
if take(&hang) {
tokio::time::sleep(Duration::from_secs(30)).await;
}
if take(&fail) {
let busy = r#"{"error":"InternalServerError","message":"busy"}"#;
let mut res = axum::response::Response::new(axum::body::Body::from(busy));
*res.status_mut() = axum::http::StatusCode::INTERNAL_SERVER_ERROR;
return res;
}
if take(&limit) {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
let reset = (now.unwrap().as_secs() + 1).to_string();
let refused = r#"{"error":"RateLimitExceeded","message":"Rate Limit Exceeded"}"#;
let mut res = axum::response::Response::new(axum::body::Body::from(refused));
*res.status_mut() = axum::http::StatusCode::TOO_MANY_REQUESTS;
let reset = axum::http::HeaderValue::from_str(&reset).unwrap();
res.headers_mut().insert("ratelimit-reset", reset);
return res;
}
let (parts, body) = req.into_parts();
let body = axum::body::to_bytes(body, usize::MAX).await.unwrap();
let method = reqwest::Method::from_bytes(parts.method.as_str().as_bytes()).unwrap();
let mut out = reqwest::Client::new().request(method, format!("{}{path}", net().pds));
for (k, v) in parts.headers.iter().filter(|(k, _)| *k != "host") {
out = out.header(k.as_str(), v.as_bytes());
}
let res = out.body(body).send().await.unwrap();
let status = res.status().as_u16();
let kind = res.headers().get("content-type").cloned();
let mut back =
axum::response::Response::new(axum::body::Body::from(res.bytes().await.unwrap()));
*back.status_mut() = axum::http::StatusCode::from_u16(status).unwrap();
if let Some(kind) = kind {
let kind = axum::http::HeaderValue::from_bytes(kind.as_bytes()).unwrap();
back.headers_mut().insert("content-type", kind);
}
back
}
};
let router = axum::Router::new().fallback(pass);
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
Flaky {
url,
fail,
hang,
limit,
}
}
/// Writes the PDS fails, or never answers, wedge nothing: a read that
/// failed for them is made again by itself, a write that hung is given up
/// on and made again, one a rate limit refused waits until it lets writes
/// in, and the runs after start as ever. A stop waits for a
/// run that hangs only so long, then cancels it, killing its steps. And a
/// push Jetstream never rang for is found by the sweep.
#[test]
fn a_failing_pds_wedges_nothing() {
let rt = tokio::runtime::Runtime::new().unwrap();
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("flaky", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("flaky")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(
dir.join("quick.yml"),
"on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n",
)
.unwrap();
let stuck = format!(
"on: {{push: {{refs: [refs/heads/stuck]}}}}\nsteps:\n - run: |\n echo $$ > {go}/stuck\n sleep 600\n",
go = go.display()
);
std::fs::write(dir.join("stuck.yml"), stuck).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
let js = jetstream(&rt);
let opts = knotless_run::Options {
jetstream: js.url.clone(),
repos: vec![format!("{}/flaky", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: None,
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 1,
pull_jobs: 1,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(1),
stop_wait: Duration::from_secs(2),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let pds = flaky(&rt);
let ctx = common::ctx(&runner);
let reporter = rt.block_on(check::Reporter::new(&ctx)).unwrap();
// Its writes alone wait only so long for an answer: past what a busy
// PDS takes for one that passes, as 3 s was not, short of the hang.
let mut env = runner.env();
env.timeout = Duration::from_secs(10);
let impatient = knotless::env::Ctx::new(env);
let reporter = reporter.through(&impatient, &pds.url);
let started = knotless_run::Runner::start_as(opts, ctx, reporter);
let (run, watches) = rt.block_on(started).unwrap();
let watch = watches[0].clone();
let key = watch.key.clone();
rt.block_on(run.reconcile(&watch)).unwrap();
rt.spawn(knotless_run::watch(run.clone(), watches));
let of = |r: &str, sha: &str, wf: &str| {
let rkey = check::rkey(&owner.account.did, &key, r, sha, None, wf);
runner.record(check::NSID, &rkey)
};
// Every write failing, the push's read fails, each write tried a few
// times; once they pass, it is read again, rung by nothing, and runs.
pds.fail.store(1000, SeqCst);
owner.git(&work, &["push", "-q", "origin", "main"]);
let first = owner.git(&work, &["rev-parse", "HEAD"]);
wait_until(|| pds.fail.load(SeqCst) <= 1000 - 8);
pds.fail.store(0, SeqCst);
wait_until(|| of("refs/heads/main", &first, "quick")["status"] == "success");
// Writes that never answer are given up on and made again.
pds.hang.store(1, SeqCst);
let second = owner.commit_file(&work, "f", "f\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
wait_until(|| of("refs/heads/main", &second, "quick")["status"] == "success");
assert_eq!(pds.hang.load(SeqCst), 0);
// Writes a rate limit refuses, more of them than a busy PDS's are
// tried, wait until it lets writes in, and the run passes.
pds.limit.store(6, SeqCst);
let third = owner.commit_file(&work, "g", "g\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
wait_until(|| of("refs/heads/main", &third, "quick")["status"] == "success");
assert_eq!(pds.limit.load(SeqCst), 0);
// A stop waits for a run that hangs only so long, then cancels it, and
// its steps are killed.
owner.git(&work, &["push", "-q", "origin", "HEAD:stuck"]);
wait_until(|| go.join("stuck").exists());
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
});
promptly(&rt, stopping);
let cut = of("refs/heads/stuck", &third, "stuck");
assert_eq!(cut["status"], "cancelled", "{cut}");
let log = blob(
&runner.account.did,
cut["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.ends_with("==> cancelled as the runner stopped, past what a stop waits for\n"),
"{log}"
);
let pid = std::fs::read_to_string(go.join("stuck")).unwrap();
assert!(!alive(pid.trim()));
}
/// Whether the process `pid` is there.
fn alive(pid: &str) -> bool {
let signalled = std::process::Command::new("kill")
.args(["-0", pid])
.stderr(std::process::Stdio::null())
.status();
signalled.unwrap().success()
}
/// Waits for `stopping`, a stop that has only its last few seconds left,
/// rather than a backlog's minutes.
fn promptly(rt: &tokio::runtime::Runtime, stopping: tokio::task::JoinHandle<()>) {
let within = async { tokio::time::timeout(Duration::from_secs(20), stopping).await };
rt.block_on(within).expect("the stop waited").unwrap();
}
#[test]
fn the_binary() {
let s = Sandbox::new();
let exe = env!("CARGO_BIN_EXE_knotless-run");
let out = s.command(exe, s.dir.path()).output().unwrap();
assert_eq!(out.status.code(), Some(2));
assert!(String::from_utf8_lossy(&out.stderr).starts_with("name a repository"));
// Nobody to run as.
let out = s
.command(exe, s.dir.path())
.arg(format!("{}/proj", s.account.handle))
.output()
.unwrap();
assert_eq!(out.status.code(), Some(1));
let err = String::from_utf8_lossy(&out.stderr);
assert!(err.contains("knotless-run: not logged in"), "{err}");
// Said as a log line too, as JSON.
assert!(err.contains(r#""message":"knotless-run failed""#), "{err}");
// Logged in and watching, it stops cleanly when asked to.
let mut s = s;
let handle = s.account.handle.clone();
s.set("KNOTLESS_IDENTIFIER", &handle);
let work = s.work("proj", 1);
s.git(&work, &["push", "-q", &s.url("proj"), "main"]);
let mut child = s
.command(exe, s.dir.path())
.args(["--jetstream", "ws://127.0.0.1:9", "--executor", "shell"])
.arg(format!("{handle}/proj"))
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
// It logs, as JSON lines, that it started, and then what it watches.
let lines = std::io::BufRead::lines(std::io::BufReader::new(child.stderr.take().unwrap()));
let mut lines = lines.map(Result::unwrap);
let first: serde_json::Value = serde_json::from_str(&lines.next().unwrap()).unwrap();
assert_eq!(first["message"], "knotless-run starting");
assert!(first["version"].is_string());
let watching = lines
.find(|l| l.contains(r#""message":"watching""#))
.unwrap();
assert!(watching.contains(&format!("{handle}/proj")), "{watching}");
let killed = std::process::Command::new("kill")
.args(["-TERM", &child.id().to_string()])
.status()
.unwrap();
assert!(killed.success());
assert_eq!(child.wait().unwrap().code(), Some(0));
}
/// An appview over the reference PDS, following its firehose, at the URL
/// returned.
fn appview(rt: &tokio::runtime::Runtime, s: &Sandbox) -> String {
listening(rt, s).0
}
/// An appview as [`appview`] makes, and its following, which stops when
/// aborted: the appview then hears of nothing written after.
fn listening(rt: &tokio::runtime::Runtime, s: &Sandbox) -> (String, tokio::task::JoinHandle<()>) {
serving(rt, s, None)
}
/// An appview as [`listening`] makes, naming `runner` for new
/// repositories, and then a gateway too, for the service tokens of
/// invites to private ones.
fn serving(
rt: &tokio::runtime::Runtime,
s: &Sandbox,
runner: Option<&str>,
) -> (String, tokio::task::JoinHandle<()>) {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
// A gateway's did:web names a host, which an IP address is not.
let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
let index = knotless_serve::index::Index::open(&s.path("index.db")).unwrap();
let gateway = runner.map(|_| knotless_serve::gateway::Gateway::new(&base).unwrap());
let mut app = knotless_serve::App::build(common::ctx(s), Duration::ZERO, gateway, Some(index));
Arc::get_mut(&mut app).unwrap().default_runner = runner.map(str::to_string);
let router = knotless_serve::router(app.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
let ears = rt.spawn(knotless_serve::firehose::watch(
app,
net().pds.replace("http://", "ws://"),
));
(base, ears)
}
const PULLS: &str = r#"
on: {pull: {base: [refs/heads/main]}}
secrets: [TOKEN]
cache: [target]
steps:
- run: |
echo "pull $KNOTLESS_RUN_PULL at $KNOTLESS_RUN_REF on $KNOTLESS_RUN_BASE token=${TOKEN:-none}"
for u in CACHES/runs/*/*/upper; do echo pull > $u/pull; done
test -f f
test ! -f h || echo "with h"
"#;
#[test]
fn a_pull_runs_its_head() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut contrib = Sandbox::new();
let mut runner = Sandbox::new();
let base = appview(&rt, &Sandbox::new());
owner.set("KNOTLESS_APPVIEW", &base);
contrib.set("KNOTLESS_APPVIEW", &base);
let theirs = contrib.account.handle.clone();
contrib.set("KNOTLESS_IDENTIFIER", &theirs);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("proj", 1);
std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
let caches = runner.path("caches");
let pulls = PULLS.replace("CACHES", &caches.display().to_string());
std::fs::write(work.join(".knotless/workflows/pulls.yml"), pulls).unwrap();
// A push's, of a branch nothing here pushes: no push runs.
std::fs::write(
work.join(".knotless/workflows/push.yml"),
"on: {push: {refs: [refs/heads/release]}}\nsteps: [{run: 'true'}]\n",
)
.unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let (code, fork) = contrib.cli(&["fork", &owner.url("proj")], b"");
assert_eq!(code, 0, "{fork}");
let fork = fork.trim().to_string();
let fork_name = fork.rsplit('/').next().unwrap().to_string();
let fork_key = contrib.record("dev.knotless.repo", &fork_name)["key"]
.as_str()
.unwrap()
.to_string();
// The pulls of whoever may push to it run as they move: the owner's,
// and a collaborator's, from their own fork.
let collaborators = |named: serde_json::Value| {
let mut record = owner.record("dev.knotless.repo", "proj");
record["collaborators"] = named;
owner.put_record("dev.knotless.repo", "proj", record);
};
collaborators(serde_json::json!([{"did": contrib.account.did}]));
let clone = contrib.path("clone");
contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
let pull = |branch: &str| {
let (code, out) = contrib.cli(
&["pull", &owner.url("proj"), "main", &fork, branch, branch],
b"",
);
assert_eq!(code, 0, "{out}");
out.trim().to_string()
};
contrib.commit_file(&clone, "f", "f\n");
contrib.git(&clone, &["push", "-q", "origin", "HEAD:topic"]);
let topic = pull("topic");
// Into a branch the workflow does not run on.
owner.git(&work, &["push", "-q", "origin", "main:side"]);
contrib.git(&clone, &["push", "-q", "origin", "HEAD:side"]);
let (code, out) = contrib.cli(
&["pull", &owner.url("proj"), "side", &fork, "side", "side"],
b"",
);
assert_eq!((code, out.is_empty()), (0, false), "{out}");
let lists = |uri: &str| {
let (_, out) = owner.cli(&["list", &owner.url("proj"), "--pulls"], b"");
out.contains(uri.rsplit('/').next().unwrap())
};
wait_until(|| lists(&topic));
let main = owner.git(&work, &["rev-parse", "main"]);
let js = jetstream(&rt);
let mut opts = runner_opts(&runner, &owner, &js);
opts.pulls = true;
// It finds pulls through an appview, and has none: KNOTLESS_APPVIEW is empty.
let err = rt
.block_on(knotless_run::run(opts.clone(), common::ctx(&runner)))
.unwrap_err();
assert!(
err.to_string().contains("KNOTLESS_APPVIEW is set empty"),
"{err}"
);
runner.set("KNOTLESS_APPVIEW", &base);
rt.spawn(knotless_run::run(opts.clone(), common::ctx(&runner)));
let check = |uri: &str| check_on(&runner, &owner, &key, uri, "pulls");
// Watching starts where the pulls are: the open one has not run.
wait_until(|| {
std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&topic))
});
assert!(check(&topic).is_null());
// Its head moving, in the fork, runs it: the pull's workflows, without
// the secrets a push gets.
let sha = contrib.commit_file(&clone, "g", "g\n");
contrib.git(&clone, &["push", "-q", "origin", "HEAD:topic"]);
let fork_ref = ref_rkey(&fork_key, "refs/heads/topic");
wait_until(|| {
js.ring(&contrib.account.did, &fork_ref);
check(&topic)["finishedAt"].is_string()
});
let ran = check(&topic);
assert_eq!(ran["status"], "success", "{ran}");
assert_eq!(ran["pull"], topic.as_str());
assert_eq!(ran["ref"], "refs/heads/topic");
assert_eq!(ran["commit"], sha.as_str());
assert_eq!(ran["baseCommit"], main.as_str());
assert_eq!(ran["repo"]["key"], key.as_str());
let log = blob(
&runner.account.did,
ran["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.contains(&format!(
"pull {topic} at refs/heads/topic on {main} token=none"
)),
"{log}"
);
// A push workflow does not run on a pull.
assert!(check_on(&runner, &owner, &key, &topic, "push").is_null());
// What it wrote to its cache went with it, its layers too: the base
// is as the default branch left it, with nothing.
let held = |dir: &std::path::Path| {
let all = std::fs::read_dir(dir).unwrap();
let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
all.sort();
all
};
wait_until(|| held(&caches.join("runs")).is_empty());
let bases = || {
let all = held(&caches.join("bases"));
all.iter().flat_map(|c| held(c)).collect::<Vec<_>>()
};
assert_eq!(bases().len(), 1);
assert!(held(&bases()[0]).is_empty());
// The base moving runs it again, at its head merged with the base, as
// merging it would leave the base.
let main_ref = ref_rkey(&key, "refs/heads/main");
let at_base = |tip: &str| {
wait_until(|| {
js.ring(&owner.account.did, &main_ref);
let c = check(&topic);
c["baseCommit"] == tip && c["finishedAt"].is_string()
});
check(&topic)
};
// A pull whose base is deleted has nothing to merge into, and is passed
// over.
owner.git(&work, &["push", "-q", "origin", ":side"]);
let moved = owner.commit_file(&work, "h", "h\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
let ran = at_base(&moved);
assert_eq!(
(&ran["status"], &ran["commit"]),
(&"success".into(), &sha.as_str().into()),
"{ran}"
);
let log = blob(
&runner.account.did,
ran["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(log.contains("with h"), "{log}");
// One that conflicts with it fails without running, naming where.
let clash = owner.commit_file(&work, "f", "the owner's f\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
let failed = at_base(&clash);
assert_eq!(failed["status"], "failure", "{failed}");
let error = failed["error"].as_str().unwrap();
let said = format!("conflicts with main at {} in f: merge main", &clash[..12]);
assert!(error.contains(&said), "{error}");
// A pull opened rings for the repository it is into, and runs.
contrib.git(&clone, &["checkout", "-q", "-b", "bare", "HEAD~2"]);
contrib.git(&clone, &["rm", "-q", "f"]);
contrib.git(&clone, &["commit", "-q", "-m", "no f"]);
contrib.git(&clone, &["push", "-q", "origin", "bare"]);
let bare = pull("bare");
let record = serde_json::json!({"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key}});
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.pull",
"p",
Some(record.clone()),
);
check(&bare)["finishedAt"].is_string()
});
assert_eq!(check(&bare)["status"], "failure");
// The owner's checks of it, by the runner the repository names.
let me = runner.account.handle.clone();
assert_eq!(
owner
.cli(&["runners", &owner.url("proj"), "add", &me], b"")
.0,
0
);
let (code, out) = owner.cli(&["checks", &owner.url("proj"), "--pull", &bare], b"");
assert_eq!(code, 0, "{out}");
assert!(
out.starts_with(&format!("failure\tpulls\t{bare}\t")),
"{out}"
);
// Asked to, it runs a pull's head again.
let finished = check(&bare)["finishedAt"].clone();
let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &bare], b"");
assert_eq!(code, 0, "{asked}");
let asked = asked.rsplit('/').next().unwrap().to_string();
wait_until(|| {
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
// Read once: read twice, the first could be the last run's end and
// the second this run's start, whose finishedAt is null.
let now = check(&bare)["finishedAt"].clone();
now.is_string() && now != finished
});
// Noted again once it has run.
wait_until(|| {
std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&bare))
});
// Asked for one of its workflows by name, it runs that one where it is.
let finished = check(&bare)["finishedAt"].clone();
let url = owner.url("proj");
let args = ["rerun", &url, "--pull", &bare, "--workflow", "pulls"];
let (code, asked) = owner.cli(&args, b"");
assert_eq!(code, 0, "{asked}");
let asked = asked.rsplit('/').next().unwrap().to_string();
wait_until(|| {
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
let now = check(&bare)["finishedAt"].clone();
now.is_string() && now != finished
});
let args = [
"rerun",
&url,
"--pull",
&bare,
"--workflow",
"pulls",
"-e",
"A=1",
];
// Variables go only to a run at a ref.
assert_eq!(owner.cli(&args, b"").0, 1);
// Closed, it drops out. Reopened where it ran, rung by its state
// record alone, it is noted, not run again: its checks of that head
// and base are there. Reopened at a head that moved while it was
// closed, it runs.
let noted =
|| std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&bare));
let set_state = |cmd: &str, noted_after: bool| {
let (code, out) = contrib.cli(&[cmd, &owner.url("proj"), &bare], b"");
assert_eq!(code, 0, "{out}");
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.state",
"s",
Some(record.clone()),
);
noted() == noted_after
});
};
wait_until(|| check(&bare)["finishedAt"].is_string());
let finished = check(&bare)["finishedAt"].clone();
set_state("close", false);
set_state("reopen", true);
std::thread::sleep(Duration::from_millis(500));
assert_eq!(check(&bare)["finishedAt"], finished);
set_state("close", false);
let reopened = contrib.commit_file(&clone, "r", "r\n");
contrib.git(&clone, &["push", "-q", "origin", "bare"]);
set_state("reopen", true);
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.state",
"s",
Some(record.clone()),
);
let c = check(&bare);
c["commit"] == reopened.as_str() && c["finishedAt"].is_string()
});
// A head branch deleted has nothing to run, and drops out. Any run
// still going ends first, so what follows is only what the deletion
// does.
wait_until(|| check(&bare)["finishedAt"].is_string());
let finished = check(&bare)["finishedAt"].clone();
contrib.git(&clone, &["push", "-q", "origin", ":bare"]);
wait_until(|| {
js.ring(
&contrib.account.did,
&ref_rkey(&fork_key, "refs/heads/bare"),
);
std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| !s.contains(&bare))
});
assert_eq!(check(&bare)["finishedAt"], finished);
// Asked to run it now, with no branch to run, it lets the request go
// rather than keep it for a pull that cannot run.
let (code, gone) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &bare], b"");
assert_eq!(code, 0, "{gone}");
let gone = gone.rsplit('/').next().unwrap().to_string();
wait_until(|| {
js.send(&owner.account.did, "dev.knotless.rerun", &gone, None);
std::fs::read_to_string(runner.path("state.json"))
.is_ok_and(|s| s.contains(&gone) && !s.contains(&bare))
});
assert_eq!(check(&bare)["finishedAt"], finished);
// Anyone else's pull is noted but not run, its code unvouched for,
// until the owner asks; then it runs where it is.
collaborators(serde_json::json!([]));
contrib.git(&clone, &["fetch", "-q", &owner.url("proj"), "main"]);
contrib.git(&clone, &["checkout", "-q", "-b", "late", "FETCH_HEAD"]);
contrib.commit_file(&clone, "l", "late\n");
contrib.git(&clone, &["push", "-q", "origin", "late"]);
let late = pull("late");
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.pull",
"p",
Some(record.clone()),
);
std::fs::read_to_string(runner.path("state.json")).is_ok_and(|s| s.contains(&late))
});
assert!(check(&late).is_null());
let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &late], b"");
assert_eq!(code, 0, "{asked}");
let asked = asked.rsplit('/').next().unwrap().to_string();
wait_until(|| {
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
check(&late)["finishedAt"].is_string()
});
assert_eq!(check(&late)["status"], "success");
// Its workflow asks for secrets, which a push's run of it may have left
// in its cache: anyone else's pull starts from a cache of its own,
// which nothing writes.
assert_eq!(held(&caches.join("bases")).len(), 2);
assert!(bases().iter().all(|b| held(b).is_empty()));
// Nor is what anyone else's pull wrote ever kept for main to take.
assert!(held(&caches.join("kept")).is_empty());
// An agent the repository names runs its pulls by opening them, as a
// collaborator does, and even one that is a collaborator too runs as
// code no one vouches for: what it writes is never kept, even by a
// workflow asking for no secrets, whose trusted pass would be.
let agents = |named: serde_json::Value| {
let mut record = owner.record("dev.knotless.repo", "proj");
record["agents"] = named;
owner.put_record("dev.knotless.repo", "proj", record);
};
agents(serde_json::json!([{"did": contrib.account.did}]));
let bot_yml = format!(
"on: {{pull: {{base: [refs/heads/main]}}}}\ncache: [target]\nsteps:\n - run: for u in {}/runs/*/*/upper; do echo bot > $u/bot; done\n",
caches.display()
);
for (branch, collaborating) in [("bot", false), ("bot-too", true)] {
let named = if collaborating {
serde_json::json!([{"did": contrib.account.did}])
} else {
serde_json::json!([])
};
collaborators(named);
contrib.git(&clone, &["checkout", "-q", "-b", branch, "late"]);
std::fs::write(clone.join(".knotless/workflows/bot.yml"), &bot_yml).unwrap();
contrib.git(&clone, &["add", "."]);
contrib.git(&clone, &["commit", "-q", "-m", branch]);
contrib.git(&clone, &["push", "-q", "origin", branch]);
let bot = pull(branch);
let ran = |wf: &str| check_on(&runner, &owner, &key, &bot, wf);
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.pull",
"p",
Some(record.clone()),
);
["pulls", "bot"]
.iter()
.all(|wf| ran(wf)["finishedAt"].is_string())
});
for wf in ["pulls", "bot"] {
assert_eq!(ran(wf)["status"], "success", "{branch} {wf}");
}
wait_until(|| held(&caches.join("runs")).is_empty());
assert!(bases().iter().all(|b| held(b).is_empty()), "{branch}");
assert!(held(&caches.join("kept")).is_empty(), "{branch}");
}
agents(serde_json::json!([]));
collaborators(serde_json::json!([]));
contrib.git(&clone, &["checkout", "-q", "late"]);
// A collaborator's pull whose lock adds shared steps from an account the
// repository does not vouch for runs as code no one vouches for, and
// says so; the owner's own add nothing.
collaborators(serde_json::json!([{"did": contrib.account.did}]));
contrib.git(&clone, &["checkout", "-q", "-b", "locked", "late"]);
let (mine, theirs) = (&owner.account.handle, &runner.account.handle);
let commit = "a".repeat(40);
let lock = format!(
"{mine}/steps/a@1: {{version: 1.0.0, commit: {commit}}}\n{theirs}/steps/b@1: {{version: 1.0.0, commit: {commit}}}\n"
);
std::fs::write(clone.join(".knotless/uses.lock"), lock).unwrap();
contrib.git(&clone, &["add", "."]);
contrib.git(&clone, &["commit", "-q", "-m", "locked"]);
contrib.git(&clone, &["push", "-q", "origin", "locked"]);
let locked = pull("locked");
wait_until(|| {
js.send(
&contrib.account.did,
"dev.knotless.pull",
"p",
Some(record.clone()),
);
check(&locked)["finishedAt"].is_string()
});
let ran = check(&locked);
assert_eq!(ran["status"], "success", "{ran}");
let log = blob(
&runner.account.did,
ran["log"]["ref"]["$link"].as_str().unwrap(),
);
let said = format!(
"==> runs as code no one vouches for, as its lock adds shared steps from {theirs}\n"
);
assert!(log.contains(&said), "{log}");
collaborators(serde_json::json!([]));
contrib.git(&clone, &["checkout", "-q", "late"]);
// A pull's run of a head and base it is no longer at is cancelled, as
// when its base moves on, which runs anyone else's pull no further.
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
for (wf, extra) in [("hold", ""), ("keep", "superseded: finish\n")] {
let text = format!(
"{extra}on: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n - name: hold\n run: |\n echo $$ > {go}/pid-{wf}\n touch {go}/started-{wf}\n until [ -f {go}/{wf} ]; do sleep 0.1; done\n",
go = go.display()
);
std::fs::write(clone.join(format!(".knotless/workflows/{wf}.yml")), text).unwrap();
}
contrib.git(&clone, &["add", "."]);
contrib.git(&clone, &["commit", "-q", "-m", "hold"]);
contrib.git(&clone, &["push", "-q", "origin", "late"]);
let (code, asked) = owner.cli(&["rerun", &owner.url("proj"), "--pull", &late], b"");
assert_eq!(code, 0, "{asked}");
let asked = asked.rsplit('/').next().unwrap().to_string();
wait_until(|| {
js.send(&owner.account.did, "dev.knotless.rerun", &asked, None);
["hold", "keep"]
.iter()
.all(|wf| go.join(format!("started-{wf}")).exists())
});
owner.commit_file(&work, "m", "m\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
let of = |wf: &str| check_on(&runner, &owner, &key, &late, wf);
// Both, even one whose runs finish, which a newer commit's run never
// cuts off.
wait_until(|| {
js.ring(&owner.account.did, &main_ref);
["hold", "keep"]
.iter()
.all(|wf| of(wf)["finishedAt"].is_string())
});
for wf in ["hold", "keep"] {
let moved = of(wf);
assert_eq!(moved["status"], "cancelled", "{moved}");
let log = blob(
&runner.account.did,
moved["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.ends_with("==> cancelled as the pull is no longer at that head and base\n"),
"{log}"
);
// Its steps are killed, not left to run on.
let pid = std::fs::read_to_string(go.join(format!("pid-{wf}"))).unwrap();
assert!(!alive(pid.trim()), "{wf} still runs");
}
}
/// A pull merged as it was checked leaves its base at the tree its run
/// passed at: the push that follows files those checks as passed, naming
/// the pull's run, rather than running them again. A tree no pull passed
/// at runs, as does a deploy, which runs for what it does, and one that
/// asks for secrets, which no pull's run gets.
#[test]
fn a_merged_pull_passes_its_push() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
// At a did:web, for the service tokens an owner's secrets are left
// and read with.
let did = runner.account.did.clone();
let (base, _ears) = serving(&rt, &Sandbox::new(), Some(&did));
owner.set("KNOTLESS_APPVIEW", &base);
owner.set("KNOTLESS_DID_WEB_SCHEME", "http");
runner.set("KNOTLESS_DID_WEB_SCHEME", "http");
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
runner.set("KNOTLESS_APPVIEW", &base);
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let work = owner.work("proj", 1);
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
// Each run notes the commit it is of; ci fails while `break` is there,
// and writes it to its cache, which each run's layer is under `runs`.
let both = "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/main]}}";
let caches = runner.path("caches");
let warm = format!(
" - name: warm\n run: for u in {}/runs/*/*/upper; do echo \"$KNOTLESS_RUN_SHA\" > $u/sha; done\n",
caches.display()
);
for (wf, extra) in [
("ci", format!("{both}\ncache: [target]\n")),
("secret", format!("{both}\nsecrets: [TOKEN]\n")),
(
"deploy",
"on: {push: {refs: [refs/heads/main]}}\nsuperseded: finish\n".to_string(),
),
] {
let text = format!(
"{extra}steps:\n - name: note\n run: |\n echo \"$KNOTLESS_RUN_SHA\" >> {go}/{wf}\n test ! -f {go}/break\n",
go = go.display()
);
let text = match wf {
"ci" => text + &warm,
"secret" => text + " - name: token\n run: echo \"token=$TOKEN\"\n",
_ => text,
};
std::fs::write(dir.join(format!("{wf}.yml")), text).unwrap();
}
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let url = owner.url("proj");
assert_eq!(owner.cli(&["protect", &url, "main"], b"").0, 0);
assert_eq!(owner.cli(&["runners", &url, "add", &handle], b"").0, 0);
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![format!("{}/proj", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: None,
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 4,
pull_jobs: 4,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: true,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let (run, watches) = rt
.block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
.unwrap();
let watch = &watches[0];
let key = watch.key.clone();
let pass = || {
rt.block_on(run.reconcile(watch)).unwrap();
rt.block_on(run.idle());
};
pass();
// Its owner seals it a secret, which only a push to main reaches, and
// another its workflows do not ask for.
for (name, value) in [("TOKEN", "t0ken\n"), ("UNASKED", "u")] {
let set = ["secret", &url, "set", name, "--ref", "refs/heads/main"];
let (code, out) = owner.cli(&set, value.as_bytes());
assert_eq!(code, 0, "{out}");
}
let ran = |wf: &str| std::fs::read_to_string(go.join(wf)).unwrap_or_default();
let at = |sha: &str, wf: &str| {
let rkey = check::rkey(&owner.account.did, &key, "refs/heads/main", sha, None, wf);
runner.record(check::NSID, &rkey)
};
let log_of = |check: &serde_json::Value| {
blob(
&runner.account.did,
check["log"]["ref"]["$link"].as_str().unwrap(),
)
};
let lists = |uri: &str| {
let (_, out) = owner.cli(&["list", &url, "--pulls"], b"");
out.contains(uri.rsplit('/').next().unwrap())
};
// A pull from a branch of main as it is now, run once it is listed.
let open = |branch: &str, file: &str| {
owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
let head = owner.commit_file(&work, file, &format!("{file}\n"));
owner.git(&work, &["push", "-q", "origin", branch]);
let (code, uri) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
assert_eq!(code, 0, "{uri}");
wait_until(|| lists(&uri));
pass();
(uri, head)
};
// Main as a merge left it, checked out to work on.
let merged = |args: &[&str]| {
let (code, out) = owner.cli(args, b"");
assert_eq!(code, 0, "{out}");
owner.git(&work, &["fetch", "-q", "origin", "main"]);
owner.git(&work, &["checkout", "-q", "-B", "main", "FETCH_HEAD"]);
let sha = owner.git(&work, &["rev-parse", "HEAD"]);
pass();
sha
};
// The runs whose layers are kept, and what ci's cache now starts from.
let held = |dir: std::path::PathBuf| {
let all = std::fs::read_dir(dir).into_iter().flatten();
let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
all.sort();
all
};
let kept = || held(caches.join("kept"));
let base_sha = || {
let cache = held(caches.join("bases")).pop().unwrap();
let newest = held(cache).into_iter().max_by_key(|g| {
let name = g.file_name().unwrap().to_string_lossy().into_owned();
name.parse::<u64>().unwrap()
});
std::fs::read_to_string(newest.unwrap().join("sha")).unwrap()
};
let (topic, head) = open("topic", "a");
let pulled = check_on(&runner, &owner, &key, &topic, "ci");
assert_eq!(pulled["status"], "success", "{pulled}");
assert_eq!(ran("ci"), format!("{head}\n"));
// A trusted pull's run that passed keeps what it wrote to its cache.
assert_eq!(kept().len(), 1);
// A pull's run of what asks for secrets gets none, and says so.
let secret = log_of(&check_on(&runner, &owner, &key, &topic, "secret"));
let none = "==> TOKEN is not granted to a pull's or a merge-queue batch's run, which gets no secrets\n";
assert!(secret.contains(none), "{secret}");
// Squashed into main: its tree is the one the pull ran at.
let squashed = merged(&["merge", &url, &topic]);
assert_ne!(squashed, head);
let ci = at(&squashed, "ci");
assert_eq!(ci["status"], "success", "{ci}");
assert_eq!(ci["commit"], squashed.as_str());
assert!(ci.get("pull").is_none(), "{ci}");
assert_eq!(ci["steps"], pulled["steps"]);
let log = log_of(&ci);
assert!(log.starts_with("==> passed: the same tree, "), "{log}");
assert!(
log.contains(&format!("as pull {topic} at {head} merged with ")),
"{log}"
);
assert!(log.contains("; not run again\n"), "{log}");
assert_eq!(ran("ci"), format!("{head}\n"), "not run again");
// And what the pull's run wrote to its cache is what main's runs start
// from now, as main's own run's would be.
let said = "==> caches: what that run wrote to /work/target is kept, as this run's would be\n";
assert!(log.contains(said), "{log}");
// Then the pull's log, so main's check stands alone once the pull's is
// pruned.
let theirs = log_of(&pulled);
assert!(log.ends_with(&format!("{said}{theirs}")), "{log}");
assert_eq!(base_sha(), format!("{head}\n"));
assert!(kept().is_empty());
// What a deploy's wait for the commit's checks reads.
let (code, out) = owner.cli(
&[
"checks",
&url,
"--ref",
"refs/heads/main",
"--commit",
&squashed,
"--workflow",
"ci",
],
b"",
);
assert_eq!(code, 0, "{out}");
assert!(out.starts_with("success\tci\trefs/heads/main\t"), "{out}");
// The deploy runs, and so does what asks for secrets.
for wf in ["deploy", "secret"] {
let pushed = at(&squashed, wf);
assert_eq!(pushed["status"], "success", "{pushed}");
assert!(ran(wf).contains(&squashed), "{wf}: {}", ran(wf));
assert!(!log_of(&pushed).contains("not run again"));
}
// The push's run opens the secret its owner sealed to the runner, and
// masks it in the log.
let log = log_of(&at(&squashed, "secret"));
assert!(log.contains("token=***\n"), "{log}");
assert!(
!log.contains("t0ken") && !log.contains("not granted"),
"{log}"
);
// A tree no pull ran at runs.
let direct = owner.commit_file(&work, "b", "b\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
pass();
assert_eq!(at(&direct, "ci")["status"], "success");
assert!(ran("ci").contains(&direct), "{}", ran("ci"));
assert_eq!(base_sha(), format!("{direct}\n"));
// A pull that moves on drops what its run there kept, for its run
// where it is now.
let (moving, _) = open("moving", "d");
let first = kept();
assert_eq!(first.len(), 1);
let moved = owner.commit_file(&work, "e", "e\n");
owner.git(&work, &["push", "-q", "origin", "moving"]);
pass();
let pulled = check_on(&runner, &owner, &key, &moving, "ci");
assert_eq!(pulled["commit"], moved.as_str(), "{pulled}");
let then = kept();
assert_eq!(then.len(), 1);
assert_ne!(then, first);
// Main run again meanwhile moves its cache's base on, and what the
// pull's run wrote over the one before is not taken when it merges.
let again = ["rerun", &url, "--ref", "refs/heads/main"];
assert_eq!(owner.cli(&again, b"").0, 0);
pass();
assert_eq!(ran("ci").matches(&direct).count(), 2, "{}", ran("ci"));
let taken = merged(&["merge", &url, &moving]);
let log = log_of(&at(&taken, "ci"));
assert!(log.contains("; not run again\n"), "{log}");
let said = "==> caches: what that run wrote to /work/target is not kept: the base it was written over has moved on since, and the next run here warms it\n";
// Before the pull's own log, which it copies.
assert!(log.contains(said), "{log}");
assert_eq!(base_sha(), format!("{direct}\n"));
assert!(kept().is_empty());
// A pull whose run passed and then failed when run again, at the same
// head and base, is not reused: its check says what it last did.
let (late, _) = open("late", "c");
assert_eq!(
check_on(&runner, &owner, &key, &late, "ci")["status"],
"success"
);
std::fs::write(go.join("break"), "").unwrap();
assert_eq!(owner.cli(&["rerun", &url, "--pull", &late], b"").0, 0);
pass();
let failed = check_on(&runner, &owner, &key, &late, "ci");
assert_eq!(failed["status"], "failure", "{failed}");
std::fs::remove_file(go.join("break")).unwrap();
let forced = merged(&["merge", &url, &late, "--force"]);
let ci = at(&forced, "ci");
assert_eq!(ci["status"], "success", "{ci}");
assert!(ran("ci").contains(&forced), "{}", ran("ci"));
assert!(!log_of(&ci).contains("not run again"));
}
#[test]
fn pulls_warm_where_pushes_run_elsewhere() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let base = appview(&rt, &Sandbox::new());
owner.set("KNOTLESS_APPVIEW", &base);
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
runner.set("KNOTLESS_APPVIEW", &base);
// ci notes the commit it is of in its cache, whose layers are each
// run's under `runs`.
let work = owner.work("proj", 1);
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let caches = runner.path("caches");
let ci = format!(
"on: {{push: {{refs: [refs/heads/main]}}, pull: {{base: [refs/heads/main]}}}}\ncache: [target]\nsteps:\n - run: for u in {}/runs/*/*/upper; do echo \"$KNOTLESS_RUN_SHA\" > $u/sha; done\n",
caches.display()
);
std::fs::write(dir.join("ci.yml"), ci).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let url = owner.url("proj");
// Its pulls alone: another runner runs its pushes.
let repos = vec![format!("{mine}/proj")];
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
pulls_only: repos.clone(),
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
caches: Some(caches.clone()),
pulls: true,
keep_checks: None,
..knotless_run::Options::new(repos)
};
let (run, watches) = rt
.block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
.unwrap();
let watch = &watches[0];
let key = watch.key.clone();
let pass = || {
rt.block_on(run.reconcile(watch)).unwrap();
rt.block_on(run.idle());
};
pass();
let lists = |uri: &str| {
let (_, out) = owner.cli(&["list", &url, "--pulls"], b"");
out.contains(uri.rsplit('/').next().unwrap())
};
let open = |branch: &str, file: &str| {
owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
let head = owner.commit_file(&work, file, &format!("{file}\n"));
owner.git(&work, &["push", "-q", "origin", branch]);
let (code, uri) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
assert_eq!(code, 0, "{uri}");
wait_until(|| lists(&uri));
pass();
(uri, head)
};
let held = |dir: std::path::PathBuf| {
let all = std::fs::read_dir(dir).into_iter().flatten();
let mut all: Vec<_> = all.map(|e| e.unwrap().path()).collect();
all.sort();
all
};
let base_sha = || {
let cache = held(caches.join("bases")).pop().unwrap();
let newest = held(cache).into_iter().max_by_key(|g| {
let name = g.file_name().unwrap().to_string_lossy().into_owned();
name.parse::<u64>().unwrap()
});
std::fs::read_to_string(newest.unwrap().join("sha")).unwrap()
};
// A trusted pull's run that passes is what the next runs start from,
// as a push's would be where pushes run: none here takes kept layers.
let (topic, head) = open("topic", "a");
let pulled = check_on(&runner, &owner, &key, &topic, "ci");
assert_eq!(pulled["status"], "success", "{pulled}");
let log = blob(
&runner.account.did,
pulled["log"]["ref"]["$link"].as_str().unwrap(),
);
let said =
"==> caches: what this run wrote to /work/target is what the runs here start from now\n";
assert!(log.contains(said), "{log}");
assert_eq!(base_sha(), format!("{head}\n"));
assert!(held(caches.join("kept")).is_empty());
// The next pull goes on from it.
let (_, other) = open("other", "b");
assert_eq!(base_sha(), format!("{other}\n"));
// A push to main runs nothing here, nor the pulls it moves the base
// of, as the queue checks them against it.
owner.git(&work, &["checkout", "-q", "main"]);
let tip = owner.commit_file(&work, "c", "c\n");
owner.git(&work, &["push", "-q", "origin", "main"]);
pass();
assert!(check_of(&runner, &owner, &key, "ci").is_null());
assert_eq!(check_on(&runner, &owner, &key, &topic, "ci"), pulled);
// Its head moving runs it, merged with main as it is now, and goes on
// from the base as it was.
owner.git(&work, &["checkout", "-q", "topic"]);
let moved = owner.commit_file(&work, "a2", "a2\n");
owner.git(&work, &["push", "-q", "origin", "topic"]);
pass();
let again = check_on(&runner, &owner, &key, &topic, "ci");
assert_eq!(again["baseCommit"], tip.as_str(), "{again}");
let now = base_sha();
assert!(
[moved, other].iter().any(|s| now == format!("{s}\n")),
"{now}"
);
}
/// Runs on every push to main, and on a pull only when it changes `web`.
const WEB: &str = r#"
on:
push: {refs: [refs/heads/main]}
pull: {base: [refs/heads/main], paths: [web/**, .knotless/workflows/web.yml]}
steps:
- name: build
run: test -d web
- name: check
run: 'true'
"#;
#[test]
fn paths_leave_out_what_a_change_misses() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let base = appview(&rt, &Sandbox::new());
owner.set("KNOTLESS_APPVIEW", &base);
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
runner.set("KNOTLESS_APPVIEW", &base);
let work = owner.work("proj", 1);
for dir in [".knotless/workflows", "web", "crates"] {
std::fs::create_dir_all(work.join(dir)).unwrap();
}
for (file, text) in [
("web.yml", WEB),
// Every pull, whatever it changes, as check and test do.
(
"all.yml",
"on: {pull: {base: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n",
),
// A push to any branch that changes `web`, or one new.
(
"branches.yml",
"on: {push: {paths: [web/**]}}\nsteps: [{name: look, run: 'true'}]\n",
),
] {
std::fs::write(work.join(".knotless/workflows").join(file), text).unwrap();
}
std::fs::write(work.join("web/page.svelte"), "page\n").unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
let url = owner.url("proj");
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let main_ref = ref_rkey(&key, "refs/heads/main");
let js = jetstream(&rt);
let mut opts = runner_opts(&runner, &owner, &js);
opts.pulls = true;
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
// Watching starts where the refs and pulls are.
let id = format!("{}/{key}", owner.account.did);
wait_until(|| {
let state = std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
let state: serde_json::Value = serde_json::from_str(&state).unwrap_or_default();
state["pulls"][&id].is_object()
});
let record = serde_json::json!({"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key}});
// A branch pushed, and a pull of it opened, each rung for.
let open = |branch: &str, file: &str| {
owner.git(&work, &["checkout", "-q", "-b", branch, "main"]);
let sha = owner.commit_file(&work, file, &format!("{branch}\n"));
owner.git(&work, &["push", "-q", "origin", branch]);
let (code, pull) = owner.cli(&["pull", &url, "main", &url, branch, branch], b"");
assert_eq!(code, 0, "{pull}");
let pull = pull.trim().to_string();
let ran = |wf: &str| check_on(&runner, &owner, &key, &pull, wf)["finishedAt"].is_string();
wait_until(|| {
js.ring(
&owner.account.did,
&ref_rkey(&key, &format!("refs/heads/{branch}")),
);
js.send(
&owner.account.did,
"dev.knotless.pull",
"p",
Some(record.clone()),
);
ran("web") && ran("all")
});
(pull, sha)
};
// A pull that changes only Rust passes `web` without running it, saying
// why, its steps each skipped; what names no paths runs.
let (rust, rust_sha) = open("rust", "crates/a.rs");
let web = check_on(&runner, &owner, &key, &rust, "web");
assert_eq!(web["status"], "success", "{web}");
assert_eq!(web["commit"], rust_sha.as_str());
assert_eq!(
web["skipped"],
"skipped: nothing changed in web/**, .knotless/workflows/web.yml"
);
let steps: Vec<_> = web["steps"].as_array().unwrap().iter().collect();
assert_eq!(steps.len(), 2, "{web}");
assert!(steps.iter().all(|s| s["status"] == "skipped"), "{web}");
assert!(web["log"].is_null(), "{web}");
let all = check_on(&runner, &owner, &key, &rust, "all");
assert_eq!(all["status"], "success", "{all}");
assert!(all["skipped"].is_null() && all["log"].is_object(), "{all}");
// A branch new to the runner is all change: a push filtered runs on it.
let branch = check_on(&runner, &owner, &key, "refs/heads/rust", "branches");
assert!(
branch["skipped"].is_null() && branch["finishedAt"].is_string(),
"{branch}"
);
// One that changes `web` runs it.
let (pages, _) = open("pages", "web/other.svelte");
let web = check_on(&runner, &owner, &key, &pages, "web");
assert_eq!(web["status"], "success", "{web}");
assert!(web["skipped"].is_null() && web["log"].is_object(), "{web}");
assert_eq!(web["steps"][0]["status"], "success", "{web}");
// The merge gate takes one left out as passed, and the CLI says why.
let me = runner.account.handle.clone();
assert_eq!(owner.cli(&["runners", &url, "add", &me], b"").0, 0);
let (code, out) = owner.cli(&["checks", &url, "--pull", &rust, "--workflow", "web"], b"");
assert_eq!(code, 0, "{out}");
assert!(out.starts_with(&format!("success\tweb\t{rust}\t")), "{out}");
assert!(
out.trim_end()
.ends_with("\tskipped: nothing changed in web/**, .knotless/workflows/web.yml"),
"{out}"
);
let (code, merged) = owner.cli(&["merge", &url, &rust], b"");
assert_eq!(code, 0, "{merged}");
let tip = merged.lines().next().unwrap().to_string();
// Main moving runs what a push to it runs, whatever it changed; a push
// filtered to `web` is left out, as only Rust changed since.
let of = |wf: &str| check_on(&runner, &owner, &key, "refs/heads/main", wf);
wait_until(|| {
js.ring(&owner.account.did, &main_ref);
["web", "branches"]
.iter()
.all(|wf| of(wf)["commit"] == tip.as_str() && of(wf)["finishedAt"].is_string())
});
let web = of("web");
assert!(web["skipped"].is_null() && web["log"].is_object(), "{web}");
assert_eq!(web["status"], "success", "{web}");
let branches = of("branches");
assert_eq!(branches["status"], "success", "{branches}");
assert_eq!(branches["skipped"], "skipped: nothing changed in web/**");
}
/// What runs on a pull into main, in a repository whose merge queue a
/// test lands: it fails one that brings `bad`, unless `good` too, and
/// otherwise leaves
/// `go/ran.<commit>` and holds until `go/go` is there.
fn held(go: &std::path::Path) -> String {
let go = go.display();
format!(
"on: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n - run: |\n test ! -f bad || test -f good\n touch {go}/ran.$KNOTLESS_RUN_SHA\n until [ -f {go}/go ]; do sleep 0.1; done\n",
)
}
/// A repository whose merge queue a runner lands as its owner, and the
/// owner, in a checkout of it, opening pulls and queueing them.
struct Line {
owner: Sandbox,
url: String,
work: std::path::PathBuf,
go: std::path::PathBuf,
/// The appview, and its following of the PDS.
base: String,
ears: tokio::task::JoinHandle<()>,
}
impl Line {
fn new(rt: &tokio::runtime::Runtime) -> Line {
Line::served(rt, None)
}
/// A [`Line`] whose appview is a gateway too if it names `runner`.
fn served(rt: &tokio::runtime::Runtime, runner: Option<&str>) -> Line {
let mut owner = Sandbox::new();
let (base, ears) = serving(rt, &Sandbox::new(), runner);
owner.set("KNOTLESS_APPVIEW", &base);
let handle = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &handle);
let url = owner.url("line");
let go = owner.path("go");
std::fs::create_dir_all(&go).unwrap();
std::fs::write(go.join("go"), "").unwrap();
let work = owner.work("line", 1);
let flows = work.join(".knotless/workflows");
std::fs::create_dir_all(&flows).unwrap();
std::fs::write(flows.join("ci.yml"), held(&go)).unwrap();
// Not a pull's, so no batch runs it.
std::fs::write(flows.join("push.yml"), "steps: [{run: 'true'}]\n").unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let line = Line {
owner,
url,
work,
go,
base,
ears,
};
line.ok(&["runners", &line.url, "add", &handle]);
line
}
fn ok(&self, args: &[&str]) -> String {
let (code, out) = self.owner.cli(args, b"");
assert_eq!(code, 0, "{args:?}: {out}");
out
}
/// Commits `file` on `branch`, off main as it is now unless `on`.
fn branch(&self, branch: &str, file: &str, text: &str, on: bool) {
let o = &self.owner;
o.git(&self.work, &["fetch", "-q", "origin"]);
let from = if on { branch } else { "origin/main" };
o.git(&self.work, &["checkout", "-q", "-B", branch, from]);
o.commit_file(&self.work, file, text);
o.git(&self.work, &["push", "-q", "-f", "origin", branch]);
}
/// A pull of a branch off main as it is now that writes `file`.
fn pull(&self, branch: &str, file: &str, text: &str) -> String {
self.branch(branch, file, text, false);
self.ok(&["pull", &self.url, "main", &self.url, branch, branch])
}
/// Queues pulls, and waits until the queue as read has them all.
fn queue(&self, pulls: &[&str]) {
for pull in pulls {
let out = self.ok(&["merge", &self.url, pull, "--queue"]);
assert!(out.starts_with("queued into main: "), "{out}");
}
wait_until(|| {
let line = self.ok(&["queue", &self.url]);
pulls.iter().all(|p| line.contains(&format!("\t{p}\t")))
});
}
fn main(&self) -> String {
let refs = self.ok(&["refs", &self.url]);
let main = refs.lines().find(|l| l.ends_with("\trefs/heads/main"));
main.unwrap().split('\t').next().unwrap().to_string()
}
/// The subjects of main's newest `n` commits, newest first.
fn log(&self, n: usize) -> Vec<String> {
let o = &self.owner;
o.git(&self.work, &["fetch", "-q", "origin"]);
let count = format!("-{n}");
let out = o.git(&self.work, &["log", &count, "--format=%s", "origin/main"]);
out.lines().map(str::to_string).collect()
}
/// Waits until `knotless show` of `pull` says `what`.
fn shows(&self, pull: &str, what: &str) -> String {
wait_until(|| self.ok(&["show", &self.url, pull]).contains(what));
self.ok(&["show", &self.url, pull])
}
/// The queue's batches' checks, newest first.
fn batches(&self) -> Vec<serde_json::Value> {
let out = self.ok(&["checks", &self.url, "--queue", "main", "--json"]);
serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
}
/// Holds the runs of what is queued from here on, forgetting which
/// ran before.
fn hold(&self) {
for ran in self.ran() {
std::fs::remove_file(self.go.join(format!("ran.{ran}"))).unwrap();
}
std::fs::remove_file(self.go.join("go")).unwrap();
}
/// The commits whose runs have started since the last hold.
fn ran(&self) -> Vec<String> {
let names = std::fs::read_dir(&self.go).unwrap();
let names = names.map(|e| e.unwrap().file_name().to_string_lossy().to_string());
names
.filter_map(|n| Some(n.strip_prefix("ran.")?.to_string()))
.collect()
}
/// The check of the newest batch on `base`.
fn batch_on(&self, base: &str) -> serde_json::Value {
let all = self.batches().into_iter();
let on = all.filter(|c| c["baseCommit"] == base);
let newest = on.max_by_key(|c| c["startedAt"].as_str().unwrap().to_string());
newest.unwrap()
}
fn release(&self) {
std::fs::write(self.go.join("go"), "").unwrap();
}
fn opts(&self, state: &str) -> knotless_run::Options {
let label = format!("{}/line", self.owner.account.handle);
let mut opts = knotless_run::Options::new(vec![label]);
opts.jetstream = "ws://127.0.0.1:9".into();
opts.state = Some(self.owner.path(state));
opts.work = Some(self.owner.path("work"));
opts.queue = true;
opts
}
/// A runner as the owner, taking batches of `batch` to `depth`.
fn runner(&self, rt: &tokio::runtime::Runtime, batch: usize, depth: usize) -> Lander {
self.lander(rt, "state.json", |opts| {
(opts.batch, opts.depth) = (batch, depth);
})
}
/// A runner as the owner, keeping its state in `state`, as `with`
/// sets it up.
fn lander(
&self,
rt: &tokio::runtime::Runtime,
state: &str,
with: impl FnOnce(&mut knotless_run::Options),
) -> Lander {
let mut opts = self.opts(state);
opts.pulls = true;
with(&mut opts);
let ctx = common::ctx(&self.owner);
let (run, mut watches) = rt.block_on(knotless_run::Runner::start(opts, ctx)).unwrap();
// Watching starts where the repository is.
rt.block_on(run.reconcile(&watches[0])).unwrap();
Lander {
run,
watch: watches.remove(0),
}
}
}
struct Lander {
run: Arc<knotless_run::Runner>,
watch: knotless_run::Watch,
}
impl Lander {
/// A pass over the repository, with what is going still going.
fn step(&self, rt: &tokio::runtime::Runtime) {
rt.block_on(self.run.reconcile(&self.watch)).unwrap();
}
/// A pass, once what is going has ended.
fn pass(&self, rt: &tokio::runtime::Runtime) {
rt.block_on(self.run.idle());
self.step(rt);
}
}
/// A runner of a repository's pulls alone leaves its queue, and its
/// pushes, to the runner that lands it, and runs pulls as that one would:
/// main moving runs no open pull again until its head moves, and a pull
/// queued runs only with its batch.
#[test]
fn a_runner_of_pulls_alone_leaves_the_queue_be() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let url = q.url.clone();
let l = q.lander(&rt, "pulls.json", |opts| {
opts.queue = false;
opts.pulls_only = opts.repos.clone();
});
let key = q.owner.record("dev.knotless.repo", "line")["key"]
.as_str()
.unwrap()
.to_string();
let ci = |pull: &str| check_on(&q.owner, &q.owner, &key, pull, "ci");
let lists = |pull: &str| {
let rkey = pull.rsplit('/').next().unwrap();
wait_until(|| q.ok(&["list", &url, "--pulls"]).contains(rkey));
};
let twice = || {
l.pass(&rt);
l.pass(&rt);
};
// A pull opened runs, at its head merged with main.
let start = q.main();
let a = q.pull("a", "a", "a\n").trim().to_string();
lists(&a);
twice();
let ran = ci(&a);
assert_eq!(ran["status"], "success", "{ran}");
assert_eq!(ran["baseCommit"], start.as_str());
// Main moving runs it no further, nor a push.
q.branch("main", "m", "m\n", false);
let moved = q.main();
twice();
assert_eq!(ci(&a), ran);
let push = check_on(&q.owner, &q.owner, &key, "refs/heads/main", "push");
assert!(push.is_null(), "{push}");
// Its head moving runs it, merged with main as it is now.
q.branch("a", "a2", "a2\n", true);
twice();
let again = ci(&a);
assert_eq!(again["status"], "success", "{again}");
assert_eq!(again["baseCommit"], moved.as_str());
assert_ne!(again["commit"], ran["commit"]);
// A pull queued is noted, and left to run with its batch.
let b = q.pull("b", "b", "b\n").trim().to_string();
q.queue(&[&b]);
twice();
let state = std::fs::read_to_string(q.owner.path("pulls.json")).unwrap();
assert!(state.contains(&b), "{state}");
assert!(ci(&b).is_null());
}
#[test]
fn a_merge_queue_lands_in_batches() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let url = q.url.clone();
let l = q.runner(&rt, 8, 1);
// A runner that may not push here leaves the queue be.
let mut other = Sandbox::new();
let theirs = other.account.handle.clone();
other.set("KNOTLESS_IDENTIFIER", &theirs);
let mut opts = q.opts("stranger.json");
opts.state = Some(other.path("state.json"));
let (stranger, watches) = rt
.block_on(knotless_run::Runner::start(opts, common::ctx(&other)))
.unwrap();
// Three pulls land together, each a commit of its own, in the order
// they were queued, once one run of the batch passes.
assert_eq!(q.ok(&["queue", &url]), "the merge queue is empty");
let start = q.main();
// Two of them name an issue to close, which landing them closes.
let issue = q.ok(&["issue", &url, "Needs a"]);
let a = q.pull("a", "a", "a\n");
q.branch("b", "b", "b\n", false);
let fixes = format!("Fixes {issue}");
let b = q.ok(&["pull", &url, "main", &url, "b", "b", "-m", &fixes]);
q.branch("c", "c", "c\n", false);
let c = q.ok(&["pull", &url, "main", &url, "c", "c", "-m", &fixes]);
q.queue(&[&a, &b, &c]);
rt.block_on(stranger.reconcile(&watches[0])).unwrap();
assert!(!rt.block_on(stranger.noticed(&watches[0])).unwrap());
l.pass(&rt);
assert_eq!(q.main(), start);
l.pass(&rt);
assert_eq!(q.log(4), ["c", "b", "a", "workflows"]);
let tip = q.main();
let ran = &q.batches()[0];
assert_eq!(ran["status"], "success", "{ran}");
assert_eq!(
(&ran["commit"], &ran["baseCommit"]),
(&tip.as_str().into(), &start.as_str().into())
);
let pulls: Vec<&str> = (ran["batch"].as_array().unwrap().iter())
.map(|b| b["pull"].as_str().unwrap())
.collect();
assert_eq!(pulls, [a.as_str(), b.as_str(), c.as_str()]);
assert_eq!(ran["batch"][2]["commit"], tip.as_str());
assert_eq!(ran["ref"], "refs/heads/main");
assert_eq!(q.batches().len(), 1, "push.yml is not a pull's");
for pull in [&a, &b, &c] {
q.shows(pull, "merged pull");
}
let closed = q.shows(&issue, "closed issue");
assert!(closed.contains(&format!("\nclosed by {b}")), "{closed}");
// Their branches went with them, and none ran alone: the batch is
// their run.
let refs = q.ok(&["refs", &url]);
assert!(!refs.contains("refs/heads/a"), "{refs}");
let key = q.owner.record("dev.knotless.repo", "line")["key"]
.as_str()
.unwrap()
.to_string();
assert!(check_on(&q.owner, &q.owner, &key, &a, "ci").is_null());
// One that conflicts with those ahead of it is ejected at once, naming
// where; one that breaks the branch is found by halving the batch.
let d = q.pull("d", "f", "d\n");
let e = q.pull("e", "f", "e\n");
let x = q.pull("x", "bad", "x\n");
let g = q.pull("g", "g", "g\n");
q.queue(&[&d, &e, &x, &g]);
l.pass(&rt);
let shown = q.shows(&e, "merge queue: ejected");
assert!(
shown.contains(&format!(
"it conflicts in f with {d} (d), queued ahead of it on main at "
)),
"{shown}"
);
assert!(shown.contains(". Merge that branch into it"), "{shown}");
// [d, x, g] fails, then [d, x]; [d] passes and lands, so x alone is
// what broke it, and is ejected without running again; then g lands.
for _ in 0..4 {
l.pass(&rt);
}
let shown = q.shows(&x, "merge queue: ejected");
assert!(shown.contains("Halving the batch found it"), "{shown}");
assert!(shown.contains("\nlogs: knotless checks "), "{shown}");
assert!(shown.contains(" --queue main --log"), "{shown}");
assert!(!shown.contains("Their logs"), "{shown}");
assert!(shown.contains(". Failed: ci (failure)"), "{shown}");
// It names the check that failed, as the runner filed it.
let listed = q.ok(&["queue", &url, "--json"]);
let listed: Vec<serde_json::Value> = serde_json::from_str(&listed).unwrap();
let ejected = listed.iter().find(|w| w["uri"] == x.as_str()).unwrap();
let failed = &ejected["queue"]["checks"][0];
assert_eq!(
(&failed["workflow"], &failed["status"]),
(&"ci".into(), &"failure".into())
);
let uris: Vec<serde_json::Value> = q.batches().iter().map(|c| c["uri"].clone()).collect();
assert!(uris.contains(&failed["check"]), "{failed} {uris:?}");
assert_eq!(q.log(3), ["g", "d", "c"]);
let line = q.ok(&["queue", &url]);
assert!(
line.contains(&format!("-\tmain\tejected\t{x}\tx\n\t")),
"{line}"
);
// One alone that fails is ejected at the head it was: here, as a
// workflow it brings does not load.
let y = q.pull("y", ".knotless/workflows/broken.yml", "steps: 3\n");
q.queue(&[&y]);
l.pass(&rt);
l.pass(&rt);
let shown = q.shows(&y, "merge queue: ejected");
assert!(shown.contains("its batch failed its checks at"), "{shown}");
assert!(shown.contains("broken (error)"), "{shown}");
// One whose branch goes once queued has nothing to land; queued again,
// it is turned away.
let gone = q.pull("gone", "gone", "gone\n");
q.queue(&[&gone]);
q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
l.pass(&rt);
q.shows(&gone, "has no branch gone");
let (queued, _, err) = q.owner.knotless(&["merge", &url, &gone, "--queue"], b"");
assert!(!queued && err.contains("has no branch gone"), "{err}");
// While its batch runs, the queue says so. The branch moving any
// other way drops the batch, which is built again on it.
q.hold();
let h = q.pull("h", "h", "h\n");
q.queue(&[&h]);
let first = q.main();
l.step(&rt);
wait_until(|| q.ok(&["queue", &url]).contains("1\tmain\tchecking at "));
q.owner
.git(&q.work, &["checkout", "-q", "-B", "main", "origin/main"]);
q.owner.commit_file(&q.work, "m", "m\n");
q.owner.git(&q.work, &["push", "-q", "origin", "main"]);
l.step(&rt);
q.release();
l.pass(&rt);
assert_eq!(q.log(2), ["h", "write m"]);
let batches = q.batches();
let dropped = batches
.iter()
.find(|c| c["baseCommit"] == first.as_str())
.unwrap();
assert_eq!(dropped["status"], "cancelled", "{dropped}");
let log = blob(
&q.owner.account.did,
dropped["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.ends_with("==> cancelled as the merge queue dropped its batch\n"),
"{log}"
);
// A pull whose head moves while its batch runs is not landed as it
// was, but built again as it is.
q.hold();
let k = q.pull("k", "k", "k\n");
q.queue(&[&k]);
l.step(&rt);
q.branch("k", "k2", "k2\n", true);
q.release();
l.pass(&rt);
l.pass(&rt);
assert_eq!(q.log(1), ["k"]);
q.owner.git(&q.work, &["checkout", "-q", "origin/main"]);
assert!(q.work.join("k2").exists());
// One taken out of the queue while its batch runs is not landed.
q.hold();
let out = q.pull("out", "out", "out\n");
q.queue(&[&out]);
l.step(&rt);
assert_eq!(
q.ok(&["queue", &url, "--leave", &out]),
"left the merge queue"
);
wait_until(|| !q.ok(&["queue", &url]).contains(&out));
l.step(&rt);
q.release();
l.pass(&rt);
assert_eq!(q.log(1), ["k"]);
q.shows(&out, "open pull");
// A queue into a branch that is gone waits.
let side = ["push", "-q", "origin", "origin/main:refs/heads/side"];
q.owner.git(&q.work, &side);
q.branch("sidework", "s", "s\n", false);
let into = q.ok(&["pull", &url, "side", &url, "sidework", "Into side"]);
let out = q.ok(&["merge", &url, &into, "--queue"]);
assert!(out.starts_with("queued into side: 1 of 1\n"), "{out}");
wait_until(|| q.ok(&["queue", &url]).contains(&into));
q.owner.git(&q.work, &["push", "-q", "origin", ":side"]);
l.pass(&rt);
q.shows(&into, "merge queue: queued");
}
/// A runner the repository names that may not push lands its queue
/// through the appview's gateway, which takes a batch from it only once the
/// owner has delegated writes there: until then, each batch it would land
/// is ejected, saying so, rather than checked and refused for ever.
#[test]
fn a_runner_named_lands_through_the_gateway() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut named = Sandbox::new();
let handle = named.account.handle.clone();
let q = Line::served(&rt, Some(&named.account.did));
named.set("KNOTLESS_IDENTIFIER", &handle);
named.set("KNOTLESS_APPVIEW", &q.base);
// The gateway's did:web names localhost, served over http.
named.set("KNOTLESS_DID_WEB_SCHEME", "http");
let url = q.url.clone();
q.ok(&["runners", &url, "add", &handle]);
q.ok(&["runners", &url, "remove", &q.owner.account.handle]);
let mut opts = q.opts("named.json");
opts.state = Some(named.path("state.json"));
opts.work = Some(named.path("work"));
opts.pulls = true;
let ctx = common::ctx(&named);
let (run, mut watches) = rt.block_on(knotless_run::Runner::start(opts, ctx)).unwrap();
let l = Lander {
run,
watch: watches.remove(0),
};
l.step(&rt);
let start = q.main();
let a = q.pull("a", "a", "a\n");
q.queue(&[&a]);
for _ in 0..3 {
l.pass(&rt);
}
assert_eq!(q.main(), start);
let shown = q.shows(&a, "merge queue: ejected");
assert!(
shown.contains("the gateway would not land its batch: "),
"{shown}"
);
assert!(shown.contains("has not delegated writes"), "{shown}");
assert!(
shown.contains("under Delegated writes in their settings"),
"{shown}"
);
}
#[test]
fn a_merge_queue_waits_for_preflights() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let url = q.url.clone();
// A pre-flight on main: it fails a pull that brings `nobuild`, and
// holds one that brings `slow` until `fly` is there.
let fly = q.go.join("fly");
let first = format!(
"queue: first\non: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n - run: |\n test ! -f nobuild\n until [ ! -f slow ] || [ -f {} ]; do sleep 0.1; done\n",
fly.display()
);
q.branch("main", ".knotless/workflows/first.yml", &first, false);
let l = q.runner(&rt, 8, 1);
// The newest run of a pull's pre-flight, by its status.
let first_of = |pull: &str| {
let args = ["checks", &url, "--pull", pull, "--workflow", "first"];
let out = q.ok(&[&args[..], &["--json"]].concat());
let all: Vec<serde_json::Value> = serde_json::from_str(&out).unwrap();
all.first().cloned().unwrap_or_default()
};
let in_batch = |pull: &str| {
(q.batches().iter())
.flat_map(|c| c["batch"].as_array().cloned().unwrap_or_default())
.any(|b| b["pull"] == pull)
};
// One that comes to conflict with main, one whose branch goes, one
// that does not build, one held, and one that passes, queued in that
// order: none has run. Queueing turns away the first two as they end
// up, so main moves and the branch goes once they are in.
let c = q.pull("c", "m", "c\n");
let gone = q.pull("gone", "gone", "gone\n");
let p = q.pull("p", "nobuild", "p\n");
let s = q.pull("s", "slow", "s\n");
let r = q.pull("r", "r", "r\n");
q.queue(&[&c, &gone, &p, &s, &r]);
q.branch("main", "m", "main\n", false);
q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
let start = q.main();
// The lander runs their pre-flights itself, as pulls' runs, and
// batches none yet. The one that conflicts it does not run, nor the
// one with no head: each is ejected as the batch is built, saying why.
l.step(&rt);
q.shows(&c, "it does not apply on main");
q.shows(&gone, "has no branch gone");
assert!(first_of(&c).is_null());
wait_until(|| {
let line = q.ok(&["queue", &url]);
line.contains(&format!("\tmain\tpre-flight first\t{s}\t"))
});
wait_until(|| first_of(&p)["status"] == "failure" && first_of(&r)["status"] == "success");
let ran = first_of(&r);
assert_eq!(ran["queue"], "first", "{ran}");
assert_eq!(ran["baseCommit"], start.as_str(), "{ran}");
assert!(q.batches().is_empty(), "no batch has run");
q.shows(&s, "Queued, 2nd of 3, waiting for its pre-flight: first");
// One whose pre-flight failed is ejected without a batch, naming it;
// one still running holds up none behind it, which goes on alone.
l.step(&rt);
let shown = q.shows(&p, "merge queue: ejected");
assert!(
shown.contains("its pre-flight failed at its head "),
"{shown}"
);
assert!(
shown.contains(&format!(
"\nlogs: knotless checks {} --pull {p} --log",
q.url
)),
"{shown}"
);
assert!(shown.contains(". Failed: first (failure)"), "{shown}");
assert!(in_batch(&r) && !in_batch(&s) && !in_batch(&p));
// Once it passes, it goes in too, after the one behind it landed.
std::fs::write(&fly, "").unwrap();
l.pass(&rt);
assert_eq!(q.log(2), ["r", "write m"]);
l.pass(&rt);
assert_eq!(q.log(3), ["s", "r", "write m"]);
assert!(!in_batch(&p));
}
/// Into a branch whose rule asks for reviews, a pull queued before it
/// did, and not reviewed as it asks, is ejected saying what it lacks; one
/// whose branch went is ejected as a batch is built.
#[test]
fn a_merge_queue_ejects_what_is_not_reviewed() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let url = q.url.clone();
let l = q.runner(&rt, 8, 1);
let a = q.pull("a", "a", "a\n");
let gone = q.pull("gone", "gone", "gone\n");
q.queue(&[&a, &gone]);
q.owner.git(&q.work, &["push", "-q", "origin", ":gone"]);
q.ok(&["protect", &url, "main", "--reviews", "1"]);
l.step(&rt);
let shown = q.shows(&a, "merge queue: ejected");
assert!(
shown.contains(&format!("{a} is not reviewed as main asks: reviews 0 of 1")),
"{shown}"
);
q.shows(&gone, "has no branch gone");
}
/// Another runner's pre-flight only queued, waiting for a job there, is
/// run by the lander too; and whichever starts second leaves it to the
/// other's, taking its own check back rather than filing it cancelled.
#[test]
fn a_preflight_only_queued_elsewhere_is_run_here_once() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let url = q.url.clone();
// A pre-flight that holds a pull that brings `slow` until `fly` is
// there, and leaves `flew.<sha>` once it runs.
let fly = q.go.join("fly");
let first = format!(
"queue: first\non: {{pull: {{base: [refs/heads/main]}}}}\nsteps:\n - run: |\n touch {go}/flew.$KNOTLESS_RUN_SHA\n until [ ! -f slow ] || [ -f {fly} ]; do sleep 0.1; done\n",
go = q.go.display(),
fly = fly.display()
);
q.branch("main", ".knotless/workflows/first.yml", &first, false);
// Another runner the repository names, whose checks are written here.
let other = Sandbox::new();
q.ok(&["runners", &url, "add", &other.account.handle]);
let l = q.lander(&rt, "state.json", |opts| opts.jobs = 1);
let key = q.owner.record("dev.knotless.repo", "line")["key"]
.as_str()
.unwrap()
.to_string();
let repo = format!("at://{}/dev.knotless.repo/line", q.owner.account.did);
// One held takes the one job.
let s = q.pull("s", "slow", "s\n");
q.queue(&[&s]);
l.step(&rt);
let flew = |sha: &str| q.go.join(format!("flew.{sha}")).exists();
let tip = q.main();
let s_head = q.owner.git(&q.work, &["rev-parse", "s"]);
wait_until(|| flew(&s_head));
// Another's run of the next one's is only queued: the lander runs it
// too, and it waits for the job.
let d = q.pull("d", "d", "d\n");
let head = q.owner.git(&q.work, &["rev-parse", "d"]);
let rkey = check::rkey(&q.owner.account.did, &key, &d, &head, Some(&tip), "first");
let theirs = |status: &str| {
let record = serde_json::json!({
"$type": check::NSID,
"repo": {"uri": repo, "key": key},
"ref": "refs/heads/d", "commit": head, "baseCommit": tip, "pull": d,
"queue": "first", "workflow": "first", "status": status,
"startedAt": "2026-01-01T00:00:00.000Z",
});
other.put_record(check::NSID, &rkey, record);
};
theirs("queued");
q.queue(&[&d]);
l.step(&rt);
let mine = || q.owner.record(check::NSID, &rkey);
wait_until(|| mine()["status"] == "queued");
q.shows(&d, "waiting for its pre-flight: first");
// The other's starts first: given the job, the lander's is not run,
// and its check goes, so the queue waits for the other's.
theirs("running");
std::fs::write(&fly, "").unwrap();
rt.block_on(l.run.idle());
assert!(mine().is_null(), "{}", mine());
assert!(!flew(&head));
l.step(&rt);
q.shows(&d, "waiting for its pre-flight: first");
// Once it passes, the pull goes into a batch.
theirs("success");
l.pass(&rt);
l.pass(&rt);
assert!(q.log(2).contains(&"d".to_string()), "{:?}", q.log(2));
}
#[test]
fn a_merge_queue_runs_ahead() {
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let l = q.runner(&rt, 2, 2);
// While a batch runs, the next runs on it: both have started before
// either lands. The second lands on the commit the first landed as.
let start = q.main();
q.hold();
let (n1, n2, n3) = (
q.pull("n1", "n1", "1\n"),
q.pull("n2", "n2", "2\n"),
q.pull("n3", "n3", "3\n"),
);
q.queue(&[&n1, &n2, &n3]);
l.step(&rt);
wait_until(|| q.ran().len() == 2);
assert_eq!(q.main(), start);
let first = q.batch_on(&start);
let ahead = q.batch_on(first["commit"].as_str().unwrap());
q.release();
l.pass(&rt);
assert_eq!(q.log(3), ["n3", "n2", "n1"]);
assert_eq!(q.batches().len(), 2);
let first = q.batch_on(&start);
let landed = first["commit"].as_str().unwrap();
let second = q.batch_on(landed);
assert_eq!(second["status"], "success", "{second}");
assert_eq!(second["startedAt"], ahead["startedAt"]);
assert_eq!(second["commit"], q.main().as_str());
// The measure of it: the second started before the first ended.
let (began, ended) = (second["startedAt"].as_str(), first["finishedAt"].as_str());
assert!(began < ended, "{second} {first}");
// Nothing lands but on the very commit its batch was built on: not on
// one the branch has moved on from, even under the batch's tip.
let stale = knotless::queue::Stack {
base: start.clone(),
stacked: Vec::new(),
conflicts: Vec::new(),
};
let url = knotless::url::RepoUrl::parse(&q.url).unwrap();
let git = knotless::git::Git::new(q.work.join(".git"));
let ctx = common::ctx(&q.owner);
let landing = knotless::queue::land(&ctx, &url, &git, "main", &stale);
let refused = rt.block_on(landing).unwrap_err().to_string();
let (at, base) = (&q.main()[..12], &start[..12]);
let said = format!("main is at {at}, not at {base}, where the batch was built");
assert_eq!(refused, said);
assert_eq!(q.log(1), ["n3"]);
// A pull leaving the batch being checked drops the one ahead on it
// too, cancelling its run, and what is left is built again.
q.hold();
let tip = q.main();
let (o1, o2, o3) = (
q.pull("o1", "o1", "1\n"),
q.pull("o2", "o2", "2\n"),
q.pull("o3", "o3", "3\n"),
);
q.queue(&[&o1, &o2, &o3]);
l.step(&rt);
wait_until(|| q.ran().len() == 2);
let ahead = q.batch_on(q.batch_on(&tip)["commit"].as_str().unwrap());
q.ok(&["queue", &q.url, "--leave", &o1]);
wait_until(|| !q.ok(&["queue", &q.url]).contains(&o1));
l.step(&rt);
q.release();
l.pass(&rt);
assert_eq!(q.log(2), ["o3", "o2"]);
q.shows(&o1, "open pull");
let dropped = q.batches().into_iter().find(|c| c["uri"] == ahead["uri"]);
let dropped = dropped.unwrap();
assert_eq!(dropped["status"], "cancelled", "{dropped}");
let did = &q.owner.account.did;
let log = blob(did, dropped["log"]["ref"]["$link"].as_str().unwrap());
assert!(
log.ends_with("cancelled as the merge queue dropped its batch\n"),
"{log}"
);
let rebuilt = q.batch_on(&tip);
assert_eq!(rebuilt["batch"].as_array().unwrap().len(), 2, "{rebuilt}");
// The branch moving any other way drops both, and they are built
// again on it.
q.hold();
let (r1, r2, r3) = (
q.pull("r1", "r1", "1\n"),
q.pull("r2", "r2", "2\n"),
q.pull("r3", "r3", "3\n"),
);
q.queue(&[&r1, &r2, &r3]);
l.step(&rt);
wait_until(|| q.ran().len() == 2);
q.owner
.git(&q.work, &["checkout", "-q", "-B", "main", "origin/main"]);
q.owner.commit_file(&q.work, "moved", "moved\n");
q.owner.git(&q.work, &["push", "-q", "origin", "main"]);
let moved = q.main();
l.step(&rt);
q.release();
l.pass(&rt);
assert_eq!(q.log(4), ["r3", "r2", "r1", "write moved"]);
let again = q.batch_on(&moved);
assert_eq!(again["status"], "success", "{again}");
let cancelled = q
.batches()
.into_iter()
.filter(|c| c["status"] == "cancelled");
assert_eq!(cancelled.count(), 4, "the two before, and these two");
// One that conflicts with a batch ahead waits for it, rather than be
// ejected for what may not land; once it has, it is.
let (n4, n5, n6) = (
q.pull("n4", "c", "4\n"),
q.pull("n5", "n5", "5\n"),
q.pull("n6", "c", "6\n"),
);
q.queue(&[&n4, &n5, &n6]);
l.pass(&rt);
q.shows(&n6, "merge queue: queued");
l.pass(&rt);
assert_eq!(q.log(2), ["n5", "n4"]);
q.shows(&n6, "merge queue: ejected");
// Halving goes no further once one suspected leaves: those left may
// all be sound.
let (m1, m2) = (q.pull("m1", "bad", "m1\n"), q.pull("m2", "m2", "m2\n"));
q.queue(&[&m1, &m2]);
l.pass(&rt);
l.pass(&rt);
q.ok(&["queue", &q.url, "--leave", &m1]);
wait_until(|| !q.ok(&["queue", &q.url]).contains(&m1));
l.pass(&rt);
l.pass(&rt);
assert_eq!(q.log(1), ["m2"]);
// A batch that fails drops the one on it, cancelling its run, and
// halving it does not run ahead. n9 brings good, so the batch ahead,
// with n7's bad, holds rather than fail too.
q.hold();
let tip = q.main();
let (n7, n8, n9) = (
q.pull("n7", "bad", "7\n"),
q.pull("n8", "n8", "8\n"),
q.pull("n9", "good", "9\n"),
);
q.queue(&[&n7, &n8, &n9]);
l.step(&rt);
wait_until(|| q.batch_on(&tip)["status"] == "failure" && q.ran().len() == 1);
let ahead = q.batch_on(q.batch_on(&tip)["commit"].as_str().unwrap());
l.step(&rt);
q.release();
l.pass(&rt);
let dropped = q.batches().into_iter().find(|c| c["uri"] == ahead["uri"]);
let dropped = dropped.unwrap();
assert_eq!(dropped["status"], "cancelled", "{dropped}");
l.pass(&rt);
q.shows(&n7, "merge queue: ejected");
l.pass(&rt);
assert_eq!(q.log(2), ["n9", "n8"]);
}
#[test]
fn a_merge_queue_runs_ahead_with_a_job_to_spare() {
// With one job, the batch ahead waits for the one it is on to let it
// go: it ranks as a pull's run, after the batch that lands next.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let l = q.lander(&rt, "state.json", |opts| {
(opts.batch, opts.jobs) = (1, 1);
});
let start = q.main();
q.hold();
let (j1, j2) = (q.pull("j1", "j1", "1\n"), q.pull("j2", "j2", "2\n"));
q.queue(&[&j1, &j2]);
l.step(&rt);
wait_until(|| q.ran().len() == 1);
let first = q.batch_on(&start);
let ahead = q.batch_on(first["commit"].as_str().unwrap());
assert_eq!(ahead["status"], "queued", "{ahead}");
std::thread::sleep(Duration::from_millis(500));
assert_eq!(q.ran(), [first["commit"].as_str().unwrap()]);
q.release();
l.pass(&rt);
assert_eq!(q.log(2), ["j2", "j1"]);
let second = q.batch_on(first["commit"].as_str().unwrap());
assert_eq!(second["uri"], ahead["uri"]);
assert_eq!(second["status"], "success", "{second}");
}
#[test]
fn a_batch_passing_as_the_runner_stops_lands() {
// A batch whose runs pass once the runner is stopping lands before it
// stops, rather than be checked again when it starts: and nothing
// else goes on, no batch after it started.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let l = q.runner(&rt, 1, 1);
q.hold();
let (s1, s2) = (q.pull("s1", "s1", "1\n"), q.pull("s2", "s2", "2\n"));
q.queue(&[&s1, &s2]);
l.step(&rt);
wait_until(|| q.ran().len() == 1);
let stopping = rt.spawn({
let run = l.run.clone();
async move { run.stop().await }
});
// Its stop begun, the batch's run passes.
std::thread::sleep(Duration::from_millis(300));
q.release();
promptly(&rt, stopping);
assert_eq!(q.log(2), ["s1", "workflows"]);
q.shows(&s1, "merged pull");
assert_eq!(q.batches().len(), 1);
}
#[test]
fn a_merge_queue_notices_what_was_missed() {
// A pull queued is noticed from its writer's repo, though neither
// Jetstream nor the appview hears of it, and lands; merged, it is
// not landed again by a runner starting afresh.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let z = q.pull("z", "z", "z\n");
wait_until(|| q.ok(&["list", &q.url]).contains(&z));
let l = q.lander(&rt, "state.json", |opts| {
opts.notice = Duration::from_millis(500);
});
q.ears.abort();
let door = Arc::new(tokio::sync::Notify::new());
let (run, watch) = (l.run.clone(), l.watch.clone());
rt.spawn(knotless_run::notice(run, watch, door.clone()));
let wait = Duration::from_secs(20);
let rung = || rt.block_on(async { tokio::time::timeout(wait, door.notified()).await });
// The first look rings, as nothing was seen before it.
rung().unwrap();
let out = q.ok(&["merge", &q.url, &z, "--queue"]);
assert!(out.starts_with("queued into main: 1 of 1"), "{out}");
rung().unwrap();
l.pass(&rt);
l.pass(&rt);
assert_eq!(q.log(1), ["z"]);
assert_eq!(q.ok(&["queue", &q.url]), "the merge queue is empty");
let tip = q.main();
let again = q.lander(&rt, "again.json", |_| ());
again.pass(&rt);
again.pass(&rt);
assert_eq!(q.main(), tip);
}
#[test]
fn a_merge_queue_goes_on_by_itself() {
// Queued, a pull lands with nobody waiting: each batch's runs ending
// ring the repository's door, and its queue goes on.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let z = q.pull("z", "z", "z\n");
q.queue(&[&z]);
// Each pass over it that runs past a few milliseconds says where it is.
let mut opts = q.opts("state.json");
opts.watchdog = Duration::from_millis(5);
rt.spawn(knotless_run::run(opts, common::ctx(&q.owner)));
q.shows(&z, "merged pull");
assert_eq!(q.log(1), ["z"]);
}
#[test]
fn a_merge_queue_goes_first() {
// Landing the queue, a pull not in it runs as its head moves, after
// the queue's runs, and not again each time the queue moves main.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let l = q.runner(&rt, 8, 1);
let key = q.owner.record("dev.knotless.repo", "line")["key"]
.as_str()
.unwrap()
.to_string();
let ran = |p: &str| check_on(&q.owner, &q.owner, &key, p, "ci");
let p = q.pull("p", "p", "p\n");
l.pass(&rt);
l.pass(&rt);
let first = ran(&p);
assert_eq!(first["status"], "success", "{first}");
// What it lands brings a workflow pushes to main run too: the push
// that lands the batch takes the batch's pass.
let both = "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n";
let z = q.pull("z", ".knotless/workflows/both.yml", both);
q.queue(&[&z]);
l.pass(&rt);
l.pass(&rt);
assert_eq!(q.log(1), ["z"]);
l.pass(&rt);
assert_eq!(ran(&p)["startedAt"], first["startedAt"]);
let pushed = check_on(&q.owner, &q.owner, &key, "refs/heads/main", "both");
assert_eq!(
(&pushed["status"], &pushed["commit"]),
(&"success".into(), &q.main().into())
);
let log = blob(
&q.owner.account.did,
pushed["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(log.contains("as the merge queue's batch at "), "{log}");
q.branch("p", "p2", "p2\n", true);
l.pass(&rt);
l.pass(&rt);
let again = ran(&p);
assert_ne!(again["commit"], first["commit"], "{again}");
assert_eq!(again["status"], "success", "{again}");
}
#[test]
fn a_workflow_asked_for_by_name() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("proj", 1);
let url = owner.url("proj");
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main"]);
assert_eq!(owner.cli(&["runners", &url, "add", &handle], b"").0, 0);
let label = format!("{mine}/proj");
let secrets = runner.path("secrets.json");
let granted = serde_json::json!({ label.clone(): {
"MAIN_ONLY": {"value": "m4inonly", "refs": ["refs/heads/main"]},
} });
std::fs::write(&secrets, granted.to_string()).unwrap();
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![label],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: Some(secrets),
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
// Shorter than a step of `costly`, which only a run asked for may
// take past it.
step_timeout: Duration::from_secs(2),
run_timeout: None,
jobs: 4,
pull_jobs: 2,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let start = || {
let started = knotless_run::Runner::start(opts.clone(), common::ctx(&runner));
let (run, mut watches) = rt.block_on(started).unwrap();
(run, watches.remove(0))
};
let (run, watch) = start();
let key = watch.key.clone();
rt.block_on(run.reconcile(&watch)).unwrap();
// `ci` runs on every push; `costly` and `wait` only when asked for at
// main, and `capped` on a push, its own timeout held to the runner's.
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let go = go.display();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
for (file, text) in [
(
"ci.yml",
format!("steps: [{{run: 'echo $KNOTLESS_RUN_REF >> {go}/ci'}}]\n"),
),
(
"costly.yml",
"on: {asked: {refs: [refs/heads/main]}}\nsecrets: [MAIN_ONLY]\nenv: {TRACKS: all, KEPT: 'yes'}\nsteps:\n - name: show\n run: echo \"tracks=$TRACKS kept=$KEPT ref=$KNOTLESS_RUN_REF main=$MAIN_ONLY\"\n - name: long\n timeout: 1\n run: sleep 3\n".into(),
),
(
"capped.yml",
"on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: sleep 3, timeout: 1}]\n".into(),
),
(
"wait.yml",
format!("on: {{asked: {{refs: [refs/heads/main]}}}}\nsteps:\n - run: |\n echo $TRACKS >> {go}/waits\n until [ -f {go}/$TRACKS ]; do sleep 0.1; done\n timeout: 1\n"),
),
] {
std::fs::write(dir.join(file), text).unwrap();
}
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["push", "-q", "origin", "main", "HEAD:other"]);
let sha = owner.git(&work, &["rev-parse", "HEAD"]);
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
let ran =
|name: &str| std::fs::read_to_string(runner.path("go").join(name)).unwrap_or_default();
let of = |r: &str, wf: &str| check_on(&runner, &owner, &key, r, wf);
let main = "refs/heads/main";
assert_eq!(of(main, "ci")["status"], "success");
let mut pushed: Vec<String> = ran("ci").lines().map(str::to_string).collect();
pushed.sort();
assert_eq!(pushed, ["refs/heads/main", "refs/heads/other"]);
let capped = of(main, "capped");
assert_eq!(capped["status"], "failure", "{capped}");
assert!(capped["steps"][0]["exitCode"].is_null(), "{capped}");
let log = blob(
&runner.account.did,
capped["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(log.contains("==> killed after 2s"), "{log}");
// A push runs none only asked for.
assert!(of(main, "costly").is_null());
assert!(of(main, "wait").is_null());
// Asked for, only it runs, with the variables given over its own, and
// its secrets; none named as the runner's own or as a secret.
let asked = serde_json::json!({
"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key},
"ref": main,
"workflow": "costly",
"env": [
{"name": "TRACKS", "value": "team"},
{"name": "KNOTLESS_RUN_REF", "value": "x"},
{"name": "MAIN_ONLY", "value": "leak"},
],
"createdAt": knotless::records::now(),
});
let rkey = format!("{key}.{}", knotless::keys::tid());
owner.put_record("dev.knotless.rerun", &rkey, asked);
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
let costly = of(main, "costly");
assert_eq!(costly["status"], "success", "{costly}");
assert_eq!(costly["commit"], sha.as_str());
// Its long step past the runner's limit, as its own allows.
assert_eq!(costly["steps"][1]["status"], "success", "{costly}");
let log = blob(
&runner.account.did,
costly["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
log.starts_with("==> asked for by name TRACKS=team\n"),
"{log}"
);
assert!(
log.contains("tracks=team kept=yes ref=refs/heads/main main=***"),
"{log}"
);
assert_eq!(ran("ci").lines().count(), 2);
assert!(of(main, "wait").is_null());
// One that runs on a push may be asked for by name too; one not there,
// or not to be asked for at that ref, says so; a ref that is not there
// is refused by the CLI, and, asked for anyway, as a branch since
// deleted may be, forgotten.
let rerun = |r: &str, wf: &str| {
let (code, out) = owner.cli(&["rerun", &url, "--ref", r, "--workflow", wf], b"");
assert_eq!(code, 0, "{out}");
};
rerun(main, "ci");
rerun(main, "nope");
rerun("refs/heads/other", "costly");
let gone = [
"rerun",
&url,
"--ref",
"refs/heads/gone",
"--workflow",
"ci",
];
// Refused, saying so on stderr, which `cli` does not keep.
let (code, out) = owner.cli(&gone, b"");
assert_eq!(code, 1, "{out}");
let asked = serde_json::json!({
"repo": {"uri": format!("at://{}/dev.knotless.repo/proj", owner.account.did), "key": key},
"ref": "refs/heads/gone",
"workflow": "ci",
"createdAt": knotless::records::now(),
});
let rkey = format!("{key}.{}", knotless::keys::tid());
owner.put_record("dev.knotless.rerun", &rkey, asked);
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
assert_eq!(ran("ci").lines().count(), 3);
let nope = of(main, "nope");
assert_eq!(nope["status"], "error", "{nope}");
assert_eq!(
nope["error"],
format!(
"asked for, but there is no workflow named nope at {}",
&sha[..12]
)
);
let other = of("refs/heads/other", "costly");
assert_eq!(other["status"], "error", "{other}");
assert_eq!(
other["error"],
"asked for, but it runs neither on a push to refs/heads/other nor when asked there"
);
let state = std::fs::read_to_string(runner.path("state.json")).unwrap();
let state: serde_json::Value = serde_json::from_str(&state).unwrap();
let id = format!("{}/{key}", owner.account.did);
assert_eq!(state["wanted"][&id], serde_json::json!({}), "{state}");
// A stop does not wait for one asked for: it is cancelled, and runs
// again on the next start.
let (code, out) = owner.cli(
&[
"rerun",
&url,
"--ref",
main,
"--workflow",
"wait",
"-e",
"TRACKS=w1",
],
b"",
);
assert_eq!(code, 0, "{out}");
rt.block_on(run.reconcile(&watch)).unwrap();
wait_until(|| ran("waits") == "w1\n");
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
});
promptly(&rt, stopping);
let wait = of(main, "wait");
assert_eq!(wait["status"], "cancelled", "{wait}");
let (run, watch) = start();
rt.block_on(run.reconcile(&watch)).unwrap();
wait_until(|| ran("waits") == "w1\nw1\n");
std::fs::write(runner.path("go").join("w1"), "").unwrap();
rt.block_on(run.idle());
assert_eq!(of(main, "wait")["status"], "success");
// Once.
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
assert_eq!(ran("waits"), "w1\nw1\n");
assert_eq!(ran("ci").lines().count(), 3);
}
/// A private repository's runs, read in its space and filed in the
/// runner's repo there, nothing of them public: a push rings no
/// Jetstream, so the runner hears of it asking the space, and of the
/// repository from its invite. These need a PDS on the spaces alpha, and
/// pass trivially without one.
#[test]
fn a_private_repository_runs_in_its_space() {
if !net().spaces {
eprintln!("skipped: the PDS is not on the spaces alpha");
return;
}
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
// It reads as itself, logged in, as a space credential needs.
let mut runner = Sandbox::new();
runner.set("KNOTLESS_PASSWORD", "");
let handle = runner.account.handle.clone();
let password = runner.account.password.clone();
let login = runner.cli(&["login", &handle, "--password-stdin"], password.as_bytes());
assert_eq!(login.0, 0, "{}", login.1);
// The appview names it for new repositories, and passes on invites.
let base = serving(&rt, &Sandbox::new(), Some(&runner.account.did)).0;
owner.set("KNOTLESS_APPVIEW", &base);
runner.set("KNOTLESS_APPVIEW", &base);
// Made private, it names the runner, a member, to read it and file
// there, and invited.
let url = owner.url("proj");
let (code, out) = owner.cli(&["create", &url, "--private"], b"");
assert_eq!(code, 0, "{out}");
assert!(
out.contains(&format!("{handle} reads it, to run its checks")),
"{out}"
);
assert!(!out.contains("warning"), "{out}");
let work = owner.work("proj", 1);
std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
std::fs::write(work.join(".knotless/workflows/fail.yaml"), FAIL).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["push", "-q", &url, "main"]);
// One named once made is invited then.
let later = owner.url("later");
let made = owner.cli(&["create", &later, "--private", "--no-runner"], b"");
assert_eq!(made.0, 0, "{}", made.1);
let (code, out) = owner.cli(&["runners", &later, "add", &handle], b"");
assert_eq!(code, 0, "{out}");
assert_eq!(out, "added");
// With no appview to tell, the runner is named and a member still.
let alone = Sandbox::with(owner.account.clone());
let lone = owner.url("lone");
let (code, out) = alone.cli(&["create", &lone, "--private", "--runner", &handle], b"");
assert_eq!(code, 0, "{out}");
assert!(
out.contains("was not told of it: invites go through an appview"),
"{out}"
);
let (code, out) = alone.cli(&["runners", &lone, "add", &handle], b"");
assert_eq!(code, 0, "{out}");
assert!(out.starts_with("added\nwarning: "), "{out}");
// Its account reads it.
let (code, out) = runner.cli(&["refs", &url], b"");
assert_eq!(code, 0, "{out}");
let js = jetstream(&rt);
let opts = knotless_run::Options {
repos: Vec::new(),
discover: true,
sweep: Duration::from_secs(1),
poll: Duration::from_secs(1),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
..runner_opts(&runner, &owner, &js)
};
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
let state = || std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
let did = &owner.account.did;
wait_until(|| {
let state = state();
["proj", "later"]
.iter()
.all(|n| state.contains(&format!("{did}/{n}")))
});
let read = || owner.cli(&["checks", &url, "--json"], b"").1;
// Found, the commit that made it runs.
wait_until(|| read().contains("\"finishedAt\""));
// Pushed, with no ring: the space says something was written.
let sha = owner.commit_file(&work, "f", "moved\n");
owner.git(&work, &["push", "-q", &url, "main"]);
wait_until(|| {
read().contains(&format!("\"commit\": \"{sha}\"")) && !read().contains("\"running\"")
});
let checks: serde_json::Value = serde_json::from_str(&read()).unwrap();
assert_eq!(checks[0]["status"], "failure", "{checks}");
assert_eq!(checks[0]["commit"], sha.as_str());
assert_eq!(checks[0]["runner"], runner.account.did.as_str());
// Its log and pictures are read from the space too.
let saved = owner.path("pictures");
let dir = saved.to_str().unwrap();
let (code, out) = owner.cli(&["checks", &url, "--log", "--images", dir], b"");
assert_eq!(code, 0, "{out}");
assert!(out.contains("==> break\nbroken\n"), "{out}");
// Of both runs: the commit that made it, as found, and the push.
let shown = out.lines().filter(|l| l.starts_with("image\t")).count();
assert_eq!(shown, 64, "{out}");
// Nothing of it is in the runner's public repo.
let public = reqwest::blocking::get(format!(
"{}/xrpc/com.atproto.repo.listRecords?repo={}&collection=dev.knotless.check",
net().pds,
runner.account.did
))
.unwrap()
.text()
.unwrap();
assert_eq!(public, r#"{"records":[]}"#);
// Deleted, its space with it, it is left rather than asked forever.
// The space is asked itself once a minute while the appview passes on
// its writes, so this may take longer than a wait.
let (code, out) = owner.cli(&["delete", &later, "--yes"], b"");
assert_eq!(code, 0, "{out}");
let until = std::time::Instant::now() + 3 * common::WAIT;
while state().contains(&format!("{did}/later")) {
assert!(std::time::Instant::now() < until, "never left: {}", state());
std::thread::sleep(Duration::from_millis(500));
}
assert!(state().contains(&format!("{did}/proj")), "{}", state());
// Its poll, waking, finds it left and ends.
std::thread::sleep(3 * Duration::from_secs(1));
}
#[test]
fn a_dropped_pull_runs_on_a_base_moved_since() {
// Landing the queue, a runner runs a pull again as its head moves,
// not its base; but one whose run was dropped as the runner stopped
// runs when it starts again, on the base as it is then.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let one = |opts: &mut knotless_run::Options| {
opts.jobs = 1;
opts.stop_wait = Duration::ZERO;
};
let l = q.lander(&rt, "state.json", one);
q.hold();
// One pull's run holds the only job; the other's waits for it.
let a = q.pull("a", "a", "a\n");
wait_until(|| {
l.step(&rt);
q.ran().len() == 1
});
let b = q.pull("b", "b", "b\n");
let checks = |pull: &str| {
let out = q.ok(&["checks", &q.url, "--pull", pull.trim(), "--json"]);
serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
};
wait_until(|| {
l.step(&rt);
!checks(&b).is_empty()
});
rt.block_on(l.run.stop());
let dropped = &checks(&b)[0];
assert_eq!(dropped["status"], "cancelled", "{dropped}");
drop(l);
// Main moves on while the runner is down.
q.branch("main", "elsewhere", "moved\n", true);
let moved = q.main();
q.release();
let l = q.lander(&rt, "state.json", one);
rt.block_on(l.run.idle());
let ran = &checks(&b)[0];
assert_eq!(ran["baseCommit"], moved.as_str(), "{ran}");
assert_eq!(ran["status"], "success", "{ran}");
// The one cut off, not dropped, is not run again until its head moves.
assert_ne!(checks(&a)[0]["baseCommit"], moved.as_str());
}
#[test]
fn checks_of_no_more_use_are_pruned() {
// Here a runner prunes on every pass, with no grace, keeping of main
// only each workflow's newest.
let rt = tokio::runtime::Runtime::new().unwrap();
let q = Line::new(&rt);
let secrets = q.owner.path("secrets.json");
let label = format!("{}/line", q.owner.account.handle);
let granted = serde_json::json!({ label.clone(): {"TOKEN": "s3cret"}, "x.test/other": {"TOKEN": "kept"} });
std::fs::write(&secrets, granted.to_string()).unwrap();
let l = q.lander(&rt, "state.json", |opts| {
opts.keep_checks = Some(Duration::ZERO);
(opts.prune_grace, opts.prune_every) = (Duration::ZERO, Duration::ZERO);
opts.secrets = Some(secrets.clone());
});
let checks = |by: &str, of: &str| {
let out = q.ok(&["checks", &q.url, by, of, "--json"]);
serde_json::from_str::<Vec<serde_json::Value>>(&out).unwrap()
};
let workflows = |found: &[serde_json::Value]| {
let mut names: Vec<&str> = found
.iter()
.map(|c| c["workflow"].as_str().unwrap())
.collect();
names.sort();
names.join(" ")
};
let a = q.pull("a", "a", "a\n");
let b = q.pull("b", "b", "b\n");
l.pass(&rt);
assert_eq!(workflows(&checks("--pull", &a)), "ci");
q.queue(&[&a]);
l.pass(&rt);
l.pass(&rt);
assert_eq!(q.log(2), ["a", "workflows"]);
// b's head moves, and runs again.
q.branch("b", "b2", "b2\n", true);
l.pass(&rt);
l.pass(&rt);
// The batch's run is gone, as is the merged pull's; the open one keeps
// only its newest.
assert!(q.batches().is_empty());
assert!(checks("--pull", &a).is_empty());
let of_b = checks("--pull", &b);
assert_eq!(workflows(&of_b), "ci");
let o = &q.owner;
o.git(&q.work, &["fetch", "-q", "origin"]);
let head = o.git(&q.work, &["rev-parse", "origin/b"]);
assert_eq!(of_b[0]["commit"], head.trim(), "{of_b:?}");
// A branch keeps its newest of each workflow, as main does here: its
// pull's and batches' runs are theirs, not the branch's, and ci runs
// only on those.
assert_eq!(workflows(&checks("--ref", "refs/heads/b")), "push");
let main = checks("--ref", "refs/heads/main");
assert_eq!(workflows(&main), "push");
assert!(
main.iter().all(|c| c["commit"] == q.main().as_str()),
"{main:?}"
);
let filed = || q.owner.records("dev.knotless.check").len();
let before = filed();
assert!(before > 0);
// Deleted, the checks filed of it go, and the secrets granted to its
// name, which a repository made under it later is not to have; it is
// left, and a pass over it again finds nothing to do.
q.ok(&["delete", &q.url, "--yes"]);
l.pass(&rt);
assert_eq!(filed(), 0);
let left: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&secrets).unwrap()).unwrap();
assert_eq!(left, serde_json::json!({"x.test/other": {"TOKEN": "kept"}}));
l.step(&rt);
}
#[test]
fn a_run_asked_for_writes_back() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
let work = owner.work("proj", 1);
let url = owner.url("proj");
let go = owner.path("go");
std::fs::create_dir_all(&go).unwrap();
// `take` writes a file it may write back and one it may not; `held`
// waits to be let go first.
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let take = "on: {asked: {commit: [shots/**]}}\nsteps:\n - run: mkdir -p shots && echo $SHOT > shots/a.txt && echo x > stray.txt\nenv: {SHOT: one}\n";
std::fs::write(dir.join("take.yml"), take).unwrap();
let held = format!(
"on: {{asked: {{commit: [shots/**]}}}}\nsteps:\n - run: |\n touch {go}/started\n until [ -f {go}/go ]; do sleep 0.1; done\n mkdir -p shots && echo held > shots/a.txt\n",
go = go.display()
);
std::fs::write(dir.join("held.yml"), held).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main", "main:topic"]);
let ok = |args: &[&str]| {
let (code, out) = owner.cli(args, b"");
assert_eq!(code, 0, "{args:?}: {out}");
out
};
ok(&["runners", &url, "add", &mine]);
let label = format!("{mine}/proj");
let mut opts = knotless_run::Options::new(vec![label]);
opts.jetstream = "ws://127.0.0.1:9".into();
opts.state = Some(owner.path("state.json"));
opts.work = Some(owner.path("work"));
let started = knotless_run::Runner::start(opts, common::ctx(&owner));
let (run, mut watches) = rt.block_on(started).unwrap();
let watch = watches.remove(0);
let key = watch.key.clone();
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
let tip = |branch: &str| {
let refs = ok(&["refs", &url]);
let line = refs
.lines()
.find(|l| l.ends_with(&format!("\trefs/heads/{branch}")));
line.unwrap().split('\t').next().unwrap().to_string()
};
let asked = |r: &str, wf: &str| {
ok(&["rerun", &url, "--ref", r, "--workflow", wf]);
rt.block_on(run.reconcile(&watch)).unwrap();
};
let of = |r: &str, wf: &str| check_on(&owner, &owner, &key, r, wf);
let log_of = |c: &serde_json::Value| {
blob(
&owner.account.did,
c["log"]["ref"]["$link"].as_str().unwrap(),
)
};
let topic = "refs/heads/topic";
let was = tip("topic");
// Asked for at a branch, what it wrote among its globs is committed
// onto it, and only that.
asked(topic, "take");
rt.block_on(run.idle());
let took = of(topic, "take");
assert_eq!(took["status"], "success", "{took}");
let now = tip("topic");
assert_ne!(now, was);
assert!(log_of(&took).contains(&format!("==> wrote back 1 file to topic as {}", &now[..12])));
owner.git(&work, &["fetch", "-q", "origin"]);
let parent = owner.git(&work, &["rev-parse", "origin/topic^"]);
assert_eq!(parent, was);
let files = owner.git(&work, &["diff", "--name-only", &was, "origin/topic"]);
assert_eq!(files, "shots/a.txt");
let said = owner.git(&work, &["log", "-1", "--format=%an %s", "origin/topic"]);
assert_eq!(said, "knotless-run take: what it wrote, as asked");
assert_eq!(tip("main"), was);
// Asked again, it writes the same, so nothing.
asked(topic, "take");
rt.block_on(run.idle());
let again = of(topic, "take");
assert_eq!(again["status"], "success", "{again}");
assert!(log_of(&again).contains("==> nothing to write back in shots/**"));
assert_eq!(tip("topic"), now);
// Never the default branch, nor a protected one.
asked("refs/heads/main", "take");
rt.block_on(run.idle());
let main = of("refs/heads/main", "take");
assert_eq!(main["status"], "failure", "{main}");
let error = main["error"].as_str().unwrap();
assert!(error.starts_with("not written back: main is "), "{error}");
assert!(error.ends_with("'s default branch"), "{error}");
// From before what was written, so there is something to write.
let from = format!("{was}:refs/heads/kept");
owner.git(&work, &["push", "-q", "origin", &from]);
ok(&["protect", &url, "kept"]);
asked("refs/heads/kept", "take");
rt.block_on(run.idle());
let kept = of("refs/heads/kept", "take");
assert_eq!(
kept["error"], "not written back: kept is protected",
"{kept}"
);
// Nor over a push made while it ran.
asked(topic, "held");
wait_until(|| go.join("started").exists());
owner.git(&work, &["checkout", "-q", "-B", "topic", "origin/topic"]);
owner.commit_file(&work, "b.txt", "b\n");
owner.git(&work, &["push", "-q", "origin", "topic"]);
let pushed = owner.git(&work, &["rev-parse", "HEAD"]);
std::fs::write(go.join("go"), "").unwrap();
rt.block_on(run.idle());
let held = of(topic, "held");
assert_eq!(held["status"], "failure", "{held}");
assert_eq!(
held["error"],
format!("not written back: topic has moved on from {}", &now[..12])
);
assert_eq!(tip("topic"), pushed);
}
#[test]
fn a_runner_finds_the_repositories_naming_it() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
// The appview names it for new repositories.
let named = [runner.account.did.clone()];
let base = serving(&rt, &Sandbox::new(), Some(&named[0])).0;
owner.set("KNOTLESS_APPVIEW", &base);
runner.set("KNOTLESS_APPVIEW", &base);
let me = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &me);
let did = owner.account.did.clone();
let runners = |name: &str| owner.record("dev.knotless.repo", name)["runners"].clone();
// A repository made names those, unless its owner names others.
let made = owner.url("made");
assert_eq!(owner.cli(&["create", &made], b"").0, 0);
let dids = |r: serde_json::Value| -> Vec<String> {
let r = r.as_array().cloned().unwrap_or_default();
r.iter()
.map(|r| r["did"].as_str().unwrap().to_string())
.collect()
};
assert_eq!(dids(runners("made")), named);
let own = [
"create",
&owner.url("own"),
"--runner",
&owner.account.handle,
];
assert_eq!(owner.cli(&own, b"").0, 0);
assert_eq!(dids(runners("own")), std::slice::from_ref(&did));
assert_eq!(
owner
.cli(&["create", &owner.url("none"), "--no-runner"], b"")
.0,
0
);
assert!(dids(runners("none")).is_empty());
// Each below is made by its first push, so names the runner too, and
// has a workflow every push runs.
let make = |name: &str| {
let url = owner.url(name);
let work = owner.work(name, 1);
std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
let flow = work.join(".knotless/workflows/ci.yml");
std::fs::write(flow, "steps: [{run: 'true'}]\n").unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main"]);
assert_eq!(dids(runners(name)), named);
let record = owner.record("dev.knotless.repo", name);
(work, record["key"].as_str().unwrap().to_string())
};
let state = || -> serde_json::Value {
let text = std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
serde_json::from_str(&text).unwrap_or_default()
};
let found = || -> Vec<String> {
let state = state();
let labels = state["discovered"].as_array().cloned().unwrap_or_default();
labels
.iter()
.map(|l| l.as_str().unwrap().to_string())
.collect()
};
let js = jetstream(&rt);
// Once it finds one, the commit that made it runs, and then a push.
let ran = |key: &str, sha: &str| {
wait_until(|| {
let check = check_of(&runner, &owner, key, "ci");
check["commit"] == sha && check["status"] == "success"
});
};
let runs = |work: &std::path::Path, key: &str| {
ran(key, &owner.git(work, &["rev-parse", "HEAD"]));
let sha = owner.commit_file(work, "f", "f\n");
owner.git(work, &["push", "-q", "origin", "main"]);
js.ring(&did, &ref_rkey(key, "refs/heads/main"));
ran(key, &sha);
};
// Those named before it starts the appview lists, but one named on
// the command line is watched as that.
let (first, first_key) = make("first");
make("both");
let listing = format!(
"{base}/xrpc/dev.knotless.listRepos?runner={}",
runner.account.did
);
let listed = || reqwest::blocking::get(&listing).unwrap().text().unwrap();
wait_until(|| {
["first", "both", "made"]
.iter()
.all(|n| listed().contains(&format!("\"{n}\"")))
});
let opts = knotless_run::Options {
repos: vec![format!("{}/both", owner.account.handle)],
discover: true,
sweep: Duration::from_secs(1),
notice: Duration::from_millis(200),
..runner_opts(&runner, &owner, &js)
};
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
let at = |names: &[&str]| -> Vec<String> {
let mut labels: Vec<String> = names.iter().map(|n| format!("{did}/{n}")).collect();
labels.sort();
labels
};
wait_until(|| found() == at(&["first", "made"]));
runs(&first, &first_key);
// One named once it has started is found as its record is written; a
// record of no repository is passed over.
let (second, second_key) = make("second");
let record = owner.record("dev.knotless.repo", "second");
js.send(&did, "dev.knotless.repo", "second", Some(record));
js.send(
&did,
"dev.knotless.repo",
"nothing",
Some(serde_json::json!({})),
);
wait_until(|| found().contains(&format!("{did}/second")));
runs(&second, &second_key);
// Renamed, it is watched by its new name.
let (code, out) = owner.cli(&["rename", &owner.url("first"), "renamed"], b"");
assert_eq!(code, 0, "{out}");
let record = owner.record("dev.knotless.repo", "renamed");
js.send(&did, "dev.knotless.repo", "renamed", Some(record));
wait_until(|| {
let found = found();
found.contains(&format!("{did}/renamed")) && !found.contains(&format!("{did}/first"))
});
// No longer naming it, it is left.
let (code, out) = owner.cli(&["runners", &owner.url("second"), "remove", &me], b"");
assert_eq!(code, 0, "{out}");
let record = owner.record("dev.knotless.repo", "second");
js.send(&did, "dev.knotless.repo", "second", Some(record));
wait_until(|| found() == at(&["made", "renamed"]));
// Its loops end with it, each at its next look.
std::thread::sleep(Duration::from_millis(500));
}
/// The server-sent events at `url`, gathered as they come, from
/// `last` if it says where the last were had to.
fn listen(url: &str, last: Option<&str>) -> Arc<std::sync::Mutex<String>> {
use std::io::Read;
let client = reqwest::blocking::Client::builder()
.timeout(None)
.build()
.unwrap();
let mut req = client.get(url);
if let Some(last) = last {
req = req.header("last-event-id", last);
}
let mut res = req.send().unwrap();
assert_eq!(res.status(), 200, "{url}");
let heard = Arc::new(std::sync::Mutex::new(String::new()));
let gathered = heard.clone();
std::thread::spawn(move || {
let mut buf = [0u8; 4096];
while let Ok(n @ 1..) = res.read(&mut buf) {
let more = String::from_utf8_lossy(&buf[..n]);
gathered.lock().unwrap().push_str(&more);
}
});
heard
}
/// What a run prints shows at the appview as it prints it, masked as its
/// log is, for whoever may read its repository, and its steps as each
/// ends; its repository's pages hear of each write about it. A page that
/// lost its place goes on from it.
#[test]
fn a_run_shows_as_it_runs() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let base = serving(&rt, &Sandbox::new(), Some(&runner.account.did)).0;
owner.set("KNOTLESS_APPVIEW", &base);
runner.set("KNOTLESS_APPVIEW", &base);
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let flow = format!(
"secrets: [TOKEN]\nsteps:\n - name: first\n run: |\n echo \"first $TOKEN\"\n until [ -f {go}/go ]; do sleep 0.1; done\n - name: second\n run: echo second\n",
go = go.display()
);
let work = owner.work("proj", 1);
std::fs::create_dir_all(work.join(".knotless/workflows")).unwrap();
std::fs::write(
work.join(".knotless/workflows/live.yml"),
"steps: [{run: 'true'}]\n",
)
.unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let ref_key = ref_rkey(&key, "refs/heads/main");
let js = jetstream(&rt);
rt.spawn(knotless_run::run(
runner_opts(&runner, &owner, &js),
common::ctx(&runner),
));
wait_until(|| runner.path("state.json").exists());
// Its pages follow it; a push is heard of.
let repo = format!("at://{}/dev.knotless.repo/proj", owner.account.did);
let changed = listen(&format!("{base}/_live/repo?repo={repo}"), None);
std::fs::write(work.join(".knotless/workflows/live.yml"), flow).unwrap();
owner.git(&work, &["commit", "-q", "-am", "it holds"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
// With data, as a browser hears no event without.
wait_until(|| {
changed
.lock()
.unwrap()
.contains("event: changed\ndata: null")
});
// Running, what it printed shows, masked; and not before.
let running = || {
let checks = runner.records(check::NSID);
let live = checks
.iter()
.find(|(_, c)| c["workflow"] == "live" && c["status"] == "running");
live.map(|(k, _)| format!("at://{}/{}/{k}", runner.account.did, check::NSID))
};
wait_until(|| running().is_some());
let uri = running().unwrap();
// Once the appview has it, as it shows only the checks it has.
let shown = format!("{base}/xrpc/dev.knotless.getCheck?repo={repo}&uri={uri}");
wait_until(|| reqwest::blocking::get(&shown).unwrap().status() == 200);
let follow = format!("{base}/_live/check?repo={repo}&uri={uri}");
let said = listen(&follow, None);
wait_until(|| said.lock().unwrap().contains(r#"first ***\n"#));
// The step going is named, as its `==>` line may be gone from the tail.
wait_until(|| {
said.lock()
.unwrap()
.contains("event: going\ndata: \"first\"")
});
assert!(!said.lock().unwrap().contains("event: ended"));
std::fs::write(go.join("go"), "").unwrap();
wait_until(|| said.lock().unwrap().contains("event: ended\ndata: null"));
let said = said.lock().unwrap().clone();
assert!(!said.contains("s3cret"), "{said}");
assert!(said.contains("event: steps"), "{said}");
assert!(said.contains(r#""name":"second""#), "{said}");
// Each piece is named by where it ends: a page that had it all is sent
// only that it ended.
let last = said
.lines()
.rev()
.find_map(|l| l.strip_prefix("id: "))
.unwrap();
let again = listen(&follow, Some(last));
wait_until(|| again.lock().unwrap().contains("event: ended"));
assert!(!again.lock().unwrap().contains("event: output"));
// A check its repository's runners did not file is no one's to follow.
let other = format!("at://{}/{}/x", owner.account.did, check::NSID);
let status = |url: String| reqwest::blocking::get(url).unwrap().status().as_u16();
assert_eq!(
status(format!("{base}/_live/check?repo={repo}&uri={other}")),
404
);
// Nor does a runner send another's.
let did = reqwest::blocking::get(format!("{base}/.well-known/did.json"))
.unwrap()
.json::<serde_json::Value>()
.unwrap()["id"]
.as_str()
.unwrap()
.to_string();
let ctx = common::ctx(&runner);
let token = rt.block_on(async {
let me = knotless::session::me(&ctx).await.unwrap();
knotless::auth::service_token(&ctx, &me, &did, "dev.knotless.putRunOutput")
.await
.unwrap()
});
let sent = reqwest::blocking::Client::new()
.post(format!("{base}/xrpc/dev.knotless.putRunOutput"))
.bearer_auth(&token)
.json(&serde_json::json!({ "check": other, "offset": 0, "text": "forged" }))
.send()
.unwrap();
assert_eq!(sent.status(), 400);
// Run again, as a rerun at the same commit is, the check's output
// starts anew rather than showing the last run's.
let anew = serde_json::json!({
"check": uri, "startedAt": "2099-01-01T00:00:00.000Z", "offset": 0, "text": "anew\n"
});
let sent = reqwest::blocking::Client::new()
.post(format!("{base}/xrpc/dev.knotless.putRunOutput"))
.bearer_auth(&token)
.json(&anew)
.send()
.unwrap();
assert_eq!(sent.status(), 200);
let fresh = listen(&follow, None);
wait_until(|| fresh.lock().unwrap().contains(r#""anew\n""#));
assert!(
!fresh.lock().unwrap().contains("first"),
"{}",
fresh.lock().unwrap()
);
}
/// A step another repository shares: it takes inputs, needs a command, may
/// be given one secret, and runs a script kept beside it.
const GREET: &str = r#"
description: Says who it greets.
inputs:
who: {required: true}
how: {default: hello}
needs: [sh]
secrets: [TOKEN]
steps:
- name: say
run: |
echo "$INPUT_HOW $INPUT_WHO token=${TOKEN:-none} other=${OTHER:-none}"
- name: script
run: sh "$KNOTLESS_STEP_DIR/greet.sh"
"#;
#[test]
fn a_workflow_uses_a_shared_step() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let h = owner.account.handle.clone();
// The shared steps, in a repository of their own, at the commit the
// lock pins.
let steps = owner.work("steps", 1);
let files = steps.join(".knotless/steps");
std::fs::create_dir_all(&files).unwrap();
let write = |name: &str, text: &str| std::fs::write(files.join(name), text).unwrap();
write("greet.yml", GREET);
std::fs::write(steps.join("greet.sh"), "echo a script of its own\n").unwrap();
// One that uses another, passing on what it is given; one that uses
// itself; and a chain past the deepest.
let outer = format!(
"secrets: [TOKEN]\nsteps:\n - {{uses: {h}/steps/greet@1, with: {{who: nested}}, secrets: [TOKEN]}}\n"
);
write("outer.yml", &outer);
// One asking for a secret by a name of its own, which says how long
// what it sees is, as the log hides the value; and one passing it a
// secret of its own under that name.
write(
"keyed.yml",
"secrets: [KEY]\nsteps:\n - run: echo \"key=${KEY:-none} len=${#KEY} token=${TOKEN:-none} other=${OTHER:-none}\"\n",
);
let renaming = format!(
"secrets: [INNER]\nsteps:\n - {{uses: {h}/steps/keyed@1, secrets: {{KEY: INNER}}}}\n"
);
write("renaming.yml", &renaming);
// One with a step of its own for main alone.
write(
"gated.yml",
"steps:\n - {name: any, run: echo inner any}\n - {name: main, run: echo inner main, on: {refs: [refs/heads/main]}}\n",
);
write(
"loop.yml",
&format!("steps: [{{uses: {h}/steps/loop@1}}]\n"),
);
for d in 1..=5 {
let next = format!("steps: [{{uses: {h}/steps/d{}@1}}]\n", d + 1);
write(&format!("d{d}.yml"), &next);
}
owner.git(&steps, &["add", "."]);
owner.git(&steps, &["commit", "-q", "-m", "steps"]);
let pinned = owner.git(&steps, &["rev-parse", "HEAD"]);
owner.git(&steps, &["remote", "add", "origin", &owner.url("steps")]);
owner.git(&steps, &["push", "-q", "origin", "main"]);
// What it says later never runs where the lock pins the earlier.
let later = "steps: [{run: exit 9}]\n";
owner.commit_file(&steps, ".knotless/steps/greet.yml", later);
owner.git(&steps, &["push", "-q", "origin", "main"]);
let uses = format!("{h}/steps/greet@1");
let work = owner.work("proj", 1);
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let on = "on: {push: {refs: [refs/heads/main]}}\nsecrets: [TOKEN, OTHER]\n";
let shared = format!(
"{on}steps:\n - run: echo \"own token=${{TOKEN:-none}} other=${{OTHER:-none}}\"\n - uses: {uses}\n with: {{who: world}}\n secrets: [TOKEN]\n - {{name: quiet, uses: {uses}, with: {{who: you, how: hi}}}}\n"
);
std::fs::write(dir.join("shared.yml"), shared).unwrap();
// Steps for some refs: a shared step's `on` holds for each of its
// steps, beside their own.
let gated = format!(
"on: {{push: {{refs: [refs/heads/main, refs/heads/x]}}}}\nsteps:\n - run: echo own any\n - {{name: own main, run: echo own main, on: {{refs: [refs/heads/main]}}}}\n - {{uses: {uses}, with: {{who: x}}, on: {{refs: [refs/heads/x]}}}}\n - {{uses: {h}/steps/gated@1, on: {{refs: [refs/heads/*]}}}}\n"
);
std::fs::write(dir.join("gated.yml"), gated).unwrap();
let using =
|step: &str, more: &str| format!("{on}steps:\n - {{uses: {h}/steps/{step}{more}}}\n");
let workflows = [
("unpinned", using("other@1", "")),
("newer", using("greet@2", ", with: {who: x}")),
("gone", using("gone@1", "")),
("nested", using("outer@1", ", secrets: [TOKEN]")),
("looped", using("loop@1", "")),
("deep", using("d1@1", "")),
("renamed", using("keyed@1", ", secrets: {KEY: OTHER}")),
(
"renamings",
using("renaming@1", ", secrets: {INNER: TOKEN}"),
),
(
"unrenamed",
format!(
"on: {{push: {{refs: [refs/heads/main]}}}}\nsecrets: [MISSING]\nsteps:\n - {{uses: {h}/steps/keyed@1, secrets: {{KEY: MISSING}}}}\n"
),
),
];
for (name, text) in &workflows {
std::fs::write(dir.join(format!("{name}.yml")), text).unwrap();
}
let mut lock = String::new();
let pin = |lock: &mut String, step: &str, version: &str| {
let line = format!("{h}/steps/{step}: {{version: {version}, commit: {pinned}}}\n");
lock.push_str(&line);
};
pin(&mut lock, "greet@1", "1.0.0");
pin(&mut lock, "greet@2", "1.0.0");
pin(&mut lock, "gone@1", "1.1.0");
let pinned_steps = [
"outer", "gated", "loop", "d1", "d2", "d3", "d4", "d5", "keyed", "renaming",
];
for step in pinned_steps {
pin(&mut lock, &format!("{step}@1"), "1.0.0");
}
std::fs::write(work.join(".knotless/uses.lock"), lock).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "shared steps"]);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let ref_key = ref_rkey(&key, "refs/heads/main");
let rt = tokio::runtime::Runtime::new().unwrap();
let js = jetstream(&rt);
let opts = runner_opts(&runner, &owner, &js);
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
wait_until(|| runner.path("state.json").exists());
owner.commit_file(&work, "f", "again");
owner.git(&work, &["push", "-q", "origin", "main"]);
js.ring(&owner.account.did, &ref_key);
let all = [
"shared",
"unpinned",
"newer",
"gone",
"nested",
"looped",
"deep",
"gated",
"renamed",
"renamings",
"unrenamed",
];
wait_until(|| {
all.iter()
.all(|wf| check_of(&runner, &owner, &key, wf)["finishedAt"].is_string())
});
let ran = check_of(&runner, &owner, &key, "shared");
assert_eq!(ran["status"], "success", "{ran}");
let names: Vec<&str> = ran["steps"]
.as_array()
.unwrap()
.iter()
.map(|s| s["name"].as_str().unwrap())
.collect();
assert_eq!(
names,
[
"step 1",
"greet / needs",
"greet / say",
"greet / script",
"quiet / needs",
"quiet / say",
"quiet / script",
]
);
let log_of = |check: &serde_json::Value| {
blob(
&runner.account.did,
check["log"]["ref"]["$link"].as_str().unwrap(),
)
};
let log = log_of(&ran);
// Its own step sees every secret granted; a shared one only those it
// is given, and none it is not.
assert!(log.contains("own token=*** other=***"), "{log}");
assert!(log.contains("hello world token=*** other=none"), "{log}");
assert!(log.contains("hi you token=none other=none"), "{log}");
assert!(log.contains("a script of its own"), "{log}");
// A shared step's own shared steps, given what it passes on.
let nested = check_of(&runner, &owner, &key, "nested");
assert_eq!(nested["status"], "success", "{nested}");
assert_eq!(nested["steps"][1]["name"], "outer / greet / say");
let log = log_of(&nested);
assert!(log.contains("hello nested token=*** other=none"), "{log}");
// A secret passed under the name the step file asks for is seen by
// that name alone: OTHER's four letters as KEY, and no OTHER.
let renamed = check_of(&runner, &owner, &key, "renamed");
assert_eq!(renamed["status"], "success", "{renamed}");
let log = log_of(&renamed);
assert!(log.contains("key=*** len=4 token=none other=none"), "{log}");
// Through a step file passing on its own name, TOKEN's six as KEY.
let renamings = check_of(&runner, &owner, &key, "renamings");
assert_eq!(renamings["status"], "success", "{renamings}");
let log = log_of(&renamings);
assert!(log.contains("key=*** len=6 token=none other=none"), "{log}");
// One the runner does not grant is named as the workflow names it.
let unrenamed = check_of(&runner, &owner, &key, "unrenamed");
assert_eq!(unrenamed["status"], "success", "{unrenamed}");
let log = log_of(&unrenamed);
assert!(log.contains("==> MISSING is not granted to"), "{log}");
assert!(log.contains("key=none len=0"), "{log}");
// On main, what is for main runs, and what is for x passes unrun.
let gated = check_of(&runner, &owner, &key, "gated");
assert_eq!(gated["status"], "success", "{gated}");
let steps = gated["steps"].as_array().unwrap().iter();
let unrun = steps.filter(|s| s["status"] == "skipped");
let unrun: Vec<&str> = unrun.map(|s| s["name"].as_str().unwrap()).collect();
assert_eq!(unrun, ["greet / needs", "greet / say", "greet / script"]);
let log = log_of(&gated);
assert!(log.contains("own any") && !log.contains("==> own main\n==> skipped"));
assert!(
log.contains("inner any") && log.contains("inner main"),
"{log}"
);
assert!(!log.contains("hello x"), "{log}");
let why = "==> greet / say\n==> skipped, as refs/heads/main is not refs/heads/x\n";
assert!(log.contains(why), "{log}");
// On x, the other way about.
owner.git(&work, &["push", "-q", "origin", "HEAD:refs/heads/x"]);
js.ring(&owner.account.did, &ref_rkey(&key, "refs/heads/x"));
wait_until(|| {
check_on(&runner, &owner, &key, "refs/heads/x", "gated")["finishedAt"].is_string()
});
let on_x = check_on(&runner, &owner, &key, "refs/heads/x", "gated");
assert_eq!(on_x["status"], "success", "{on_x}");
let log = log_of(&on_x);
assert!(
log.contains("hello x") && log.contains("inner any"),
"{log}"
);
assert!(!log.contains("inner main"), "{log}");
let why = "==> own main\n==> skipped, as refs/heads/x is not refs/heads/main\n";
assert!(log.contains(why), "{log}");
let inner = "==> gated / main\n==> skipped, as refs/heads/x is not refs/heads/main\n";
assert!(log.contains(inner), "{log}");
let said = |wf: &str| {
let check = check_of(&runner, &owner, &key, wf);
assert_eq!(check["status"], "error", "{check}");
check["error"].as_str().unwrap().to_string()
};
for (wf, says) in [
("unpinned", "does not pin".to_string()),
("newer", "which @2 does not take".to_string()),
("gone", format!(".knotless/steps/gone.yml at {pinned} of")),
("looped", "loop@1 uses itself, through".to_string()),
("deep", "d5@1 is past 4 shared steps deep".to_string()),
] {
let error = said(wf);
assert!(error.contains(&says), "{wf}: {error}");
}
// Without a lock, nothing shared runs.
owner.git(&work, &["rm", "-q", ".knotless/uses.lock"]);
owner.git(&work, &["commit", "-q", "-m", "no lock"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let unlocked = owner.git(&work, &["rev-parse", "HEAD"]);
js.ring(&owner.account.did, &ref_key);
wait_until(|| {
let check = check_of(&runner, &owner, &key, "shared");
check["commit"] == unlocked.as_str() && check["finishedAt"].is_string()
});
let error = said("shared");
assert!(error.contains("there is no .knotless/uses.lock"), "{error}");
}