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

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

The owner cancels a ref's or a pull's runs

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:run/cancel
.changeset/run-cancel.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# The owner cancels runs
`knotless run cancel <url> --ref <name> | --pull <at-uri> [--workflow <name>]`, and the MCP `run_cancel` tool, ask a repository's runners to cancel a ref's or a pull's runs going or waiting, a deploy's too, or only one workflow's.
crates/knotless/src/cli.rs+17−0
View file
@@ -375,6 +375,9 @@ run
or one workflow there, which may run
only when asked (`on: {asked: {}}`),
its steps seeing the variables given
run cancel <url> --ref <name> | --pull <at-uri> [--workflow <name>]
ask the runners to cancel its runs
going or waiting, or one workflow's
workflow
workflow list <url> [<ref>] the workflows at a ref (the default
@@ -2374,6 +2377,20 @@ async fn command(
let asked = manage::rerun(ctx, &url, refname, pull, workflow, env).await?;
writeln!(out, "{asked}")?;
}
("run", "cancel") => {
let url = args.url()?;
let (mut refname, mut pull, mut workflow) = (None, None, None);
while let Some(flag) = args.rest.next() {
match flag.as_str() {
"--ref" => refname = Some(args.next("<ref>")?),
"--pull" => pull = Some(args.next("<at-uri>")?),
"--workflow" => workflow = Some(args.next("<name>")?),
other => bail!("unexpected {other}"),
}
}
let asked = manage::cancel(ctx, &url, refname, pull, workflow).await?;
writeln!(out, "{asked}")?;
}
("workflow", "list") => {
let url = args.url()?;
let name = args.rest.next().map(String::as_str);
crates/knotless/src/jetstream.rs+2−0
View file
@@ -68,6 +68,7 @@ pub fn knotless(collection: &str) -> bool {
nsid::CHECK,
nsid::MINUTES,
nsid::RERUN,
nsid::CANCEL,
nsid::STAR,
nsid::FOLLOW,
nsid::REACTION,
@@ -293,6 +294,7 @@ mod tests {
for collection in nsid::ALL.iter().chain(nsid::FORGE.iter()).chain([
&nsid::CHECK,
&nsid::RERUN,
&nsid::CANCEL,
&nsid::STAR,
&nsid::FOLLOW,
&nsid::REACTION,
crates/knotless/src/manage.rs+92−28
View file
@@ -864,39 +864,102 @@ pub async fn rerun(
);
}
}
let remote = Remote::open(ctx, url).await?;
let auth = session::ensure(ctx, &remote.identity).await?;
let Some(repo) = remote.repo().await? else {
bail!("{url}: no such repository");
};
if repo.value.runners.is_empty() {
bail!(
"{url} names no runners to ask: name the account a runner reports as with `knotless runner add {url} <handle>`"
);
}
// A branch or tag may be named bare, as git takes it; a ref the
// repository does not have would be asked of no runner.
let refname = match refname {
Some(name) => {
let snap = remote.snapshot(ctx).await?;
let found =
full_ref(&snap, name).with_context(|| format!("{url} has no ref {name}"))?;
Some(found)
}
None => None,
};
let asking = Asking::of(ctx, url, refname, pull).await?;
let record = records::RerunRecord {
repo: RepoRef::new(remote.did(), &repo.rkey, &repo.value.key),
ref_name: refname,
pull: pull.map(|p| crate::aturi::resolve(p, remote.space_ref().as_ref())),
repo: asking.repo.clone(),
ref_name: asking.refname.clone(),
pull: asking.pull.clone(),
workflow: workflow.map(str::to_string),
env,
created_at: records::now(),
};
let rkey = format!("{}.{}", repo.value.key, keys::tid());
let write = remote::create(nsid::RERUN, &rkey, records::typed(nsid::RERUN, &record));
remote.apply(vec![write], &auth.auth()).await?;
Ok(remote.uri_of(nsid::RERUN, &rkey))
asking.ask(nsid::RERUN, &record).await
}
/// Asks the repository's runners, as its owner, to cancel their runs of
/// the ref `refname`, or of the pull `pull`, going or waiting when they
/// read the request: only `workflow`'s, if named. Gives the request's
/// AT-URI.
pub async fn cancel(
ctx: &Ctx,
url: &RepoUrl,
refname: Option<&str>,
pull: Option<&str>,
workflow: Option<&str>,
) -> Result<String> {
if refname.is_some() == pull.is_some() {
bail!("name one ref or one pull whose runs to cancel");
}
if let Some(name) = workflow.filter(|w| !crate::workflow::valid_name(w)) {
bail!("a workflow's name is 1 to 64 letters, digits, `-` and `_`, not {name:?}");
}
let asking = Asking::of(ctx, url, refname, pull).await?;
let record = records::CancelRecord {
repo: asking.repo.clone(),
ref_name: asking.refname.clone(),
pull: asking.pull.clone(),
workflow: workflow.map(str::to_string),
created_at: records::now(),
};
asking.ask(nsid::CANCEL, &record).await
}
/// A request of a repository's runners, as its owner: the repository,
/// and the ref or pull it is about, as the runners read them.
struct Asking {
remote: Remote,
auth: session::Session,
repo: RepoRef,
refname: Option<String>,
pull: Option<String>,
}
impl Asking {
async fn of(
ctx: &Ctx,
url: &RepoUrl,
refname: Option<&str>,
pull: Option<&str>,
) -> Result<Asking> {
let remote = Remote::open(ctx, url).await?;
let auth = session::ensure(ctx, &remote.identity).await?;
let Some(repo) = remote.repo().await? else {
bail!("{url}: no such repository");
};
if repo.value.runners.is_empty() {
bail!(
"{url} names no runners to ask: name the account a runner reports as with `knotless runner add {url} <handle>`"
);
}
// A branch or tag may be named bare, as git takes it; a ref the
// repository does not have would be asked of no runner.
let refname = match refname {
Some(name) => {
let snap = remote.snapshot(ctx).await?;
let found =
full_ref(&snap, name).with_context(|| format!("{url} has no ref {name}"))?;
Some(found)
}
None => None,
};
let pull = pull.map(|p| crate::aturi::resolve(p, remote.space_ref().as_ref()));
Ok(Asking {
repo: RepoRef::new(remote.did(), &repo.rkey, &repo.value.key),
refname,
pull,
remote,
auth,
})
}
/// Writes `record` into `collection`, keyed by the repository and a
/// TID, and gives its AT-URI.
async fn ask(&self, collection: &str, record: &impl serde::Serialize) -> Result<String> {
let rkey = format!("{}.{}", self.repo.key, keys::tid());
let write = remote::create(collection, &rkey, records::typed(collection, record));
self.remote.apply(vec![write], &self.auth.auth()).await?;
Ok(self.remote.uri_of(collection, &rkey))
}
}
/// The ref `name` names in `snap`: itself, else the branch or the tag of
@@ -1329,6 +1392,7 @@ pub async fn delete(ctx: &Ctx, url: &RepoUrl) -> Result<u64> {
nsid::PACK,
nsid::RELEASE,
nsid::RERUN,
nsid::CANCEL,
] {
for r in remote
.scan_all::<serde_json::Value>(collection, &key)
crates/knotless/src/oauth.rs+3−1
View file
@@ -58,8 +58,10 @@ fn scope(spelled: bool) -> String {
&nsid::ORG_MEMBER,
&nsid::ORG_ACCEPTANCE,
&nsid::REACTION,
// An owner asking their runners to run again, from the pages.
// An owner asking their runners to run again, from the pages, or
// to cancel what they run.
&nsid::RERUN,
&nsid::CANCEL,
// Versions of a package, published from a repository.
&nsid::RELEASE,
// Themes for the pages, published from their settings.
crates/knotless/src/records.rs+19−0
View file
@@ -53,6 +53,9 @@ pub mod nsid {
/// An owner asking their repository's runners to run a ref or a pull
/// again, in the owner's account.
pub const RERUN: &str = "dev.knotless.rerun";
/// An owner asking their repository's runners to cancel a ref's or a
/// pull's runs still going, in the owner's account.
pub const CANCEL: &str = "dev.knotless.cancel";
/// The minutes a runner's steps took for one owner's public
/// repositories, in the runner's account, keyed by the owner's DID.
pub const MINUTES: &str = "dev.knotless.minutes";
@@ -809,6 +812,22 @@ pub struct Freeing {
pub minutes: u64,
}
/// A repository owner's request that its runners cancel a ref's or a
/// pull's runs going or waiting, or one workflow's of them
/// (`lexicons/dev/knotless/cancel.json`). Keyed `<repo key>.<tid>`.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct CancelRecord {
pub repo: RepoRef,
#[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")]
pub ref_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pull: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workflow: Option<String>,
pub created_at: String,
}
/// A repository owner's request that its runners run a ref's or a pull's
/// workflows again, where it points now, or one workflow at a ref, with
/// variables of their own (`lexicons/dev/knotless/rerun.json`). Keyed
crates/knotless-mcp/src/lib.rs+32−0
View file
@@ -463,6 +463,19 @@ pub struct Rerun {
pub env: BTreeMap<String, String>,
}
#[derive(Deserialize, schemars::JsonSchema)]
pub struct RunCancel {
/// `handle/name`, `did/name` or `knotless://handle/name`.
pub repo: String,
/// The ref whose runs to cancel, such as `refs/heads/main`.
#[serde(rename = "ref")]
pub refname: Option<String>,
/// Or the pull whose runs to cancel, by its AT-URI.
pub pull: Option<String>,
/// Only this workflow's runs, by name.
pub workflow: Option<String>,
}
#[derive(Deserialize, schemars::JsonSchema)]
pub struct ReleaseFile {
/// Its name, as it is downloaded: letters, digits, `.`, `_`, `+` and
@@ -3703,6 +3716,25 @@ impl Server {
)
}
#[tool(
description = "Ask a repository's runners, as its owner, to cancel a ref's or a pull's runs going or waiting for their turn when they read it: give one of ref or pull. workflow cancels only that one's runs. Runs of a workflow whose runs finish, as a deploy's, are cancelled too; a merge queue's batches are not (queue_leave takes a pull out). Their checks are filed as cancelled, under run_list."
)]
async fn run_cancel(
&self,
Parameters(p): Parameters<RunCancel>,
) -> Result<CallToolResult, McpError> {
answer(
async {
let url = repo_url(&p.repo)?;
let (refname, pull) = (p.refname.as_deref(), p.pull.as_deref());
let workflow = p.workflow.as_deref();
let asked = manage::cancel(&self.ctx, &url, refname, pull, workflow).await?;
Ok(json!({"asked": asked}))
}
.await,
)
}
#[tool(
description = "The workflows of a repository at a ref (the default branch if none), in file name order, each with what runs it there: push, a push to it; pull, a pull into it; asked, whether its owner may ask for it by name with run_rerun, ref, workflow and env, as one that runs on a push there, whose `on: {asked: {refs: [...]}}` names it, or that has a schedule, at the default branch; schedule, the `[{cron}]` it runs on there, at the default branch only; and error, why it does not load. Gives the ref in full too."
)]
crates/knotless-mcp/tests/mcp.rs+11−0
View file
@@ -395,6 +395,17 @@ fn an_agent_works_on_a_repository() {
let rerun = s.record("dev.knotless.rerun", rkey);
assert_eq!(rerun["workflow"], "scenarios");
assert_eq!(rerun["env"], json!([{"name": "TRACKS", "value": "team"}]));
// Or to cancel what a ref or a pull has going, one workflow's or all.
let off = json!({"repo": repo, "ref": "main", "workflow": "deploy"});
let asked = c.json("run_cancel", off);
let rkey = asked["asked"].as_str().unwrap().rsplit('/').next().unwrap();
let cancel = s.record("dev.knotless.cancel", rkey);
assert_eq!(
(&cancel["ref"], &cancel["workflow"]),
(&json!("refs/heads/main"), &json!("deploy"))
);
let (text, err) = c.call("run_cancel", json!({"repo": repo}));
assert!(err && text.contains("name one ref or one pull"), "{text}");
// The workflows at a ref, and what runs each there: it has none.
let listed = c.json("workflow_list", json!({"repo": repo}));
crates/knotless-run/src/lib.rs+65−19
View file
@@ -25,7 +25,9 @@
//! after every other waiting, and its steps may take as long as their own
//! `timeout`s say, past the runner's limit; so as not to hold up a stop, as
//! a deploy's, a stop cancels it, and it is asked for again on the next
//! start.
//! start. A `dev.knotless.cancel` record of the owner's, read the same
//! way, cancels the runs of a ref or pull going or waiting then, or one
//! workflow's of them, a deploy's too; not a merge queue's batches.
//!
//! Runs go side by side, `--jobs` at once over every repository, so a
//! quick workflow's check is never kept waiting behind a slow one's, and
@@ -106,8 +108,8 @@ use knotless::jetstream;
use knotless::keys;
use knotless::mirror::Mirror;
use knotless::records::{
CheckImage, CheckOutput, Freeing, MinutesRecord, QUEUE_FIRST, RefRecord, RepoRecord, RepoRef,
RerunRecord, RerunVar, now, nsid,
CancelRecord, CheckImage, CheckOutput, Freeing, MinutesRecord, QUEUE_FIRST, RefRecord,
RepoRecord, RepoRef, RerunRecord, RerunVar, now, nsid,
};
use knotless::remote::{Remote, Snapshot, Stored};
use knotless::sealed;
@@ -526,6 +528,9 @@ struct State {
pulls: BTreeMap<String, BTreeMap<String, String>>,
#[serde(default)]
reruns: BTreeMap<String, String>,
/// The newest request to cancel acted on, by `<did>/<key>`.
#[serde(default)]
cancels: BTreeMap<String, String>,
#[serde(default)]
asked: BTreeMap<String, BTreeSet<String>>,
/// Of the pulls asked for, those asked for only some workflows, by
@@ -1131,7 +1136,7 @@ fn cancel(cancel: &Cancel, why: &'static str) {
/// Whether `why` cancels even a run kept, of a workflow whose runs finish.
fn heeded(why: &str) -> bool {
why == MOVED || why == DROPPED || why == CUT_OFF
why == MOVED || why == DROPPED || why == CUT_OFF || why == ASKED_OFF
}
/// What says a run is cancelled, and why.
@@ -1172,6 +1177,10 @@ const MOVED: &str = "as the pull is no longer at that head and base";
/// the branch moved, or a pull in it moved or left, or a batch it was on
/// failed.
const DROPPED: &str = "as the merge queue dropped its batch";
/// Why a run is cancelled: its repository's owner asked, with a
/// `dev.knotless.cancel` record.
const ASKED_OFF: &str = "as the repository's owner asked";
/// Why a run not started yet is cancelled as the runner stops, for the
/// next one to run.
const WAITED: &str = "as the runner stopped, to run once it starts again,";
@@ -1362,6 +1371,8 @@ impl Runner {
self.caches.drop_branches(&watch.id(), &live)?;
watch.at("reading reruns");
self.reruns(watch).await?;
watch.at("reading cancels");
self.cancels(watch).await?;
let moved = self.moved(watch, &snap)?;
if !moved.is_empty() && !watch.pulls_only {
watch.at("running refs");
@@ -1742,20 +1753,7 @@ impl Runner {
/// Forgets where each ref or pull the owner has asked to run again since
/// the last request acted on was last run, so the passes after run it.
async fn reruns(&self, watch: &Watch) -> Result<()> {
let found = watch
.remote
.scan_all::<serde_json::Value>(nsid::RERUN, &watch.key)
.await?;
// Keyed by a TID, so the newest acted on sorts before every request
// after it.
let asked: Vec<_> = found
.into_iter()
.filter(|r| {
keys::suffix(&watch.key, &r.rkey)
.and_then(keys::tid_micros)
.is_some()
})
.collect();
let asked = requests(watch, nsid::RERUN).await?;
let newest = asked.iter().map(|r| r.rkey.clone()).max();
let id = watch.id();
let mut state = self.state.lock().unwrap();
@@ -1816,6 +1814,39 @@ impl Runner {
self.save(&state)
}
/// Cancels the runs of each ref or pull the owner has asked to cancel
/// since the last request acted on, going or waiting for their turn,
/// even those of a workflow whose runs finish: not a merge queue's
/// batches.
async fn cancels(&self, watch: &Watch) -> Result<()> {
let asked = requests(watch, nsid::CANCEL).await?;
let newest = asked.iter().map(|r| r.rkey.clone()).max();
let id = watch.id();
let since = self.state.lock().unwrap().cancels.get(&id).cloned();
let wanted = asked
.into_iter()
.filter(|r| since.as_ref().is_some_and(|s| r.rkey > *s))
.filter_map(|r| serde_json::from_value::<CancelRecord>(r.value).ok());
for asked in wanted {
let Some(keyed) = asked.pull.or(asked.ref_name) else {
continue;
};
let of = format!("{id}\n{keyed}\n");
let named = |wf: &str| asked.workflow.as_deref().is_none_or(|w| w == wf);
for (run, going) in self.going.lock().unwrap().iter() {
if !going.batch && run.strip_prefix(&of).is_some_and(named) {
cancel(&going.cancel, ASKED_OFF);
}
}
}
let mut state = self.state.lock().unwrap();
let since = since.unwrap_or_default();
state
.cancels
.insert(id, newest.unwrap_or_default().max(since));
self.save(&state)
}
/// Where the writes in the space of `watch`, a private repository,
/// are now, if any were made after `since`.
async fn space_moved(&self, watch: &Watch, since: Option<&str>) -> Result<Option<String>> {
@@ -4253,6 +4284,21 @@ impl jetstream::Follower for Doorbell<'_> {
}
}
/// The requests of `collection` in the owner's account about `watch`'s
/// repository, keyed by its key and a TID, so the newest acted on sorts
/// before every request after it.
async fn requests(watch: &Watch, collection: &str) -> Result<Vec<Stored<serde_json::Value>>> {
let found = (watch.remote)
.scan_all::<serde_json::Value>(collection, &watch.key)
.await?;
let tid = |r: &Stored<serde_json::Value>| {
keys::suffix(&watch.key, &r.rkey)
.and_then(keys::tid_micros)
.is_some()
};
Ok(found.into_iter().filter(tid).collect())
}
fn ring(doors: &Doors, heads: &Heads, event: &jetstream::Event) {
let Some(did) = &event.did else {
return;
@@ -4260,7 +4306,7 @@ fn ring(doors: &Doors, heads: &Heads, event: &jetstream::Event) {
for op in &event.ops {
let rung = match op.collection.as_str() {
// A request to run again is keyed by the repository, as a ref is.
nsid::REF | nsid::RERUN => {
nsid::REF | nsid::RERUN | nsid::CANCEL => {
let key = op.rkey.split('.').next().unwrap_or_default();
let repo = (did.clone(), key.to_string());
let mut rung: Vec<_> = heads.get(&repo).into_iter().flatten().cloned().collect();
crates/knotless-run/tests/run.rs+152−16
View file
@@ -1186,22 +1186,12 @@ fn what_goes_wrong_is_reported() {
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 {
/// A runner of `owner`'s repository `name` that runs one job at a time,
/// on this machine's shell, its passes made by the test.
fn one_job(runner: &Sandbox, owner: &Sandbox, name: &str) -> knotless_run::Options {
knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![format!("{}/ship", owner.account.handle)],
repos: vec![format!("{}/{name}", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
@@ -1235,7 +1225,23 @@ fn a_deploy_is_never_cut_off() {
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
}
}
/// 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 = one_job(&runner, &owner, "ship");
let rt = tokio::runtime::Runtime::new().unwrap();
let (run, watches) = rt
.block_on(knotless_run::Runner::start(opts, common::ctx(&runner)))
@@ -1322,6 +1328,136 @@ fn a_deploy_is_never_cut_off() {
);
}
/// The owner cancels what a ref has going: one workflow's run waiting for
/// the job, then a deploy's running, which a newer commit's never cuts
/// off. A request is acted on once, and one naming nothing asks nothing.
#[test]
fn an_owner_cancels_runs() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("halt", 1);
let url = owner.url("halt");
owner.git(&work, &["remote", "add", "origin", &url]);
owner.git(&work, &["push", "-q", "origin", "main"]);
assert_eq!(owner.cli(&["runner", "add", &url, &handle], b"").0, 0);
let opts = one_job(&runner, &owner, "halt");
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();
// Asked wrongly, it asks nothing.
for wrong in [
&["--workflow", "ci"][..],
&[
"--ref",
"main",
"--pull",
"at://did:plc:a/dev.knotless.pull/b",
],
&["--ref", "main", "--workflow", "no good"],
&["--ref", "nowhere"],
&["--ref", "main", "--bogus"],
] {
let args = [&["run", "cancel", url.as_str()][..], wrong].concat();
assert_eq!(owner.cli(&args, b"").0, 1, "{wrong:?}");
}
assert!(owner.records("dev.knotless.cancel").is_empty());
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()
);
let ci = "on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n";
std::fs::write(dir.join("block.yml"), block).unwrap();
std::fs::write(dir.join("deploy.yml"), deploy).unwrap();
std::fs::write(dir.join("ci.yml"), ci).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());
owner.git(&work, &["push", "-q", "origin", "HEAD:main"]);
let sha = owner.git(&work, &["rev-parse", "HEAD"]);
rt.block_on(run.reconcile(watch)).unwrap();
let at = |sha: &str, workflow: &str| {
let rkey = check::rkey(
&owner.account.did,
&key,
"refs/heads/main",
sha,
None,
workflow,
);
runner.record(check::NSID, &rkey)
};
let log = |check: &serde_json::Value| {
let cid = check["log"]["ref"]["$link"].as_str().unwrap();
blob(&runner.account.did, cid)
};
wait_until(|| at(&sha, "ci")["status"] == "queued");
// One naming neither a ref nor a pull is passed over.
let repo = format!("at://{}/dev.knotless.repo/halt", owner.account.did);
let nothing = serde_json::json!({"$type": "dev.knotless.cancel", "repo": {"uri": repo, "key": key}, "createdAt": "2026-11-04T00:00:00Z"});
let rkey = format!("{key}.{}", knotless::keys::tid());
owner.put_record("dev.knotless.cancel", &rkey, nothing);
// One workflow's run, waiting for the job.
let args = ["run", "cancel", &url, "--ref", "main", "--workflow", "ci"];
let (code, asked) = owner.cli(&args, b"");
assert_eq!(code, 0, "{asked}");
assert!(asked.contains("/dev.knotless.cancel/"), "{asked}");
rt.block_on(run.reconcile(watch)).unwrap();
wait_until(|| at(&sha, "ci")["finishedAt"].is_string());
let off = at(&sha, "ci");
assert_eq!(off["status"], "cancelled", "{off}");
assert_eq!(
log(&off),
"==> cancelled as the repository's owner asked before it started\n"
);
assert_eq!(at(&sha, "deploy")["status"], "queued");
// Then all of the ref's: the deploy, running, though its runs finish.
std::fs::write(go.join("block"), "").unwrap();
wait_until(|| at(&sha, "deploy")["status"] == "running");
let args = ["run", "cancel", &url, "--ref", "refs/heads/main"];
assert_eq!(owner.cli(&args, b"").0, 0);
rt.block_on(run.reconcile(watch)).unwrap();
wait_until(|| at(&sha, "deploy")["finishedAt"].is_string());
let off = at(&sha, "deploy");
assert_eq!(off["status"], "cancelled", "{off}");
let said = log(&off);
assert!(
said.ends_with("==> cancelled as the repository's owner asked\n"),
"{said}"
);
// Acted on once: a newer commit's runs go as ever.
let next = owner.commit_file(&work, "f", "next\n");
owner.git(&work, &["push", "-q", "origin", "HEAD:main"]);
std::fs::write(go.join(&next), "").unwrap();
rt.block_on(run.reconcile(watch)).unwrap();
rt.block_on(run.idle());
for workflow in ["ci", "deploy"] {
let ran = at(&next, workflow);
assert_eq!(ran["status"], "success", "{ran}");
}
}
/// A workflow that runs `after` others waits for their checks at its
/// commit holding no job, so the one job goes to them; runs once they
/// pass; and does not run when one fails or is cancelled.
crates/knotless-serve/src/index.rs+1−1
View file
docs/design/runner.md+23−6
View file
docs/design/security.md+1−1
View file
docs/reference/actions.md+1−0
View file
docs/reference/cli.md+10−0
View file
docs/reference/mcp.md+1−0
View file
docs/reference/runner-internals.md+2−0
View file
lexicons/dev/knotless/cancel.jsonnew+23−0
View file