pub(crate) mod failpoint;
pub(crate) mod profile;
pub mod activate;
pub mod gc;
pub(crate) mod hooks;
pub mod manifest;
pub mod plan;
pub mod publish;
pub mod release;
pub mod snapshot;
#[cfg(test)]
mod tests;
mod verify;
pub(crate) use verify::{
COMPLETION_FILE, PrivateGeneration, Publication, WorkerJoin, build_plan, compute_dir_size,
mount_release_export, unchanged_shortcut, write_www,
};
use rayon::prelude::*;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::Path;
use std::sync::{Arc, Mutex};
use anyhow::Context as _;
use knot_index::Index;
use knot_types::{OwnerDid, RepoDid, RepoName};
use self::plan::{GlobalInputs, RepoOutcome};
use crate::config::ValidatedConfig;
use crate::git_export;
use crate::identity;
use crate::manifest::{
BuildStats, GENERATOR_PROFILE_VERSION, GenerationManifest, PhaseTimings, RepoEntry, short_id,
};
use crate::out::out;
use crate::release::ReleaseBuild;
use crate::store::RepoStore;
pub(crate) struct RepoPlan {
pub(crate) snapshot: crate::snapshot::RepoSnapshot,
pub(crate) identity: String,
pub(crate) outcome: RepoOutcome,
pub(crate) prior_release: Option<(String, String)>,
pub(crate) prior_export: Option<git_export::ExportPrior>,
}
impl RepoPlan {
pub(crate) fn unchanged(&self) -> bool {
matches!(self.outcome, RepoOutcome::Unchanged { .. })
}
}
#[derive(Debug)]
pub enum BuildOutcome {
Unchanged {
generation_id: String,
},
Built {
generation_id: String,
/// keep it until the generation is activated
pin: crate::gc::Pin,
},
}
fn entry_release_sealed(output_root: &Path, entry: &RepoEntry) -> bool {
let Some(release_id) = &entry.release_id else {
return false;
};
if entry.git_export_hash.is_none() {
return false;
}
let release = output_root
.join("repos")
.join(plan::repo_key(&entry.repo_did))
.join("releases")
.join(release_id);
crate::release::PublishedRelease::open(&release).is_ok_and(|release| release.validate().is_ok())
}
/// the fingerprints only hold under the same global config. any mismatch recaptures.
fn carried_candidate_dids(
output_root: &Path,
active: &GenerationManifest,
global: &GlobalInputs,
hosted: &BTreeSet<String>,
skip_did: Option<&str>,
) -> BTreeSet<String> {
if !global.matches_manifest(active) {
return BTreeSet::new();
}
active
.repos
.iter()
.filter(|entry| {
Some(entry.repo_did.as_str()) != skip_did
&& hosted.contains(entry.repo_did.as_str())
&& entry_release_sealed(output_root, entry)
})
.map(|entry| entry.repo_did.clone())
.collect()
}
pub(crate) struct Tracer {
enabled: bool,
}
impl Tracer {
pub(crate) fn new() -> Self {
Self {
enabled: std::env::var_os("KNOT_PLAN_TRACE").is_some(),
}
}
pub(crate) fn time<R>(&self, label: &'static str, work: impl FnOnce() -> R) -> R {
if !self.enabled {
return work();
}
let started = std::time::Instant::now();
let result = work();
let ms = started.elapsed().as_secs_f64() * 1000.0;
out!("[trace] {label}: {ms:.1}ms");
result
}
fn record(&self, label: &'static str, opened: std::time::Instant) {
if !self.enabled {
return;
}
let ms = opened.elapsed().as_secs_f64() * 1000.0;
out!("[trace] {label}: {ms:.1}ms");
}
}
pub async fn run(
config: Arc<ValidatedConfig>,
repo_hint: Option<&str>,
) -> anyhow::Result<BuildOutcome> {
run_with(config, repo_hint, None).await
}
/// a write hands over the index it kept current, which saves replaying every cob of its repo again
pub(crate) async fn run_with(
config: Arc<ValidatedConfig>,
repo_hint: Option<&str>,
synced: Option<&crate::atproto::SyncedIndex>,
) -> anyhow::Result<BuildOutcome> {
let started = std::time::Instant::now();
let t_plan = started;
let trace = Tracer::new();
let layout = trace.time("bootstrap: layout", || config.layout())?;
let meta_repo = layout
.bootstrap_meta(&config.knot_did)
.context("bootstrap meta repo")?;
let meta_path = layout.meta_path(&config.knot_did)?;
let fresh;
let index = match synced {
Some(synced) => &synced.index,
None => {
fresh = Index::new(meta_path, layout.clone());
&fresh
}
};
trace.time("bootstrap: index rebuild", || {
index.rebuild().context("rebuild registry index")
})?;
let output_root = config.output_root.clone();
let generations_root = output_root.join("generations");
std::fs::create_dir_all(&generations_root)?;
let repos_root = output_root.join("repos");
std::fs::create_dir_all(&repos_root)?;
// the active generation is read twice: unlocked here only to pick the repo locks,
// and the read under the publication lock is the one planning trusts.
// a tombstone means the active generation may point at bytes that never
// reached disk. rebuild everything.
let forced_rebuild = crate::tombstone::present(&output_root);
let active = trace.time("plan: scope active view", || {
plan::load_active_generation(&output_root)
})?;
let active = if forced_rebuild {
tracing::warn!(
"tombstone present: previous activation may be incomplete; forcing full rebuild"
);
None
} else {
active
};
let global = GlobalInputs::of_config(&config);
let active_manifest = active.as_ref().map(|a| &a.manifest);
let hosted: BTreeSet<String> = index
.hosted_repos()
.iter()
.map(|did| did.as_str().to_string())
.collect();
let carried_scope: BTreeSet<String> = match (repo_hint, active_manifest) {
(Some(hint), Some(manifest)) => {
carried_candidate_dids(&output_root, manifest, &global, &hosted, Some(hint))
}
_ => BTreeSet::new(),
};
// the source locks must outlive the snapshot and every reader of the source objects.
// repo locks always come before the publication lock.
let captured_dids: Vec<String> = match repo_hint {
Some(hint) => hosted
.iter()
.filter(|did| did.as_str() == hint || !carried_scope.contains(did.as_str()))
.cloned()
.collect(),
None => hosted.iter().cloned().collect(),
};
let _source_locks = trace.time("plan: source lifetime locks", || {
crate::repolock::acquire_dids(
&output_root,
captured_dids.iter().map(String::as_str),
crate::repolock::LockMode::Lifetime,
)
})?;
tracing::debug!(
target: "knot_static::generate",
repos = _source_locks.len(),
keys = ?_source_locks.keys(),
"holding the captured repositories' source lifetime locks"
);
let _output_lock = crate::output::lock(&output_root)?;
// a half-written generation does not count as unchanged. the next run repairs it.
let mut active = trace.time("plan: load_active_generation", || {
plan::load_active_generation(&output_root)
})?;
if forced_rebuild {
active = None;
}
let t_snapshots = std::time::Instant::now();
let registry_state = crate::registry::load_state(&meta_repo)?;
let snapshots: Vec<crate::snapshot::RepoSnapshot> = match repo_hint {
Some(hint) => {
let mut captured = Vec::new();
let mut hint_found = false;
for did in index.hosted_repos().iter() {
if did.as_str() == hint {
hint_found = true;
captured.push(crate::snapshot::snapshot_repo(
&layout,
index,
registry_state.as_ref(),
did,
)?);
} else if !carried_scope.contains(did.as_str()) {
captured.push(crate::snapshot::snapshot_repo(
&layout,
index,
registry_state.as_ref(),
did,
)?);
}
}
anyhow::ensure!(
hint_found,
"repo hint {hint} does not resolve to a hosted repo"
);
captured
}
None => crate::snapshot::snapshot_all(&layout, index, registry_state.as_ref())?,
};
trace.record("plan: snapshot capture", t_snapshots);
// a removed repo must force a rebuild here, not pass as unchanged.
let t_outcomes = std::time::Instant::now();
let outcomes: Vec<RepoOutcome> = snapshots
.iter()
.map(
|snapshot| match plan::plan_repo(snapshot, &global, active.as_ref()) {
RepoOutcome::Unchanged { generation } => {
let sealed = active.as_ref().is_some_and(|active| {
active
.manifest
.repos
.iter()
.find(|entry| entry.repo_did == snapshot.repo_did.as_str())
.is_some_and(|entry| entry_release_sealed(&output_root, entry))
});
if sealed {
RepoOutcome::Unchanged { generation }
} else {
RepoOutcome::Rebuild
}
}
RepoOutcome::Rebuild => RepoOutcome::Rebuild,
},
)
.collect();
trace.record("plan: unchanged pass", t_outcomes);
let handles = crate::index_page::served_digest(
&crate::index_page::cached_handles(&output_root),
&output_root.join(crate::xrpc::social::AVATARS_DIR),
);
let answers_handles = active
.as_ref()
.is_some_and(|active| active.manifest.handles_digest.as_deref() == Some(handles.as_str()));
if answers_handles
&& unchanged_shortcut(
&snapshots,
&outcomes,
&carried_scope,
&hosted,
active.as_ref(),
)
{
let generation = match outcomes.first() {
Some(RepoOutcome::Unchanged { generation }) => generation.clone(),
_ => active
.as_ref()
.expect("checked above")
.manifest
.generation_id
.clone(),
};
if !crate::out::is_transport() {
eprintln!(
"unchanged, generation {} already active ({:.1?})",
short_id(&generation),
started.elapsed(),
);
}
return Ok(BuildOutcome::Unchanged {
generation_id: generation,
});
}
let carried: Vec<RepoEntry> = active
.as_ref()
.map(|active| {
active
.manifest
.repos
.iter()
.filter(|entry| carried_scope.contains(&entry.repo_did))
.cloned()
.collect()
})
.unwrap_or_default();
let t_build_plan = std::time::Instant::now();
let plans: Arc<Vec<RepoPlan>> = Arc::new(
snapshots
.iter()
.zip(outcomes)
.map(|(snapshot, outcome)| build_plan(snapshot, outcome, &global, active.as_ref()))
.collect(),
);
trace.record("plan: build_plan", t_build_plan);
let t_per_repo = std::time::Instant::now();
let per_repo: Arc<Vec<(RepoStore, Arc<Mutex<ReleaseBuild>>)>> = Arc::new(
plans
.iter()
.map(|plan| {
let repo_dir = repos_root.join(plan::repo_key(plan.snapshot.repo_did.as_str()));
let store = RepoStore::open(&repo_dir);
let release = ReleaseBuild::new(&repo_dir.join("releases"))?;
Ok((store, Arc::new(Mutex::new(release))))
})
.collect::<anyhow::Result<Vec<_>>>()?,
);
trace.record("plan: per_repo stores", t_per_repo);
let private = PrivateGeneration::new(&generations_root)?;
let build_dir = private.path().to_path_buf();
let git_root = build_dir.join("www/_git");
std::fs::create_dir_all(&git_root)?;
let planning_seconds = t_plan.elapsed().as_secs_f64();
if trace.enabled {
out!("[trace] planning total: {:.1}ms", planning_seconds * 1000.0);
}
// carried exports are mounted from their releases, never copied or rehashed here.
let export_config = Arc::clone(&config);
let export_plans = Arc::clone(&plans);
let export_repo_state = Arc::clone(&per_repo);
let export_worker =
WorkerJoin::spawn(move || -> anyhow::Result<(f64, HashMap<String, String>)> {
let t_export_start = std::time::Instant::now();
let results = export_plans
.par_iter()
.enumerate()
.filter(|(_, plan)| !plan.unchanged())
.map(|(index, plan)| {
let (store, staging) = &export_repo_state[index];
let prior = plan.prior_export.as_ref();
let fingerprint = git_export::export_fingerprint(
&plan.snapshot,
export_config.default_branch.as_str(),
prior,
);
let export_dir = staging.lock().unwrap().path().join("export");
git_export::materialize_export(
&export_config,
&plan.snapshot,
&export_dir,
&fingerprint,
Some(store),
prior,
)?;
let mut release = staging.lock().unwrap();
release.record_input("export", &fingerprint);
Tracer::new().time("release: finalize export", || {
release.finalize_tree("export")
})?;
Ok((plan.snapshot.repo_did.as_str().to_string(), fingerprint))
})
.collect::<anyhow::Result<HashMap<_, _>>>()?;
Ok((t_export_start.elapsed().as_secs_f64(), results))
});
// only fires with `KNOT_FAILPOINT=route_emission`, while the export worker runs.
crate::failpoint::check("route_emission")?;
let (git_export_seconds, exported) = export_worker.join()?;
// a release is visible only once complete.
// the manifest keeps these ids, which means nothing re-derives inputs after a crash.
let mut release_ids: BTreeMap<String, String> = BTreeMap::new();
{
let t_releases = std::time::Instant::now();
for (index, plan) in plans.iter().enumerate() {
let did = plan.snapshot.repo_did.as_str();
let releases_root = repos_root.join(plan::repo_key(did)).join("releases");
let release_id = if exported.contains_key(did) {
let (_store, staging) = &per_repo[index];
let mut build = staging.lock().unwrap();
let release_id = crate::release::release_id(did, build.inputs());
trace.time("release: publish", || {
build.publish(&releases_root, &release_id)
})?;
release_id
} else {
let (prior_id, _) = plan.prior_release.as_ref().with_context(|| {
format!("repo {did} is unchanged but the active generation has no release")
})?;
prior_id.clone()
};
mount_release_export(
&releases_root.join(&release_id),
&git_root.join(plan::repo_token(did)),
)?;
release_ids.insert(did.to_string(), release_id);
}
tracing::debug!(
target: "knot_static::build",
releases_seconds = t_releases.elapsed().as_secs_f64(),
"published repo releases"
);
}
for entry in &carried {
let release_id = entry.release_id.as_ref().with_context(|| {
format!(
"active generation has no release for carried repo {}; run a full generate to repair",
entry.repo_did
)
})?;
mount_release_export(
&repos_root
.join(plan::repo_key(&entry.repo_did))
.join("releases")
.join(release_id),
&git_root.join(plan::repo_token(&entry.repo_did)),
)?;
}
let mut entries: Vec<RepoEntry> = plans
.iter()
.map(|plan| {
let snapshot = &plan.snapshot;
let did = snapshot.repo_did.as_str();
let replies = git_export::fetch_replies(&git_root.join(plan::repo_token(did)))?;
Ok(RepoEntry {
repo_did: did.to_string(),
owner: snapshot.owner.as_str().to_string(),
rkey: snapshot.rkey.as_str().to_string(),
name: snapshot.name.as_ref().map(|n| n.as_str().to_string()),
head: snapshot.head_capture.resolved.clone(),
public_refs: snapshot
.public_refs
.iter()
.map(|r| (r.name.clone(), r.raw.clone()))
.collect(),
cob_refs: snapshot.cob_refs.clone(),
object_format: snapshot.object_format.capability().to_string(),
fingerprint: plan.identity.clone(),
git_export_hash: exported
.get(did)
.or_else(|| {
plan.prior_release
.as_ref()
.map(|(_, fingerprint)| fingerprint)
})
.cloned(),
release_id: release_ids.get(did).cloned(),
fetch_bases: replies.bases,
shallow_branches: replies.shallow_branches,
shallow_head: replies.shallow_head,
moved_wants: replies.moved_wants,
})
})
.collect::<anyhow::Result<_>>()?;
entries.extend(carried.iter().cloned());
let captured: BTreeSet<&str> = plans
.iter()
.map(|plan| plan.snapshot.repo_did.as_str())
.collect();
let atproto_sources: Vec<crate::atproto::RepoSource> = entries
.iter()
.map(|entry| {
let release_id = entry
.release_id
.as_deref()
.with_context(|| format!("no release for {}", entry.repo_did))?;
let did = RepoDid::new(entry.repo_did.as_str())?;
let repo_dir = repos_root.join(plan::repo_key(&entry.repo_did));
let store = RepoStore::open(&repo_dir);
let cobs = captured
.contains(entry.repo_did.as_str())
.then(|| {
let key = plan::repo_key(&entry.repo_did);
let lock = _source_locks.get(&key).with_context(|| {
format!("the source lifetime lock of {} is not held", entry.repo_did)
})?;
crate::atproto::cob_state_synced(
index,
&layout.open(&did)?,
&did,
&store,
&crate::datastore::Datastore::at(&output_root),
lock,
synced
.filter(|synced| synced.repo == did)
.map(|synced| &synced.tips),
)
})
.transpose()?;
Ok(crate::atproto::RepoSource {
did,
cobs,
served: RepoStore::served(&repo_dir),
verify: forced_rebuild,
owner: OwnerDid::new(entry.owner.as_str())?,
name: RepoName::new(entry.name.as_deref().unwrap_or(entry.rkey.as_str()))?,
export_dir: repos_root
.join(plan::repo_key(&entry.repo_did))
.join("releases")
.join(release_id)
.join("export"),
})
})
.collect::<anyhow::Result<_>>()?;
let atproto_repos = {
let signer = identity::load_knot_signer(&config)?;
let chain_root = output_root.join("atproto");
// keep the chain stores off this tokio worker, since they drive their own executor
std::thread::scope(|scope| {
scope
.spawn(|| {
atproto_sources
.iter()
.map(|source| crate::atproto::settle(&chain_root, &signer, source))
.collect::<anyhow::Result<Vec<_>>>()
})
.join()
.map_err(|_| anyhow::anyhow!("atproto settle panicked"))?
})?
};
crate::index_page::verify_unseen(&config, &crate::xrpc::social::accounts(&atproto_repos)).await;
let social = crate::xrpc::social::Social::of(&config);
crate::atproto::emit_with(&build_dir, &atproto_repos, &social)?;
crate::blobs::emit(&output_root, &build_dir, &atproto_repos, &captured)?;
let git_xrpc_repos: Vec<crate::git_xrpc::SealedRepo> = {
let shared_blobs = output_root.join(crate::git_xrpc::SHARED_BLOBS);
let inputs: Vec<crate::git_xrpc::RepoInput> = atproto_sources
.iter()
.zip(&entries)
.map(|(source, entry)| {
Ok(crate::git_xrpc::RepoInput {
did: source.did.clone(),
name: source.name.as_str().to_string(),
export_dir: source.export_dir.clone(),
served: source.served.clone(),
shared_blobs: shared_blobs.clone(),
patches: source.cobs.as_ref().map(crate::git_xrpc::patch_ranges),
refs_digest: serde_json::to_string(&(
&entry.public_refs,
&entry.head,
config.default_branch.as_str(),
))?,
verify: forced_rebuild,
})
})
.collect::<anyhow::Result<_>>()?;
inputs
.par_iter()
.map(crate::git_xrpc::seal)
.collect::<anyhow::Result<_>>()?
};
crate::git_xrpc::emit(&build_dir, &git_xrpc_repos)?;
let card_repos = crate::xrpc::social::card_repos(&atproto_repos, &social);
let card_users = crate::xrpc::social::card_users(&card_repos, &social);
let card_git: Vec<crate::og_card::GitRepo> = atproto_sources
.iter()
.zip(&git_xrpc_repos)
.map(|(source, sealed)| crate::og_card::GitRepo {
did: source.did.as_str(),
export: &source.export_dir,
served: source.served.clone(),
seal: &sealed.key,
})
.collect();
let cards = crate::og_card::plan(
&output_root,
config.hostname.as_str(),
card_repos,
card_users,
&card_git,
)?;
crate::og_card::emit(&output_root, &build_dir, cards).context("emit og cards")?;
// a blob write waits on the sweep. the source locks drop right after it
drop(_source_locks);
// chain heads are generation identity, since a ref flipped back still moves the rev
let atproto_heads: BTreeMap<&str, String> = atproto_repos
.iter()
.map(|repo| (repo.did.as_str(), repo.commit.to_string()))
.collect();
let atproto_entries: BTreeMap<&str, &str> = atproto_repos
.iter()
.map(|repo| (repo.did.as_str(), repo.entry_key.as_str()))
.collect();
let git_xrpc_entries: BTreeMap<&str, &str> = git_xrpc_repos
.iter()
.map(|repo| (repo.did.as_str(), repo.key.as_str()))
.collect();
let t_stage = std::time::Instant::now();
let rebuilt_repos: Vec<String> = plans
.iter()
.filter(|plan| plan.outcome == RepoOutcome::Rebuild)
.map(|plan| plan.snapshot.repo_did.as_str().to_string())
.collect();
let mut manifest = GenerationManifest {
generation_id: String::new(),
generator_profile: GENERATOR_PROFILE_VERSION.to_string(),
conf_template: crate::manifest::NGINX_TEMPLATE_VERSION.to_string(),
knot_did: config.knot_did.as_str().to_string(),
hostname: config.hostname.as_str().to_string(),
owner: config.owner.as_str().to_string(),
static_version: identity::STATIC_VERSION.to_string(),
built_by: identity::built_by().to_string(),
profile: serde_json::json!({
"atproto": atproto_heads,
"atproto_entries": atproto_entries,
"git_xrpc_entries": git_xrpc_entries,
}),
repos: entries,
rebuilt_repos,
global_input_digest: Some(global.config_digest.clone()),
handles_digest: Some(social.handles_digest.clone()),
stats: BuildStats::default(),
};
// the top-level exports are mounts. size what is inside them, not the link
let mut git_export_bytes = 0;
for entry in std::fs::read_dir(&git_root)? {
git_export_bytes += compute_dir_size(&entry?.path());
}
let pre_seal_seconds = started.elapsed().as_secs_f64();
manifest.stats = BuildStats {
repos: manifest.repos.len(),
public_refs: manifest
.repos
.iter()
.map(|entry| entry.public_refs.len())
.sum(),
git_export_bytes,
build_seconds: Some(pre_seal_seconds),
timings: Some(PhaseTimings {
planning_seconds: Some(planning_seconds),
git_export_seconds: Some(git_export_seconds),
}),
created_at_unix: crate::publish::unix_now(),
};
let did_document = identity::knot_did_document(&config)?;
write_www(
&build_dir,
".well-known/did.json",
&serde_json::to_vec_pretty(&did_document)?,
)?;
write_www(
&build_dir,
"_health.json",
&serde_json::to_vec(&identity::health_body())?,
)?;
write_www(
&build_dir,
"_describe.json",
&serde_json::to_vec(&identity::describe_body(&config))?,
)?;
if let Ok(metadata) = crate::oauth::client_metadata_document(&config) {
write_www(
&build_dir,
"oauth/client-metadata.json",
&serde_json::to_vec_pretty(&metadata)?,
)?;
}
manifest.finalize_and_write(&build_dir)?;
crate::index_page::emit(&config, &manifest, &build_dir)
.await
.context("emit front page")?;
let generation_dir = generations_root.join(&manifest.generation_id);
crate::nginx::emit_at(&config, &manifest, &build_dir, &generation_dir)
.context("emit nginx configuration")?;
let staging_seconds = t_stage.elapsed().as_secs_f64();
let t_pub = std::time::Instant::now();
let publication = private.publish(&generation_dir, &manifest)?;
// clear only after the publication, which re-materialized every export from git
crate::tombstone::clear(&output_root)?;
let publication_seconds = t_pub.elapsed().as_secs_f64();
let total_wall_seconds = started.elapsed().as_secs_f64();
tracing::info!(
target: "knot_static::build",
generation_id = %manifest.generation_id,
publication = ?publication,
planning_seconds = planning_seconds,
git_export_seconds = git_export_seconds,
staging_seconds = staging_seconds,
publication_seconds = publication_seconds,
total_wall_seconds = total_wall_seconds,
"completed generation build"
);
if !crate::out::is_transport() {
eprintln!(
"{} {} ({} repos, {:.1}s) at {}",
publication.verb(),
&manifest.generation_id[..16],
manifest.repos.len(),
total_wall_seconds,
generation_dir.display()
);
}
let pin = crate::gc::pin(&output_root, &manifest.generation_id)?;
Ok(BuildOutcome::Built {
generation_id: manifest.generation_id,
pin,
})
}