~ 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

A check's log is read ahead as it is filed, and many more are kept

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/check-log-cache
crates/knotless-serve/src/card.rs+45−7
View file
@@ -807,27 +807,40 @@ pub fn render(svg: &str, picture: Option<Arc<Vec<u8>>>) -> Vec<u8> {
encode(&raster(svg, picture))
}
/// Values by key, the oldest dropped past a number of them.
/// Values by key, the oldest dropped past a number of them, or past a
/// number of bytes when each is weighed.
pub struct Newest<V> {
kept: Mutex<Entries<V>>,
cap: usize,
weigh: fn(&V) -> usize,
}
struct Entries<V> {
by_key: HashMap<String, V>,
/// Oldest first.
order: VecDeque<String>,
/// What all of them weigh together.
weight: usize,
}
impl<V: Clone> Newest<V> {
/// Keeping `cap` of them.
pub fn new(cap: usize) -> Newest<V> {
Newest::weighed(cap, |_| 1)
}
/// Keeping as many as `weigh` says weigh `cap` together, and always the
/// newest.
pub fn weighed(cap: usize, weigh: fn(&V) -> usize) -> Newest<V> {
let keys = Entries {
by_key: HashMap::new(),
order: VecDeque::new(),
weight: 0,
};
Newest {
kept: Mutex::new(keys),
cap,
weigh,
}
}
@@ -837,13 +850,20 @@ impl<V: Clone> Newest<V> {
pub fn put(&self, key: &str, value: V) {
let mut kept = self.kept.lock().unwrap();
let Entries { by_key, order } = &mut *kept;
if by_key.insert(key.to_string(), value).is_none() {
order.push_back(key.to_string());
let Entries {
by_key,
order,
weight,
} = &mut *kept;
*weight += (self.weigh)(&value);
match by_key.insert(key.to_string(), value) {
Some(was) => *weight -= (self.weigh)(&was),
None => order.push_back(key.to_string()),
}
while order.len() > self.cap {
let oldest = order.pop_front().expect("longer than the cap");
by_key.remove(&oldest);
while *weight > self.cap && order.len() > 1 {
let oldest = order.pop_front().expect("more than one");
let was = by_key.remove(&oldest).expect("each in order is kept");
*weight -= (self.weigh)(&was);
}
}
@@ -898,6 +918,24 @@ impl Cache {
mod tests {
use super::*;
#[test]
fn newest_by_weight() {
let kept: Newest<Vec<u8>> = Newest::weighed(10, Vec::len);
kept.put("a", vec![0; 4]);
kept.put("b", vec![0; 4]);
assert_eq!(kept.len(), 2);
// Put again, it weighs what it weighs now, not both.
kept.put("b", vec![0; 6]);
assert_eq!((kept.len(), kept.get("b").map(|b| b.len())), (2, Some(6)));
// Past the cap, the oldest go.
kept.put("c", vec![0; 3]);
assert!(kept.get("a").is_none());
assert_eq!(kept.len(), 2);
// One heavier than the cap is still kept, alone.
kept.put("d", vec![0; 20]);
assert_eq!((kept.len(), kept.get("d").map(|d| d.len())), (1, Some(20)));
}
#[test]
fn wrapping() {
let w = |t: &str, width: f32, max: usize| wrap(t, 40.0, Weight::Regular, width, max);
crates/knotless-serve/src/lib.rs+13−0
View file
@@ -439,6 +439,7 @@ impl App {
telemetry::indexed(&did, index_inline(&index, &did, op, rev));
}
accounts::repos_deleted(self, &did, &deleted);
self.warm_checks(&did, &event.ops);
return;
}
let (app, ctx) = (self.clone(), self.ctx.clone());
@@ -448,10 +449,22 @@ impl App {
telemetry::indexed(&did, index_op(&ctx, &index, &did, op, rev).await);
}
accounts::repos_deleted(&app, &did, &deleted);
app.warm_checks(&did, &event.ops);
});
}
}
/// Reads ahead the logs of the checks `ops` filed for `did`, once they
/// are indexed ([`web::warm_check_log`]).
fn warm_checks(self: &Arc<App>, did: &str, ops: &[firehose::Op]) {
let check = knotless::records::nsid::CHECK;
let filed = ops.iter().filter(|op| op.collection == check);
for op in filed.filter(|op| op.action != "delete") {
let (app, did, rkey) = (self.clone(), did.to_string(), op.rkey.clone());
tokio::spawn(web::warm_check_log(app, did, rkey));
}
}
/// Whether the account `did` is taken down here, or by a labeler this
/// server follows: never without an index, and not when the index
/// cannot answer.
crates/knotless-serve/src/web.rs+93−7
View file
@@ -2037,11 +2037,20 @@ async fn in_space(reader: &knotless::env::Ctx, uri: &str, did: &str) -> anyhow::
crate::spaces::remote(reader, &joined, did).await
}
/// What check runs made, as shown, by kind, runner and CID: a log's tail,
/// or a picture. A blob is the same for good, so it is read once; one in a
/// space is read each time, as whoever asks may read it.
/// Pictures check runs made and blobs attached, by kind, author and CID.
/// A blob is the same for good, so it is read once; one in a space is
/// read each time, as whoever asks may read it.
static RUN_BLOBS: LazyLock<card::Newest<Arc<Vec<u8>>>> = LazyLock::new(|| card::Newest::new(64));
/// The tails of check logs, kept as [`RUN_BLOBS`] are: a few kilobytes
/// most, and at most [`LOG_SHOWN`], so many more of them, up to
/// [`LOG_TAILS_KEPT`] bytes in all.
static LOG_TAILS: LazyLock<card::Newest<Arc<Vec<u8>>>> =
LazyLock::new(|| card::Newest::weighed(LOG_TAILS_KEPT, |tail| tail.len()));
/// What the kept tails of check logs weigh at most.
const LOG_TAILS_KEPT: usize = 64 * 1024 * 1024;
/// The blob `cid` of the runner `remote`, as `shown` makes it of `kind`.
async fn run_blob(
remote: &Remote,
@@ -2051,25 +2060,71 @@ async fn run_blob(
) -> anyhow::Result<Arc<Vec<u8>>> {
let public = remote.space.is_none();
let key = format!("{kind}/{}/{cid}", remote.did());
let kept = public.then(|| RUN_BLOBS.get(&key)).flatten();
let cache = [&RUN_BLOBS, &LOG_TAILS][usize::from(kind == "log")];
let kept = public.then(|| cache.get(&key)).flatten();
if let Some(kept) = looked("run blobs", kept) {
return Ok(kept);
}
let blob = Arc::new(shown(remote.get_blob(cid).await?));
if public {
RUN_BLOBS.put(&key, blob.clone());
cache.put(&key, blob.clone());
}
Ok(blob)
}
/// A log as a check's page shows it: its tail.
fn shown_tail(log: Vec<u8>) -> Vec<u8> {
log_tail(&log).into_bytes()
}
/// How many check logs are read ahead at once, as their checks come.
static WARMING: LazyLock<tokio::sync::Semaphore> = LazyLock::new(|| tokio::sync::Semaphore::new(4));
/// The CID of the log of `did`'s check `rkey` as the index has it, when
/// its repository names `did` among its runners, so that its page shows it.
fn log_to_warm(index: &Index, did: &str, rkey: &str) -> Option<String> {
let uri = format!("at://{did}/{}/{rkey}", nsid::CHECK);
let value = index.check_record(&uri).ok().flatten()?;
let check: CheckRecord = serde_json::from_value(value).ok()?;
let cid = check.log?.link.cid;
let repo = index.repo(&check.repo.uri).ok().flatten()?;
let repo: RepoRecord = serde_json::from_value(repo).ok()?;
repo.is_runner(did).then_some(cid)
}
/// Reads ahead the tail of the log of `did`'s check `rkey`, as the check
/// comes from the network, so its page's first look need not wait on the
/// runner's PDS. A few at once, and none past them: a burst of checks
/// leaves the rest to be read as they are looked at.
pub(crate) async fn warm_check_log(app: Arc<App>, did: String, rkey: String) {
let index = app.index.as_ref();
let cid = index.and_then(|index| log_to_warm(index, &did, &rkey));
let permit = cid.as_ref().and_then(|_| WARMING.try_acquire().ok());
let (Some(cid), Some(_permit)) = (cid, permit) else {
return;
};
let at = RepoUrl {
authority: did,
name: "_".into(),
};
let read = async {
let remote = Remote::connect(&app.ctx, &at).await?;
run_blob(&remote, "log", &cid, shown_tail).await
};
fill(
"check logs",
&format!("{}/{cid}", at.authority),
&read.await,
);
}
/// `dev.knotless.getCheck`.
async fn get_check(State(app): State<Arc<App>>, Query(p): Query<CheckParams>) -> Result<Response> {
let (repo, remote, mut c) = trusted_check(&app, &p).await?;
let tail = |b: Vec<u8>| log_tail(&b).into_bytes();
let read = c
.log
.as_ref()
.map(|b| run_blob(&remote, "log", &b.link.cid, tail));
.map(|b| run_blob(&remote, "log", &b.link.cid, shown_tail));
let blob = futures::future::OptionFuture::from(read).await;
let log = blob.and_then(|b| b.ok());
let log = log.map(|b| String::from_utf8_lossy(&b).into_owned());
@@ -3216,6 +3271,37 @@ async fn item_card_png(
mod tests {
use super::*;
#[test]
fn only_a_named_runners_logs_are_read_ahead() {
let dir = tempfile::tempdir().unwrap();
let index = Index::open(&dir.path().join("i.db")).unwrap();
let ran = serde_json::json!({"name": "r", "key": "k", "runners": [{"did": "did:ci"}],
"createdAt": "2026-01-01T00:00:00Z"});
index.put("did:a", nsid::REPO, "r", &ran).unwrap();
let repo = serde_json::json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
let check = |by: &str, rkey: &str, log: bool| {
let mut run = serde_json::json!({"repo": repo, "ref": "refs/heads/main",
"commit": "c", "workflow": "test", "status": "success",
"startedAt": "2026-01-02T00:00:00Z"});
if log {
run["log"] = serde_json::json!({"$type": "blob", "ref": {"$link": "bafylog"},
"mimeType": "text/plain", "size": 3});
}
index.put(by, nsid::CHECK, rkey, &run).unwrap();
};
check("did:ci", "1", true);
check("did:ci", "2", false);
check("did:other", "3", true);
assert_eq!(
log_to_warm(&index, "did:ci", "1").as_deref(),
Some("bafylog")
);
// No log yet, a runner the repository does not name, or no check.
assert_eq!(log_to_warm(&index, "did:ci", "2"), None);
assert_eq!(log_to_warm(&index, "did:other", "3"), None);
assert_eq!(log_to_warm(&index, "did:ci", "none"), None);
}
#[test]
fn first_files_drawn_stop_at_ten_or_the_lines() {
let file = |changed: usize, binary: bool| diff::File {
docs/design/serve.md+6−1
View file
@@ -653,7 +653,12 @@ again, so a member removed meanwhile stops hearing.
log stays scrolled to its end as output comes, unless the reader has
scrolled up from it. Once the run ends the
record is filed, the repository's stream says so, and the page draws the
record's log.
record's log. Its tail, the last 256 KiB the page shows, is read from
the runner's PDS as the record comes from the network, a few at once,
and kept in memory, 64 MiB of tails in all, so the page's first look
does not wait on that PDS; one the server had no time for is read as
it is looked at, and kept the same way. A private repository's is read
each time, as whoever asks may read it.
Any page left out of sight a minute or more asks its answers again as it
is looked at, and any page asks again as the browser is back online, so a