use std::path::Path;
use anyhow::Context as _;
use serde::{Deserialize, Serialize};
use crate::datastore::{Datastore, schema};
use crate::manifest::GenerationManifest;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ActivateOptions {
pub rollback: bool,
/// the hooks and gc run in a process of their own, so a write answers without them
pub detached: bool,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct ActivationTelemetry {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub config_test_seconds: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reload_seconds: Option<f64>,
#[serde(default, skip_serializing_if = "is_zero")]
pub nginx_config_bytes: u64,
#[serde(default, skip_serializing_if = "is_zero")]
pub fragments_total: usize,
#[serde(default, skip_serializing_if = "is_zero")]
pub fragments_reused: usize,
}
fn is_zero<T: Default + PartialEq>(v: &T) -> bool {
*v == T::default()
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BackendRecord {
pub revision: u64,
pub generation: String,
#[serde(default)]
pub rollback: bool,
#[serde(default)]
pub reconciled: bool,
pub activated_at_unix: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub telemetry: Option<ActivationTelemetry>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub superseded: Vec<Superseded>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Superseded {
pub generation: String,
pub at_unix: u64,
}
pub fn supersede(previous: Option<&BackendRecord>, next: &str, now: u64) -> Vec<Superseded> {
let mut superseded = previous.map_or_else(Vec::new, |record| record.superseded.clone());
if let Some(previous) = previous.filter(|record| record.generation != next) {
superseded.retain(|entry| entry.generation != previous.generation);
superseded.push(Superseded {
generation: previous.generation.clone(),
at_unix: now,
});
}
superseded.retain(|entry| entry.generation != next);
superseded
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StalePlan {
candidate: String,
reason: String,
}
impl std::fmt::Display for StalePlan {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"stale publication candidate {}: {}",
self.candidate, self.reason
)
}
}
impl std::error::Error for StalePlan {}
impl StalePlan {
pub fn injected() -> Self {
Self {
candidate: "injected".into(),
reason: "KNOT_FAILPOINT=merge_rebuild".into(),
}
}
}
pub fn check_activation_freshness(
candidate: &GenerationManifest,
active: &GenerationManifest,
hosted: &std::collections::BTreeSet<String>,
) -> Result<(), StalePlan> {
if candidate.generation_id == active.generation_id {
return Ok(());
}
let rebuilt = |repo_did: &str| candidate.rebuilt_repos.iter().any(|did| did == repo_did);
for entry in &active.repos {
if candidate.repos.iter().any(|e| e.repo_did == entry.repo_did)
|| !hosted.contains(&entry.repo_did)
{
continue;
}
return Err(StalePlan {
candidate: candidate.generation_id.clone(),
reason: format!(
"candidate drops repo {} which the active generation serves",
entry.repo_did
),
});
}
for entry in &candidate.repos {
if rebuilt(&entry.repo_did) {
continue;
}
let Some(active_entry) = active.repos.iter().find(|e| e.repo_did == entry.repo_did) else {
return Err(StalePlan {
candidate: candidate.generation_id.clone(),
reason: format!(
"candidate carries repo {} which the active generation no longer serves",
entry.repo_did
),
});
};
if active_entry != entry {
let bases = |entry: &crate::manifest::RepoEntry| format!("{:?}", entry.fetch_bases);
let (bases_before, bases_after) = (bases(active_entry), bases(entry));
let shallow =
|entry: &crate::manifest::RepoEntry| format!("{:?}", entry.shallow_branches);
let (shallow_before, shallow_after) = (shallow(active_entry), shallow(entry));
let fields: Vec<String> = [
("owner", active_entry.owner.as_str(), entry.owner.as_str()),
("rkey", active_entry.rkey.as_str(), entry.rkey.as_str()),
(
"name",
active_entry.name.as_deref().unwrap_or(""),
entry.name.as_deref().unwrap_or(""),
),
(
"fingerprint",
active_entry.fingerprint.as_str(),
entry.fingerprint.as_str(),
),
(
"git_export_hash",
active_entry.git_export_hash.as_deref().unwrap_or(""),
entry.git_export_hash.as_deref().unwrap_or(""),
),
(
"release_id",
active_entry.release_id.as_deref().unwrap_or(""),
entry.release_id.as_deref().unwrap_or(""),
),
("fetch_bases", &bases_before, &bases_after),
("shallow_branches", &shallow_before, &shallow_after),
(
"shallow_head",
active_entry.shallow_head.as_deref().unwrap_or(""),
entry.shallow_head.as_deref().unwrap_or(""),
),
]
.into_iter()
.filter(|(_, before, after)| before != after)
.map(|(name, before, after)| format!("{name}: {before} -> {after}"))
.collect();
return Err(StalePlan {
candidate: candidate.generation_id.clone(),
reason: format!(
"candidate carries stale state for repo {} (entry differs ({}); the active generation moved past the plan base)",
entry.repo_did,
fields.join(", ")
),
});
}
}
Ok(())
}
/// an absent record is rebuilt by `reconcile`, an unreadable one is not.
/// its superseded list is the only record of what retired workers may still read.
pub(crate) fn load_backend_record(output_root: &Path) -> anyhow::Result<Option<BackendRecord>> {
Datastore::at(output_root)
.record::<schema::Backend>(&())
.read()
.context(
"the backend publication record cannot be read, so which generations are still \
being served is unknown; move it aside only if you accept that a gc may remove \
a generation a retired nginx worker is reading",
)
}
pub fn write_backend_record(output_root: &Path, record: &BackendRecord) -> anyhow::Result<()> {
Datastore::at(output_root)
.record::<schema::Backend>(&())
.put(record)
}
pub fn current_generation(output_root: &Path) -> Option<String> {
Datastore::at(output_root).pointer().generation()
}
pub fn reconcile(output_root: &Path) -> anyhow::Result<Option<BackendRecord>> {
let now = unix_now();
let Some(current) = current_generation(output_root) else {
// without a current pointer, a leftover record would claim something is served.
Datastore::at(output_root)
.record::<schema::Backend>(&())
.remove()?;
return Ok(None);
};
let generation_dir = output_root.join("generations").join(¤t);
if !generation_dir
.join(crate::generate::COMPLETION_FILE)
.is_file()
{
return Ok(None);
}
let existing = load_backend_record(output_root)?;
let record = match existing {
Some(record) if record.generation == current => record,
previous => {
let revision = previous.as_ref().map_or(0, |record| record.revision);
let record = BackendRecord {
revision: revision + 1,
generation: current.clone(),
rollback: false,
reconciled: true,
activated_at_unix: now,
telemetry: None,
superseded: supersede(previous.as_ref(), ¤t, now),
};
write_backend_record(output_root, &record)?;
record
}
};
Ok(Some(record))
}
pub(crate) fn unix_now() -> u64 {
crate::atproto::unix_now().get() as u64
}
pub fn measure_fragment_reuse(
active_generation: Option<&Path>,
candidate_generation: &Path,
) -> (usize, usize) {
let candidate = fragment_digests(&candidate_generation.join("nginx/maps"));
let total = candidate.len();
let reused = match active_generation {
Some(active) => {
let prior: std::collections::HashSet<String> =
fragment_digests(&active.join("nginx/maps"))
.into_iter()
.collect();
candidate.iter().filter(|d| prior.contains(*d)).count()
}
None => 0,
};
(reused, total)
}
fn fragment_digests(maps_dir: &Path) -> Vec<String> {
let mut digests = Vec::new();
let Ok(entries) = std::fs::read_dir(maps_dir) else {
return digests;
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_file() {
if let Ok(digest) = crate::cas::Digest::of_file(&path) {
digests.push(digest.hex());
}
}
}
digests
}
#[cfg(test)]
mod tests {
use super::*;
use crate::manifest::{BuildStats, RepoEntry};
fn entry(did: &str, fingerprint: &str, release_id: Option<&str>) -> RepoEntry {
RepoEntry {
repo_did: did.to_string(),
owner: "did:plc:owner".to_string(),
rkey: "r".to_string(),
name: None,
head: Some("1111111111111111111111111111111111111111".to_string()),
public_refs: vec![(
"refs/heads/main".to_string(),
"1111111111111111111111111111111111111111".to_string(),
)],
cob_refs: Vec::new(),
object_format: "sha1".to_string(),
fingerprint: fingerprint.to_string(),
git_export_hash: Some("export-1".to_string()),
release_id: release_id.map(str::to_string),
fetch_bases: Vec::new(),
shallow_branches: Vec::new(),
shallow_head: None,
moved_wants: Vec::new(),
}
}
fn manifest(id: &str, did: &str, fingerprint: &str, rebuilt: &[&str]) -> GenerationManifest {
GenerationManifest {
generation_id: id.to_string(),
generator_profile: "test".to_string(),
conf_template: "test".to_string(),
knot_did: "did:web:knot.test".to_string(),
hostname: "knot.test".to_string(),
owner: "did:plc:owner".to_string(),
static_version: "test".to_string(),
built_by: String::new(),
profile: serde_json::json!({}),
repos: vec![entry(did, fingerprint, Some("release-1"))],
global_input_digest: Some("cfg".to_string()),
handles_digest: None,
rebuilt_repos: rebuilt.iter().map(|s| s.to_string()).collect(),
stats: BuildStats::default(),
}
}
#[test]
fn fresh_candidate_passes_and_stale_carried_state_is_refused() {
let hosted = std::collections::BTreeSet::from(["did:plc:a".to_string()]);
let active = manifest("gen-1", "did:plc:a", "fp-1", &["did:plc:a"]);
let candidate = manifest("gen-1", "did:plc:a", "fp-1", &["did:plc:a"]);
assert!(check_activation_freshness(&candidate, &active, &hosted).is_ok());
let drifted = manifest("gen-1", "did:plc:a", "fp-2", &[]);
assert!(check_activation_freshness(&drifted, &active, &hosted).is_ok());
let carried = manifest("gen-2", "did:plc:a", "fp-1", &[]);
assert!(check_activation_freshness(&carried, &active, &hosted).is_ok());
let moved_active = manifest("gen-2", "did:plc:a", "fp-9", &["did:plc:a"]);
let stale = manifest("gen-3", "did:plc:a", "fp-1", &[]);
let error = check_activation_freshness(&stale, &moved_active, &hosted).unwrap_err();
assert!(error.reason.contains("stale state for repo"));
assert!(error.reason.contains("entry differs"));
let mut dropped = moved_active.clone();
dropped.generation_id = "gen-4".to_string();
dropped.repos.clear();
let error = check_activation_freshness(&dropped, &moved_active, &hosted).unwrap_err();
assert!(error.reason.contains("drops repo"));
let deleted = Default::default();
assert!(
check_activation_freshness(&dropped, &moved_active, &deleted).is_ok(),
"a repo the knot no longer hosts drops out"
);
}
#[test]
fn stale_carried_state_names_the_fetch_reply_fields() {
let hosted = std::collections::BTreeSet::from(["did:plc:a".to_string()]);
let active = manifest("gen-1", "did:plc:a", "fp-1", &[]);
let mut stale = manifest("gen-2", "did:plc:a", "fp-1", &[]);
stale.repos[0].fetch_bases = vec![vec!["1".repeat(40)]];
stale.repos[0].shallow_branches = vec!["3".repeat(40)];
stale.repos[0].shallow_head = Some("2".repeat(40));
let error = check_activation_freshness(&stale, &active, &hosted).unwrap_err();
for field in [
"fetch_bases: [] -> ",
"shallow_branches: [] -> [\"3333333333333333333333333333333333333333\"]",
"shallow_head: -> 2",
] {
assert!(error.reason.contains(field), "{}", error.reason);
}
}
#[test]
fn superseding_appends_the_previous_generation_in_activation_order() {
let record = |generation: &str, superseded: Vec<Superseded>| BackendRecord {
revision: 1,
generation: generation.to_string(),
rollback: false,
reconciled: false,
activated_at_unix: 0,
telemetry: None,
superseded,
};
let entry = |generation: &str, at_unix| Superseded {
generation: generation.to_string(),
at_unix,
};
let now = 1_000;
assert!(supersede(None, "g1", now).is_empty());
assert_eq!(
supersede(Some(&record("g1", Vec::new())), "g2", now),
[entry("g1", now)]
);
assert!(
supersede(Some(&record("g1", Vec::new())), "g1", now).is_empty(),
"reactivating the current generation supersedes nothing"
);
let earlier = vec![entry("g0", 1), entry("g2", 5), entry("g3", 9)];
assert_eq!(
supersede(Some(&record("g1", earlier)), "g2", now),
[entry("g0", 1), entry("g3", 9), entry("g1", now)],
"the newly current generation leaves the history, the rest keeps its order"
);
}
#[test]
fn records_round_trip_and_revision_bumps() {
let root = tempfile::tempdir().unwrap();
let output = root.path().join("output");
std::fs::create_dir_all(&output).unwrap();
assert!(load_backend_record(&output).unwrap().is_none());
let record = BackendRecord {
revision: 3,
generation: "gen-1".to_string(),
rollback: true,
reconciled: false,
activated_at_unix: 42,
telemetry: Some(ActivationTelemetry {
config_test_seconds: Some(0.5),
reload_seconds: Some(1.25),
nginx_config_bytes: 1024,
fragments_total: 20,
fragments_reused: 18,
}),
superseded: vec![Superseded {
generation: "gen-0".to_string(),
at_unix: 41,
}],
};
write_backend_record(&output, &record).unwrap();
let loaded = load_backend_record(&output).unwrap().unwrap();
assert_eq!(loaded, record);
write_backend_record(&output, &record).unwrap();
let versioned = output.join("active.json");
let mut stored: serde_json::Value =
serde_json::from_slice(&std::fs::read(&versioned).unwrap()).unwrap();
assert_eq!(
stored["version"], 1,
"the version is the record's, not a caller's"
);
stored["version"] = 2.into();
std::fs::write(&versioned, serde_json::to_vec(&stored).unwrap()).unwrap();
assert!(
load_backend_record(&output).is_err(),
"a record this release can't read is never rebuilt from"
);
let mut unversioned = stored.clone();
unversioned.as_object_mut().unwrap().remove("version");
std::fs::write(&versioned, serde_json::to_vec(&unversioned).unwrap()).unwrap();
assert!(
load_backend_record(&output).is_err(),
"a record with no version is not assumed to be this one"
);
}
#[test]
fn a_backend_record_that_does_not_read_stops_reconcile_and_gc() {
let root = tempfile::tempdir().unwrap();
let output = root.path().join("output");
std::fs::create_dir_all(output.join("generations/gen-1")).unwrap();
std::fs::write(
output.join("generations/gen-1/manifest.json"),
serde_json::json!({ "stats": { "created_at_unix": 1 } }).to_string(),
)
.unwrap();
std::fs::write(
output
.join("generations/gen-1")
.join(crate::generate::COMPLETION_FILE),
b"{}",
)
.unwrap();
std::os::unix::fs::symlink("generations/gen-1", output.join("current")).unwrap();
std::fs::write(output.join("active.json"), b"{broken").unwrap();
let error = load_backend_record(&output).expect_err("junk is not an absent record");
assert!(
format!("{error:#}").contains("active.json")
|| format!("{error:#}").contains("backend publication record")
);
assert!(
reconcile(&output).is_err(),
"reconcile must not overwrite it"
);
assert!(
crate::gc::collect(&output, 1).is_err(),
"a gc that can't read the history must not remove generations"
);
std::fs::remove_file(output.join("active.json")).unwrap();
assert!(
load_backend_record(&output).unwrap().is_none(),
"absent is the one case that is rebuilt"
);
}
}