an activation writes no record per hosted repo
Signed-off-by: dawn <90008@klbr.net>
20 unmodified lines2122232425262526272836 unmodified lines65666768717273747576777879808169707172738788899091929374757677314 unmodified linesExpand all3923933943954164174184194203963973983994004014264274024034044054064074084094354104114374384124134144154434444454164174184195 unmodified lines42542642742845946046146246346446546646746846947047147247347442943043143233 unmodified lines46646746846951651751851952052152252352452552652752847047147247370 unmodified lines544545546547607608609548549550551447 unmodified linesExpand all20 unmodified lines#[derive(Deserialize, Debug, Clone)]struct RepoEntry { repo_did: String, #[serde(default)] fingerprint: String, #[serde(default)] release_id: Option<String>, #[serde(default)] public_refs: Vec<(String, String)>,}36 unmodified lines #[serde(default)] fragments_reused: usize,}
#[derive(Deserialize, Debug, Clone, PartialEq)]struct RepoRecord { version: u32, repo_did: String, revision: u64, generation: String, #[serde(default)] release_id: Option<String>, input_identity: String,}
fn backend_record(output: &Path) -> BackendRecord { let bytes = std::fs::read(output.join("active.json")).expect("backend publication record"); serde_json::from_slice(&bytes).expect("parse backend record")}
fn repo_record(output: &Path, repo_did: &str) -> RepoRecord { let key = repo_key(repo_did); let bytes = std::fs::read(output.join("repos").join(&key).join("active.json")) .unwrap_or_else(|e| panic!("repo record for {repo_did}: {e}")); serde_json::from_slice(&bytes).expect("parse repo record")}
fn knot_static_raw( config: &Path, args: impl IntoIterator<Item = impl AsRef<std::ffi::OsStr>>,) -> Output {314 unmodified linesExpand all assert!(telemetry.fragments_total > 0); assert_eq!(telemetry.fragments_reused, 0);
let manifest = Manifest::load(&gen0); for (seed, did) in &env.dids { let record = repo_record(&env.output, did); assert_eq!(record.version, 1); assert_eq!(record.revision, 1); assert_eq!(record.generation, gen0_id); for did in env.dids.values() { let entry = manifest .repos .iter() .find(|e| &e.repo_did == did) .expect("manifest entry"); assert_eq!(record.input_identity, entry.fingerprint); let release_id = record.release_id.expect("release id recorded"); let release_id = entry.release_id.clone().expect("release id recorded"); assert!( env.output .join("repos") .join(repo_key(did)) .join("releases") .join(&release_id) .is_dir(), "record must name a published release directory" "the manifest must name a published release directory" ); assert_eq!(entry.release_id.as_deref(), Some(release_id.as_str())); let _ = seed; }
activate_ok(&env, &gen0_id); assert_eq!(backend_record(&env.output).revision, 2); for did in env.dids.values() { assert_eq!(repo_record(&env.output, did).revision, 2); }}
#[test]fn failed_activation_leaves_active_generation_byte_identical_and_servable() {5 unmodified lines
let before_files = snapshot_tree(&gen0); let before_conf = std::fs::read(env.output.join("active.conf")).unwrap(); let before_backend = std::fs::read(env.output.join("active.json")).unwrap(); let before_repo_records: BTreeMap<String, Vec<u8>> = env .dids .values() .map(|did| { ( did.clone(), std::fs::read( env.output .join("repos") .join(repo_key(did)) .join("active.json"), ) .unwrap(), ) }) .collect();
push_commit(&env, "alpha", "failed-activation.txt"); knot_static(&env.config, ["generate"]); let gen1 = generations(&env.output).pop().unwrap();33 unmodified lines before_backend, std::fs::read(env.output.join("active.json")).unwrap(), "backend record must be untouched by the failed activation" ); for (did, bytes) in &before_repo_records { assert_eq!( &std::fs::read( env.output .join("repos") .join(repo_key(did)) .join("active.json") ) .unwrap(), bytes, "repo record for {did} must be untouched" ); }
activate_ok(&env, &gen0_id); assert_eq!( current_target(&env.output).as_deref(),70 unmodified lines assert_eq!( current_target(&env.output).as_deref(), Some(gen2_id.as_str()) ); for did in env.dids.values() { assert!(repo_record(&env.output, did).revision >= 2); }}
#[test]fn concurrent_repo_updates_never_lose_one_another() {447 unmodified linesExpand all14 unmodified lines15161718192021222395 unmodified lines14 unmodified lines let encodes_itself = crate::plan::repo_key(&decode(name)?) == name; encodes_itself.then(|| Self(name.to_string())) }
#[cfg(test)] pub(crate) fn parse_did(did: &str) -> anyhow::Result<Self> { Ok(Self::of(&RepoDid::new(did)?)) }
95 unmodified lines33 unmodified lines343536373839404142434445464748495051525354555657585960616238394041190 unmodified linesExpand all33 unmodified lines}
impl Put for Backend {}
pub(crate) struct RepoPublication;
impl Record for RepoPublication { type Area = OutputRoot; const NAME: &'static str = "repo publication record"; const VERSION: u32 = 1; const DURABILITY: Durability = Durability::Fsync; type Key = RepoKey; type Value = crate::publish::RepoRecord;
fn relative(key: &Self::Key) -> PathBuf { key.dir().join("active.json") }
fn encode(value: &Self::Value) -> anyhow::Result<Vec<u8>> { encode_json(Self::VERSION, value) }
fn decode(bytes: &[u8]) -> anyhow::Result<Self::Value> { decode_json(Self::VERSION, Self::NAME, bytes) }}
impl Put for RepoPublication {}
pub(crate) struct SpindleOutbox;
impl Record for SpindleOutbox { type Area = OutputRoot;190 unmodified linesExpand all3 unmodified lines456788910121112131415260 unmodified linesExpand all276277278279281280283284285286287288289290291292293294295296297298299300281282283284206 unmodified linesExpand all3 unmodified lines
use anyhow::Context as _;
use crate::config::ValidatedConfig;use crate::datastore::keys::RepoKey;use crate::datastore::{Datastore, Durability};use crate::manifest::{GENERATOR_PROFILE_VERSION, GenerationManifest};use crate::nginx::conf::Block;use crate::publish::{self, ActivateOptions, ActivationTelemetry, BackendRecord, RepoRecord};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)}260 unmodified linesExpand all Datastore::at(&config.output_root) .pointer() .point_at(generation)?;
// rollbacks still advance the revision counter let activated_at_unix = publish::unix_now(); for entry in &manifest.repos { let repo_key = RepoKey::parse_did(entry.repo_did.as_str())?; let previous_revision = publish::load_repo_record(&config.output_root, &repo_key)? .map_or(0, |record| record.revision); publish::write_repo_record( &config.output_root, &repo_key, &RepoRecord { repo_did: entry.repo_did.clone(), revision: previous_revision + 1, generation: manifest.generation_id.clone(), release_id: entry.release_id.clone(), input_identity: entry.fingerprint.clone(), activated_at_unix, }, ) .with_context(|| format!("record activation for {}", entry.repo_did))?; }
let previous_generation_dir = previous_generation_name.map(|name| config.output_root.join("generations").join(name)); let (fragments_reused, fragments_total) =206 unmodified linesExpand all1 unmodified line23456678954 unmodified lines64656667697071727374757677787968697071130 unmodified linesExpand all20220320420521821922022122222322422522622722822923023123223323423523623723820620720820932 unmodified lines242243244245279246247248283284285286287288289290291292293294295296297298299300301302303304305306307308249250251252189 unmodified linesExpand all44244344444550650750850951051151251351451551651751851952052152244644744844960 unmodified lines1 unmodified line
use anyhow::Context as _;use serde::{Deserialize, Serialize};
use crate::datastore::keys::RepoKey;use crate::datastore::{Datastore, schema};use crate::manifest::GenerationManifest;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]54 unmodified lines superseded.retain(|entry| entry.generation != next); superseded}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]pub struct RepoRecord { pub repo_did: String, pub revision: u64, pub generation: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub release_id: Option<String>, pub input_identity: String, pub activated_at_unix: u64,}
#[derive(Debug, Clone, PartialEq, Eq)]pub struct StalePlan { candidate: String, reason: String,130 unmodified linesExpand all .record::<schema::Backend>(&()) .put(record)}
/// a rebuilt repo record could reuse a revision, since no generation carries it.pub fn load_repo_record( output_root: &Path, repo_key: &RepoKey,) -> anyhow::Result<Option<RepoRecord>> { Datastore::at(output_root) .record::<schema::RepoPublication>(repo_key) .read() .with_context(|| format!("the publication record of {repo_key} cannot be read"))}
pub fn write_repo_record( output_root: &Path, repo_key: &RepoKey, record: &RepoRecord,) -> anyhow::Result<()> { Datastore::at(output_root) .record::<schema::RepoPublication>(repo_key) .put(record)}
pub fn current_generation(output_root: &Path) -> Option<String> { Datastore::at(output_root).pointer().generation()}
32 unmodified lines record } };
reconcile_repo_records(output_root, &generation_dir)?; Ok(Some(record))}
/// a crash between the reference swap and the record writes leaves them behind.fn reconcile_repo_records(output_root: &Path, generation_dir: &Path) -> anyhow::Result<()> { let manifest = GenerationManifest::load(generation_dir) .with_context(|| format!("load active generation manifest in {generation_dir:?}"))?; for entry in &manifest.repos { let key = RepoKey::parse_did(&entry.repo_did)?; let existing = load_repo_record(output_root, &key)?; let up_to_date = existing .as_ref() .is_some_and(|record| record.generation == manifest.generation_id); if up_to_date { continue; } let record = RepoRecord { repo_did: entry.repo_did.clone(), revision: existing.as_ref().map_or(0, |record| record.revision) + 1, generation: manifest.generation_id.clone(), release_id: entry.release_id.clone(), input_identity: entry.fingerprint.clone(), activated_at_unix: unix_now(), }; write_repo_record(output_root, &key, &record)?; } Ok(())}
pub(crate) fn unix_now() -> u64 { crate::atproto::unix_now().get() as u64}
189 unmodified linesExpand all write_backend_record(&output, &record).unwrap(); let loaded = load_backend_record(&output).unwrap().unwrap(); assert_eq!(loaded, record);
let repo_record = RepoRecord { repo_did: "did:plc:a".to_string(), revision: 7, generation: "gen-1".to_string(), release_id: Some("release-1".to_string()), input_identity: "fp-1".to_string(), activated_at_unix: 42, }; let key = RepoKey::parse_did("did:plc:a").unwrap(); write_repo_record(&output, &key, &repo_record).unwrap(); assert_eq!( load_repo_record(&output, &key).unwrap().unwrap(), repo_record ); let other = RepoKey::parse_did("did:plc:b").unwrap(); assert!(load_repo_record(&output, &other).unwrap().is_none());
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();60 unmodified lines