use std::path::Path;
use std::sync::Arc;
use std::time::Instant;
use anyhow::Context as _;
use crate::config::ValidatedConfig;
use crate::datastore::{Datastore, Durability};
use crate::manifest::{GENERATOR_PROFILE_VERSION, GenerationManifest};
use crate::nginx::conf::Block;
use crate::publish::{self, ActivateOptions, ActivationTelemetry, BackendRecord};
fn write_active_conf(active_conf: &Path, bytes: &[u8]) -> std::io::Result<()> {
crate::datastore::write::publish(active_conf, bytes, Durability::Fsync)
}
fn rollback_active_conf(previous: Option<&[u8]>, active_conf: &Path) {
if let Some(bytes) = previous {
let _ = write_active_conf(active_conf, bytes);
} else {
let _ = std::fs::remove_file(active_conf);
let _ = crate::datastore::write::sync_directory(
active_conf.parent().unwrap_or(active_conf),
Durability::Fsync,
);
}
}
fn carries_git_maps(
config: &ValidatedConfig,
previous: Option<&str>,
generation_dir: &Path,
) -> bool {
let Some(previous) = previous else {
return false;
};
let maps =
|dir: &Path| std::fs::read(dir.join("nginx/maps").join(crate::git_xrpc::MAPS_FILE)).ok();
let previous_dir = config.output_root.join("generations").join(previous);
generation_dir
.join("nginx")
.join(crate::nginx::CHECK_FILE)
.is_file()
&& maps(&previous_dir).is_some_and(|before| maps(generation_dir) == Some(before))
}
fn abort_activation(
config: &ValidatedConfig,
previous_active: Option<&[u8]>,
active_conf: &Path,
) -> anyhow::Result<()> {
rollback_active_conf(previous_active, active_conf);
crate::tombstone::clear(&config.output_root)
}
#[derive(Debug, Clone)]
pub(crate) struct ActivationResult {
pub generation: String,
pub changed_repos: Vec<String>,
}
pub async fn activate(
config: Arc<ValidatedConfig>,
generation: &str,
) -> anyhow::Result<ActivationResult> {
let options = ActivateOptions {
detached: true,
..ActivateOptions::default()
};
activate_with(config, generation, options).await
}
fn resolve_generation(output_root: &Path, given: &str) -> anyhow::Result<String> {
let full = given.len() == 64;
if full || given.is_empty() || !given.bytes().all(|b| b.is_ascii_hexdigit()) {
return Ok(given.to_string());
}
let mut matches: Vec<String> = std::fs::read_dir(output_root.join("generations"))
.into_iter()
.flatten()
.flatten()
.filter_map(|entry| entry.file_name().into_string().ok())
.filter(|name| name.starts_with(given))
.collect();
matches.sort();
match matches.as_slice() {
[only] => Ok(only.clone()),
[] => anyhow::bail!("no generation starts with {given}"),
many => anyhow::bail!(
"{given} matches {} generations: {}",
many.len(),
many.iter()
.map(|id| &id[..16])
.collect::<Vec<_>>()
.join(", ")
),
}
}
async fn activate_with(
config: Arc<ValidatedConfig>,
generation: &str,
options: ActivateOptions,
) -> anyhow::Result<ActivationResult> {
let _output_lock = crate::output::lock(&config.output_root)?;
activate_locked(config, generation, options)
}
fn activate_locked(
config: Arc<ValidatedConfig>,
generation: &str,
options: ActivateOptions,
) -> anyhow::Result<ActivationResult> {
let t_activation = Instant::now();
if crate::tombstone::present(&config.output_root) {
anyhow::bail!(
"activation tombstone present in {}: the previous activation may be incomplete; run `knot-static generate` first",
config.output_root.display()
);
}
let previous_record = publish::reconcile(&config.output_root)?;
anyhow::ensure!(
generation.len() == 64 && generation.bytes().all(|b| b.is_ascii_hexdigit()),
"generation must be a 64-character hexadecimal id"
);
let generation_dir = config.output_root.join("generations").join(generation);
anyhow::ensure!(
generation_dir.exists(),
"generation directory does not exist: {}",
generation_dir.display()
);
let manifest = GenerationManifest::load(&generation_dir)
.with_context(|| format!("load manifest from {}", generation_dir.display()))?;
anyhow::ensure!(
manifest.generation_id == generation,
"manifest generation_id ({}) does not match requested generation ({generation})",
manifest.generation_id
);
let computed_id = manifest.compute_id();
anyhow::ensure!(
computed_id == manifest.generation_id,
"manifest generation_id ({}) does not match computed semantic hash ({computed_id})",
manifest.generation_id
);
anyhow::ensure!(
manifest.generator_profile == GENERATOR_PROFILE_VERSION,
"generation {generation} was built by {}, not {GENERATOR_PROFILE_VERSION}; run `knot-static generate`",
manifest.generator_profile
);
for repo in &manifest.repos {
let release_id = repo
.release_id
.as_ref()
.with_context(|| format!("missing release id for {}", repo.repo_did))?;
let expected_release = config
.output_root
.join("repos")
.join(crate::plan::repo_key(&repo.repo_did))
.join("releases")
.join(release_id);
let release = crate::release::PublishedRelease::open(&expected_release)?;
release.validate()?;
let expected_export = std::fs::canonicalize(expected_release.join("export"))?;
let mount = generation_dir
.join("www/_git")
.join(crate::plan::repo_token(&repo.repo_did));
anyhow::ensure!(
std::fs::canonicalize(&mount).ok().as_ref() == Some(&expected_export),
"git mount {} does not resolve to sealed release {}",
mount.display(),
expected_export.display()
);
}
let nginx_conf = generation_dir.join("nginx/knot-static.conf");
anyhow::ensure!(
nginx_conf.is_file(),
"generation is missing its nginx configuration; regenerate it"
);
if !options.rollback {
if let Some(active) = crate::plan::load_active_generation(&config.output_root)? {
let hosted = crate::registry::hosted(&config)?
.iter()
.map(|did| did.as_str().to_string())
.collect();
publish::check_activation_freshness(&manifest, &active.manifest, &hosted)
.map_err(anyhow::Error::new)?;
}
}
let active_conf = config.output_root.join("active.conf");
let previous_active = active_conf
.exists()
.then(|| std::fs::read(&active_conf).context("read existing active.conf"))
.transpose()?;
let previous_generation_name = publish::current_generation(&config.output_root);
let previous_manifest = previous_generation_name.as_ref().and_then(|id| {
GenerationManifest::load(&config.output_root.join("generations").join(id)).ok()
});
let nginx_conf_canon =
std::fs::canonicalize(&nginx_conf).unwrap_or_else(|_| nginx_conf.clone());
let pid_path = config.output_root.join("runtime/nginx.pid");
let new_conf_content = Block::file()
.directive("pid", [pid_path.display()])
.directive("include", [nginx_conf_canon.display()])
.to_string();
crate::tombstone::write(&config.output_root, generation)?;
// sync before nginx is pointed at the generation, or a power cut leaves files with no bytes.
crate::datastore::write::sync_filesystem(&config.output_root)
.with_context(|| format!("sync {} before activating", config.output_root.display()))?;
write_active_conf(&active_conf, new_conf_content.as_bytes())
.with_context(|| format!("write active.conf: {}", active_conf.display()))?;
// nginx refuses a bad conf on reload and keeps serving the old generation, so -t checks it
// first. git read maps byte-identical to the active generation's passed it then, and their
// fragments are immutable, so only the rest is checked again.
let t_test = Instant::now();
let carried = !options.rollback
&& carries_git_maps(
&config,
previous_generation_name.as_deref(),
&generation_dir,
);
let checked = if carried {
let check_conf = config.output_root.join("runtime/nginx-check.conf");
let check = Block::file()
.directive("pid", [pid_path.display()])
.directive(
"include",
[generation_dir
.join("nginx")
.join(crate::nginx::CHECK_FILE)
.display()],
)
.to_string();
crate::cas::atomic_write(&check_conf, check.as_bytes())?;
check_conf
} else {
active_conf.clone()
};
let test_res = std::process::Command::new(&config.activate.nginx_binary)
.arg("-t")
.arg("-c")
.arg(&checked)
.output();
match test_res {
Ok(output) if !output.status.success() => {
abort_activation(&config, previous_active.as_deref(), &active_conf)?;
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
anyhow::bail!(
"nginx configuration test failed (-t) [binary: {}, conf_path: {}, conf: {}]:\nstderr: {}\nstdout: {}",
config.activate.nginx_binary,
config.config_path.display(),
active_conf.display(),
stderr.trim(),
stdout.trim()
);
}
Err(err) => {
abort_activation(&config, previous_active.as_deref(), &active_conf)?;
anyhow::bail!(
"failed to execute nginx binary '{}': {err}",
config.activate.nginx_binary
);
}
_ => {}
}
let config_test_seconds = Some(t_test.elapsed().as_secs_f64());
crate::provision::chown_tree_to_dir_owner(
&config.output_root.join("runtime"),
&config.scan_path,
);
crate::provision::chown_tree_to_dir_owner(&generation_dir, &config.scan_path);
crate::provision::chown_tree_to_dir_owner(&active_conf, &config.scan_path);
let mut reload_seconds: Option<f64> = None;
if let Some(reload_cmd) = &config.activate.reload_command {
let workers_before = crate::gc::active_workers(&config.output_root);
let t_reload = Instant::now();
let reload_res = std::process::Command::new("sh")
.arg("-c")
.arg(reload_cmd)
.output();
match reload_res {
Ok(output) if !output.status.success() => {
abort_activation(&config, previous_active.as_deref(), &active_conf)?;
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
anyhow::bail!(
"nginx reload command failed:\nstderr: {}\nstdout: {}",
stderr.trim(),
stdout.trim()
);
}
Err(err) => {
abort_activation(&config, previous_active.as_deref(), &active_conf)?;
anyhow::bail!("failed to execute reload command '{reload_cmd}': {err}");
}
_ => {}
}
crate::gc::wait_for_reload(&config.output_root, &workers_before);
reload_seconds = Some(t_reload.elapsed().as_secs_f64());
}
Datastore::at(&config.output_root)
.pointer()
.point_at(generation)?;
let activated_at_unix = publish::unix_now();
let previous_generation_dir =
previous_generation_name.map(|name| config.output_root.join("generations").join(name));
let (fragments_reused, fragments_total) =
publish::measure_fragment_reuse(previous_generation_dir.as_deref(), &generation_dir);
let telemetry = ActivationTelemetry {
config_test_seconds,
reload_seconds,
nginx_config_bytes: std::fs::metadata(&nginx_conf).map(|m| m.len()).unwrap_or(0),
fragments_total,
fragments_reused,
};
let revision = previous_record.as_ref().map_or(0, |record| record.revision) + 1;
publish::write_backend_record(
&config.output_root,
&BackendRecord {
revision,
generation: manifest.generation_id.clone(),
rollback: options.rollback,
reconciled: false,
activated_at_unix,
telemetry: Some(telemetry),
superseded: publish::supersede(
previous_record.as_ref(),
&manifest.generation_id,
activated_at_unix,
),
},
)?;
let firehose = crate::firehose::append(&config.output_root, crate::firehose::heads(&manifest));
let previous_fingerprints: std::collections::HashMap<&str, &str> = previous_manifest
.as_ref()
.map(|m| {
m.repos
.iter()
.map(|repo| (repo.repo_did.as_str(), repo.fingerprint.as_str()))
.collect()
})
.unwrap_or_default();
let changed_repos: Vec<String> = manifest
.repos
.iter()
.filter(|repo| {
previous_fingerprints.get(repo.repo_did.as_str()) != Some(&repo.fingerprint.as_str())
})
.map(|repo| repo.repo_did.clone())
.collect();
let mut summary = format!(
"activated generation {} revision {revision} (repos: {}",
generation,
manifest.repos.len(),
);
if options.rollback {
summary.push_str(", rollback");
}
if let Some(seconds) = reload_seconds {
summary.push_str(&format!(", reload {seconds:.3}s"));
}
summary.push_str(&format!(
", fragments {}/{} reused, {:.1?})",
fragments_reused,
fragments_total,
t_activation.elapsed(),
));
crate::tombstone::clear(&config.output_root)?;
let collected = if options.detached {
let at = activated_at_unix.to_string();
let mut args = vec!["after-activation", generation, "--at", &at];
if let Some(previous) = &previous_manifest {
args.extend(["--previous", &previous.generation_id]);
}
if let Err(error) = crate::git_ssh::spawn_detached(&config, args) {
tracing::warn!("hooks and gc skipped: {error:#}");
}
Ok(crate::gc::Removed::default())
} else {
crate::generate::hooks::after_activation(
&config,
&crate::generate::hooks::Activation {
generation_dir: &generation_dir,
manifest: &manifest,
previous: previous_manifest.as_ref(),
rollback: options.rollback,
activated_at_unix,
},
);
crate::gc::collect(&config.output_root, config.activate.keep_generations)
};
if !crate::out::is_transport() {
match firehose {
Ok(0) => {}
Ok(appended) => eprintln!("firehose: {appended} frames"),
Err(error) => eprintln!("firehose append skipped: {error:#}"),
}
eprintln!("{summary}");
match collected {
Ok(removed) if !removed.is_empty() => eprintln!("gc: {removed}"),
Ok(_) => {}
Err(error) => eprintln!("gc skipped: {error:#}"),
}
}
Ok(ActivationResult {
generation: generation.to_string(),
changed_repos,
})
}
pub async fn post_activate(
config: Arc<ValidatedConfig>,
changed_repos: &[String],
declarations: bool,
) {
crate::crawl::announce(&config, changed_repos).await;
if declarations && !changed_repos.is_empty() {
crate::pds::publish_all(config.clone(), changed_repos).await;
}
}
/// a newer activation starts its own hook and collects after it, so a superseded one does neither
pub fn after_activation(
config: &ValidatedConfig,
generation: &str,
previous: Option<&str>,
activated_at_unix: u64,
) -> anyhow::Result<()> {
let superseded =
|| publish::current_generation(&config.output_root).as_deref() != Some(generation);
if superseded() {
return Ok(());
}
// a burst of writes shares one gc: the one waiting collects for everyone activated before it
let gc_slot = crate::gc::queue(&config.output_root)?;
if gc_slot.is_none() && config.activate.hooks.is_empty() {
return Ok(());
}
let generations = config.output_root.join("generations");
let hook = {
let _output_lock = crate::output::lock(&config.output_root)?;
let collect = gc_slot.is_some();
drop(gc_slot);
let hook = if !config.activate.hooks.is_empty() && !superseded() {
let generation_dir = generations.join(generation);
let manifest = GenerationManifest::load(&generation_dir)?;
let previous = previous
.and_then(|previous| GenerationManifest::load(&generations.join(previous)).ok());
let activation = crate::generate::hooks::Activation {
generation_dir: &generation_dir,
manifest: &manifest,
previous: previous.as_ref(),
rollback: false,
activated_at_unix,
};
crate::generate::hooks::write_hook(config, &activation)
.inspect_err(|error| tracing::warn!("hooks skipped: {error:#}"))
.ok()
} else {
None
};
if collect {
let removed =
crate::gc::collect(&config.output_root, config.activate.keep_generations)?;
if !removed.is_empty() {
eprintln!("gc: {removed}");
}
}
hook
};
if let Some(path) = hook {
crate::generate::hooks::start_all(config, generation, &path, true);
}
Ok(())
}
pub async fn run(
config: Arc<ValidatedConfig>,
generation: &str,
rollback: bool,
) -> anyhow::Result<()> {
let generation = resolve_generation(&config.output_root, generation)?;
let result = activate_with(
config.clone(),
&generation,
ActivateOptions {
rollback,
..ActivateOptions::default()
},
)
.await?;
post_activate(config, &result.changed_repos, true).await;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::datastore::write::syncs;
#[test]
fn active_conf_is_written_and_restored_durably() {
let root = tempfile::tempdir().unwrap();
let active_conf = root.path().join("active.conf");
let before = syncs();
write_active_conf(&active_conf, b"include new;\n").unwrap();
assert_eq!(syncs(), before + 2, "the file and its directory");
assert_eq!(std::fs::read(&active_conf).unwrap(), b"include new;\n");
let before = syncs();
rollback_active_conf(Some(b"include previous;\n"), &active_conf);
assert_eq!(syncs(), before + 2, "the restored conf is durable too");
assert_eq!(std::fs::read(&active_conf).unwrap(), b"include previous;\n");
let before = syncs();
rollback_active_conf(None, &active_conf);
assert_eq!(syncs(), before + 1, "the removal is durable as well");
assert!(!active_conf.exists());
}
}