johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! A read accelerator for knotless repositories.
//!
//! A PDS can hand out records and blobs and nothing else: no partial clone,
//! no shallow fetch, no "what is in this file at this commit". This server
//! keeps a bare mirror per repository and answers those from it:
//!
//! - git smart HTTP (protocol v0 and v2, `--filter`, `--depth`) at
//! `/<handle or did>/<name>`, read only;
//! - a JSON and raw API at `/api/<handle or did>/<name>/...` for agents that
//! want a file, a tree, a log or a diff without a clone.
//!
//! Everything it holds is a cache of the owners' records, which stay the
//! truth: each mirror is revalidated against them (one round trip) when
//! older than the TTL, or kept warm by the firehose or Jetstream, and the
//! index (a SQLite file) is built from them and can be rebuilt by a backfill.
//! Every object it serves is named by its hash, so a client can check it.
pub mod accounts;
mod api;
mod appview;
pub mod archive;
pub mod card;
pub mod code;
pub mod diff;
pub mod firehose;
pub mod gateway;
pub mod highlight;
pub mod index;
mod invites;
pub mod jetstream;
pub mod languages;
pub mod live;
pub mod login;
pub mod logs;
pub mod moderation;
pub mod objects;
pub mod oci;
mod orgs;
pub mod plans;
pub mod private;
pub mod releases;
mod secrets;
pub mod shared;
pub mod sites;
mod smart;
pub mod spaces;
pub mod status;
pub mod telemetry;
pub mod usage;
pub mod viewer;
pub mod web;
use anyhow::Result;
use axum::Router;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use knotless::env::Ctx;
use knotless::git::Git;
use knotless::mirror::Mirror;
use knotless::records::RepoRecord;
use knotless::records::nsid::PROFILES;
use knotless::remote::Remote;
use knotless::telemetry::{Fields, Level, millis, note, slow};
use knotless::url::RepoUrl;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
pub struct App {
pub ctx: Ctx,
/// How long a mirror is trusted without asking the PDS, when the
/// firehose is not watching.
pub ttl: Duration,
repos: Mutex<HashMap<(String, String), Shared<Entry>>>,
/// The private repositories viewers were shown, each viewer's apart, by
/// viewer, owner and name ([`private`]).
viewed: Mutex<HashMap<(String, String, String), Shared<Entry>>>,
/// How long, in milliseconds, a viewer's reading of a private
/// repository is taken as allowed before its space is asked again.
pub reprove: std::sync::atomic::AtomicU64,
/// The domain repositories' sites are served under, as `knotless.page`,
/// once a [`sites::Sites`] serves them: their pages show the address.
pub pages_domain: std::sync::OnceLock<String>,
/// Firehose events seen per DID.
events: Mutex<HashMap<String, u64>>,
/// Of those, the ones that could change any of the account's
/// repositories, per DID ([`App::stamp`]).
account_events: Mutex<HashMap<String, u64>>,
/// And those that change one repository, by its owner's DID and key.
repo_events: Mutex<HashMap<(String, String), u64>>,
/// Writes made from the pages (or by this server for an account), per
/// DID: a read after one waits for its sync ([`App::served`]).
writes: Mutex<HashMap<String, u64>>,
/// Each public repository's mirror as last synced, by owner and name,
/// for a read to take without waiting on the lock a sync holds.
last: Mutex<HashMap<(String, String), Entry>>,
/// Each repository's storage as last counted, kept until it moves
/// (`dev.knotless.getRepoStorage`).
pub(crate) counts: web::Counts,
/// The last commit, by its rev, that wrote to each DID's repository
/// collections ([`first_of`]).
revs: Mutex<HashMap<String, String>>,
pub watching: AtomicBool,
/// Syncs of each space that found something written there, by URI:
/// shared with a sync run aside from the read that asked for it.
moves: Arc<Mutex<HashMap<String, u64>>>,
/// Set while the spaces kept are synced ([`spaces::follow`]): a
/// private repository's mirror is then fresh until its space moves.
pub following: AtomicBool,
/// The feeds followed, as their watchers tell of them, by name.
feeds: Mutex<std::collections::BTreeMap<&'static str, status::Followed>>,
/// How long a feed may stay silent before `/_status` calls it dead,
/// rather than each feed's own default.
pub stale: Option<Duration>,
/// The directory whose disk `/_status` watches for room: the cache's,
/// where mirrors go. None in tests, which share whatever disk they get.
pub disk: Option<std::path::PathBuf>,
started: Instant,
/// Set when this server also takes writes for owners who delegate them.
pub gateway: Option<gateway::Gateway>,
/// Set when this server is an appview, indexing what the firehose says.
pub index: Option<Arc<index::Index>>,
/// Where people log in, this server's own index of the private
/// repositories read here, synced from their spaces ([`spaces`]): kept
/// apart from `index`, so nothing of a space reaches an appview's.
pub spaces: Option<Arc<index::Index>>,
/// When each account's spaces were last listed, for private
/// repositories of theirs not yet in `spaces`, and those shared with
/// them asked after ([`spaces::discover`]).
pub discovered: Mutex<HashMap<String, Instant>>,
/// Whether each owner's PDS has spaces, as their listing last found.
pub with_spaces: Mutex<HashMap<String, bool>>,
/// The spaces their authorities told of writes in since they were last
/// synced, to sync now ([`spaces::written`]).
due: Mutex<std::collections::HashSet<String>>,
/// Wakes [`spaces::follow`] for what is due.
told: tokio::sync::Notify,
/// When each space was last synced in full, as anyone here synced it:
/// a read syncs it itself, aside unless it never was, once that is
/// [`spaces::STALE`] ago, as when
/// the session of the member it is synced as no longer reads it. Its
/// entries taken out, every space counts as not synced lately.
pub last_synced: Arc<Mutex<HashMap<String, Instant>>>,
/// When this server last asked each space's authority to tell it of
/// writes there.
registered: tokio::sync::Mutex<HashMap<String, Instant>>,
/// Set when this server searches the code of public repositories
/// ([`code`]).
pub code: Option<Arc<code::Code>>,
/// Set when this server counts how it is used, for its operator.
pub usage: Option<Arc<usage::Usage>>,
/// Set when this server keeps the lines it logs, for its operator.
pub logs: Option<Arc<logs::Logs>>,
/// The labels this server gives out, to those following them.
pub labels: tokio::sync::broadcast::Sender<(i64, serde_json::Value)>,
/// When each check still running was last read again from its PDS.
rechecked: Mutex<HashMap<String, Instant>>,
/// The delegation tokens queries brought, by `jti`, each until it ends:
/// spent once ([`App::spend`]).
spent: Mutex<HashMap<String, u64>>,
/// The runner, by DID, a new public repository names unless its
/// owner names others, as `dev.knotless.describeServer` says.
pub default_runner: Option<String>,
/// The oldest client version this appview asks for, and why, as
/// `dev.knotless.describeServer` says.
pub client_minimum: Option<(String, Option<String>)>,
/// The oldest build of the iOS app this appview lets run, and why, as
/// `dev.knotless.describeServer` says.
pub ios_minimum: Option<(u64, Option<String>)>,
/// The output of the checks running now, and the spaces written to,
/// for whoever follows them.
pub live: live::Live,
/// One lock per organization, held while this gateway changes its
/// records: a space's records take no compare and swap, so two changes
/// at once would each write over the other.
orgs: Mutex<HashMap<String, Shared<()>>>,
/// One lock per account, held while its spaces are listed
/// ([`spaces::discover`]): two listings at once could each keep what
/// the other's authority said after it, a removal undone by an answer
/// asked before it.
discovers: Mutex<HashMap<String, Shared<()>>>,
}
type Shared<T> = Arc<tokio::sync::Mutex<T>>;
#[derive(Default, Clone)]
struct Entry {
git: Option<Git>,
/// The repository's owner, key and name now, which differ from what
/// was asked for after a move or a rename.
found: Option<(String, String, String)>,
record: RepoRecord,
/// The space it is in, if it is private.
space: Option<String>,
synced: Option<Instant>,
/// When its viewer was last found allowed to read it, for a private
/// repository a viewer was shown.
proved: Option<Instant>,
/// Where a private repository was then found, read until it is
/// proved again.
remote: Option<Remote>,
/// The events it was last synced as of ([`App::stamp`]).
events: (u64, u64),
/// Its space's moves when it was last synced, for a private one.
moves: u64,
/// The writes made from the pages for its owner when it was last
/// synced ([`App::writes_for`]).
writes: u64,
/// When the first event came that made it stale, if one has since it
/// was synced: kept with [`App::last`] only.
stale_since: Option<Instant>,
}
/// How long a mirror that events have made stale is still served, to a
/// read that wrote nothing, while its sync catches up.
const STALE_FOR: Duration = Duration::from_secs(10);
/// Whether a read is served `entry`, a mirror as it was last synced,
/// without waiting on a sync, when its events now stamp as `stamp` and
/// `wrote` writes were made from the pages: if no event has come since it
/// synced, or the first came within [`STALE_FOR`], and the sync it started
/// is then still catching up; never after a write from the pages that it
/// may not have seen.
fn serves(entry: &Entry, stamp: (u64, u64), wrote: u64) -> bool {
let catching_up = entry.stale_since.is_some_and(|at| at.elapsed() < STALE_FOR);
entry.writes == wrote && (entry.events == stamp || catching_up)
}
/// A repository found by name, and its mirror.
pub struct Located {
pub git: Git,
pub did: String,
pub key: String,
/// Its name now: after a rename, not the name it was asked for by.
pub name: String,
/// Whether it is in another account than the one asked for: it moved.
pub moved: bool,
/// Its record, as the PDS had it when the mirror last synced.
pub record: RepoRecord,
/// The space it is in, if it is private.
pub space: Option<String>,
/// What reads it: this server, or for a private repository the viewer
/// it was found for, so that its records are read as theirs.
pub reader: Ctx,
}
/// The index of the private repositories this server may keep one of,
/// kept with its cache: none if it cannot be opened.
fn spaces_index(ctx: &Ctx) -> Option<Arc<index::Index>> {
let dir = ctx.env.cache_dir.join("serve");
std::fs::create_dir_all(&dir).ok()?;
let index = index::Index::open(&dir.join("private.sqlite")).ok()?;
Some(Arc::new(index))
}
impl App {
pub fn new(ctx: Ctx, ttl: Duration) -> Arc<App> {
App::build(ctx, ttl, None, None)
}
pub fn build(
ctx: Ctx,
ttl: Duration,
gateway: Option<gateway::Gateway>,
index: Option<index::Index>,
) -> Arc<App> {
let spaces = gateway.as_ref().and_then(|_| spaces_index(&ctx));
// Asking itself, as the pages ask its appview and gateway for a
// logged-in account, it signs the token: none from their PDS. Failing
// that, it asks their PDS as anyone else would.
if let Some(gw) = &gateway {
let kept = ctx.env.config_dir.join("loopback.json");
let _ = knotless::auth::loopback(&kept, &gw.did);
}
Arc::new(App {
ctx,
ttl,
repos: Mutex::default(),
viewed: Mutex::default(),
reprove: std::sync::atomic::AtomicU64::new(private::REPROVE.as_millis() as u64),
pages_domain: std::sync::OnceLock::new(),
events: Mutex::default(),
account_events: Mutex::default(),
repo_events: Mutex::default(),
writes: Mutex::default(),
last: Mutex::default(),
counts: web::Counts::default(),
revs: Mutex::default(),
watching: AtomicBool::new(false),
moves: Default::default(),
following: AtomicBool::new(false),
feeds: Mutex::default(),
stale: None,
disk: None,
started: Instant::now(),
gateway,
index: index.map(Arc::new),
spaces,
discovered: Mutex::default(),
with_spaces: Mutex::default(),
due: Mutex::default(),
told: tokio::sync::Notify::new(),
last_synced: Default::default(),
registered: tokio::sync::Mutex::default(),
code: None,
usage: None,
logs: None,
labels: tokio::sync::broadcast::channel(256).0,
rechecked: Mutex::default(),
spent: Mutex::default(),
default_runner: None,
client_minimum: None,
ios_minimum: None,
live: live::Live::default(),
orgs: Mutex::default(),
discovers: Mutex::default(),
})
}
/// Whether `did`'s spaces are to be listed now: not within
/// [`spaces::DISCOVER`] of the last time.
pub fn discovering(&self, did: &str) -> bool {
let mut discovered = self.discovered.lock().unwrap();
let recent = discovered
.get(did)
.is_some_and(|at| at.elapsed() < spaces::DISCOVER);
if !recent {
discovered.insert(did.to_string(), Instant::now());
}
!recent
}
/// The lock held while the organization `org`'s records change.
pub fn org_lock(&self, org: &str) -> Shared<()> {
let mut orgs = self.orgs.lock().unwrap();
orgs.entry(org.to_string()).or_default().clone()
}
/// The lock held while `did`'s spaces are listed.
pub(crate) fn discover_lock(&self, did: &str) -> Shared<()> {
let mut discovers = self.discovers.lock().unwrap();
discovers.entry(did.to_string()).or_default().clone()
}
/// Whether `did`'s PDS has spaces, for private repositories: unknown
/// until their spaces are first listed ([`spaces::discover`]).
pub fn has_spaces(&self, did: &str) -> Option<bool> {
self.with_spaces.lock().unwrap().get(did).copied()
}
pub(crate) fn found_spaces(&self, did: &str, has: bool) {
let mut with_spaces = self.with_spaces.lock().unwrap();
with_spaces.insert(did.to_string(), has);
}
/// Reads again from its PDS each of `checks` the index still has
/// queued or running a minute after it was filed so, once a minute at
/// most: a stream can miss the write that started or finished it, as one written while serve is
/// itself rolling over to a new build. In the background, so a later
/// read has it.
pub fn recheck(&self, index: &Arc<index::Index>, checks: &[knotless::forge::Entry]) {
let ago = std::time::SystemTime::now() - RECHECK;
let before = humantime::format_rfc3339_millis(ago).to_string();
let mut rechecked = self.rechecked.lock().unwrap();
rechecked.retain(|_, at| at.elapsed() < RECHECK);
let stale = checks.iter().filter(|c| {
let started = c.value["startedAt"].as_str().unwrap_or_default();
let going = c.value["status"]
.as_str()
.is_some_and(knotless::records::status::going);
going && started < before.as_str()
});
// Not again within the minute, however often it is read meanwhile.
let uris = stale.filter(|c| {
!rechecked.contains_key(&c.uri)
&& rechecked.insert(c.uri.clone(), Instant::now()).is_none()
});
let uris: Vec<String> = uris.map(|c| c.uri.clone()).collect();
drop(rechecked);
for (did, collection, rkey) in uris
.iter()
.filter_map(|u| knotless::forge::parse_uri(u).ok())
{
let op = firehose::Op {
action: "update".into(),
collection: collection.into(),
rkey: rkey.into(),
record: None,
};
let (ctx, index, did) = (self.ctx.clone(), index.clone(), did.to_string());
tokio::spawn(async move {
let _ = index_op(&ctx, &index, &did, &op, None).await;
});
}
}
/// Takes in a firehose or Jetstream event: refreshes the mirrors of an account whose
/// repositories moved, indexes what it wrote, and hides or drops an
/// account its host says is no longer active or deleted ([`accounts`]).
pub fn observe(self: &Arc<App>, mut event: firehose::Event) {
let Some(did) = event.did else {
return;
};
if let Some(status) = &event.account {
accounts::changed(self, &did, status);
// Nothing of one deleted is synced again.
if status == "deleted" {
return;
}
}
// A handle changed: the one kept is looked up again, as the event
// only says what it is. Kept only for accounts that were looked up.
if event.identity
&& let Some(index) = self.index.clone()
&& index.handle(&did).ok().flatten().is_some()
{
tokio::spawn(web::rename(self.ctx.clone(), index, did.clone()));
}
// Profiles change all over the network: an edit to one is taken in
// only for an account whose profile is kept, and is no sign that
// its repositories moved. A knotless profile is always taken in,
// for whether it says the account is a bot.
let profile = |op: &firehose::Op| {
let bluesky = op.collection != knotless::records::nsid::PROFILE;
bluesky && PROFILES.contains(&op.collection.as_str())
};
let kept = |index: &Arc<index::Index>| index.profile_kept(&did).unwrap_or(false);
let unkept = event.ops.iter().any(profile) && !self.index.as_ref().is_some_and(kept);
if unkept {
event.ops.retain(|op| !profile(op));
}
if unkept && event.ops.is_empty() {
return;
}
let repository = event
.ops
.iter()
.any(|op| knotless::records::nsid::ALL.contains(&op.collection.as_str()));
// Jetstream sends each record a commit wrote as an event of its own,
// and all of them are there once the first is: a push's pack, claim
// and ref are one sync, not a sync and then another for the rest.
// A record of one repository makes only its mirrors stale; anything
// else, all of the account's.
let moved = event.ops.is_empty() || repository;
let keys = scopes(&event.ops);
for key in keys.iter().filter(|_| moved) {
let seen = format!("{did} {}", key.as_deref().unwrap_or_default());
if first_of(&self.revs, &seen, event.rev.as_deref()) {
self.heard_of(&did, key.as_deref(), false);
}
}
if let Some(index) = self.index.clone()
&& !event.ops.is_empty()
{
// The keys of the repositories whose records it deletes, read
// before they go, for what is kept of them to go after.
let repo = knotless::records::nsid::REPO;
let deleted: Vec<String> = event
.ops
.iter()
.filter(|op| op.action == "delete" && op.collection == repo)
.filter_map(|op| {
let uri = format!("at://{did}/{repo}/{}", op.rkey);
let found = index.repo(&uri).ok().flatten();
found.and_then(|r| r["key"].as_str().map(str::to_string))
})
.collect();
// What the event carries in full is indexed now, in order, so a
// later write cannot land before an earlier one.
if event
.ops
.iter()
.all(|op| op.action == "delete" || op.record.is_some())
{
for op in &event.ops {
let rev = event.rev.as_deref();
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());
tokio::spawn(async move {
let rev = event.rev.as_deref();
for op in &event.ops {
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.
pub fn taken_down(&self, did: &str) -> bool {
let index = self.index.as_ref();
index.is_some_and(|i| i.down(&[did]).unwrap_or(false))
}
/// Firehose events seen for `did`.
pub fn events_for(&self, did: &str) -> u64 {
self.events.lock().unwrap().get(did).copied().unwrap_or(0)
}
/// What a mirror of `did`'s repository with `key` goes stale with: the
/// events that could change any of the account's repositories, and
/// those of that one. One of a key not yet known goes stale with every
/// event of the account.
pub(crate) fn stamp(&self, did: &str, key: Option<&str>) -> (u64, u64) {
let Some(key) = key else {
return (self.events_for(did), u64::MAX);
};
let account = self.account_events.lock().unwrap().get(did).copied();
let repos = self.repo_events.lock().unwrap();
let repo = repos.get(&(did.to_string(), key.to_string())).copied();
(account.unwrap_or(0), repo.unwrap_or(0))
}
/// What `entry`, synced now, is fresh as of: `stamp`, taken before its
/// sync for the key it had then; for a key the sync first found, as
/// of now, but only if no event of the account came meanwhile, as
/// `any`, its events then, says. Otherwise none, to sync again.
pub(crate) fn settled(
&self,
did: &str,
entry_key: Option<&str>,
stamp: (u64, u64),
any: u64,
found: Option<&str>,
) -> (u64, u64) {
if found == entry_key {
return stamp;
}
let quiet = self.events_for(did) == any;
let stale = (u64::MAX, u64::MAX);
[stale, self.stamp(did, found)][usize::from(quiet)]
}
/// Syncs of the space `uri` that found something written there.
pub fn moves_of(&self, uri: &str) -> u64 {
self.moves.lock().unwrap().get(uri).copied().unwrap_or(0)
}
/// Notes that the space `uri` was synced just now.
pub(crate) fn synced_now(&self, uri: &str) {
let mut last = self.last_synced.lock().unwrap();
last.insert(uri.to_string(), Instant::now());
}
/// Whether the space `uri` was not synced within [`spaces::STALE`],
/// or not since this server started.
pub(crate) fn sync_stale(&self, uri: &str) -> bool {
let last = self.last_synced.lock().unwrap();
last.get(uri).is_none_or(|at| at.elapsed() >= spaces::STALE)
}
/// Records that the space `uri` was written to, or is no longer synced,
/// so that mirrors of it sync again as they are next read.
pub fn moved(&self, uri: &str) {
*self
.moves
.lock()
.unwrap()
.entry(uri.to_string())
.or_default() += 1;
}
/// Records that `did` wrote to a repository collection, and refreshes its
/// mirrors in the background.
pub fn touch(self: &Arc<App>, did: &str) {
self.heard_of(did, None, true);
}
/// The writes from the pages recorded for `did` ([`App::touch`]).
fn writes_for(&self, did: &str) -> u64 {
self.writes.lock().unwrap().get(did).copied().unwrap_or(0)
}
/// The mirror of `did`'s repository `name` as last synced, if a read
/// may have it now ([`serves_last`]): never unless the firehose is
/// watching, which is what makes the sync of a stale one certain.
fn served(&self, did: &str, name: &str, wrote: u64) -> Option<Entry> {
if !self.watching.load(Ordering::Relaxed) {
return None;
}
let last = self.last.lock().unwrap();
let entry = last.get(&(did.to_string(), name.to_string()))?;
let stamp = self.stamp(did, entry.key(did).as_deref());
serves(entry, stamp, wrote).then(|| entry.clone())
}
/// Keeps `entry`, just synced, as what [`App::served`] gives, or none
/// if the repository is not there.
fn remember(&self, did: &str, name: &str, entry: &Entry) {
let mut last = self.last.lock().unwrap();
let at = (did.to_string(), name.to_string());
match entry.git {
Some(_) => last.insert(at, entry.clone()),
None => last.remove(&at),
};
}
/// Records that `did` wrote to its repository with `key`, or with none
/// to what could change any of them, and refreshes its mirrors in the
/// background: those it did not write to are found fresh, and left.
fn heard_of(self: &Arc<App>, did: &str, key: Option<&str>, own: bool) {
*self
.events
.lock()
.unwrap()
.entry(did.to_string())
.or_default() += 1;
let now = Instant::now();
for (_, entry) in self
.last
.lock()
.unwrap()
.iter_mut()
.filter(|(at, _)| at.0 == did)
{
entry.stale_since.get_or_insert(now);
}
if own {
*self.writes.lock().unwrap().entry(did.into()).or_default() += 1;
}
match key {
Some(key) => {
let mut repos = self.repo_events.lock().unwrap();
*repos.entry((did.into(), key.into())).or_default() += 1;
}
None => {
*self
.account_events
.lock()
.unwrap()
.entry(did.into())
.or_default() += 1
}
}
let names: Vec<String> = self
.repos
.lock()
.unwrap()
.keys()
.filter(|(d, _)| d == did)
.map(|(_, n)| n.clone())
.collect();
for name in names {
let app = self.clone();
let did = did.to_string();
tokio::spawn(async move {
let _ = app.located_with(&did, &name, true).await;
});
}
}
/// The mirror of `authority/name`, synced if it may be stale.
pub async fn mirror(&self, authority: &str, name: &str) -> Result<Git, AppError> {
Ok(self.located(authority, name).await?.git)
}
/// The mirror of `authority/name`, with the repository's key and the
/// name it has now: a public repository, or a private one the viewer
/// of the request being answered may read ([`private`]).
pub async fn located(&self, authority: &str, name: &str) -> Result<Located, AppError> {
self.located_with(authority, name, true).await
}
/// [`App::located`] for a read of the repository that answers a request:
/// a mirror a change has made stale may be served while it syncs
/// ([`App::served`]). Nothing that derives and keeps something from the
/// mirror may read it so.
pub async fn located_for_reader(
&self,
authority: &str,
name: &str,
) -> Result<Located, AppError> {
self.located_with(authority, name, false).await
}
/// [`App::located`], waiting for the sync of a stale mirror when `wait`,
/// as the refresh a change starts does: it is the sync.
async fn located_with(
&self,
authority: &str,
name: &str,
wait: bool,
) -> Result<Located, AppError> {
let url = RepoUrl::parse(&format!("knotless://{authority}/{name}"))
.map_err(|e| AppError(StatusCode::BAD_REQUEST, e.to_string()))?;
let remote = Remote::connect(&self.ctx, &url).await;
let remote = remote.map_err(AppError::upstream)?;
let did = remote.did().to_string();
let gateway = self.gateway.is_some();
// A delegation token for this one's space proves its asker a member,
// with no session of theirs here.
let delegated = private::delegated(&knotless::space::uri(&did, &url.name));
let delegated = delegated.filter(|_| gateway);
let viewer = private::viewer().filter(|_| gateway);
let viewer = viewer.or_else(|| delegated.as_ref().map(|d| d.did.clone()));
// One the viewer was shown as private is read as theirs again. Not
// found so, it is looked for as anyone looks: shown here may be only
// a look under way, such as the new repository form's check of a
// name, which a repository made public meanwhile is not in.
let shown = viewer
.as_ref()
.filter(|v| self.shown_private(v, &did, &url.name));
if let Some(viewer) = shown {
let read = self.as_viewer(viewer, &did, &url, None, delegated.as_ref());
let read = read.await;
if !read.as_ref().is_err_and(|e| e.0 == StatusCode::NOT_FOUND) {
return read;
}
}
let owner = wait || viewer.as_deref() == Some(did.as_str());
let missing = match self.public(remote, &url, owner).await {
Err(e) if e.0 == StatusCode::NOT_FOUND => e,
found => return found,
};
match &viewer {
Some(viewer) => {
let read = self.as_viewer(viewer, &did, &url, Some(missing), delegated.as_ref());
read.await
}
None => Err(missing),
}
}
/// [`App::located`] for a repository anyone may read.
async fn public(
&self,
remote: Remote,
url: &RepoUrl,
owner: bool,
) -> Result<Located, AppError> {
let did = remote.did().to_string();
// Watched, a mirror events have made stale is still served for a
// moment, while the sync they started finishes: a read does not
// wait on the lock that sync holds, unless its owner made it, who
// may have just written with a tool of their own.
let wrote = self.writes_for(&did);
let view = self.served(&did, &url.name, wrote).filter(|_| !owner);
if let Some(view) = view {
return self.admit(url, self.found(&view, &did, self.ctx.clone()));
}
let entry = self
.repos
.lock()
.unwrap()
.entry((did.clone(), url.name.clone()))
.or_default()
.clone();
let mut entry = entry.lock().await;
let key = entry.key(&did);
let any = self.events_for(&did);
let events = self.stamp(&did, key.as_deref());
let fresh = match (&entry.git, entry.synced) {
(Some(_), Some(_)) if self.watching.load(Ordering::Relaxed) => entry.events == events,
// A write from the pages touches it either way. Not there, it is
// asked after again on the same terms as one that is.
(_, Some(at)) => at.elapsed() < self.ttl && entry.events == events,
_ => false,
};
if fresh && entry.git.is_none() {
return Err(AppError(
StatusCode::NOT_FOUND,
format!("{url}: no such repository"),
));
}
if !fresh {
let synced = self.sync(&self.ctx, remote, url, &mut entry).await;
// Found, or found not there, as of these events.
if synced.is_ok() || entry.git.is_none() {
let found = entry.key(&did);
entry.events = self.settled(&did, key.as_deref(), events, any, found.as_deref());
}
if synced.is_ok() {
entry.writes = wrote;
}
self.remember(&did, &url.name, &entry);
synced?;
}
self.admit(url, self.found(&entry, &did, self.ctx.clone()))
}
/// `found`, unless it is taken down.
fn admit(&self, url: &RepoUrl, found: Located) -> Result<Located, AppError> {
let uri = knotless::forge::repo_uri(&found.did, &found.name);
if let Some(index) = &self.index
&& index
.down(&[&found.did, &uri])
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?
{
return Err(AppError(
StatusCode::NOT_FOUND,
format!("{url}: taken down"),
));
}
Ok(found)
}
/// Reads `remote`'s records afresh with `ctx`, and syncs `entry`'s
/// mirror, kept under `ctx`'s cache, to them.
async fn sync(
&self,
ctx: &Ctx,
remote: Remote,
url: &RepoUrl,
entry: &mut Entry,
) -> Result<(), AppError> {
let started = Instant::now();
// Its record and refs, not its packs: those are read only when the
// mirror lacks an object they hold.
let snap = remote.snapshot(ctx).await.map_err(AppError::upstream)?;
let (remote, snap) = remote.follow(ctx, snap).await.map_err(AppError::upstream)?;
let records = started.elapsed();
let Some(repo) = &snap.repo else {
// Not there, as of now.
entry.git = None;
entry.synced = Some(Instant::now());
return Err(AppError(
StatusCode::NOT_FOUND,
format!("{url}: no such repository"),
));
};
let key = repo.value.key.as_str();
let mirror = Mirror::open(ctx, remote.did(), key)
.await
.map_err(AppError::internal)?;
mirror
.refresh(&remote, &snap)
.await
.map_err(AppError::upstream)?;
aside(&mirror);
entry.found = Some((remote.did().to_string(), key.to_string(), repo.rkey.clone()));
entry.record = repo.value.clone();
entry.space = remote.space.as_ref().map(|s| s.uri.clone());
entry.git = Some(mirror.git);
entry.synced = Some(Instant::now());
// Every read of the repository waits on its sync: how long each
// takes, and how much of that is its records, at warn when slow.
let took = started.elapsed();
let f = Fields {
repo: Some(&url.to_string()),
duration_ms: Some(millis(took)),
records_ms: Some(millis(records)),
..Fields::default()
};
let level = [Level::INFO, Level::WARN][usize::from(slow(took))];
note(level, "mirror sync", &f);
Ok(())
}
/// What a synced `entry` holds, asked for under `did`, read with
/// `reader`.
fn found(&self, entry: &Entry, did: &str, reader: Ctx) -> Located {
let (now, key, name) = entry.found.clone().expect("a synced entry has a mirror");
let record = self.with_members(&now, entry.space.as_deref(), entry.record.clone());
Located {
git: entry.git.clone().expect("a synced entry has a mirror"),
moved: now != did,
did: now,
key,
name,
record,
space: entry.space.clone(),
reader,
}
}
}
/// Tidies `mirror` in a task of its own, out of the way of the request
/// that synced it and of every other waiting on that sync.
pub(crate) fn aside(mirror: &Mirror) {
let mirror = Mirror {
git: mirror.git.clone(),
};
tokio::spawn(async move { mirror.tidy().await.ok() });
}
/// The repositories `ops` write to, by key, each once: a record of one
/// repository, kept under its key, names it; any other, or none at all,
/// is `None`, for all of them.
fn scopes(ops: &[firehose::Op]) -> Vec<Option<String>> {
use knotless::records::nsid;
let one = [nsid::REF, nsid::CLAIM, nsid::PACK];
let mut keys: Vec<Option<String>> = ops
.iter()
.map(|op| {
let key = op.rkey.split_once('.').map(|(key, _)| key.to_string());
key.filter(|_| one.contains(&op.collection.as_str()))
})
.collect();
if keys.is_empty() {
keys.push(None);
}
keys.sort();
keys.dedup();
keys
}
impl Entry {
/// The key of the repository it last found, if it is still `did`'s.
fn key(&self, did: &str) -> Option<String> {
let found = self.found.as_ref().filter(|(now, _, _)| now == did);
found.map(|(_, key, _)| key.clone())
}
}
/// Whether `rev` is a commit of `did`'s that `seen` has not had yet, and
/// from now on has. An event naming no commit always is one.
fn first_of(seen: &Mutex<HashMap<String, String>>, did: &str, rev: Option<&str>) -> bool {
rev.is_none_or(|rev| {
let last = seen.lock().unwrap().insert(did.into(), rev.into());
last.as_deref() != Some(rev)
})
}
/// Indexes a write the event carried in full, made in the commit at `rev`
/// when it says: a deletion, or a record delivered inline. `false` when
/// the record has to be fetched.
fn index_inline(
index: &index::Index,
did: &str,
op: &firehose::Op,
rev: Option<&str>,
) -> rusqlite::Result<bool> {
let (collection, rkey) = (op.collection.as_str(), op.rkey.as_str());
match (&op.record, op.action.as_str()) {
(_, "delete") => index.delete_as_of(did, collection, rkey, rev)?,
(Some(record), _) => index.put_as_of(did, collection, rkey, record, rev)?,
(None, _) => return Ok(false),
}
Ok(true)
}
/// Indexes one write, made in the commit at `rev` when it is known: the
/// record the event carried, else the record as its PDS now has it, which
/// is as of that commit or later.
pub async fn index_op(
ctx: &Ctx,
index: &index::Index,
did: &str,
op: &firehose::Op,
rev: Option<&str>,
) -> Result<()> {
if index_inline(index, did, op, rev)? {
return Ok(());
}
let id = knotless::identity::resolve(ctx, did).await?;
let out: serde_json::Value = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds)
.query(
"com.atproto.repo.getRecord",
&[
("repo", did),
("collection", &op.collection),
("rkey", &op.rkey),
],
None,
)
.await?;
index.put_as_of(did, &op.collection, &op.rkey, &out["value"], rev)?;
Ok(())
}
/// Backfills from `service`, saying how it went.
pub async fn backfilled(
ctx: &Ctx,
index: &index::Index,
service: &str,
sources: &Sources,
) -> String {
match backfill(ctx, index, service, sources).await {
Ok(f) => format!(
"backfilled {} records of {} accounts from {service}, {} unchanged, {} failed",
f.records, f.read, f.unchanged, f.failed
),
Err(err) => format!("backfilling from {service}: {err:#}"),
}
}
/// The collections a backfill reads.
fn indexed_collections() -> impl Iterator<Item = &'static str> {
[
knotless::records::nsid::REPO,
knotless::records::nsid::REF,
knotless::records::nsid::CLAIM,
knotless::records::nsid::REDIRECT,
knotless::records::nsid::CHECK,
knotless::records::nsid::MINUTES,
knotless::sealed::KEY_NSID,
knotless::records::nsid::STAR,
knotless::records::nsid::FOLLOW,
knotless::records::nsid::REACTION,
knotless::records::nsid::RELEASE,
knotless::records::nsid::RELEASE_YANK,
knotless::records::nsid::SITE,
knotless::records::nsid::THEME,
knotless::records::nsid::ORG,
knotless::records::nsid::ORG_MEMBER,
knotless::records::nsid::ORG_ACCEPTANCE,
knotless::records::nsid::PROFILE,
]
.into_iter()
.chain(knotless::records::nsid::FORGE)
}
/// The accounts on `xrpc`'s service to index: from a relay, those it lists
/// as holding an indexed collection, a few among millions; from a PDS,
/// which lists accounts only all at once, every one.
///
/// Each of `directories` is asked first, in turn, and the first to answer
/// for every collection is believed.
async fn accounts(ctx: &Ctx, service: &str, directories: &[String]) -> Result<Vec<String>> {
for directory in directories {
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), directory);
if let Ok(dids) = holders(&xrpc).await {
return Ok(dids);
}
}
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), service);
match holders(&xrpc).await {
Ok(dids) => Ok(dids),
Err(_) => pages(&xrpc, "com.atproto.sync.listRepos", None).await,
}
}
/// The accounts `xrpc`'s service lists as holding an indexed collection.
async fn holders(xrpc: &knotless::xrpc::Xrpc) -> Result<Vec<String>> {
let mut dids = std::collections::BTreeSet::new();
for collection in indexed_collections() {
let by = "com.atproto.sync.listReposByCollection";
dids.extend(pages(xrpc, by, Some(collection)).await?);
}
Ok(dids.into_iter().collect())
}
/// The DIDs every page of `method` lists, with `collection` if given.
async fn pages(
xrpc: &knotless::xrpc::Xrpc,
method: &str,
collection: Option<&str>,
) -> Result<Vec<String>> {
#[derive(serde::Deserialize)]
struct Repos {
repos: Vec<Account>,
cursor: Option<String>,
}
#[derive(serde::Deserialize)]
struct Account {
did: String,
}
let mut dids = Vec::new();
let mut cursor: Option<String> = None;
loop {
let mut params = vec![("limit", "1000")];
params.extend(collection.map(|c| ("collection", c)));
params.extend(cursor.as_deref().map(|c| ("cursor", c)));
let page: Repos = xrpc.query(method, ¶ms, None).await?;
let done = page.repos.is_empty();
dids.extend(page.repos.into_iter().map(|r| r.did));
match page.cursor {
Some(next) if !done => cursor = Some(next),
_ => break,
}
}
Ok(dids)
}
/// Where a backfill finds the accounts to read and reads them, besides the
/// PDS or relay it is from and each account's PDS.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Sources {
/// Directories asked, in turn, which accounts hold knotless records
/// (`com.atproto.sync.listReposByCollection`), before the relay is.
pub directories: Vec<String>,
/// A mirror each account's repository is read from first, as
/// STAR-lite or a CAR ([`archive::fetch`]).
pub mirror: Option<String>,
}
/// What a backfill did: the records it indexed, from the accounts it
/// read; the accounts it found unchanged since it last read them, or put
/// off as one failed not long ago; and those it failed to read.
#[derive(Debug, Default)]
pub struct Filled {
pub records: usize,
pub read: usize,
pub unchanged: usize,
pub failed: usize,
}
/// How many accounts a backfill reads at once.
const READERS: usize = 4;
/// Indexes every account on `service`, a PDS or relay, that holds knotless
/// records: each account's repositories, checks and forge records, read
/// whole from its PDS and checked ([`archive`]).
///
/// An account whose latest commit is the one last read is passed over. A
/// record the feed indexed from a later commit than the one read stands
/// ([`index::Index::put_as_of`]). An account that fails is tried again
/// by a later backfill, a minute later for each time it failed, up to an
/// hour, and does not stop the rest.
pub async fn backfill(
ctx: &Ctx,
index: &index::Index,
service: &str,
sources: &Sources,
) -> Result<Filled> {
use futures::StreamExt;
let found = accounts(ctx, service, &sources.directories).await?;
let mirror = sources.mirror.as_deref();
let reads = futures::stream::iter(found)
.map(|did| async move {
let read = read_account(ctx, index, &did, mirror).await;
(did, read)
})
.buffer_unordered(READERS);
let reads: Vec<(String, Result<Option<usize>>)> = reads.collect().await;
let mut filled = Filled::default();
for (did, read) in reads {
match read {
Ok(Some(records)) => {
(filled.records, filled.read) = (filled.records + records, filled.read + 1)
}
Ok(None) => filled.unchanged += 1,
Err(err) => {
filled.failed += 1;
telemetry::backfill_failed(&did, &err);
let tries = u64::from(index.backfilled(&did)?.tries);
index.backfill_failed(&did, telemetry::now_secs() + 60 * (tries + 1).min(60))?;
}
}
}
Ok(filled)
}
/// Reads `did` into the index, from `mirror` when it has it, and how many
/// records it held; `None` when it is as last read, or a failure put it off
/// until later.
async fn read_account(
ctx: &Ctx,
index: &index::Index,
did: &str,
mirror: Option<&str>,
) -> Result<Option<usize>> {
#[derive(serde::Deserialize)]
struct Latest {
rev: String,
}
let was = index.backfilled(did)?;
if was.retry_at > telemetry::now_secs() {
return Ok(None);
}
let id = knotless::identity::resolve(ctx, did).await?;
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds);
let asked = [("did", did)];
let latest = xrpc.query::<Latest>("com.atproto.sync.getLatestCommit", &asked, None);
if was.rev.as_deref() == Some(latest.await?.rev.as_str()) {
return Ok(None);
}
let kept = |c: &str| indexed_collections().any(|i| i == c);
let read = archive::fetch(ctx, did, &id.pds, mirror, archive::MAX, &kept).await?;
for r in &read.records {
index.put_as_of(did, &r.collection, &r.rkey, &r.value, Some(&read.rev))?;
}
index.read_through(did, &read.rev)?;
index.backfill_read(did, &read.rev)?;
Ok(Some(read.records.len()))
}
/// How often the index is filled again from the relay.
const SWEEP: Duration = Duration::from_secs(600);
/// How often the spaces this server keeps an index of are asked what was
/// written there, besides as their authorities tell of writes: for those
/// whose notifications went missing.
const SPACES: Duration = Duration::from_secs(30);
/// How often devices are told of what is new.
const PUSH_ROUND: Duration = Duration::from_secs(30);
/// How often what was counted is summed up in the log.
const SUMMARY: Duration = Duration::from_secs(60);
/// Backfills from `relay` every `every`, until the process ends. A record
/// whose event a feed never delivered, one Jetstream dropped or a commit
/// the relay resynced over, is otherwise missing from the index for good.
/// And puts back, each time, what labels that lapsed took down.
pub async fn sweep(
ctx: Ctx,
index: Arc<index::Index>,
relay: String,
sources: Sources,
every: Duration,
) {
loop {
tokio::time::sleep(every).await;
let _ = index.lapse();
telemetry::sweep(&ctx, &index, &relay, &sources).await;
}
}
/// `err` as the pages and the app say it: what to do about a refusal in
/// their own controls, never a command ([`knotless::fix`]).
pub(crate) fn shown(err: &anyhow::Error) -> String {
knotless::fix::said(err, knotless::fix::Words::Pages)
}
/// A status and a message, as the response.
#[derive(Debug)]
pub struct AppError(pub StatusCode, pub String);
impl AppError {
/// 502 for what another server failed at; 404 for a name that is no
/// one's handle, as it cannot be one or a resolver found no one.
pub(crate) fn upstream(err: anyhow::Error) -> AppError {
let status = match err.downcast_ref::<knotless::identity::NoSuchHandle>() {
Some(_) => StatusCode::NOT_FOUND,
None => StatusCode::BAD_GATEWAY,
};
AppError(status, shown(&err))
}
/// 400 for input the request got wrong, else 500.
pub(crate) fn input(err: anyhow::Error) -> AppError {
let status = match err.downcast_ref::<knotless::read::BadInput>() {
Some(_) => StatusCode::BAD_REQUEST,
None => StatusCode::INTERNAL_SERVER_ERROR,
};
AppError(status, shown(&err))
}
pub(crate) fn internal(err: anyhow::Error) -> AppError {
AppError(StatusCode::INTERNAL_SERVER_ERROR, shown(&err))
}
}
impl IntoResponse for AppError {
fn into_response(self) -> Response {
telemetry::failed(self.0, self.1)
}
}
pub fn router(app: Arc<App>) -> Router {
// The queries, which know who asks when a service token says.
let queries = Router::new()
.route("/xrpc/dev.knotless.getForge", get(appview::get_forge))
.route("/xrpc/dev.knotless.listRepos", get(appview::list_repos))
.route(
"/xrpc/dev.knotless.searchOwners",
get(appview::search_owners),
)
.route(
"/xrpc/dev.knotless.describeServer",
get(appview::describe_server),
)
.route("/xrpc/dev.knotless.listChecks", get(appview::list_checks))
.route("/xrpc/dev.knotless.getFeed", get(web::get_feed))
.route("/xrpc/dev.knotless.getProfile", get(web::get_profile))
.route(
"/xrpc/dev.knotless.getActorRepos",
get(web::get_actor_repos),
)
.route(
"/xrpc/dev.knotless.getActorActivity",
get(web::get_actor_activity),
)
.merge(web::queries())
.route_layer(axum::middleware::from_fn_with_state(
app.clone(),
viewer::viewer,
))
.layer(viewer::cors());
Router::new()
.route("/healthz", get(|| async { "ok" }))
.route("/_status", get(status::status))
.route("/_hit", post(usage::hit))
.route("/xrpc/dev.knotless.getUsage", get(usage::report))
.route("/xrpc/dev.knotless.getStats", get(usage::stats))
.route("/xrpc/dev.knotless.getLogs", get(logs::get_logs))
.route("/{authority}/{repo}/info/refs", get(smart::info_refs))
.route(
"/{authority}/{repo}/git-upload-pack",
post(smart::upload_pack),
)
.route("/api/{authority}/{repo}/refs", get(api::refs))
.route("/api/{authority}/{repo}/tree", get(api::tree))
.route("/api/{authority}/{repo}/blob", get(api::blob))
.route("/api/{authority}/{repo}/log", get(api::log))
.route("/api/{authority}/{repo}/diff", get(api::diff))
.route("/api/{authority}/{repo}/object/{sha}", get(api::object))
// Public, and how a page elsewhere finds where the queries are.
.route(
"/.well-known/did.json",
get(gateway::did_document).layer(viewer::cors()),
)
.route("/xrpc/dev.knotless.putDelegation", post(gateway::delegate))
.route(
"/xrpc/dev.knotless.getDelegation",
get(gateway::await_delegation),
)
.route(gateway::METADATA, get(gateway::client_metadata))
.route("/oauth/callback", get(gateway::callback))
.route(gateway::BOT_METADATA, get(gateway::bot_metadata))
.route("/oauth/bot", get(gateway::bot_login))
.route("/oauth/bot-callback", get(gateway::bot_callback))
.route(login::METADATA, get(login::client_metadata))
.route("/oauth/login", post(login::login))
.route("/oauth/app-login", get(login::app_login))
.route("/oauth/app-token", post(login::app_token))
.route(login::APP_RETURN, get(login::app_return))
.route(
"/.well-known/apple-app-site-association",
get(login::app_association),
)
.route("/oauth/web-callback", get(login::callback))
.route("/oauth/switch", post(login::switch))
.route("/oauth/logout", post(login::logout))
.route(
"/xrpc/dev.knotless.pushRefs",
post(gateway::push).layer(axum::extract::DefaultBodyLimit::max(PUSH_LIMIT)),
)
.route("/xrpc/dev.knotless.createCommit", post(gateway::commit))
.route(
"/xrpc/dev.knotless.putOrgMember",
post(gateway::put_org_member),
)
.route(
"/xrpc/dev.knotless.removeOrgMember",
post(gateway::remove_org_member),
)
.route("/xrpc/dev.knotless.joinOrg", post(gateway::join_org))
.route(
"/xrpc/dev.knotless.createOrgRepo",
post(gateway::create_org_repo),
)
.route("/xrpc/dev.knotless.leaveOrg", post(gateway::leave_org))
.route(
"/xrpc/com.atproto.moderation.createReport",
post(moderation::create_report),
)
.route(
"/xrpc/com.atproto.admin.updateSubjectStatus",
post(moderation::update_subject_status),
)
.route(
"/xrpc/com.atproto.admin.getSubjectStatus",
get(moderation::get_subject_status),
)
.route(
"/xrpc/dev.knotless.listReports",
get(moderation::list_reports),
)
.route("/xrpc/dev.knotless.inviteRunner", post(invites::invite))
.route("/xrpc/com.atproto.space.notifyWrite", post(spaces::written))
.route(
"/xrpc/com.atproto.space.notifySpaceDeleted",
post(spaces::deleted),
)
.route("/xrpc/dev.knotless.listRunnerInvites", get(invites::list))
.route("/xrpc/dev.knotless.putSecret", post(secrets::put))
.route("/xrpc/dev.knotless.deleteSecret", post(secrets::delete))
.route("/xrpc/dev.knotless.listSecrets", get(secrets::list))
.route(
"/xrpc/dev.knotless.putRunOutput",
post(live::put_run_output).layer(axum::extract::DefaultBodyLimit::max(live::BODY)),
)
.route(
"/xrpc/dev.knotless.awaitSpaceWrites",
get(live::await_space_writes),
)
.route(
"/xrpc/com.atproto.label.queryLabels",
get(moderation::query_labels),
)
.route(
"/xrpc/com.atproto.label.subscribeLabels",
get(moderation::subscribe_labels),
)
.merge(queries)
.merge(web::routes())
.layer(axum::middleware::from_fn(web::xrpc_failures))
.layer(axum::middleware::from_fn(web::origin))
.with_state(app)
}
/// How long a check may run before the index's word for it is read
/// again from its PDS, and how often again after.
const RECHECK: Duration = Duration::from_secs(60);
/// The largest push a gateway takes.
const PUSH_LIMIT: usize = 512 << 20;
/// `name` with a trailing `.git` dropped, as clone URLs often carry one.
fn repo_name(name: &str) -> &str {
name.strip_suffix(".git").unwrap_or(name)
}
/// The arguments `knotless-serve` takes.
#[derive(Debug, PartialEq)]
pub struct Options {
pub listen: String,
pub ttl: Duration,
pub firehose: Option<String>,
/// A Jetstream v2 host, followed like the firehose.
pub jetstream: Option<String>,
/// The URL this server is reached at, which makes it a gateway.
pub public_url: Option<String>,
/// Where to keep the appview index, which makes it an appview.
pub index: Option<std::path::PathBuf>,
/// A PDS or relay to index every account of at startup.
pub backfill: Option<String>,
/// A relay to index the accounts holding knotless records from when
/// the index starts empty, so a new version of its tables loses
/// nothing older than what Jetstream replays.
pub relay: Option<String>,
/// Where a backfill finds accounts and reads them first: directories,
/// and a mirror.
pub sources: Sources,
/// Labelers, by DID, whose takedowns hold here too.
pub labelers: Vec<String>,
/// The runner, by DID, a new public repository names unless its
/// owner names others: one, as every runner named runs each check.
pub default_runner: Option<String>,
/// The oldest client version to ask for, when one harms what it
/// touches, and why: clients older refuse to run until upgraded.
pub client_minimum: Option<String>,
pub client_minimum_reason: Option<String>,
/// The oldest build of the iOS app to let run, once a newer one is in
/// people's reach, and why: builds older show only a way to update.
pub ios_minimum_build: Option<u64>,
pub ios_minimum_reason: Option<String>,
/// The SvelteKit app that draws the web pages, which this server hands
/// every request it does not answer itself.
pub web: Option<String>,
/// Where the client's builds are, a directory or a repository's
/// releases, handed out under `/releases` with `/install.sh` to fetch
/// one.
pub releases: Option<releases::Source>,
/// How long a feed may stay silent before `/_status` calls it dead.
pub stale: Option<Duration>,
/// Where to keep the counts of how this server is used.
pub usage: Option<std::path::PathBuf>,
/// Where to keep the lines it logs, and how many at info and below.
pub logs: Option<std::path::PathBuf>,
pub logs_keep: u64,
/// Only check that the index at this path is of this build's tables,
/// so that starting would keep it rather than rebuild it.
pub check_index: Option<std::path::PathBuf>,
/// Where `/metrics` and `/log` are answered, apart from the rest.
pub metrics: Option<String>,
/// Where a gateway keeps the sessions its bots log in to.
pub bots: Option<std::path::PathBuf>,
/// The PDS where the pages send someone making an account:
/// [`gateway::SIGN_UP_PDS`] unless `--sign-up-pds` names another.
pub sign_up_pds: String,
/// Where the PDS sends a gateway's OAuth logins back to, when not
/// `--public-url`: an https front for it, as the PDS asks of a client.
pub oauth_url: Option<String>,
/// Where code search keeps its Zoekt shards, if it is on.
pub code_index: Option<std::path::PathBuf>,
/// The domain repositories' sites are answered on, each owner at a
/// label of it, as `knotless.page`: none answers none.
pub pages_domain: Option<String>,
/// The key Apple gave for its push service, a `.p8` file, which with
/// its id and team tells devices of their accounts' notifications.
pub apns_key: Option<std::path::PathBuf>,
pub apns_key_id: Option<String>,
pub apns_team: Option<String>,
}
/// What `--private`, which is gone, says instead.
pub const PRIVATE_GONE: &str = "--private is gone: knotless-serve shows a private repository to each member who logs in to its pages (--public-url), read as them, so run it without --private and log in, or browse https://knotless.dev logged in";
pub fn parse_args(args: &[String]) -> Result<Options, String> {
let mut opts = Options {
listen: "127.0.0.1:2590".into(),
ttl: Duration::from_secs(5),
firehose: None,
jetstream: None,
public_url: None,
index: None,
backfill: None,
relay: None,
sources: Sources::default(),
labelers: Vec::new(),
default_runner: None,
client_minimum: None,
client_minimum_reason: None,
ios_minimum_build: None,
ios_minimum_reason: None,
web: None,
releases: None,
stale: None,
usage: None,
logs: None,
logs_keep: logs::KEEP,
check_index: None,
metrics: None,
bots: None,
sign_up_pds: gateway::SIGN_UP_PDS.into(),
oauth_url: None,
code_index: None,
pages_domain: None,
apns_key: None,
apns_key_id: None,
apns_team: None,
};
let mut it = args.iter().skip(1);
while let Some(flag) = it.next() {
let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value"));
match flag.as_str() {
"--listen" => opts.listen = value()?,
"--ttl" => {
opts.ttl = Duration::from_secs(
value()?
.parse()
.map_err(|_| "--ttl takes seconds".to_string())?,
)
}
// An empty stream or backfill is none, so a setting can turn
// one off: `--firehose "$KNOTLESS_FIREHOSE"`, left unset.
"--firehose" => opts.firehose = Some(value()?).filter(|v| !v.is_empty()),
"--jetstream" => opts.jetstream = Some(value()?).filter(|v| !v.is_empty()),
"--public-url" => opts.public_url = Some(value()?),
"--pages-domain" => opts.pages_domain = Some(value()?).filter(|v| !v.is_empty()),
"--apns-key" => {
opts.apns_key = Some(value()?).filter(|v| !v.is_empty()).map(Into::into)
}
"--apns-key-id" => opts.apns_key_id = Some(value()?).filter(|v| !v.is_empty()),
"--apns-team" => opts.apns_team = Some(value()?).filter(|v| !v.is_empty()),
"--index" => opts.index = Some(value()?.into()),
"--backfill" => opts.backfill = Some(value()?).filter(|v| !v.is_empty()),
"--relay" => opts.relay = Some(value()?),
"--directory" => opts
.sources
.directories
.extend(Some(value()?).filter(|v| !v.is_empty())),
"--mirror" => opts.sources.mirror = Some(value()?).filter(|v| !v.is_empty()),
"--labeler" => opts.labelers.push(value()?),
"--default-runner" => {
let did = value()?;
if !did.is_empty() && !did.starts_with("did:") {
return Err(format!("--default-runner takes a DID, not {did}"));
}
if opts.default_runner.is_some() {
return Err("--default-runner names one runner".into());
}
opts.default_runner = Some(did).filter(|d| !d.is_empty());
}
"--client-minimum" => {
let version = value()?;
if !version.is_empty() {
knotless::release::version(&version).map_err(|e| e.to_string())?;
}
opts.client_minimum = Some(version).filter(|v| !v.is_empty());
}
"--client-minimum-reason" => {
opts.client_minimum_reason = Some(value()?).filter(|v| !v.is_empty())
}
"--ios-minimum-build" => {
let build = value()?;
let number = build.parse::<u64>();
let number = number.map_err(|_| "--ios-minimum-build takes a build number");
opts.ios_minimum_build = match build.is_empty() {
true => None,
false => Some(number?),
};
}
"--ios-minimum-reason" => {
opts.ios_minimum_reason = Some(value()?).filter(|v| !v.is_empty())
}
// Gone: the pages read private repositories as whoever logs
// in to them.
"--private" => return Err(PRIVATE_GONE.into()),
"--web" => opts.web = Some(value()?),
"--releases" => opts.releases = Some(releases::Source::parse(&value()?)?),
"--stale" => {
let secs = value()?.parse();
let secs = secs.map_err(|_| "--stale takes seconds".to_string())?;
opts.stale = Some(Duration::from_secs(secs));
}
"--usage" => opts.usage = Some(value()?.into()),
"--logs" => opts.logs = Some(value()?.into()),
"--logs-keep" => {
let n = value()?.parse();
opts.logs_keep = n.map_err(|_| "--logs-keep takes a count of lines".to_string())?;
}
"--check-index" => opts.check_index = Some(value()?.into()),
"--metrics" => opts.metrics = Some(value()?),
"--bots" => opts.bots = Some(value()?.into()),
"--sign-up-pds" => opts.sign_up_pds = value()?,
"--oauth-url" => opts.oauth_url = Some(value()?),
"--code-index" => opts.code_index = Some(value()?.into()),
other => {
return Err(format!(
"unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--ios-minimum-build <build> [--ios-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--logs <file> [--logs-keep <lines>]] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>] [--code-index <dir>] [--apns-key <file> --apns-key-id <id> --apns-team <id>] [--pages-domain <domain>]"
));
}
}
}
Ok(opts)
}
/// Whether the index at `path` is missing or of this build's tables: an
/// error when starting with it would rebuild it.
fn check_index(path: &std::path::Path) -> Result<()> {
let found = index::version_at(path)?;
if let Some(v) = found.filter(|v| *v != index::VERSION) {
anyhow::bail!(
"{}: its tables are version {v}, this build's {}: starting would rebuild it",
path.display(),
index::VERSION
);
}
println!("{}: this build keeps it", path.display());
Ok(())
}
/// Serves until SIGTERM or SIGINT. Prints the bound address first.
pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
// Asked by deploy/deploy.sh before starting this build beside a running
// one: both have the index open while one hands over to the other,
// which is safe only if this one would not rebuild it.
if let Some(path) = opts.check_index {
return check_index(&path);
}
if opts.public_url.is_none() && opts.bots.is_some() {
anyhow::bail!("--bots takes --public-url, where bots log in");
}
if opts.public_url.is_none() && opts.pages_domain.is_some() {
anyhow::bail!(
"--pages-domain takes --public-url, where a site's owner is sent from its root"
);
}
let apns = match &opts.apns_key {
Some(path) => {
let (key_id, team) = (opts.apns_key_id.as_deref(), opts.apns_team.as_deref());
Some(web::push::Apns::open(path, key_id, team)?)
}
None => None,
};
let gateway = match &opts.public_url {
Some(url) => {
let mut gw = gateway::Gateway::new(url).map_err(anyhow::Error::msg)?;
gw.bots = opts.bots.clone();
gw.sign_up_pds = opts.sign_up_pds.clone();
if let Some(url) = &opts.oauth_url {
gw.oauth_url = url.trim_end_matches('/').to_string();
}
Some(gw)
}
None => None,
};
let opening = Instant::now();
let index = match &opts.index {
Some(path) => Some(index::Index::open(path)?),
None => None,
};
telemetry::opened(index.as_ref(), opening.elapsed());
let usage = match &opts.usage {
Some(path) => Some(Arc::new(usage::Usage::open(path)?)),
None => None,
};
let logs = match &opts.logs {
Some(path) => Some(Arc::new(logs::Logs::open(path, opts.logs_keep)?)),
None => None,
};
// Every line logged from here on, kept to read back.
if let Some(logs) = logs.clone() {
knotless::telemetry::keep_lines(move |line| logs.keep(line));
}
// The index's slow statements are counted in it.
usage.iter().for_each(usage::sink);
// MCP over HTTP, for any account not taken down, made once
// the address it reads this server's index at is known.
let mcp_ctx = ctx.clone();
let code = match &opts.code_index {
Some(dir) => Some(Arc::new(code::Code::start(dir).await?)),
None => None,
};
let mut app = App::build(ctx, opts.ttl, gateway, index);
let reason = opts.client_minimum_reason.clone();
let client_minimum = opts.client_minimum.clone().map(|m| (m, reason));
let reason = opts.ios_minimum_reason.clone();
let ios_minimum = opts.ios_minimum_build.map(|b| (b, reason));
if let Some(app) = Arc::get_mut(&mut app) {
app.stale = opts.stale;
app.disk = Some(app.ctx.env.cache_dir.clone());
app.usage = usage.clone();
app.logs = logs.clone();
app.default_runner = opts.default_runner.clone();
app.client_minimum = client_minimum;
app.ios_minimum = ios_minimum;
app.code = code;
}
// An index built anew is filled from the relay, unless asked otherwise.
let rebuilt = app.index.as_ref().is_some_and(|i| i.rebuilt);
let backfill = opts.backfill.or(opts.relay.clone().filter(|_| rebuilt));
if let (Some(service), Some(index)) = (backfill, app.index.clone()) {
let ctx = app.ctx.clone();
let sources = opts.sources.clone();
tokio::spawn(async move { telemetry::sweep(&ctx, &index, &service, &sources).await });
}
// And again now and then, for what the feeds missed.
if let (Some(relay), Some(index)) = (opts.relay, app.index.clone()) {
tokio::spawn(sweep(app.ctx.clone(), index, relay, opts.sources, SWEEP));
}
for did in opts.labelers {
tokio::spawn(moderation::follow(app.clone(), did));
}
if let Some(url) = opts.firehose {
app.watching.store(true, Ordering::Relaxed);
tokio::spawn(firehose::watch(app.clone(), url));
}
if let Some(url) = opts.jetstream {
app.watching.store(true, Ordering::Relaxed);
tokio::spawn(jetstream::watch(app.clone(), url));
}
if let (Some(apns), Some(index)) = (apns, app.index.clone()) {
tokio::spawn(web::push::follow(app.clone(), index, apns, PUSH_ROUND));
}
tokio::spawn(spaces::follow(app.clone(), SPACES));
tokio::spawn(languages::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?;
let addr = listener.local_addr()?;
let listening = format!("http://{addr}");
println!("listening on {listening}");
telemetry::said("listening", Some(&listening));
// The numbers and the log's level, on a port of their own, kept off
// the proxy: printed second, after where the pages are answered.
telemetry::serve_admin(opts.metrics.as_deref(), app.clone()).await?;
// With no pages to hand on to, what this server does not answer is not found.
let pages = opts.web.map(web::pages).unwrap_or_default();
let index = app.index.clone();
let mcp = opts
.public_url
.as_deref()
.map(|url| mcp_routes(mcp_ctx, url, index, addr, usage.clone()));
let mcp = mcp.transpose()?.unwrap_or_default();
let registry = oci::routes(app.ctx.clone(), opts.public_url.clone());
let home = opts.public_url.clone().unwrap_or_default();
let releases = releases::routes(opts.releases, opts.public_url, app.ctx.clone());
let stopping = app.live.stopping.clone();
let sites = opts
.pages_domain
.map(|d| sites::Sites::new(app.clone(), d, home));
let mut routes = router(app)
.merge(releases)
.merge(registry)
.merge(mcp)
.merge(pages);
if let Some(sites) = sites {
routes = sites::hosted(routes, sites);
}
if let Some(usage) = usage.clone() {
tokio::spawn(usage::flushing(usage.clone(), usage::FLUSH));
routes = usage::counted(routes, usage);
}
if let Some(logs) = logs.clone() {
tokio::spawn(logs::flushing(logs, logs::FLUSH));
}
axum::serve(listener, telemetry::observed(routes))
.with_graceful_shutdown(async move {
let mut term =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("installing a SIGTERM handler");
tokio::select! {
_ = term.recv() => {}
_ = tokio::signal::ctrl_c() => {}
}
telemetry::said("stopping: finishing the requests it has", None);
// What the pages follow ends, for them to follow the next build.
stopping.cancel();
})
.await?;
// What was counted since the last write.
usage.as_deref().map_or(Ok(()), usage::Usage::flush)?;
telemetry::said("stopped", None);
// The lines logged since the last write, that one with them.
logs.as_deref().map_or(Ok(()), logs::Logs::flush)?;
Ok(())
}
/// MCP over HTTP at `url`, for any account `index` has not taken down.
fn mcp_routes(
mut ctx: Ctx,
url: &str,
index: Option<Arc<index::Index>>,
addr: SocketAddr,
usage: Option<Arc<usage::Usage>>,
) -> Result<Router> {
ctx.env.appview = mcp_appview(ctx.env.appview, index.is_some(), addr);
let limits = knotless_mcp::http::Limits::default();
let mcp = knotless_mcp::http::Mcp::new(&ctx, url, limits)?;
if let Some(usage) = usage {
mcp.observe(usage::mcp_events(usage));
}
if let Some(index) = index {
mcp.refuse(taken_down(index));
}
Ok(knotless_mcp::http::router(mcp))
}
/// Whether an account is taken down here, or by a labeler this server
/// follows: one the index cannot answer for is let in.
fn taken_down(index: Arc<index::Index>) -> knotless_mcp::http::Refused {
Arc::new(move |did| index.down(&[did]).unwrap_or(false))
}
/// Where MCP over HTTP finds pulls and issues on public repositories: the
/// appview named, or else this server itself when it keeps the index, at
/// `addr` on this machine. An index of its own is read before the default
/// appview, knotless.dev's, which only an unset KNOTLESS_APPVIEW gives.
fn mcp_appview(named: Option<String>, indexed: bool, addr: SocketAddr) -> Option<String> {
let default = named.as_deref() == Some(knotless::env::DEFAULT_APPVIEW);
let named = named.filter(|_| !(indexed && default));
named.or_else(|| {
indexed.then(|| {
let ip = match addr.ip() {
IpAddr::V4(ip) if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
IpAddr::V6(ip) if ip.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
ip => ip,
};
format!("http://{}", SocketAddr::new(ip, addr.port()))
})
})
}
#[cfg(test)]
mod tests {
#[test]
fn a_record_of_one_repository_is_its_alone() {
let op = |collection: &str, rkey: &str| firehose::Op {
action: "create".into(),
collection: collection.into(),
rkey: rkey.into(),
record: None,
};
use knotless::records::nsid;
// A push's pack, claim and ref name one repository, once.
let push = [
op(nsid::PACK, "k.3kaaa"),
op(nsid::CLAIM, "k.m.1"),
op(nsid::REF, "k.m"),
];
assert_eq!(scopes(&push), [Some("k".to_string())]);
// A repository's own record, or one under no key, is of them all.
let both = [op(nsid::REF, "j.m"), op(nsid::REPO, "name")];
assert_eq!(scopes(&both), [None, Some("j".to_string())]);
assert_eq!(scopes(&[op(nsid::REF, "bare")]), [None]);
assert_eq!(scopes(&[]), [None]);
}
#[test]
fn a_stale_mirror_is_served_while_it_syncs() {
let long_ago = Instant::now().checked_sub(STALE_FOR * 2).unwrap();
let mut entry = Entry {
events: (1, 1),
..Entry::default()
};
// Quiet, it is served however long ago it synced.
entry.synced = Some(long_ago);
assert!(serves(&entry, (1, 1), 0));
// Stale with no event noted, or the first long ago, it is not.
assert!(!serves(&entry, (2, 1), 0));
entry.stale_since = Some(long_ago);
assert!(!serves(&entry, (2, 1), 0));
// The first event only now, it is, however long ago it synced.
entry.stale_since = Some(Instant::now());
assert!(serves(&entry, (2, 1), 0));
// A write from the pages it may not have seen waits for the sync.
assert!(!serves(&entry, (2, 1), 1));
assert!(!serves(&entry, (1, 1), 1));
}
#[test]
fn one_commit_is_one_touch() {
let seen = Mutex::default();
assert!(first_of(&seen, "did:a", Some("r1")));
// The rest of the commit's records, each an event of its own.
assert!(!first_of(&seen, "did:a", Some("r1")));
assert!(!first_of(&seen, "did:a", Some("r1")));
assert!(first_of(&seen, "did:b", Some("r1")));
assert!(first_of(&seen, "did:a", Some("r2")));
assert!(first_of(&seen, "did:a", None));
assert!(first_of(&seen, "did:a", None));
}
use super::*;
#[test]
fn mcp_reads_its_own_index() {
let at = |s: &str| s.parse::<SocketAddr>().unwrap();
let named = Some("https://appview.example".to_string());
assert_eq!(mcp_appview(named.clone(), true, at("0.0.0.0:2590")), named);
assert_eq!(mcp_appview(None, false, at("0.0.0.0:2590")), None);
// The default appview gives way to an index of its own.
let default = Some(knotless::env::DEFAULT_APPVIEW.to_string());
let unindexed = mcp_appview(default.clone(), false, at("0.0.0.0:2590"));
assert_eq!(unindexed, default);
let indexed = mcp_appview(default, true, at("10.0.0.2:2590"));
assert_eq!(indexed.as_deref(), Some("http://10.0.0.2:2590"));
let own = |addr| mcp_appview(None, true, at(addr)).unwrap();
assert_eq!(own("0.0.0.0:2590"), "http://127.0.0.1:2590");
assert_eq!(own("[::]:2590"), "http://[::1]:2590");
assert_eq!(own("10.0.0.2:2590"), "http://10.0.0.2:2590");
}
#[test]
fn taken_down_accounts_are_refused() {
let dir = tempfile::tempdir().unwrap();
let index = Arc::new(index::Index::open(&dir.path().join("index.db")).unwrap());
let refused = taken_down(index.clone());
assert!(!refused("did:plc:a"));
let subject = serde_json::json!({"did": "did:plc:a"});
index
.take_down("did:web:me", "did:plc:a", &subject, true)
.unwrap();
assert!(refused("did:plc:a"));
assert!(!refused("did:plc:b"));
}
#[test]
fn input_errors() {
let bad = AppError::input(knotless::read::BadInput("no".into()).into());
assert_eq!(bad.0, StatusCode::BAD_REQUEST);
let broken = AppError::input(anyhow::anyhow!("disk on fire"));
assert_eq!(broken.0, StatusCode::INTERNAL_SERVER_ERROR);
}
#[test]
fn upstream_errors() {
let nobody = knotless::identity::NoSuchHandle("alice.test".into());
assert_eq!(AppError::upstream(nobody.into()).0, StatusCode::NOT_FOUND);
let down = AppError::upstream(anyhow::anyhow!("resolver down"));
assert_eq!(down.0, StatusCode::BAD_GATEWAY);
}
// A collection a backfill reads and the firehose passes by is indexed
// once and never again: each list names it, or the index goes stale.
// A rebuilt index must still know what was yanked, or each yanked
// version reads as live again.
#[test]
fn a_backfill_reads_yanks() {
let yanks = knotless::records::nsid::RELEASE_YANK;
assert!(indexed_collections().any(|c| c == yanks));
}
#[test]
fn the_firehose_keeps_every_indexed_collection() {
for c in indexed_collections() {
let kept = knotless::jetstream::knotless(c);
assert!(kept, "{c} is backfilled but not kept live");
}
}
}