johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! `/_status`: whether this server is doing its job, for a monitor off the
//! host to poll. Every page can answer 200 while the feeds that keep it
//! current are dead, or while a mirror's objects can no longer be read, so
//! this checks those:
//!
//! - each feed followed: connected, and an event lately. One feed bringing
//! events is enough, since each brings all the index needs;
//! - the index answers a read;
//! - with `?repo=<handle or did>/<name>`, that repository's mirror reads as
//! its code page does: its default branch's commit and tree;
//! - the disk its mirrors and caches are on has room: under 15% free is told
//! of, under 5% fails. A full disk fails every push a mirror syncs, and
//! answers 502 for every page that needs one.
//!
//! 200 when all of that holds, else 503, and JSON saying which part failed.
//! `/healthz` stays the liveness check: a dead feed upstream is no reason
//! to restart this server.
use crate::{App, objects, repo_name};
use axum::Json;
use axum::extract::{Query, State};
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use knotless::status::{Check, Disk, Feed, MirrorCheck, Status};
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
/// How long the firehose may stay silent: it carries every account's
/// commits, about 400 a second (October 2026), its longest gap a sixth of a
/// second, so two minutes without one is a feed gone dead. Two minutes is
/// also how long a watcher waits on a silent connection, through a ping,
/// before it reconnects.
pub const FIREHOSE_STALE: Duration = Duration::from_secs(120);
/// How long Jetstream may stay silent. Asked only for `dev.knotless.*`, it
/// still brings every account's identity, account and sync events: 0.3 a
/// second over 15 minutes (October 2026), its longest gap 11 seconds, so
/// two minutes is ten of the longest and leaves no doubt.
pub const JETSTREAM_STALE: Duration = Duration::from_secs(120);
/// How long a feed named `name` may stay silent before it counts as dead.
pub fn stale_after(name: &str) -> Duration {
match name {
"firehose" => FIREHOSE_STALE,
_ => JETSTREAM_STALE,
}
}
/// The share of the disk free below which it is low: told of among the
/// problems, still 200. Prod's 150 GB disk then has 22 GB left, a day or
/// more of deploys and runs, time to clear it before it fills.
pub const DISK_LOW: f64 = 0.15;
/// The share free below which the disk counts as full and `/_status`
/// fails: 7.5 GB of prod's, a single cargo build's worth.
pub const DISK_FULL: f64 = 0.05;
/// How a disk with `free` of its `total` bytes is: `ok`, `low` or `full`.
pub fn disk_state(free: u64, total: u64) -> &'static str {
let share = free as f64 / total.max(1) as f64;
match share {
s if s < DISK_FULL => "full",
s if s < DISK_LOW => "low",
_ => "ok",
}
}
/// The disk `path` is on, asked of its nearest directory that exists.
pub(crate) fn disk(path: &Path) -> Disk {
let at = path.ancestors().find(|p| p.exists()).unwrap_or(path);
let mut disk = Disk {
path: path.display().to_string(),
state: "unknown".into(),
ok: false,
free_bytes: 0,
total_bytes: 0,
error: None,
};
match rustix::fs::statvfs(at) {
Ok(space) => {
disk.free_bytes = space.f_bavail * space.f_frsize;
disk.total_bytes = space.f_blocks * space.f_frsize;
let state = disk_state(disk.free_bytes, disk.total_bytes);
(disk.state, disk.ok) = (state.into(), state != "full");
}
Err(err) => disk.error = Some(format!("{}: {err}", at.display())),
}
disk
}
/// What `/_status` says of a disk that is not `ok`, among its problems.
fn disk_problem(disk: &Disk) -> Option<String> {
let free = 100 * disk.free_bytes / disk.total_bytes.max(1);
match (disk.state.as_str(), &disk.error) {
("ok", _) => None,
(_, Some(err)) => Some(format!("disk: {err}")),
(state, None) => Some(format!("disk {state}: {free}% free")),
}
}
/// A feed as its watcher last told of it.
pub(crate) struct Followed {
url: String,
connected: bool,
/// When it was first followed, which a feed that never brought an
/// event is silent since.
since: Instant,
last: Option<Instant>,
events: u64,
seq: Option<i64>,
}
impl App {
/// That the feed `name` is followed from `url`, not yet connected.
pub fn follows(&self, name: &'static str, url: &str) {
self.feeds.lock().unwrap().insert(
name,
Followed {
url: url.to_string(),
connected: false,
since: Instant::now(),
last: None,
events: 0,
seq: None,
},
);
}
/// That the feed `name` connected, or dropped. Mirrors trust the feeds
/// to tell them of writes while any is connected.
pub fn connected(&self, name: &str, connected: bool) {
let mut feeds = self.feeds.lock().unwrap();
let feed = feeds.get_mut(name).into_iter();
feed.filter(|f| f.connected != connected).for_each(|f| {
crate::telemetry::feed(name, &f.url, connected);
f.connected = connected;
});
let any = feeds.values().any(|f| f.connected);
self.watching.store(any, Ordering::Relaxed);
}
/// That the feed `name` brought an event, numbered `seq`.
pub fn heard(&self, name: &str, seq: Option<i64>) {
let mut feeds = self.feeds.lock().unwrap();
feeds.get_mut(name).into_iter().for_each(|feed| {
feed.last = Some(Instant::now());
feed.events += 1;
feed.seq = seq.or(feed.seq);
});
}
/// The feeds, as `/_status` tells of them.
pub(crate) fn feeds(&self) -> Vec<Feed> {
let feeds = self.feeds.lock().unwrap();
let mut out = Vec::new();
for (name, f) in feeds.iter() {
let stale = self.stale.unwrap_or(stale_after(name));
let silent = f.last.unwrap_or(f.since).elapsed();
out.push(Feed {
name: name.to_string(),
url: f.url.clone(),
ok: f.connected && silent <= stale,
connected: f.connected,
silent_seconds: silent.as_secs(),
stale_after_seconds: stale.as_secs(),
events: f.events,
seq: f.seq,
});
}
out
}
/// How this server is, asking after `repo`'s mirror too if given.
pub async fn status(&self, repo: Option<&str>) -> Status {
let feeds = self.feeds();
let mut problems = Vec::new();
for f in feeds.iter().filter(|f| !f.ok) {
let why = match f.connected {
true => format!("no event for {}s", f.silent_seconds),
false => "not connected".to_string(),
};
problems.push(format!("{}: {why}", f.name));
}
let fed = feeds.is_empty() || feeds.iter().any(|f| f.ok);
if !fed {
problems.push("no feed is bringing events".into());
}
let index = self.index.as_ref().map(|i| match i.check() {
Ok(_) => Check {
ok: true,
error: None,
},
Err(err) => Check {
ok: false,
error: Some(err.to_string()),
},
});
let mirror = match repo {
Some(repo) => Some(probe(self, repo).await),
None => None,
};
let failed = [
index
.as_ref()
.and_then(|i| i.error.clone())
.map(|e| format!("index: {e}")),
mirror
.as_ref()
.and_then(|m| Some(format!("mirror {}: {}", m.repo, m.error.as_ref()?))),
];
problems.extend(failed.into_iter().flatten());
let disk = self.disk.as_deref().map(disk);
problems.extend(disk.as_ref().and_then(disk_problem));
let code = match self.code.as_deref() {
Some(code) => Some(code.check().await),
None => None,
};
let failing = code.as_ref().and_then(|c| c.error.as_deref());
problems.extend(failing.map(|e| format!("code: {e}")));
let parts = index.as_ref().is_none_or(|i| i.ok)
&& mirror.as_ref().is_none_or(|m| m.ok)
&& code.as_ref().is_none_or(|c| c.ok);
Status {
ok: fed && parts && disk.as_ref().is_none_or(|d| d.ok),
problems,
uptime_seconds: self.started.elapsed().as_secs(),
feeds,
index,
mirror,
disk,
code,
}
}
}
/// What `/_status` is asked.
#[derive(serde::Deserialize)]
pub(crate) struct Ask {
repo: Option<String>,
}
/// `GET /_status`.
pub(crate) async fn status(State(app): State<Arc<App>>, Query(ask): Query<Ask>) -> Response {
let status = app.status(ask.repo.as_deref()).await;
let code = match status.ok {
true => StatusCode::OK,
false => StatusCode::SERVICE_UNAVAILABLE,
};
(code, [(header::CACHE_CONTROL, "no-store")], Json(status)).into_response()
}
/// `repo`'s mirror, read as its code page reads it.
async fn probe(app: &App, repo: &str) -> MirrorCheck {
let mut check = MirrorCheck {
repo: repo.to_string(),
ok: false,
error: None,
commit: None,
packs: None,
};
match read(app, repo).await {
Ok((commit, packs)) => {
check.ok = true;
check.commit = Some(commit);
check.packs = Some(packs);
}
Err(err) => check.error = Some(err),
}
check
}
/// The commit `repo`'s default branch is at, its tree read, and how many
/// packs its mirror holds. A mirror with more packs than its open handle
/// has room for fails here, as its pages do.
async fn read(app: &App, repo: &str) -> Result<(String, usize), String> {
let (authority, name) = repo
.split_once('/')
.ok_or("?repo= takes <handle or did>/<name>")?;
let found = app.located(authority, repo_name(name)).await;
let git = found.map_err(|e| e.1)?.git;
let packs = std::fs::read_dir(git.dir.join("objects/pack"))
.into_iter()
.flatten()
.flatten()
.filter(|e| e.path().extension().is_some_and(|x| x == "pack"))
.count();
let objects = objects::open(&git.dir).map_err(text)?;
let commit = objects.resolve("HEAD").map_err(text)?;
objects.tree(&commit, "").map_err(text)?;
Ok((commit, packs))
}
fn text(err: anyhow::Error) -> String {
format!("{err:#}")
}
#[cfg(test)]
mod tests {
use super::*;
use knotless::env::{Ctx, Env};
fn app(index: Option<crate::index::Index>) -> Arc<App> {
let ctx = Ctx::new(Env::from_vars(|_| None));
App::build(ctx, Duration::ZERO, None, index)
}
#[tokio::test]
async fn feeds() {
// Nothing followed: nothing to be stale.
let mut alone = app(None);
let s = alone.status(None).await;
assert!(s.ok, "{s:?}");
assert!(s.feeds.is_empty());
assert_eq!((s.index, s.mirror), (None, None));
let s = alone.status(Some("unnamed")).await;
assert!(!s.ok);
assert_eq!(
s.problems,
["mirror unnamed: ?repo= takes <handle or did>/<name>"]
);
Arc::get_mut(&mut alone).unwrap().stale = Some(Duration::from_millis(200));
// Followed but never connected: not ok, and mirrors fall back to the TTL.
alone.follows("jetstream", "wss://jetstream");
alone.watching.store(true, Ordering::Relaxed);
alone.connected("jetstream", false);
assert!(!alone.watching.load(Ordering::Relaxed));
let s = alone.status(None).await;
assert!(!s.ok);
assert_eq!(
s.problems,
["jetstream: not connected", "no feed is bringing events"]
);
// Connected and bringing events: ok.
alone.connected("jetstream", true);
alone.heard("jetstream", Some(7));
alone.heard("jetstream", None);
alone.heard("elsewhere", Some(1));
alone.connected("elsewhere", false);
assert!(alone.watching.load(Ordering::Relaxed));
let s = alone.status(None).await;
assert!(s.ok, "{s:?}");
let f = &s.feeds[0];
assert_eq!((f.events, f.seq, f.connected), (2, Some(7), true));
assert_eq!(
(f.url.as_str(), f.stale_after_seconds),
("wss://jetstream", 0)
);
// Connected, but silent past the threshold: stale.
tokio::time::sleep(Duration::from_millis(300)).await;
let s = alone.status(None).await;
assert!(!s.ok);
assert_eq!(s.problems[0], "jetstream: no event for 0s");
// One feed bringing events is enough; each is still told of.
alone.follows("firehose", "wss://relay");
alone.connected("firehose", true);
alone.heard("firehose", Some(1));
let s = alone.status(None).await;
assert!(s.ok, "{s:?}");
assert_eq!(s.problems, ["jetstream: no event for 0s"]);
assert_eq!(s.feeds.len(), 2);
}
#[test]
fn thresholds() {
assert_eq!(stale_after("firehose"), FIREHOSE_STALE);
assert_eq!(stale_after("jetstream"), JETSTREAM_STALE);
let app = app(None);
app.follows("firehose", "wss://relay");
assert_eq!(app.feeds()[0].stale_after_seconds, FIREHOSE_STALE.as_secs());
}
#[test]
fn disk_thresholds() {
assert_eq!(disk_state(100, 100), "ok");
assert_eq!(disk_state(15, 100), "ok");
assert_eq!(disk_state(14, 100), "low");
assert_eq!(disk_state(5, 100), "low");
assert_eq!(disk_state(4, 100), "full");
assert_eq!(disk_state(0, 0), "full");
let at = |state: &str, free: u64, error: Option<&str>| Disk {
path: "/data".into(),
state: state.into(),
ok: state != "full",
free_bytes: free,
total_bytes: 100,
error: error.map(String::from),
};
assert_eq!(disk_problem(&at("ok", 50, None)), None);
assert_eq!(
disk_problem(&at("low", 12, None)).unwrap(),
"disk low: 12% free"
);
assert_eq!(
disk_problem(&at("full", 3, None)).unwrap(),
"disk full: 3% free"
);
assert_eq!(
disk_problem(&at("unknown", 0, Some("gone"))).unwrap(),
"disk: gone"
);
}
#[tokio::test]
async fn disk_asked() {
// A directory not made yet is asked of the nearest that is.
let dir = tempfile::tempdir().unwrap();
let mut app = app(None);
let path = dir.path().join("cache/serve");
Arc::get_mut(&mut app).unwrap().disk = Some(path.clone());
let d = app.status(None).await.disk.unwrap();
assert_eq!(d.path, path.display().to_string());
assert!(d.total_bytes > 0 && d.free_bytes <= d.total_bytes, "{d:?}");
assert_eq!(d.state, disk_state(d.free_bytes, d.total_bytes));
assert_eq!(d.error, None);
// Nowhere to ask: unknown, and failing.
let d = disk(Path::new("nowhere/at/all"));
assert_eq!((d.state.as_str(), d.ok), ("unknown", false));
assert!(d.error.unwrap().contains(": "));
}
#[tokio::test]
async fn code() {
// Zoekt gone: told of, and failing.
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("meta")).unwrap();
for shard in [
"a_2fk_v16.00000.zoekt",
"a_2fk_v16.00001.zoekt",
"b_2fk_v16.00000.zoekt",
] {
std::fs::write(dir.path().join(shard), "1234").unwrap();
}
let mut app = app(None);
let code = crate::code::dead(dir.path());
Arc::get_mut(&mut app).unwrap().code = Some(Arc::new(code));
let s = app.status(None).await;
assert!(!s.ok);
let c = s.code.unwrap();
assert_eq!((c.ok, c.repos, c.shards, c.bytes), (false, 2, 3, 12));
assert_eq!(s.problems, [format!("code: {}", c.error.unwrap())]);
}
#[tokio::test]
async fn index() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("index.db");
let app = app(Some(crate::index::Index::open(&path).unwrap()));
let s = app.status(None).await;
assert!(s.ok, "{s:?}");
assert_eq!(
s.index.unwrap(),
Check {
ok: true,
error: None
}
);
// Its tables gone from under it: the index answers no more.
let other = rusqlite::Connection::open(&path).unwrap();
other.execute_batch("DROP TABLE stream").unwrap();
let s = app.status(None).await;
assert!(!s.ok);
let error = s.index.unwrap().error.unwrap();
assert_eq!(s.problems, [format!("index: {error}")]);
}
}