chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Code indexing is set up before serve spawns it, so its coverage is steady
crates/knotless-serve/src/code.rs+26−20
| @@ -20,6 +20,7 @@ | ||
use crate::{App, AppError}; | ||
use anyhow::{Context, Result}; | ||
use base64::Engine; | ||
use futures::StreamExt; | ||
use knotless::telemetry::{self, Fields, Level}; | ||
use serde::{Deserialize, Serialize}; | ||
use serde_json::json; | ||
| @@ -353,33 +354,38 @@ fn escape(s: &str) -> String { | ||
.collect() | ||
} | ||
| ||
/// Indexes the public repositories as the index asks ([`Index::reindex`]), | ||
/// after each with no shards yet. | ||
/// What indexes the public repositories as the index asks | ||
/// ([`Index::reindex`]), when serve keeps a code index: it asks first for | ||
/// each with no shards yet, then answers each ask, for as long as serve | ||
/// runs. Set up here, before it is spawned, so serve without a code | ||
/// index starts nothing. | ||
/// | ||
/// [`Index::reindex`]: crate::index::Index::reindex | ||
pub async fn follow(app: Arc<App>) { | ||
let both = app.code.clone().zip(app.index.clone()); | ||
let Some(((code, index), mut asked)) = both.and_then(|b| Some((b.clone(), b.1.indexing()?))) | ||
else { | ||
return; | ||
}; | ||
pub fn follow(app: Arc<App>) -> Option<impl Future<Output = ()>> { | ||
let (code, index) = app.code.clone().zip(app.index.clone())?; | ||
let asked = index.indexing()?; | ||
for (did, key) in index.code_repos().unwrap_or_default() { | ||
if !code.has(&did, &key) { | ||
index.reindex(&did, &key); | ||
} | ||
} | ||
while let Some((did, key)) = asked.recv().await { | ||
let error = reindex(&app, &code, &did, &key).await.err(); | ||
let error = error.as_ref().map(ToString::to_string); | ||
let level = [Level::INFO, Level::WARN][error.is_some() as usize]; | ||
let fields = Fields { | ||
did: Some(&did), | ||
key: Some(&key), | ||
error, | ||
..Fields::default() | ||
}; | ||
telemetry::note(level, "code indexed", &fields); | ||
} | ||
let asked = futures::stream::unfold(asked, |mut a| async { a.recv().await.map(|r| (r, a)) }); | ||
Some(asked.for_each(move |(did, key)| indexed(app.clone(), code.clone(), did, key))) | ||
} | ||
| ||
/// Indexes `did`'s repository with `key` again, as [`follow`] is asked, | ||
/// and notes how it went. | ||
async fn indexed(app: Arc<App>, code: Arc<Code>, did: String, key: String) { | ||
let error = reindex(&app, &code, &did, &key).await.err(); | ||
let error = error.as_ref().map(ToString::to_string); | ||
let level = [Level::INFO, Level::WARN][error.is_some() as usize]; | ||
let fields = Fields { | ||
did: Some(&did), | ||
key: Some(&key), | ||
error, | ||
..Fields::default() | ||
}; | ||
telemetry::note(level, "code indexed", &fields); | ||
} | ||
| ||
/// Indexes `did`'s repository with `key` again, or forgets it if the | ||
crates/knotless-serve/src/lib.rs+3−1
| @@ -1493,7 +1493,9 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { | ||
} | ||
tokio::spawn(spaces::follow(app.clone(), SPACES)); | ||
tokio::spawn(languages::follow(app.clone())); | ||
tokio::spawn(code::follow(app.clone())); | ||
if let Some(follow) = code::follow(app.clone()) { | ||
tokio::spawn(follow); | ||
} | ||
tokio::spawn(shared::follow(app.clone())); | ||
tokio::spawn(telemetry::summaries(app.clone(), SUMMARY)); | ||
let listener = tokio::net::TcpListener::bind(&opts.listen).await?; | ||
crates/knotless-serve/tests/web.rs+2−2
| @@ -3475,8 +3475,8 @@ fn searching_code() { | ||
s.git(&work, &["push", "-q", &s.url(name), "main"]); | ||
} | ||
wait_until(|| names(&site.search("needle")["repos"]).len() == 2); | ||
site._rt | ||
.spawn(knotless_serve::code::follow(site.app.clone())); | ||
let follow = knotless_serve::code::follow(site.app.clone()); | ||
site._rt.spawn(follow.expect("a code index")); | ||
let search = |q: &str| site.json(&format!("/xrpc/dev.knotless.searchCode?q={q}")); | ||
let found = |q: &str| search(q)["files"].as_array().unwrap().len(); | ||
wait_until(|| found("needle_") == 2); | ||