chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Relays down are asked less often, and batches skip by the pull's paths
crates/knotless-run/src/lib.rs+3−2
| @@ -2086,8 +2086,9 @@ impl Runner { | ||
continue; | ||
} | ||
// One the change leaves out has passed, saying so, for whatever | ||
// waits for it. | ||
let pull = subject.pull.is_some(); | ||
// waits for it. A batch is pulls landing, so it is left out by | ||
// the pull trigger's paths, as they are. | ||
let pull = subject.pull.is_some() || subject.batch.is_some(); | ||
let skipped = wf.skipped(pull, subject.changed.as_deref()); | ||
if let Some(why) = skipped.filter(|_| !asked) { | ||
check.status = status::SUCCESS.into(); | ||
crates/knotless-run/src/queue.rs+3−2
| @@ -21,8 +21,9 @@ | ||
//! and ejected: a half that passes lands, and what is left of a batch that | ||
//! failed is then the same tree on the tip it moved to, so it fails too, | ||
//! and a lone pull left is ejected without running again. The branch | ||
//! moving any other way, or a pull in a batch moving or leaving, drops the | ||
//! batches going, to be built again. | ||
//! moving any other way, or a pull in a batch leaving the queue, drops the | ||
//! batches going, to be built again. A pull whose head moves while its | ||
//! batch runs is caught as the batch lands, which then drops what is going. | ||
//! | ||
//! While a batch runs, the next is built on its tip and run, as if it | ||
//! passed (`--depth` 2, the default; 1 runs one at a time). It runs ahead | ||
crates/knotless-serve/src/firehose.rs+31−3
| @@ -6,6 +6,7 @@ use crate::App; | ||
use knotless::drisl::{link, record}; | ||
use knotless::jetstream::knotless; | ||
pub use knotless::jetstream::{Event, IDLE, Op, Socket, frames}; | ||
use knotless::jetstream::{RETRY, RETRY_MAX}; | ||
use std::collections::HashMap; | ||
use std::sync::Arc; | ||
use std::time::Duration; | ||
| @@ -129,12 +130,18 @@ pub fn endpoint(base: &str, cursor: Option<i64>) -> String { | ||
} | ||
| ||
/// Follows the firehose until the process ends, reconnecting from the last | ||
/// sequence number. While disconnected, mirrors fall back to the TTL. | ||
/// sequence number. While disconnected, mirrors fall back to the TTL. A | ||
/// connection that ends is made again after [`RETRY`], and one refused | ||
/// waits twice as long as the last, up to [`RETRY_MAX`], so a relay that is | ||
/// down is not asked twice a second. | ||
pub async fn watch(app: Arc<App>, base: String) { | ||
let mut cursor = None; | ||
let mut wait = RETRY; | ||
app.follows("firehose", &base); | ||
loop { | ||
if let Ok((mut ws, _)) = tokio_tungstenite::connect_async(endpoint(&base, cursor)).await { | ||
let connected = tokio_tungstenite::connect_async(endpoint(&base, cursor)).await; | ||
wait = next_wait(wait, connected.is_ok()); | ||
if let Ok((mut ws, _)) = connected { | ||
app.connected("firehose", true); | ||
frames(&mut ws, IDLE, |msg| { | ||
if let Message::Binary(frame) = msg | ||
| @@ -148,7 +155,17 @@ pub async fn watch(app: Arc<App>, base: String) { | ||
.await; | ||
} | ||
app.connected("firehose", false); | ||
tokio::time::sleep(Duration::from_millis(500)).await; | ||
tokio::time::sleep(wait).await; | ||
} | ||
} | ||
| ||
/// How long to wait before the next attempt: [`RETRY`] after a connection, | ||
/// and after a refusal twice `wait`, up to [`RETRY_MAX`]. | ||
fn next_wait(wait: Duration, connected: bool) -> Duration { | ||
if connected { | ||
RETRY | ||
} else { | ||
(wait * 2).min(RETRY_MAX) | ||
} | ||
} | ||
| ||
| @@ -157,6 +174,17 @@ mod tests { | ||
use super::*; | ||
use ciborium::Value as V; | ||
| ||
#[test] | ||
fn a_refused_relay_is_asked_less_often() { | ||
let mut wait = RETRY; | ||
for _ in 0..20 { | ||
wait = next_wait(wait, false); | ||
} | ||
assert_eq!(wait, RETRY_MAX); | ||
assert_eq!(next_wait(RETRY, false), RETRY * 2); | ||
assert_eq!(next_wait(RETRY_MAX, true), RETRY); | ||
} | ||
| ||
fn frame(header: V, body: V) -> Vec<u8> { | ||
let mut out = Vec::new(); | ||
ciborium::into_writer(&header, &mut out).unwrap(); | ||
crates/knotless-serve/src/oci.rs+79−6
| @@ -108,9 +108,55 @@ struct Registry { | ||
tokens: Mutex<HashMap<String, (Arc<Pusher>, Instant)>>, | ||
/// The uploads under way, by id. | ||
uploads: Mutex<HashMap<String, Upload>>, | ||
/// The blobs pushed and uploaded, not yet in a release, by account and | ||
/// digest. | ||
staged: Mutex<HashMap<(String, String), ReleaseFile>>, | ||
/// The blobs pushed and uploaded, not yet in a release. | ||
staged: Mutex<Staged>, | ||
} | ||
| ||
/// The most blobs one account has staged at once: past it, its oldest is | ||
/// forgotten, so one account's pushes never cost another's. | ||
const STAGED_EACH: usize = 256; | ||
| ||
/// The blobs pushed and uploaded, not yet in a release, by account and | ||
/// digest, each with when it was staged, counting up. Past [`STAGED_EACH`] | ||
/// for an account, or [`MOST_KEPT`] in all, the oldest is forgotten: a push | ||
/// still under way stages its blobs last, so it keeps them. | ||
#[derive(Default)] | ||
struct Staged { | ||
files: HashMap<(String, String), (u64, ReleaseFile)>, | ||
next: u64, | ||
} | ||
| ||
impl Staged { | ||
fn get(&self, did: &str, digest: &str) -> Option<&ReleaseFile> { | ||
let key = (did.to_string(), digest.to_string()); | ||
self.files.get(&key).map(|(_, file)| file) | ||
} | ||
| ||
fn insert(&mut self, did: &str, digest: String, file: ReleaseFile) { | ||
let own = self.files.keys().filter(|(d, _)| d == did).count(); | ||
if own >= STAGED_EACH { | ||
self.forget_oldest(Some(did)); | ||
} | ||
if self.files.len() >= MOST_KEPT { | ||
self.forget_oldest(None); | ||
} | ||
self.next += 1; | ||
self.files | ||
.insert((did.to_string(), digest), (self.next, file)); | ||
} | ||
| ||
/// Forgets the oldest staged, of `did` if given. | ||
fn forget_oldest(&mut self, did: Option<&str>) { | ||
let oldest = self | ||
.files | ||
.iter() | ||
.filter(|((d, _), _)| did.is_none_or(|did| d == did)) | ||
.min_by_key(|(_, (at, _))| *at) | ||
.map(|(key, _)| key.clone()); | ||
if let Some(key) = oldest { | ||
self.files.remove(&key); | ||
} | ||
} | ||
} | ||
| ||
/// The registry's routes, reading as `ctx`, its tokens given out at | ||
| @@ -277,7 +323,7 @@ impl Registry { | ||
/// The blob of `digest` that `did` pushed and no release holds yet. | ||
fn staged(&self, did: &str, digest: &str) -> Option<ReleaseFile> { | ||
let staged = self.staged.lock().unwrap(); | ||
staged.get(&(did.to_string(), digest.to_string())).cloned() | ||
staged.get(did, digest).cloned() | ||
} | ||
| ||
/// Uploads the file at `path`, of `size` bytes and SHA-256 `sha`, to | ||
| @@ -303,8 +349,7 @@ impl Registry { | ||
parts: blobs.into_iter().map(|blob| ReleasePart { blob }).collect(), | ||
}; | ||
let mut staged = self.staged.lock().unwrap(); | ||
bounded(&mut staged); | ||
staged.insert((pusher.did.clone(), format!("sha256:{sha}")), file); | ||
staged.insert(&pusher.did, format!("sha256:{sha}"), file); | ||
Ok(()) | ||
} | ||
| ||
| @@ -880,4 +925,32 @@ mod tests { | ||
.is_none() | ||
); | ||
} | ||
| ||
#[test] | ||
fn staging_forgets_only_the_oldest() { | ||
let file = |name: &str| ReleaseFile { | ||
name: name.into(), | ||
size: 1, | ||
sha256: name.into(), | ||
parts: Vec::new(), | ||
}; | ||
let mut staged = Staged::default(); | ||
staged.insert("did:a", "a0".into(), file("a0")); | ||
// One account past its share loses only its own oldest. | ||
for i in 0..STAGED_EACH { | ||
staged.insert("did:b", format!("b{i}"), file("b")); | ||
} | ||
staged.insert("did:b", "last".into(), file("b")); | ||
assert!(staged.get("did:b", "b0").is_none()); | ||
assert!(staged.get("did:b", "b1").is_some()); | ||
assert!(staged.get("did:b", "last").is_some()); | ||
assert!(staged.get("did:a", "a0").is_some()); | ||
// Many accounts past the whole lose the oldest of all. | ||
for i in 0..MOST_KEPT { | ||
staged.insert(&format!("did:{i}"), "x".into(), file("x")); | ||
} | ||
assert_eq!(staged.files.len(), MOST_KEPT); | ||
assert!(staged.get("did:a", "a0").is_none()); | ||
assert!(staged.get(&format!("did:{}", MOST_KEPT - 1), "x").is_some()); | ||
} | ||
} | ||
docs/design/runner.md+3−1
| @@ -246,7 +246,9 @@ A `push` or `pull` trigger may list `paths` globs, and the workflow runs | ||
only if a changed file matches one ([syntax](../reference/workflows.md#paths)). | ||
For a push, the change set runs from where the runner's mirror had the ref | ||
before this pass synced it to the new target. When that is unknown, as | ||
for a new ref, nothing is skipped. | ||
for a new ref, nothing is skipped. A merge-queue batch is | ||
skipped as a pull is, by the `pull` trigger's `paths`, against what the | ||
batch changes from the branch's tip. | ||
| ||
A workflow left out is filed at once as `success`, every step `skipped`, | ||
with a note naming the paths. It takes no job and no cache, and the merge | ||
docs/design/serve.md+2−2
| @@ -226,8 +226,8 @@ Details: | ||
connection that ends is made again in 500 ms, the wait doubling after | ||
each failed attempt up to 60 s. knotless-run and knotless-triage follow | ||
Jetstream through the same loop, `knotless::jetstream::follow`, without | ||
a cursor. The firehose is connected again 500 ms after each end or | ||
failed attempt, with no doubling. | ||
a cursor. The firehose waits the same way: 500 ms after a connection | ||
ends, doubling after each failed attempt up to 60 s. | ||
- With an index, the Jetstream cursor is saved per host, so a restart misses | ||
nothing Jetstream still keeps. The firehose's cursor is kept only in | ||
memory: a reconnect resumes from it, and a restart starts at the relay's | ||
docs/reference/workflows.md+3−2
| @@ -156,8 +156,9 @@ any changed file. What counts as changed: | ||
A workflow no changed file matches is filed at once as `success`, with its | ||
steps `skipped` and the note `skipped: nothing changed in <paths>`, cut | ||
to 1000 characters. A run asked for by name is never skipped this way. A | ||
merge-queue batch picks its workflows by `pull.base` but skips by the | ||
`push` trigger's `paths`. | ||
merge-queue batch is pulls landing: it picks its workflows by `pull.base` | ||
and skips by the `pull` trigger's `paths`, against what the batch changes | ||
from the branch's tip. | ||
| ||
### The merge queue | ||
| ||
lexicons/dev/knotless/check.json+4−4
| @@ -5,7 +5,7 @@ | ||
"defs": { | ||
"main": { | ||
"type": "record", | ||
"description": "Kept in the runner's account, keyed `<owner hash>.<repo key>.<ref hash>.<commit>.<workflow>`, where each hash is the first 26 characters of the lowercase unpadded base32 SHA-256 of the owner's DID or the full ref name (for a pull's run, the pull's AT-URI), so each commit keeps its own check of each workflow, replaced only when that commit runs again. A pull's run is keyed `<owner hash>.<repo key>.<pull hash>.<commit>.<baseCommit>.<workflow>`, so a run against a newer base is a check of its own. A runner may delete checks of no more use: a batch's, a closed pull's, ones superseded by a newer run of the same workflow, and the default branch's after a while. Readers trust only a runner the repository names.", | ||
"description": "Kept in the runner's account, keyed `<owner hash>.<repo key>.<ref hash>.<commit>.<workflow>`, where each hash is the first 26 characters of the lowercase unpadded base32 SHA-256 of the owner's DID or the full ref name (for a pull's run, the pull's AT-URI), so each commit keeps its own check of each workflow, replaced only when that commit runs again. A pull's run is keyed `<owner hash>.<repo key>.<pull hash>.<commit>.<baseCommit>.<workflow>`, so a run against a newer base is a check of its own. A merge queue's batch's run is keyed the same way, with the hash of `queue <branch>` (the branch's full ref name) in place of the pull's, its last commit and the branch's tip it was built on. A runner may delete checks of no more use: a batch's, a closed pull's, ones superseded by a newer run of the same workflow, and the default branch's after a while. Readers trust only a runner the repository names.", | ||
"key": "any", | ||
"record": { | ||
"type": "object", | ||
| @@ -21,10 +21,10 @@ | ||
"items": { "type": "ref", "ref": "#batched" } | ||
}, | ||
"commit": { "type": "string", "minLength": 40, "maxLength": 64, "description": "The hex object ID the ref pointed at." }, | ||
"baseCommit": { "type": "string", "minLength": 40, "maxLength": 64, "description": "For a pull's run, the hex object ID of the tip of the branch the pull is into that its head was merged with: the run is of that merge, as merging the pull would make it, so it is current only while both the head and the base are where they were." }, | ||
"baseCommit": { "type": "string", "minLength": 40, "maxLength": 64, "description": "For a pull's run, the hex object ID of the tip of the branch the pull is into that its head was merged with: the run is of that merge, as merging the pull would make it, so it is current only while both the head and the base are where they were. For a batch's run, the tip of the branch the batch was built on." }, | ||
"generation": { "type": "integer", "minimum": 1, "description": "The ref's generation at that commit." }, | ||
"workflow": { "type": "string", "maxLength": 64, "description": "The workflow's name, which is also the last part of the record key." }, | ||
"status": { "type": "string", "maxLength": 64, "knownValues": ["queued", "running", "success", "failure", "error", "cancelled"], "description": "`queued` is a run filed and waiting for a job on its runner, no step started yet: it is filed `running` once it starts, its `startedAt` then. `failure` is a step that exited non-zero or ran out of time, or a pull's head that conflicts with its base; `error` is the runner failing to run the workflow at all; `cancelled` is a run stopped for a newer commit's run of the workflow on the same ref or pull. A `success` with `skipped` ran no step, as the change touched none of the paths the workflow names, and passes as any `success` does." }, | ||
"status": { "type": "string", "maxLength": 64, "knownValues": ["queued", "running", "success", "failure", "error", "cancelled"], "description": "`queued` is a run filed and waiting for a job on its runner, no step started yet: it is filed `running` once it starts, its `startedAt` then. `failure` is a step that exited non-zero or ran out of time, or a pull's head that conflicts with its base; `error` is the runner failing to run the workflow at all; `cancelled` is a run stopped before it finished: for a newer commit's run of the workflow on the same ref or pull, as its pull moved to another head or base or closed, as the merge queue dropped its batch, or as the runner stopped. A `success` with `skipped` ran no step, as the change touched none of the paths the workflow names, and passes as any `success` does." }, | ||
"steps": { | ||
"type": "array", | ||
"description": "Each step the workflow has, in order. Absent for `error`.", | ||
| @@ -47,7 +47,7 @@ | ||
"properties": { | ||
"name": { "type": "string", "maxLength": 256, "description": "The step's name in the workflow, or `step <n>` for one with none." }, | ||
"status": { "type": "string", "maxLength": 64, "knownValues": ["success", "failure", "skipped", "cancelled"], "description": "`cancelled` is the step running when its run was cancelled." }, | ||
"exitCode": { "type": "integer", "description": "Absent when the step was killed for running out of time." }, | ||
"exitCode": { "type": "integer", "description": "Absent when the step was killed, for running out of time or as its run was cancelled." }, | ||
"durationMs": { "type": "integer", "minimum": 0, "description": "How long the step ran, in milliseconds." } | ||
} | ||
}, | ||