~ 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

Gateway pushes log each part's time; landings read a space side by side

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:push-timing
.changeset/private-forge-reads-side-by-side.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: patch
---
# A private repository's pulls and issues are read with fewer round trips
Reading a private repository's pulls, issues and the rest from its space reads each member's collections side by side, eight at a time, rather than one after another: a space of a few members was dozens of round trips in a row.
crates/knotless/src/auto_merge.rs+9−3
View file
@@ -161,8 +161,15 @@ pub async fn cancel(ctx: &Ctx, url: &RepoUrl, pull: &str) -> Result<()> {
/// not otherwise push, are the merge of one open pull set to merge when its
/// checks pass, as [`forge::merge`] makes it: its branch moved forward,
/// without force, by one commit on the branch's tip, with that head as a
/// second parent or none, holding the tree the two make merged. Its set must stand at that head. `git` holds the pushed commits.
pub async fn lands(ctx: &Ctx, t: &Target, git: &Git, updates: &[Update]) -> Result<bool> {
/// second parent or none, holding the tree the two make merged. Its set must stand at that head. `git` holds the pushed commits;
/// `items` are the repository's pulls, as its queue reads them.
pub async fn lands(
ctx: &Ctx,
t: &Target,
git: &Git,
updates: &[Update],
items: &[Item],
) -> Result<bool> {
let [moved] = updates else {
return Ok(false);
};
@@ -180,7 +187,6 @@ pub async fn lands(ctx: &Ctx, t: &Target, git: &Git, updates: &[Update]) -> Resu
.await?;
let parents: Vec<&str> = parents.split(' ').skip(1).collect();
let tree = git.rev_parse(&format!("{to}^{{tree}}")).await?;
let items = forge::assemble(&t.record, t.remote.did(), &t.queue_entries(ctx).await?);
let set = items.iter().filter(|i| {
let into = i.base.as_deref() == Some(branch);
into && i.state == "open" && i.auto_merge.is_some()
crates/knotless/src/forge.rs+39−25
View file
@@ -83,6 +83,10 @@ const FRESH: [&str; 3] = [nsid::QUEUE, nsid::STATE, nsid::AUTO_MERGE];
const FRESH_PAGE: usize = 25;
const FRESH_WAIT: Duration = Duration::from_secs(5);
/// How many of the collections of a private repository's members
/// [`Target::space_entries`] reads at once: as many as a fetch downloads.
const SPACE_READS: usize = 8;
/// `entries`, with each of `fresh` in place of its copy among them.
pub(crate) fn freshened(mut entries: Vec<Entry>, fresh: Vec<Entry>) -> Vec<Entry> {
let uris: std::collections::HashSet<String> = fresh.iter().map(|e| e.uri.clone()).collect();
@@ -1043,8 +1047,12 @@ impl Target {
Ok(found.filter(|f| self.is_about(&f.value)))
}
/// Reads every member's repo in the space.
/// Reads every member's repo in the space, each of its collections
/// beside the others, [`SPACE_READS`] at a time: one after another, a
/// space of a few members was dozens of round trips, as a runner's
/// landing waited on them at the gateway.
async fn space_entries(&self, ctx: &Ctx) -> Result<Vec<Entry>> {
use futures::{StreamExt, TryStreamExt};
#[derive(Deserialize)]
struct Repos {
repos: Vec<SpaceRepo>,
@@ -1054,6 +1062,16 @@ impl Target {
struct SpaceRepo {
did: String,
}
/// `member`'s records in `collection`, as entries of theirs.
async fn listed(member: &Remote, collection: &'static str) -> Result<Vec<Entry>> {
let found = member.list_all::<Value>(collection).await?;
let found = found.into_iter().map(|r| Entry {
uri: member.uri_of(collection, &r.rkey),
author: member.did().to_string(),
value: r.value,
});
Ok(found.collect())
}
let space = self.joined(ctx).await?;
let uri = space.uri.clone();
let mut dids = Vec::new();
@@ -1078,30 +1096,26 @@ impl Target {
}
dids.sort();
dids.dedup();
let mut entries = Vec::new();
let mut reactions = Vec::new();
for did in dids {
let member = self.member(ctx, &did, Some(&space)).await?;
for collection in nsid::FORGE {
for r in member.list_all::<Value>(collection).await? {
if self.is_about(&r.value) {
entries.push(Entry {
uri: member.uri_of(collection, &r.rkey),
author: did.clone(),
value: r.value,
});
}
}
}
// A reaction names what it is about, not the repository.
for r in member.list_all::<Value>(nsid::REACTION).await? {
reactions.push(Entry {
uri: member.uri_of(nsid::REACTION, &r.rkey),
author: did.clone(),
value: r.value,
});
}
}
let members = dids.iter().map(|did| self.member(ctx, did, Some(&space)));
let members = futures::future::try_join_all(members).await?;
// The reads are all made before any is awaited: a closure over the
// members held across the wait leaves a request's future unproven
// to be Send.
let collections = || nsid::FORGE.into_iter().chain([nsid::REACTION]);
let pairs = members
.iter()
.flat_map(|m| collections().map(move |c| (m, c)));
let pairs: Vec<(&Remote, &'static str)> = pairs.collect();
let reads: Vec<_> = pairs.iter().map(|&(m, c)| listed(m, c)).collect();
let reads = futures::stream::iter(reads).buffered(SPACE_READS);
let lists: Vec<Vec<Entry>> = reads.try_collect().await?;
let read: Vec<(&str, Vec<Entry>)> = pairs.into_iter().map(|(_, c)| c).zip(lists).collect();
// A reaction names what it is about, not the repository.
let (reactions, read): (Vec<_>, Vec<_>) =
read.into_iter().partition(|(c, _)| *c == nsid::REACTION);
let read = read.into_iter().flat_map(|(_, found)| found);
let mut entries: Vec<Entry> = read.filter(|e| self.is_about(&e.value)).collect();
let mut reactions: Vec<Entry> = reactions.into_iter().flat_map(|(_, r)| r).collect();
let about = |e: &Entry| entries.iter().any(|i| i.uri == e.value["subject"]["uri"]);
reactions.retain(about);
entries.extend(reactions);
crates/knotless/src/push.rs+27−3
View file
@@ -16,6 +16,7 @@ use futures::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet};
use std::time::{Duration, Instant};
/// One refspec of a push. `src: None` deletes `dst`.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -56,6 +57,16 @@ struct Accepted {
previous: Option<Stored<RefRecord>>,
}
/// How long the parts of a push took, for a caller that logs them: checking
/// the updates and making the pack, uploading it as blobs, and writing the
/// records. A part it never reached is `None`.
#[derive(Default)]
pub struct Spent {
pub pack: Option<Duration>,
pub upload: Option<Duration>,
pub write: Option<Duration>,
}
pub async fn push(
ctx: &Ctx,
remote: &Remote,
@@ -64,11 +75,14 @@ pub async fn push(
auth: &Auth,
updates: &[Update],
) -> Result<Report> {
push_by(ctx, remote, snap, git, auth, updates, None).await
let spent = &mut Spent::default();
push_by(ctx, remote, snap, git, auth, updates, None, spent).await
}
/// [`push`], for `pusher` when it is not the owner: a gateway writing for a
/// collaborator or a runner names them on each claim and ref it writes.
/// collaborator or a runner names them on each claim and ref it writes, and
/// keeps in `spent` how long each part took.
#[allow(clippy::too_many_arguments)]
pub async fn push_by(
ctx: &Ctx,
remote: &Remote,
@@ -77,7 +91,9 @@ pub async fn push_by(
auth: &Auth,
updates: &[Update],
pusher: Option<&str>,
spent: &mut Spent,
) -> Result<Report> {
let started = Instant::now();
let pusher = pusher
.filter(|&did| did != remote.did())
.map(str::to_string);
@@ -180,7 +196,9 @@ pub async fn push_by(
None
} else {
let rkey = keys::pack_rkey(&key);
let uploading = Instant::now();
let record = place(ctx, remote, auth, &key, pack, tips, bases, objects).await?;
spent.upload = Some(uploading.elapsed());
writes.push(remote::create(
nsid::PACK,
&rkey,
@@ -189,6 +207,9 @@ pub async fn push_by(
Some(rkey)
}
};
// What came before the upload: the updates checked, the pack made.
let uploaded = spent.upload.unwrap_or_default();
spent.pack = Some(started.elapsed() - uploaded);
let now = records::now();
let mut ref_writes = Vec::new();
@@ -247,13 +268,16 @@ pub async fn push_by(
bail!("the refs were read too long ago to push against: fetch, then push again");
}
let batches = batch(writes, ref_writes);
let writing = Instant::now();
let mut done = 0;
for (i, writes) in batches.iter().enumerate() {
let refs_in_batch = writes
.iter()
.filter(|w| w["collection"] == nsid::REF)
.count();
if let Err(err) = remote.apply(writes.clone(), auth).await {
let applied = remote.apply(writes.clone(), auth).await;
spent.write = Some(writing.elapsed());
if let Err(err) = applied {
// Out of write points: what landed stays, and pushing the same
// refs again once they come back writes only the rest.
if let Some(limited) = XrpcError::of(&err).filter(|e| e.status == 429) {
crates/knotless/src/queue.rs+4−2
View file
@@ -1063,7 +1063,10 @@ pub async fn landing_refusal(
git: &Git,
updates: &[Update],
) -> Result<Option<String>> {
if crate::auto_merge::lands(ctx, t, git, updates).await? {
// What the records say of the queue is read once, for both shapes a
// landing takes: in a space, that is every member's records there.
let items = forge::assemble(&t.record, t.remote.did(), &t.queue_entries(ctx).await?);
if crate::auto_merge::lands(ctx, t, git, updates, &items).await? {
return Ok(None);
}
let (moves, deletes): (Vec<&Update>, Vec<&Update>) =
@@ -1085,7 +1088,6 @@ pub async fn landing_refusal(
let line = git
.text(&["rev-list", "--reverse", "--parents", &range], None)
.await?;
let items = forge::assemble(&t.record, t.remote.did(), &t.queue_entries(ctx).await?);
let lines = by_branch(&items);
// Each pull queued into it with its head, where its branch still is.
let mut waiting: Vec<(&Item, String)> = Vec::new();
crates/knotless/src/telemetry.rs+26−1
View file
@@ -56,8 +56,24 @@ pub struct Fields<'a> {
pub max_ms: Option<u64>,
pub lag_ms: Option<u64>,
pub duration_ms: Option<u64>,
/// Of a mirror's sync, the part spent reading its records.
/// Of a mirror's sync, or a push through the gateway, the part spent
/// reading the repository's records.
pub records_ms: Option<u64>,
/// Of a push through the gateway, each part it took in turn: checking
/// the caller's token, waiting for the repository's turn, syncing the
/// gateway's mirror, indexing the pack sent, holding the updates to
/// the branch rules and the merge gate, checking them and making the
/// pack to write, uploading it as blobs, and writing the records.
pub auth_ms: Option<u64>,
pub turn_ms: Option<u64>,
pub mirror_ms: Option<u64>,
pub index_ms: Option<u64>,
pub gate_ms: Option<u64>,
pub pack_ms: Option<u64>,
pub upload_ms: Option<u64>,
pub write_ms: Option<u64>,
/// How large something was, as the pack a push sent.
pub bytes: Option<u64>,
pub exit_code: Option<i64>,
pub connected: Option<bool>,
pub rebuilt: Option<bool>,
@@ -99,6 +115,15 @@ macro_rules! event {
lag_ms = $f.lag_ms,
duration_ms = $f.duration_ms,
records_ms = $f.records_ms,
auth_ms = $f.auth_ms,
turn_ms = $f.turn_ms,
mirror_ms = $f.mirror_ms,
index_ms = $f.index_ms,
gate_ms = $f.gate_ms,
pack_ms = $f.pack_ms,
upload_ms = $f.upload_ms,
write_ms = $f.write_ms,
bytes = $f.bytes,
exit_code = $f.exit_code,
connected = $f.connected,
rebuilt = $f.rebuilt,
crates/knotless-serve/src/gateway.rs+95−0
View file
@@ -37,6 +37,7 @@ use knotless::queue;
use knotless::records::RepoRecord;
use knotless::remote::{Remote, Snapshot};
use knotless::session::{self, Session};
use knotless::telemetry::{Fields, Level, millis, note, slow};
use knotless::url::RepoUrl;
use knotless::{identity, oauth, org};
use serde::Deserialize;
@@ -576,6 +577,91 @@ async fn mirror(app: &App, remote: &Remote, snap: &Snapshot) -> Result<Mirror, A
Ok(mirror)
}
/// How long each part of a push through the gateway took, in turn, logged
/// as one `gateway push` line when it ends, however it ends: answered,
/// refused, or given up by a client that stopped waiting. A part it never
/// reached is left out, so the part it ended in is what `duration_ms` has
/// beyond the rest.
struct Timed {
started: Instant,
last: Instant,
repo: String,
caller: String,
refs: u64,
bytes: u64,
auth: u64,
turn: Option<u64>,
records: Option<u64>,
mirror: Option<u64>,
index: Option<u64>,
gate: Option<u64>,
spent: push::Spent,
}
impl Timed {
/// A push `caller` started at `started`, checked as theirs by now, of
/// `refs` updates and a pack of `bytes` to the repository `owner`'s
/// `name`.
fn new(
started: Instant,
caller: &str,
owner: &str,
name: &str,
refs: usize,
bytes: usize,
) -> Timed {
let now = Instant::now();
Timed {
started,
last: now,
repo: format!("knotless://{owner}/{name}"),
caller: caller.to_string(),
refs: refs as u64,
bytes: bytes as u64,
auth: millis(now - started),
turn: None,
records: None,
mirror: None,
index: None,
gate: None,
spent: push::Spent::default(),
}
}
/// How long since the part before ended, ending this one.
fn mark(&mut self) -> Option<u64> {
let now = Instant::now();
let took = now - self.last;
self.last = now;
Some(millis(took))
}
}
impl Drop for Timed {
fn drop(&mut self) {
let took = self.started.elapsed();
let f = Fields {
repo: Some(&self.repo),
did: Some(&self.caller),
count: Some(self.refs),
bytes: Some(self.bytes),
duration_ms: Some(millis(took)),
auth_ms: Some(self.auth),
turn_ms: self.turn,
records_ms: self.records,
mirror_ms: self.mirror,
index_ms: self.index,
gate_ms: self.gate,
pack_ms: self.spent.pack.map(millis),
upload_ms: self.spent.upload.map(millis),
write_ms: self.spent.write.map(millis),
..Fields::default()
};
let level = [Level::INFO, Level::WARN][usize::from(slow(took))];
note(level, "gateway push", &f);
}
}
pub async fn push(
State(app): State<Arc<App>>,
headers: HeaderMap,
@@ -584,6 +670,8 @@ pub async fn push(
let started = Instant::now();
let caller = caller(&app, &headers, PUSH).await?;
let (header, pack) = gateway::split_push(&body).map_err(|e| bad(format!("{e:#}")))?;
let (owner, name, refs) = (&header.repo, &header.name, header.updates.len());
let mut timed = Timed::new(started, &caller, owner, name, refs, pack.len());
let mut updates = Vec::new();
for u in &header.updates {
if u.src.as_deref().is_some_and(|s| !gateway::is_sha(s)) || !u.dst.starts_with("refs/") {
@@ -603,9 +691,12 @@ pub async fn push(
.or_default()
.clone();
let _turn = lock.lock().await;
timed.turn = timed.mark();
let found = writer(&app, &caller, &header.repo, &header.name).await?;
timed.records = timed.mark();
let ((remote, session, snap), writer) = found;
let mirror = mirror(&app, &remote, &snap).await?;
timed.mirror = timed.mark();
if !pack.is_empty() {
mirror
.git
@@ -624,6 +715,7 @@ pub async fn push(
{
return Err(bad(format!("the pack does not hold {sha}")));
}
timed.index = timed.mark();
// Protected branches hold whoever pushes here; the owner can write
// past the gateway. A collaborator's merge is held to the merge gate,
// read as the repository is now: as `writer` read it, under this
@@ -664,6 +756,7 @@ pub async fn push(
}
}
let updates = allowed;
timed.gate = timed.mark();
let mut report = push::push_by(
&app.ctx,
&remote,
@@ -672,6 +765,7 @@ pub async fn push(
&session.auth(),
&updates,
Some(&caller),
&mut timed.spent,
)
.await
.map_err(written)?;
@@ -860,6 +954,7 @@ async fn attempt(
&session.auth(),
&[update],
pusher(batch),
&mut push::Spent::default(),
)
.await
.map_err(written)?;
docs/design/gateway.md+17−5
View file
@@ -169,6 +169,17 @@ of it, if the pages keep one for them, aside: their next look on the pages
then reads only its records
([access.md](access.md#reading-a-private-repository)).
Each push logs one `gateway push` line as it ends, however it ends, with
the repository, the caller, how many refs and how large a pack it sent, its
`duration_ms`, and what each part took in turn: `auth_ms` (the caller's
token), `turn_ms` (waiting for the repository's turn), `records_ms`
(reading the repository as its owner), `mirror_ms` (syncing the mirror),
`index_ms` (indexing the pack), `gate_ms` (the branch rules, the merge
gate, and a runner's landing), `pack_ms` (checking the updates and
packing what they bring), `upload_ms` (the pack's blobs) and `write_ms`
(the records). A part it never reached is left out
([observability.md](../observability.md)).
## Runners landing a queue
A runner the repository names lands its merge queue through the gateway
@@ -197,11 +208,12 @@ the tree the two make merged. The pull must target that branch, be open,
and have its set stand at that head. Anything else falls to the checks
above.
The gateway reads the queue, and each pull's head, as the owner, with the
session they delegated: a private repository's are in its space, which
takes a member's credential to read, and nobody is logged in to the
gateway. The [merge gate](#protected-branches) a collaborator's push is
held to reads the repository as the owner too.
The gateway reads the queue once, for both, and each pull's head, as the
owner, with the session they delegated: a private repository's are in its
space, which takes a member's credential to read, and nobody is logged in
to the gateway. There it reads each member's collections side by side,
eight at a time. The [merge gate](#protected-branches) a collaborator's
push is held to reads the repository as the owner too.
So a runner, or whoever holds its account, can land what a maintainer
queued, before or without its checks passing, and nothing more.
docs/observability.md+9−0
View file
@@ -37,6 +37,15 @@ to read it is in [deploy/README.md](../deploy/README.md#logs).
requests for one repository that all end together is explained by the
sync before it. A sync started by a write the feed brought is in no
request's trace.
- **A line for each push through the gateway**: `gateway push`, with the
repository, the caller (`did`), the refs (`count`) and the pack's size
(`bytes`), `duration_ms`, and each part it took in turn: `auth_ms`,
`turn_ms`, `records_ms`, `mirror_ms`, `index_ms`, `gate_ms`, `pack_ms`,
`upload_ms` and `write_ms` ([gateway.md](design/gateway.md#push)). At
info, or at warn when slow; logged however the push ends, a part it never
reached left out, so the part it stopped in is what `duration_ms` has
beyond the rest. A slow push's calls are mostly each under a second, too
quick to be logged, and add up: this line says which part they were in.
- **A line for each call made**, to a PDS, the PLC directory, Bluesky or
a handle's host, from the one place every call goes through
(`knotless::telemetry::send`): host, path, XRPC method, status, error,