20260914 强制 TOML 配置并完善 TA 刷新与 daemon

src/cli/anchors.rs

@@ -0,0 +1,246 @@
+//! TAL-bound certificate refresh with a validated last-known certificate.
+use super::config::Config;
+use crate::model::{ta::TrustAnchor, tal::Tal};
+use crate::repository::fetch::{
+    http::{BlockingHttpFetcher, HttpFetcherConfig},
+    rsync::RsyncFetcher,
+    rsync_system::{SystemRsyncConfig, SystemRsyncFetcher},
+};
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use std::{
+    fs,
+    path::{Path, PathBuf},
+    time::Duration,
+};
+use x509_parser::prelude::{FromDer, X509Certificate};
+
+pub(crate) fn validate_certificate(tal: &Tal, bytes: &[u8]) -> Result<Value, String> {
+    TrustAnchor::bind_der(tal.clone(), bytes, None).map_err(|e| e.to_string())?;
+    let (_, cert) = X509Certificate::from_der(bytes).map_err(|e| e.to_string())?;
+    if !cert.validity().is_valid() {
+        return Err("TA certificate is expired or not yet valid".into());
+    }
+    Ok(
+        json!({"certificateSha256":hex::encode(Sha256::digest(bytes)),"notBefore":cert.validity().not_before.to_rfc2822().map_err(|e|e.to_string())?,"notAfter":cert.validity().not_after.to_rfc2822().map_err(|e|e.to_string())?}),
+    )
+}
+pub(crate) fn cache_root(state: &Path) -> PathBuf {
+    let mut name = state.as_os_str().to_os_string();
+    name.push(".ta-cache");
+    PathBuf::from(name)
+}
+
+pub(crate) fn resolve(
+    config: &Config,
+) -> Result<(Vec<crate::scheduler::types::TalInputSpec>, Vec<Value>), String> {
+    let http = BlockingHttpFetcher::new(HttpFetcherConfig {
+        timeout: Duration::from_secs(config.http_timeout_secs),
+        extra_root_certificates_pem: config
+            .http_root_cert
+            .iter()
+            .map(fs::read)
+            .collect::<Result<_, _>>()
+            .map_err(|e| e.to_string())?,
+        ..Default::default()
+    })?;
+    let rsync = SystemRsyncFetcher::new(SystemRsyncConfig::default());
+    resolve_with(config, |uri| {
+        match url::Url::parse(uri).map_err(|e| e.to_string())?.scheme() {
+            "https" => http.fetch_bytes(uri),
+            "rsync" => rsync.fetch_object(uri).map_err(|e| e.to_string()),
+            _ => Err("unsupported TA URI scheme".into()),
+        }
+    })
+}
+
+pub(crate) fn resolve_with(
+    config: &Config,
+    mut fetch: impl FnMut(&str) -> Result<Vec<u8>, String>,
+) -> Result<(Vec<crate::scheduler::types::TalInputSpec>, Vec<Value>), String> {
+    let root = cache_root(config.rrdp_state_dir.as_ref().ok_or("state path missing")?);
+    let mut inputs = Vec::new();
+    let mut records = Vec::new();
+    for anchor in &config.anchors {
+        let mut record = json!({"id":anchor.id,"mode":if anchor.ta.is_some(){"fixed"}else{"auto"},"source":Value::Null,"certificateSha256":Value::Null,"fetchedAt":Value::Null,"notBefore":Value::Null,"notAfter":Value::Null,"refreshFailed":false,"errorCode":Value::Null});
+        let attempt = (|| -> Result<PathBuf, String> {
+            let tal = Tal::decode_bytes(&fs::read(&anchor.tal).map_err(|e| e.to_string())?)
+                .map_err(|e| e.to_string())?;
+            if let Some(path) = &anchor.ta {
+                let info = validate_certificate(&tal, &fs::read(path).map_err(|e| e.to_string())?)?;
+                record
+                    .as_object_mut()
+                    .unwrap()
+                    .extend(info.as_object().unwrap().clone());
+                record["source"] = json!("file");
+                return Ok(path.clone());
+            }
+            let key = hex::encode(Sha256::digest(&tal.subject_public_key_info_der));
+            let path = root.join(format!("{key}.json"));
+            let der_path = root.join(format!("{key}.cer"));
+            let mut last_error = "no usable TA URI".to_string();
+            let mut error_code = "download_failed";
+            for uri in &tal.ta_uris {
+                match fetch(uri.as_str()) {
+                    Ok(bytes) => match validate_certificate(&tal, &bytes) {
+                        Ok(info) => {
+                            record
+                                .as_object_mut()
+                                .unwrap()
+                                .extend(info.as_object().unwrap().clone());
+                            record["source"] = json!("download");
+                            record["fetchedAt"] = json!(super::publication::now());
+                            // Certificate and metadata are one atomic cache record; the .cer is a runtime projection.
+                            super::publication::write_json(
+                                &path,
+                                &json!({"der":hex::encode(&bytes),"record":record}),
+                            )?;
+                            super::publication::write_atomic(&der_path, &bytes)?;
+                            return Ok(der_path);
+                        }
+                        Err(e) => {
+                            last_error = e;
+                            error_code = "certificate_invalid";
+                        }
+                    },
+                    Err(e) => {
+                        last_error = e;
+                    }
+                }
+            }
+            record["refreshFailed"] = json!(true);
+            record["errorCode"] = json!(error_code);
+            if config.ta_refresh_policy == "allow-valid-cache" && path.exists() {
+                let cached: Value = match fs::read(&path)
+                    .map_err(|e| e.to_string())
+                    .and_then(|bytes| serde_json::from_slice(&bytes).map_err(|e| e.to_string()))
+                {
+                    Ok(value) => value,
+                    Err(error) => {
+                        record["errorCode"] = json!("cache_unusable");
+                        return Err(format!("read cached TA metadata failed: {error}"));
+                    }
+                };
+                let bytes = match cached["der"].as_str() {
+                    Some(encoded) => match hex::decode(encoded) {
+                        Ok(bytes) => bytes,
+                        Err(error) => {
+                            record["errorCode"] = json!("cache_unusable");
+                            return Err(format!("decode cached TA failed: {error}"));
+                        }
+                    },
+                    None => {
+                        record["errorCode"] = json!("cache_unusable");
+                        return Err("cached TA metadata has no DER value".into());
+                    }
+                };
+                let info = validate_certificate(&tal, &bytes).inspect_err(|_| {
+                    record["errorCode"] = json!("cache_unusable");
+                })?;
+                record
+                    .as_object_mut()
+                    .unwrap()
+                    .extend(info.as_object().unwrap().clone());
+                record["source"] = json!("cache");
+                record["fetchedAt"] = cached["record"]["fetchedAt"].clone();
+                super::publication::write_atomic(&der_path, &bytes)?;
+                crate::logging::emit(
+                    crate::logging::Level::Warn,
+                    "ta_refresh_failed",
+                    || json!({"tal_id":anchor.id,"error":last_error,"fallback":"valid_cache"}),
+                );
+                return Ok(der_path);
+            }
+            Err(last_error)
+        })();
+        match attempt {
+            Ok(path) => inputs.push(anchor.input(&path)),
+            Err(e) => {
+                if record["errorCode"].is_null() {
+                    record["errorCode"] = json!("input_invalid");
+                }
+                record["error"] = json!(e);
+                crate::logging::emit(
+                    crate::logging::Level::Warn,
+                    "trust_anchor_failed",
+                    || json!({"tal_id":anchor.id,"error":e}),
+                );
+            }
+        }
+        records.push(record);
+    }
+    Ok((inputs, records))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{cache_root, resolve_with};
+    use crate::cli::config::{Anchor, Config};
+    use std::fs;
+
+    fn config(state: &std::path::Path) -> Config {
+        let fixture = crate::test_support::synthetic_repository();
+        Config {
+            config_version: 1,
+            mode: "validate".into(),
+            rrdp_state_dir: Some(state.to_path_buf()),
+            anchors: vec![Anchor {
+                id: "custom".into(),
+                tal: fixture.tal(),
+                ta: None,
+                constraints: None,
+            }],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn automatic_ta_download_is_bound_and_cached() {
+        let fixture = crate::test_support::synthetic_repository();
+        let directory = tempfile::tempdir().unwrap();
+        let config = config(directory.path().join("state").as_path());
+        let ta = fs::read(fixture.trust_anchor()).unwrap();
+        let (inputs, records) = resolve_with(&config, |_| Ok(ta.clone())).unwrap();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(records[0]["mode"], "auto");
+        assert_eq!(records[0]["source"], "download");
+        let mut cached = fs::read_dir(cache_root(config.rrdp_state_dir.as_ref().unwrap()))
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .collect::<Vec<_>>();
+        cached.sort();
+        assert!(
+            cached
+                .iter()
+                .any(|path| path.extension().is_some_and(|x| x == "json"))
+        );
+        assert!(
+            cached
+                .iter()
+                .any(|path| path.extension().is_some_and(|x| x == "cer"))
+        );
+    }
+
+    #[test]
+    fn valid_cache_fallback_is_partial_and_bad_cache_is_rejected() {
+        let fixture = crate::test_support::synthetic_repository();
+        let directory = tempfile::tempdir().unwrap();
+        let config = config(directory.path().join("state").as_path());
+        let ta = fs::read(fixture.trust_anchor()).unwrap();
+        resolve_with(&config, |_| Ok(ta.clone())).unwrap();
+        let (inputs, records) = resolve_with(&config, |_| Err("offline".into())).unwrap();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(records[0]["source"], "cache");
+        assert_eq!(records[0]["refreshFailed"], true);
+
+        let cache = fs::read_dir(cache_root(config.rrdp_state_dir.as_ref().unwrap()))
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .find(|path| path.extension().is_some_and(|x| x == "json"))
+            .unwrap();
+        fs::write(cache, br#"{"der":"00"}"#).unwrap();
+        let (inputs, records) = resolve_with(&config, |_| Err("offline".into())).unwrap();
+        assert!(inputs.is_empty());
+        assert_eq!(records[0]["errorCode"], "cache_unusable");
+    }
+}

src/cli/config.rs

@@ -0,0 +1,391 @@
+//! One explicit TOML file is the sole source of runtime settings.
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
+use std::collections::BTreeSet;
+use std::path::{Component, Path, PathBuf};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "kebab-case")]
+pub struct Anchor {
+    #[serde(default)]
+    pub id: String,
+    pub tal: PathBuf,
+    pub ta: Option<PathBuf>,
+    pub constraints: Option<PathBuf>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default, deny_unknown_fields, rename_all = "kebab-case")]
+pub struct Config {
+    pub config_version: u32,
+    pub mode: String,
+    pub anchors: Vec<Anchor>,
+    pub ta_refresh_policy: String,
+    pub max_ca_depth: usize,
+    pub http_timeout_secs: u64,
+    pub http_root_cert: Vec<PathBuf>,
+    pub parallel_max_repo_sync_workers_global: usize,
+    pub parallel_phase2_object_workers: usize,
+    pub parallel_phase2_worker_queue_capacity: usize,
+    pub parallel_repo_worker_queue_capacity: usize,
+    pub log_level: String,
+    pub log_format: String,
+    pub log_progress_interval_secs: u64,
+    pub require_complete: bool,
+    pub run_timeout_secs: u64,
+    pub out: Option<PathBuf>,
+    pub rrdp_state_dir: Option<PathBuf>,
+    pub rrdp_sync_mode: String,
+    pub ccr_out: Option<PathBuf>,
+    pub publish_dir: Option<PathBuf>,
+    pub state_root: Option<PathBuf>,
+    pub interval_secs: u64,
+    pub max_runs: Option<u64>,
+    pub retain_runs: usize,
+    pub shutdown_grace_secs: u64,
+}
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            config_version: 1,
+            mode: String::new(),
+            anchors: Vec::new(),
+            ta_refresh_policy: "allow-valid-cache".into(),
+            max_ca_depth: 64,
+            http_timeout_secs: 300,
+            http_root_cert: Vec::new(),
+            parallel_max_repo_sync_workers_global: 8,
+            parallel_phase2_object_workers: 8,
+            parallel_phase2_worker_queue_capacity: 256,
+            parallel_repo_worker_queue_capacity: 256,
+            log_level: "info".into(),
+            log_format: "text".into(),
+            log_progress_interval_secs: 10,
+            require_complete: false,
+            run_timeout_secs: 0,
+            out: None,
+            rrdp_state_dir: None,
+            rrdp_sync_mode: "auto".into(),
+            ccr_out: None,
+            publish_dir: None,
+            state_root: None,
+            interval_secs: 60,
+            max_runs: None,
+            retain_runs: 10,
+            shutdown_grace_secs: 30,
+        }
+    }
+}
+
+/// Resolve symlinked existing ancestors without creating a destination.
+pub(crate) fn absolute(base: &Path, path: &Path) -> Result<PathBuf, String> {
+    if path.as_os_str().is_empty() {
+        return Err("empty path".into());
+    }
+    let joined = if path.is_absolute() {
+        path.to_path_buf()
+    } else {
+        base.join(path)
+    };
+    let mut normalized = PathBuf::new();
+    for c in joined.components() {
+        match c {
+            Component::ParentDir => {
+                normalized.pop();
+            }
+            Component::CurDir => {}
+            x => normalized.push(x),
+        }
+    }
+    let mut ancestor = normalized.as_path();
+    let mut missing = Vec::new();
+    while !ancestor.exists() {
+        missing.push(ancestor.file_name().ok_or("invalid path")?.to_os_string());
+        ancestor = ancestor.parent().ok_or("invalid path")?;
+    }
+    let mut result = ancestor.canonicalize().map_err(|e| e.to_string())?;
+    for c in missing.into_iter().rev() {
+        result.push(c);
+    }
+    Ok(result)
+}
+
+impl Config {
+    pub(crate) fn load(path: &Path, mode: &str) -> Result<(Self, Value), String> {
+        let path = absolute(&std::env::current_dir().map_err(|e| e.to_string())?, path)?;
+        let raw = std::fs::read_to_string(&path)
+            .map_err(|e| format!("read config {}: {e}", path.display()))?;
+        let table: toml::Table = toml::from_str(&raw).map_err(|e| format!("invalid TOML: {e}"))?;
+        if table
+            .get("config-version")
+            .and_then(toml::Value::as_integer)
+            != Some(1)
+        {
+            return Err("config-version must be explicitly set to 1".into());
+        }
+        let mut config: Self =
+            toml::from_str(&raw).map_err(|e| format!("invalid configuration: {e}"))?;
+        if config.mode != mode {
+            return Err(format!("config mode must be {mode}"));
+        }
+        let forbidden: &[&str] = if mode == "daemon" {
+            &[
+                "out",
+                "rrdp-state-dir",
+                "rrdp-sync-mode",
+                "ccr-out",
+                "publish-dir",
+            ]
+        } else {
+            &[
+                "state-root",
+                "interval-secs",
+                "max-runs",
+                "retain-runs",
+                "shutdown-grace-secs",
+            ]
+        };
+        for key in forbidden {
+            if table.contains_key(*key) {
+                return Err(format!("{key} is not allowed in {mode} configuration"));
+            }
+        }
+        let base = path.parent().ok_or("config has no parent")?;
+        for p in [
+            &mut config.out,
+            &mut config.rrdp_state_dir,
+            &mut config.ccr_out,
+            &mut config.publish_dir,
+            &mut config.state_root,
+        ]
+        .into_iter()
+        .flatten()
+        {
+            *p = absolute(base, p)?;
+        }
+        if mode == "validate" {
+            let out = config.out.as_ref().ok_or("out is required")?;
+            if config.rrdp_state_dir.is_none() {
+                config.rrdp_state_dir = Some(out.join(".state"));
+            }
+            if let Some(publish) = &config.publish_dir {
+                for other in [out, config.rrdp_state_dir.as_ref().unwrap()] {
+                    if other.starts_with(publish) || publish.starts_with(other) {
+                        return Err("publish-dir must not overlap out or state".into());
+                    }
+                }
+            }
+        } else if config.state_root.is_none() {
+            return Err("state-root is required".into());
+        }
+        for p in config.http_root_cert.iter_mut() {
+            *p = absolute(base, p)?;
+        }
+        for anchor in &mut config.anchors {
+            anchor.tal = absolute(base, &anchor.tal)?;
+            if anchor.id.is_empty() {
+                anchor.id = anchor
+                    .tal
+                    .file_stem()
+                    .and_then(|x| x.to_str())
+                    .ok_or("invalid TAL name")?
+                    .into();
+            }
+            if let Some(p) = &mut anchor.ta {
+                *p = absolute(base, p)?;
+            }
+            if let Some(p) = &mut anchor.constraints {
+                *p = absolute(base, p)?;
+            }
+        }
+        config.validate_inputs()?;
+        let mut normalizations = Vec::new();
+        if config.http_timeout_secs < 1 {
+            normalizations.push(json!({"field":"http-timeout-secs","input":0,"effective":1}));
+            config.http_timeout_secs = 1;
+        }
+        if config.parallel_repo_worker_queue_capacity < 1024 {
+            normalizations.push(json!({"field":"parallel-repo-worker-queue-capacity","input":config.parallel_repo_worker_queue_capacity,"effective":1024}));
+            config.parallel_repo_worker_queue_capacity = 1024;
+        }
+        let mut effective = serde_json::to_value(&config).map_err(|e| e.to_string())?;
+        for key in forbidden {
+            effective.as_object_mut().unwrap().remove(*key);
+        }
+        let raw_json = serde_json::to_value(&table).map_err(|e| e.to_string())?;
+        let mut sources = serde_json::Map::new();
+        collect_sources(
+            "",
+            &effective,
+            &raw_json,
+            &format!("file:{}", path.display()),
+            &mut sources,
+        );
+        let printed = json!({"configVersion":1,"mode":mode,"effective":effective,"sources":sources,"normalizations":normalizations});
+        Ok((config, printed))
+    }
+
+    pub(crate) fn validate_inputs(&self) -> Result<(), String> {
+        if !["validate", "daemon"].contains(&self.mode.as_str()) || self.config_version != 1 {
+            return Err("invalid mode/config-version".into());
+        }
+        if self.anchors.is_empty() {
+            return Err("anchors must not be empty".into());
+        }
+        if !["allow-valid-cache", "require-fresh"].contains(&self.ta_refresh_policy.as_str()) {
+            return Err("invalid ta-refresh-policy".into());
+        }
+        if !["auto", "snapshot", "delta"].contains(&self.rrdp_sync_mode.as_str()) {
+            return Err("invalid rrdp-sync-mode".into());
+        }
+        crate::logging::parse_level(&self.log_level)?;
+        if !["text", "json"].contains(&self.log_format.as_str()) {
+            return Err("invalid log-format".into());
+        }
+        if [
+            self.parallel_max_repo_sync_workers_global,
+            self.parallel_phase2_object_workers,
+            self.parallel_phase2_worker_queue_capacity,
+            self.parallel_repo_worker_queue_capacity,
+            self.retain_runs,
+        ]
+        .contains(&0)
+            || self.max_runs == Some(0)
+        {
+            return Err("workers, queues, retain-runs and max-runs must be positive".into());
+        }
+        let mut ids = BTreeSet::new();
+        for anchor in &self.anchors {
+            if anchor.id.is_empty() || !ids.insert(&anchor.id) {
+                return Err("anchor IDs must be nonempty and unique".into());
+            }
+            let tal = crate::model::tal::Tal::decode_bytes(
+                &std::fs::read(&anchor.tal)
+                    .map_err(|e| format!("read TAL {}: {e}", anchor.tal.display()))?,
+            )
+            .map_err(|e| e.to_string())?;
+            if let Some(ta) = &anchor.ta {
+                super::anchors::validate_certificate(
+                    &tal,
+                    &std::fs::read(ta).map_err(|e| format!("read TA: {e}"))?,
+                )?;
+            }
+            let constraints = anchor
+                .constraints
+                .clone()
+                .unwrap_or_else(|| anchor.tal.with_extension("constraints"));
+            if anchor.constraints.is_some() || constraints.exists() {
+                crate::ta_constraints::TaConstraintsByTal::load_for_tals(
+                    &[anchor.input(anchor.ta.as_deref().unwrap_or(&anchor.tal))],
+                    &[format!("{}={}", anchor.id, constraints.display())],
+                )?;
+            }
+        }
+        for path in &self.http_root_cert {
+            reqwest::Certificate::from_pem(
+                &std::fs::read(path).map_err(|e| format!("read HTTP root: {e}"))?,
+            )
+            .map_err(|e| e.to_string())?;
+        }
+        for p in [
+            &self.out,
+            &self.state_root,
+            &self.rrdp_state_dir,
+            &self.publish_dir,
+        ]
+        .into_iter()
+        .flatten()
+        {
+            if p.exists() && !p.is_dir() {
+                return Err(format!("not a directory: {}", p.display()));
+            }
+            if let Some(parent) = p.parent()
+                && parent.exists()
+                && !parent.is_dir()
+            {
+                return Err(format!("parent is not a directory: {}", parent.display()));
+            }
+        }
+        if let Some(p) = &self.ccr_out {
+            if p.exists() && !p.is_file() {
+                return Err("ccr-out must be a file".into());
+            }
+            if let Some(parent) = p.parent()
+                && parent.exists()
+                && !parent.is_dir()
+            {
+                return Err(format!(
+                    "ccr-out parent is not a directory: {}",
+                    parent.display()
+                ));
+            }
+        }
+        Ok(())
+    }
+
+    pub(crate) fn child_config(
+        &self,
+        out: &Path,
+        state: &Path,
+        mode: &str,
+    ) -> Result<String, String> {
+        let mut child = self.clone();
+        child.mode = "validate".into();
+        child.out = Some(out.into());
+        child.rrdp_state_dir = Some(state.into());
+        child.ccr_out = Some(out.join("result.ccr"));
+        child.publish_dir = None;
+        child.rrdp_sync_mode = mode.into();
+        child.run_timeout_secs = 0;
+        let mut value = toml::Value::try_from(child).map_err(|e| e.to_string())?;
+        for key in [
+            "state-root",
+            "interval-secs",
+            "max-runs",
+            "retain-runs",
+            "shutdown-grace-secs",
+        ] {
+            value.as_table_mut().unwrap().remove(key);
+        }
+        toml::to_string_pretty(&value).map_err(|e| e.to_string())
+    }
+}
+impl Anchor {
+    pub(crate) fn input(&self, ta: &Path) -> crate::scheduler::types::TalInputSpec {
+        let mut input =
+            crate::scheduler::types::TalInputSpec::from_file_path_with_ta(&self.tal, ta);
+        input.tal_id = self.id.clone();
+        input
+    }
+}
+fn collect_sources(
+    prefix: &str,
+    effective: &Value,
+    raw: &Value,
+    file: &str,
+    sources: &mut serde_json::Map<String, Value>,
+) {
+    match effective {
+        Value::Object(map) => {
+            for (key, value) in map {
+                let name = if prefix.is_empty() {
+                    key.clone()
+                } else {
+                    format!("{prefix}.{key}")
+                };
+                collect_sources(&name, value, &raw[key], file, sources);
+            }
+        }
+        Value::Array(items) if !items.is_empty() => {
+            for (i, value) in items.iter().enumerate() {
+                collect_sources(&format!("{prefix}[{i}]"), value, &raw[i], file, sources);
+            }
+        }
+        _ => {
+            sources.insert(
+                prefix.into(),
+                json!(if raw.is_null() { "default" } else { file }),
+            );
+        }
+    }
+}

src/cli/mod.rs

@@ -1,17 +1,23 @@
-//! Narrow command-line surface for the v0.1.0 RFC core.
-
+//! Minimal command surface: one TOML configuration and read-only inspection.
+pub(crate) mod anchors;
+pub(crate) mod config;
+pub(crate) mod publication;
 mod validate;
+use std::path::PathBuf;
 
-/// Reject unrelated state layouts before selecting a synchronization mode.
 pub(crate) fn check_state_root(root: &std::path::Path) -> Result<(), String> {
     if !root.exists() {
         return Ok(());
     }
     for entry in std::fs::read_dir(root).map_err(|e| e.to_string())? {
         let entry = entry.map_err(|e| e.to_string())?;
-        if entry.file_name() != "repository-db"
-            || !entry.file_type().map_err(|e| e.to_string())?.is_dir()
-        {
+        let file_type = entry.file_type().map_err(|e| e.to_string())?;
+        let valid = match entry.file_name().to_string_lossy().as_ref() {
+            "repository-db" => file_type.is_dir(),
+            ".panda-rpki.lock" => file_type.is_file(),
+            _ => false,
+        };
+        if !valid {
             return Err(format!(
                 "unrecognized state layout at {}; use a fresh state directory",
                 root.display()
@@ -20,210 +26,88 @@ pub(crate) fn check_state_root(root: &std::path::Path) -> Result<(), String> {
     }
     Ok(())
 }
-
-pub(crate) use validate::{VALIDATE_OPTIONS, VALIDATE_REPEATABLE};
-
-/// Bounded default long enough for large public RIR snapshots.
-const DEFAULT_HTTP_TIMEOUT_SECS: u64 = 300;
-const DEFAULT_WORKER_COUNT: usize = 8;
-const DEFAULT_REPO_SYNC_WORKER_COUNT: usize = 8;
-const DEFAULT_WORKER_QUEUE_CAPACITY: usize = 256;
-
 pub fn usage() -> String {
-    let validation = format!(
-        "panda-rpki {}\n\nUSAGE:\n  panda-rpki validate --tal <file> --ta <file> [--tal <file> --ta <file> ...] --out <dir> [--rrdp-state-dir <dir>] [--rrdp-sync-mode auto|snapshot|delta] [--ccr-out <file>] [--parallel-phase2-object-workers <n>] [--parallel-max-repo-sync-workers-global <n>] [--parallel-phase2-worker-queue-capacity <n>] [--parallel-repo-worker-queue-capacity <n>] [--max-ca-depth <n>] [--http-timeout-secs <n>] [--http-root-cert <pem>]... [--log-level off|error|warn|info|debug|trace] [--log-format text|json]\n\nTAL/TA pairs are positional and may be repeated. RRDP state is an explicit protocol-resume directory, not a validation cache. The public v0.1.0 fast path accepts HTTPS RRDP notification/snapshot/delta URLs only.",
-        crate::VERSION
-    );
     format!(
-        "{validation}\n  --log-progress-interval-secs <n>: periodic info progress (default 10; 0 disables).\n\n{}",
-        crate::daemon::usage()
+        "panda-rpki {}\n\nUSAGE:\n  panda-rpki validate --config FILE [--check-config | --print-config]\n  panda-rpki daemon --config FILE [--check-config | --print-config]\n  panda-rpki --help | --version\n\nAll runtime settings belong in TOML. No runtime option or environment overrides.\n\nOPTIONS:\n  --config FILE     Required TOML configuration (also --config=FILE)\n  --check-config    Read-only validation, without network or state writes\n  --print-config    Print effective configuration, defaults and sources as JSON\n  -h, --help        Show this help\n  -V, --version     Show version (top level only)\n",
+        crate::VERSION
     )
 }
-
-pub fn run<I, S>(args: I) -> Result<(), String>
+pub fn run<I, S>(args: I) -> Result<i32, String>
 where
     I: IntoIterator<Item = S>,
     S: Into<String>,
 {
     let args: Vec<String> = args.into_iter().map(Into::into).collect();
-    if args.is_empty() || matches!(args.first().map(String::as_str), Some("--help" | "-h")) {
-        return Err(usage());
+    if args.is_empty() {
+        print!("{}", usage());
+        return Ok(0);
     }
-    match args[0].as_str() {
-        "validate" => validate::run(&args[1..]),
-        "daemon" => crate::daemon::run(&args[1..]),
-        _ => Err(format!("unknown command: {}", args[0])),
-    }
-}
-
-#[derive(Debug)]
-pub(crate) struct Options {
-    values: std::collections::BTreeMap<String, Vec<String>>,
-}
-
-impl Options {
-    pub(crate) fn parse(
-        args: &[String],
-        allowed: &[&str],
-        repeatable: &[&str],
-    ) -> Result<Self, String> {
-        let mut values = std::collections::BTreeMap::<String, Vec<String>>::new();
-        let mut index = 0;
-        while index < args.len() {
-            let flag = &args[index];
-            if matches!(flag.as_str(), "--help" | "-h") {
-                return Err(usage());
-            }
-            let name = flag
-                .strip_prefix("--")
-                .ok_or_else(|| format!("expected an option, got {flag}"))?;
-            if !allowed.contains(&name) {
-                return Err(format!("unsupported option: {flag}"));
-            }
-            index += 1;
-            let value = args
-                .get(index)
-                .ok_or_else(|| format!("{flag} requires a value"))?;
-            if value.starts_with("--") {
-                return Err(format!("{flag} requires a value"));
-            }
-            let entry = values.entry(name.to_string()).or_default();
-            if !repeatable.contains(&name) && !entry.is_empty() {
-                return Err(format!("option repeated: {flag}"));
-            }
-            entry.push(value.clone());
-            index += 1;
+    if args.len() == 1 && ["--help", "-h", "--version", "-V"].contains(&args[0].as_str()) {
+        if ["--version", "-V"].contains(&args[0].as_str()) {
+            println!("panda-rpki {}", crate::VERSION);
+        } else {
+            print!("{}", usage());
         }
-        Ok(Self { values })
+        return Ok(0);
     }
-
-    pub(crate) fn required(&self, name: &str) -> Result<&str, String> {
-        self.values
-            .get(name)
-            .and_then(|values| values.first())
-            .map(String::as_str)
-            .ok_or_else(|| format!("--{name} is required"))
+    let mode = &args[0];
+    if !["validate", "daemon"].contains(&mode.as_str()) {
+        return Err(format!("unknown command: {mode}"));
     }
-
-    pub(crate) fn values(&self, name: &str) -> &[String] {
-        self.values.get(name).map(Vec::as_slice).unwrap_or(&[])
+    let mut path = None;
+    let mut help = false;
+    let mut check = false;
+    let mut print = false;
+    let mut i = 1;
+    while i < args.len() {
+        let a = &args[i];
+        match a.as_str() {
+            "--help" | "-h" if !help => help = true,
+            "--check-config" if !check => check = true,
+            "--print-config" if !print => print = true,
+            "--config" if path.is_none() => {
+                i += 1;
+                let p = args
+                    .get(i)
+                    .filter(|v| !v.starts_with('-'))
+                    .ok_or("--config requires a file")?;
+                path = Some(PathBuf::from(p));
+            }
+            _ if a.starts_with("--config=") && path.is_none() => {
+                path = Some(PathBuf::from(&a[9..]))
+            }
+            _ => {
+                return Err(format!(
+                    "unsupported or repeated option: {a}; configure runtime settings in TOML and use --config FILE"
+                ));
+            }
+        }
+        i += 1;
     }
-
-    pub(crate) fn parse_or<T>(&self, name: &str, default: T) -> Result<T, String>
-    where
-        T: std::str::FromStr,
-        T::Err: std::fmt::Display,
-    {
-        self.values
-            .get(name)
-            .and_then(|values| values.first())
-            .map(|value| {
-                value
-                    .parse()
-                    .map_err(|error| format!("invalid --{name}: {error}"))
-            })
-            .unwrap_or(Ok(default))
+    if u8::from(help) + u8::from(check) + u8::from(print) > 1 {
+        return Err("help, check-config and print-config are mutually exclusive".into());
     }
-
-    pub(crate) fn optional_parse<T>(&self, name: &str) -> Result<Option<T>, String>
-    where
-        T: std::str::FromStr,
-        T::Err: std::fmt::Display,
-    {
-        self.values
-            .get(name)
-            .and_then(|values| values.first())
-            .map(|value| {
-                value
-                    .parse()
-                    .map(Some)
-                    .map_err(|error| format!("invalid --{name}: {error}"))
-            })
-            .unwrap_or(Ok(None))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn state_layout_rejects_unrecognized_entries_without_modifying_them() {
-        let root = tempfile::tempdir().unwrap();
-        check_state_root(root.path()).unwrap();
-        std::fs::create_dir(root.path().join("repository-db")).unwrap();
-        check_state_root(root.path()).unwrap();
-        let unknown = root.path().join("previous-state");
-        std::fs::create_dir(&unknown).unwrap();
-        std::fs::write(unknown.join("checkpoint"), b"retain").unwrap();
-        assert!(
-            check_state_root(root.path())
-                .unwrap_err()
-                .contains("unrecognized state layout")
-        );
-        assert_eq!(
-            std::fs::read(unknown.join("checkpoint")).unwrap(),
-            b"retain"
-        );
-    }
-
-    #[test]
-    fn parser_rejects_private_or_unknown_options() {
-        let error = Options::parse(&["--extra".to_string(), "1".to_string()], &["out"], &[])
-            .expect_err("unknown option");
-        assert!(error.contains("unsupported option"));
-    }
-
-    #[test]
-    fn parser_preserves_repeatable_tal_pairs() {
-        let options = Options::parse(
-            &[
-                "--tal".to_string(),
-                "one.tal".to_string(),
-                "--ta".to_string(),
-                "one.cer".to_string(),
-                "--tal".to_string(),
-                "two.tal".to_string(),
-                "--ta".to_string(),
-                "two.cer".to_string(),
-            ],
-            &["tal", "ta"],
-            &["tal", "ta"],
-        )
-        .expect("repeatable pairs");
-        assert_eq!(options.values("tal"), ["one.tal", "two.tal"]);
-        assert_eq!(options.values("ta"), ["one.cer", "two.cer"]);
-    }
-
-    #[test]
-    fn parser_rejects_non_repeatable_worker_option() {
-        let error = Options::parse(
-            &[
-                "--workers".to_string(),
-                "2".to_string(),
-                "--workers".to_string(),
-                "3".to_string(),
-            ],
-            &["workers"],
-            &[],
-        )
-        .expect_err("worker option should not repeat");
-        assert!(error.contains("option repeated"));
-    }
-
-    #[test]
-    fn parser_preserves_repeatable_http_root_certificates() {
-        let options = Options::parse(
-            &[
-                "--http-root-cert".to_string(),
-                "one.pem".to_string(),
-                "--http-root-cert".to_string(),
-                "two.pem".to_string(),
-            ],
-            &["http-root-cert"],
-            &["http-root-cert"],
-        )
-        .expect("repeatable HTTP roots");
-        assert_eq!(options.values("http-root-cert"), ["one.pem", "two.pem"]);
+    if help {
+        print!("{}", usage());
+        return Ok(0);
+    }
+    let path = path.ok_or("--config FILE is required")?;
+    let (config, effective) = config::Config::load(&path, mode)?;
+    if check {
+        println!("Configuration valid.");
+        return Ok(0);
+    }
+    if print {
+        println!(
+            "{}",
+            serde_json::to_string_pretty(&effective).map_err(|e| e.to_string())?
+        );
+        return Ok(0);
+    }
+    crate::logging::configure(&config.log_level, &config.log_format)?;
+    if mode == "daemon" {
+        crate::daemon::run(config)
+    } else {
+        validate::run(config)
     }
 }

src/cli/publication.rs

@@ -0,0 +1,133 @@
+//! Durable single-file writes and immutable successful result publication.
+use serde_json::{Value, json};
+use std::{
+    fs::{self, File, OpenOptions},
+    io::Write,
+    path::{Path, PathBuf},
+};
+
+pub(crate) fn now() -> String {
+    time::OffsetDateTime::now_utc()
+        .format(&time::format_description::well_known::Rfc3339)
+        .expect("UTC")
+}
+pub(crate) fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), String> {
+    let parent = path.parent().ok_or("path has no parent")?;
+    fs::create_dir_all(parent).map_err(|e| e.to_string())?;
+    let mut temp = tempfile::NamedTempFile::new_in(parent).map_err(|e| e.to_string())?;
+    temp.write_all(bytes)
+        .and_then(|()| temp.as_file().sync_all())
+        .map_err(|e| e.to_string())?;
+    temp.persist(path).map_err(|e| e.to_string())?;
+    File::open(parent)
+        .and_then(|f| f.sync_all())
+        .map_err(|e| e.to_string())
+}
+pub(crate) fn write_json(path: &Path, value: &Value) -> Result<(), String> {
+    write_atomic(
+        path,
+        &serde_json::to_vec_pretty(value).map_err(|e| e.to_string())?,
+    )
+}
+
+pub(crate) struct Locks(Vec<File>);
+impl Locks {
+    pub(crate) fn acquire(paths: &[PathBuf]) -> Result<Self, String> {
+        let mut paths = paths.to_vec();
+        paths.sort();
+        paths.dedup();
+        let mut locks = Self(Vec::new());
+        for path in paths {
+            // The lock belongs inside the protected directory. This keeps a
+            // container with a read-only root filesystem from trying to create
+            // `/output.lock` or `/state.lock` beside a bind mount.
+            fs::create_dir_all(&path).map_err(|e| e.to_string())?;
+            let lock_path = path.join(".panda-rpki.lock");
+            let f = OpenOptions::new()
+                .create(true)
+                .truncate(false)
+                .read(true)
+                .write(true)
+                .open(&lock_path)
+                .map_err(|e| e.to_string())?;
+            f.try_lock()
+                .map_err(|e| format!("path is locked: {}: {e}", path.display()))?;
+            locks.0.push(f);
+        }
+        Ok(locks)
+    }
+}
+
+pub(crate) fn switch_latest(root: &Path, target: &Path) -> Result<(), String> {
+    let tmp = root.join(format!(".latest-{}", uuid::Uuid::new_v4()));
+    std::os::unix::fs::symlink(target, &tmp).map_err(|e| e.to_string())?;
+    if let Err(e) = fs::rename(&tmp, root.join("latest")) {
+        let _ = fs::remove_file(tmp);
+        return Err(e.to_string());
+    }
+    File::open(root)
+        .and_then(|f| f.sync_all())
+        .map_err(|e| e.to_string())
+}
+pub(crate) fn attempt(root: &Path, id: &str, status: &str) -> Result<(), String> {
+    let published = fs::read_link(root.join("latest"))
+        .ok()
+        .and_then(|p| p.file_name().map(|s| s.to_string_lossy().into_owned()));
+    write_json(
+        &root.join("last-attempt.json"),
+        &json!({"runId":id,"resultStatus":status,"finishedAt":now(),"publishedRunId":published}),
+    )
+}
+pub(crate) fn publish(
+    root: &Path,
+    id: &str,
+    files: &[(PathBuf, PathBuf)],
+    summary: &Value,
+) -> Result<(), String> {
+    fs::create_dir_all(root.join("runs")).map_err(|e| e.to_string())?;
+    let staging = tempfile::Builder::new()
+        .prefix(".staging-")
+        .tempdir_in(root)
+        .map_err(|e| e.to_string())?;
+    for (source, relative) in files {
+        write_atomic(
+            &staging.path().join(relative),
+            &fs::read(source).map_err(|e| e.to_string())?,
+        )?;
+    }
+    let mut published_summary = summary.clone();
+    published_summary["publicationStatus"] = json!("published");
+    write_json(&staging.path().join("summary.json"), &published_summary)?;
+    complete(
+        staging.path(),
+        id,
+        "success",
+        &files.iter().map(|(_, p)| p.clone()).collect::<Vec<_>>(),
+    )?;
+    let dest = root.join("runs").join(id);
+    fs::rename(staging.path(), &dest).map_err(|e| e.to_string())?;
+    File::open(root.join("runs"))
+        .and_then(|f| f.sync_all())
+        .map_err(|e| e.to_string())?;
+    switch_latest(root, &Path::new("runs").join(id))?;
+    // The pointer is the commit point; bookkeeping failures cannot undo it.
+    if let Err(e) = attempt(root, id, "success") {
+        crate::logging::emit(
+            crate::logging::Level::Warn,
+            "publication_bookkeeping_failed",
+            || json!({"error":e}),
+        );
+    }
+    Ok(())
+}
+pub(crate) fn complete(
+    out: &Path,
+    id: &str,
+    status: &str,
+    files: &[PathBuf],
+) -> Result<(), String> {
+    write_json(
+        &out.join("result-complete.json"),
+        &json!({"runId":id,"resultStatus":status,"finishedAt":now(),"files":files}),
+    )
+}

src/cli/validate.rs

@@ -1,228 +1,227 @@
-//! Parse validation options into typed runtime configuration.
-
+//! Execute one validation attempt, then commit complete successful outputs.
+use super::{config::Config, publication};
+use serde_json::{Value, json};
 use std::fs::{self, File, OpenOptions};
 use std::io::{BufRead, BufReader, BufWriter, Write};
 use std::path::{Path, PathBuf};
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-enum SyncMode {
-    Auto,
-    Snapshot,
-    Delta,
+pub(super) fn run(config: Config) -> Result<i32, String> {
+    // A separate process group makes the complete attempt cancellable, including
+    // blocking native validation and rsync descendants. Fork occurs before workers.
+    super::super::daemon::process::supervise(config.run_timeout_secs, || run_once(&config))
+        .inspect_err(|error| {
+            let _ = record_failure(&config, "interrupted", error);
+        })
 }
-
-pub(crate) const VALIDATE_OPTIONS: &[&str] = &[
-    "tal",
-    "log-level",
-    "log-format",
-    "log-progress-interval-secs",
-    "ta",
-    "tal-id",
-    "ta-constraints",
-    "out",
-    "ccr-out",
-    "rrdp-state-dir",
-    "rrdp-sync-mode",
-    "max-ca-depth",
-    "http-timeout-secs",
-    "http-root-cert",
-    "parallel-max-repo-sync-workers-global",
-    "parallel-phase2-object-workers",
-    "parallel-phase2-worker-queue-capacity",
-    "parallel-repo-worker-queue-capacity",
-];
-pub(crate) const VALIDATE_REPEATABLE: &[&str] =
-    &["tal", "ta", "tal-id", "ta-constraints", "http-root-cert"];
-
-pub(super) fn run(args: &[String]) -> Result<(), String> {
-    let options = super::Options::parse(args, VALIDATE_OPTIONS, VALIDATE_REPEATABLE)?;
-    let level = resolve_setting(
-        options.values("log-level"),
-        std::env::var("PANDA_RPKI_LOG_LEVEL").ok(),
-        "info",
+fn record_failure(config: &Config, id: &str, error: &str) -> Result<(), String> {
+    if let Some(out) = &config.out
+        && out.is_dir()
+    {
+        let _ = fs::remove_file(out.join("result-complete.json"));
+    }
+    if let Some(root) = &config.publish_dir {
+        publication::attempt(root, id, "failed")?;
+    }
+    crate::logging::emit(
+        crate::logging::Level::Error,
+        "validation_failed",
+        || json!({"error":error}),
     );
-    let format = resolve_setting(
-        options.values("log-format"),
-        std::env::var("PANDA_RPKI_LOG_FORMAT").ok(),
-        "text",
-    );
-    crate::logging::configure(&level, &format)?;
-    let progress_interval = resolve_setting(
-        options.values("log-progress-interval-secs"),
-        std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").ok(),
-        "10",
-    )
-    .parse::<u64>()
-    .map_err(|_| {
-        "invalid log-progress-interval-secs: expected a nonnegative integer".to_string()
-    })?;
-    let run_id =
-        std::env::var("PANDA_RPKI_RUN_ID").unwrap_or_else(|_| uuid::Uuid::new_v4().to_string());
-    let run_seq = std::env::var("PANDA_RPKI_RUN_SEQ")
-        .ok()
-        .and_then(|value| value.parse().ok());
-    crate::logging::set_context(&run_id, run_seq);
-    let tal_paths = options.values("tal");
-    let ta_paths = options.values("ta");
-    if tal_paths.is_empty() || ta_paths.is_empty() || tal_paths.len() != ta_paths.len() {
-        return Err("--tal and --ta must be repeated as equal-length positional pairs".to_string());
-    }
-    let requested_ids = options.values("tal-id");
-    if !requested_ids.is_empty() && requested_ids.len() != tal_paths.len() {
-        return Err("--tal-id must be supplied once for each --tal when used".to_string());
-    }
-    let max_ca_depth = options.parse_or("max-ca-depth", 64_usize)?;
-    let timeout = options.parse_or("http-timeout-secs", super::DEFAULT_HTTP_TIMEOUT_SECS)?;
-    let workers = options
-        .optional_parse("parallel-phase2-object-workers")?
-        .unwrap_or(super::DEFAULT_WORKER_COUNT);
-    let repo_workers = options
-        .optional_parse("parallel-max-repo-sync-workers-global")?
-        .unwrap_or(super::DEFAULT_REPO_SYNC_WORKER_COUNT);
-    let phase2_queue = options.parse_or(
-        "parallel-phase2-worker-queue-capacity",
-        super::DEFAULT_WORKER_QUEUE_CAPACITY,
-    )?;
-    let repo_queue = options.parse_or(
-        "parallel-repo-worker-queue-capacity",
-        super::DEFAULT_WORKER_QUEUE_CAPACITY,
-    )?;
-    if workers == 0 || repo_workers == 0 || phase2_queue == 0 || repo_queue == 0 {
-        return Err("worker counts and queue capacities must be greater than zero".to_string());
-    }
-
-    let out = PathBuf::from(options.required("out")?);
-    fs::create_dir_all(&out)
-        .map_err(|error| format!("create output directory {}: {error}", out.display()))?;
-    let state_root = options
-        .values("rrdp-state-dir")
-        .first()
-        .map(PathBuf::from)
-        .unwrap_or_else(|| out.join(".state"));
-    fs::create_dir_all(&state_root)
-        .map_err(|error| format!("create state directory {}: {error}", state_root.display()))?;
-    let db_path = state_root.join("repository-db");
-    super::check_state_root(&state_root)?;
-    let sync_mode = parse_sync_mode(options.values("rrdp-sync-mode"))?;
-    if sync_mode == SyncMode::Snapshot && directory_has_entries(&db_path)? {
-        return Err(format!(
-            "snapshot mode requires a fresh fast-path state directory; existing database: {}",
-            db_path.display()
-        ));
-    }
-    if sync_mode == SyncMode::Delta && !directory_has_entries(&db_path)? {
-        return Err(format!(
-            "delta mode requires an existing fast-path state database: {}",
-            db_path.display()
-        ));
-    }
-
-    use crate::runtime::RunConfig;
-    use crate::scheduler::config::{ParallelPhase1Config, ParallelPhase2Config};
-    use crate::scheduler::types::TalInputSpec;
-    let mut tal_inputs: Vec<_> = tal_paths
-        .iter()
-        .zip(ta_paths)
-        .map(|(tal, ta)| TalInputSpec::from_file_path_with_ta(tal, ta))
-        .collect();
-    for (input, id) in tal_inputs.iter_mut().zip(requested_ids) {
-        input.tal_id = id.clone();
-    }
-    let tal_ids: Vec<String> = tal_inputs
-        .iter()
-        .map(|input| input.tal_id.clone())
-        .collect();
-    let ta_constraints = crate::ta_constraints::TaConstraintsByTal::load_for_tals(
-        &tal_inputs,
-        options.values("ta-constraints"),
-    )?;
-    let config = RunConfig {
-        db_path,
-        tal_paths: tal_paths.iter().map(PathBuf::from).collect(),
-        ta_paths: ta_paths.iter().map(PathBuf::from).collect(),
-        tal_path: tal_paths.first().map(PathBuf::from),
-        ta_path: ta_paths.first().map(PathBuf::from),
-        tal_inputs,
-        ta_constraints,
-        parallel_phase1_config: ParallelPhase1Config {
-            max_repo_sync_workers_global: repo_workers,
-            max_pending_repo_results: repo_queue.max(1024),
-            ..Default::default()
-        },
-        parallel_phase2_config: ParallelPhase2Config {
-            object_workers: workers,
-            worker_queue_capacity: phase2_queue,
-            ..Default::default()
-        },
-        max_ca_depth,
-        http_timeout_secs: timeout,
-        http_root_cert_paths: options
-            .values("http-root-cert")
-            .iter()
-            .map(PathBuf::from)
-            .collect(),
-        rsync_timeout_secs: 30,
-        summary_out_path: Some(out.join("summary.json")),
-        vrps_csv_out_path: Some(out.join(".vrps-source.csv")),
-        vaps_csv_out_path: Some(out.join("vaps.csv")),
-        analysis_out_path: Some(out.join("analysis")),
-        analyze: true,
-        ccr_out_path: options.values("ccr-out").first().map(PathBuf::from),
-        ..Default::default()
-    };
-
-    let started = std::time::Instant::now();
-    let observer = crate::logging::observer::Observer::start(progress_interval)?;
-    crate::logging::emit(crate::logging::Level::Info, "validation_started", || {
-        serde_json::json!({
-            "tal_count": tal_paths.len(), "tal_ids": tal_ids, "phase2_workers": workers, "repo_workers": repo_workers,
-            "sync_mode": format!("{sync_mode:?}").to_lowercase(),
-        })
-    });
-    let result = crate::runtime::run_config(config).and_then(|()| {
-        crate::logging::observer::phase("vrp_csv_conversion");
-        let fast_vrps = out.join(".vrps-source.csv");
-        crate::logging::artifact::write("vrps", &out.join("vrps.csv"), || {
-            convert_vrp_csv(&fast_vrps, &out.join("vrps.csv"))
-        })
-    });
-    observer.finish(result.is_ok());
-    result.inspect_err(|error| {
-        crate::logging::emit(crate::logging::Level::Error, "validation_failed", || {
-            serde_json::json!({
-                "elapsed_ms": started.elapsed().as_millis(),
-                "error": error,
-            })
-        });
-    })?;
-
-    crate::logging::emit(crate::logging::Level::Info, "validation_completed", || {
-        serde_json::json!({
-            "elapsed_ms": started.elapsed().as_millis(), "phase2_workers": workers, "repo_workers": repo_workers,
-        })
-    });
     Ok(())
 }
-
-fn parse_sync_mode(values: &[String]) -> Result<SyncMode, String> {
-    match values.first().map(String::as_str).unwrap_or("auto") {
-        "auto" => Ok(SyncMode::Auto),
-        "snapshot" => Ok(SyncMode::Snapshot),
-        "delta" => Ok(SyncMode::Delta),
-        other => Err(format!(
-            "--rrdp-sync-mode must be auto, snapshot, or delta; got {other}"
-        )),
+fn run_once(config: &Config) -> Result<i32, String> {
+    let out = config.out.as_ref().ok_or("out required")?;
+    let state = config.rrdp_state_dir.as_ref().ok_or("state required")?;
+    let mut paths = vec![out.clone(), state.clone()];
+    if let Some(p) = &config.publish_dir {
+        paths.push(p.clone());
     }
+    let _locks = publication::Locks::acquire(&paths)?;
+    fs::create_dir_all(out).map_err(|e| e.to_string())?;
+    fs::create_dir_all(state).map_err(|e| e.to_string())?;
+    match fs::remove_file(out.join("result-complete.json")) {
+        Ok(()) => {}
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+        Err(e) => return Err(e.to_string()),
+    }
+    let internal_context = std::env::var_os("PANDA_RPKI_INTERNAL_CONTEXT").is_some();
+    let run_id = if internal_context {
+        std::env::var("PANDA_RPKI_RUN_ID").unwrap_or_else(|_| uuid::Uuid::new_v4().to_string())
+    } else {
+        uuid::Uuid::new_v4().to_string()
+    };
+    let run_seq = internal_context
+        .then(|| std::env::var("PANDA_RPKI_RUN_SEQ").ok())
+        .flatten()
+        .and_then(|value| value.parse::<u64>().ok());
+    crate::logging::set_context(&run_id, run_seq);
+    let result = execute(config, &run_id);
+    if let Err(error) = &result {
+        let _ = record_failure(config, &run_id, error);
+    }
+    result
 }
-
-fn resolve_setting(cli_values: &[String], env_value: Option<String>, default: &str) -> String {
-    cli_values
-        .first()
-        .cloned()
-        .or(env_value)
-        .unwrap_or_else(|| default.to_string())
+fn execute(config: &Config, run_id: &str) -> Result<i32, String> {
+    let out = config.out.as_ref().unwrap();
+    let state = config.rrdp_state_dir.as_ref().unwrap();
+    super::check_state_root(state)?;
+    let db_path = state.join("repository-db");
+    let populated = directory_has_entries(&db_path)?;
+    if config.rrdp_sync_mode == "snapshot" && populated {
+        return Err("snapshot mode requires a fresh state directory".into());
+    }
+    if config.rrdp_sync_mode == "delta" && !populated {
+        return Err("delta mode requires an existing state database".into());
+    }
+    let observer = crate::logging::observer::Observer::start(config.log_progress_interval_secs)?;
+    let start = std::time::Instant::now();
+    crate::logging::emit(
+        crate::logging::Level::Info,
+        "validation_started",
+        || json!({"tal_count":config.anchors.len(),"sync_mode":config.rrdp_sync_mode}),
+    );
+    let (inputs, anchors) = super::anchors::resolve(config)?;
+    let failed = config.anchors.len() - inputs.len();
+    if inputs.is_empty() {
+        publication::write_json(
+            &out.join("summary.json"),
+            &json!({"resultStatus":"failed","trustAnchors":anchors,"taFailedCount":failed,"publicationStatus":"skipped"}),
+        )?;
+        observer.finish(false);
+        return Err("all trust anchors failed".into());
+    }
+    let overrides: Vec<String> = config
+        .anchors
+        .iter()
+        .filter(|a| inputs.iter().any(|i| i.tal_id == a.id))
+        .filter_map(|a| {
+            a.constraints
+                .as_ref()
+                .map(|p| format!("{}={}", a.id, p.display()))
+        })
+        .collect();
+    let constraints =
+        crate::ta_constraints::TaConstraintsByTal::load_for_tals(&inputs, &overrides)?;
+    // All runtime artifacts are first written under an isolated attempt directory.
+    let staging = tempfile::Builder::new()
+        .prefix(".validation-")
+        .tempdir_in(out)
+        .map_err(|e| e.to_string())?;
+    let root = staging.path();
+    let runtime = crate::runtime::RunConfig {
+        db_path,
+        tal_inputs: inputs,
+        ta_constraints: constraints,
+        parallel_phase1_config: crate::scheduler::config::ParallelPhase1Config {
+            max_repo_sync_workers_global: config.parallel_max_repo_sync_workers_global,
+            max_pending_repo_results: config.parallel_repo_worker_queue_capacity,
+            ..Default::default()
+        },
+        parallel_phase2_config: crate::scheduler::config::ParallelPhase2Config {
+            object_workers: config.parallel_phase2_object_workers,
+            worker_queue_capacity: config.parallel_phase2_worker_queue_capacity,
+            ..Default::default()
+        },
+        max_ca_depth: config.max_ca_depth,
+        http_timeout_secs: config.http_timeout_secs,
+        http_root_cert_paths: config.http_root_cert.clone(),
+        rsync_timeout_secs: 30,
+        summary_out_path: Some(root.join("summary.json")),
+        vrps_csv_out_path: Some(root.join(".vrps-source.csv")),
+        vaps_csv_out_path: Some(root.join("vaps.csv")),
+        analysis_out_path: Some(root.join("analysis")),
+        analyze: true,
+        ccr_out_path: config.ccr_out.as_ref().map(|_| root.join("result.ccr")),
+        ..Default::default()
+    };
+    let result = crate::runtime::run_config(runtime)
+        .and_then(|()| convert_vrp_csv(&root.join(".vrps-source.csv"), &root.join("vrps.csv")));
+    observer.finish(result.is_ok());
+    result?;
+    let mut summary: Value =
+        serde_json::from_slice(&fs::read(root.join("summary.json")).map_err(|e| e.to_string())?)
+            .map_err(|e| e.to_string())?;
+    let repository_failed = summary["repositoryFailedCount"].as_u64().unwrap_or(0);
+    let cached = anchors.iter().filter(|a| a["source"] == "cache").count();
+    let status = if failed > 0 || repository_failed > 0 || cached > 0 {
+        "partial"
+    } else {
+        "success"
+    };
+    summary["resultStatus"] = json!(status);
+    summary["taFailedCount"] = json!(failed);
+    summary["taCacheFallbackCount"] = json!(cached);
+    summary["trustAnchors"] = json!(anchors);
+    summary["publicationStatus"] = json!(if config.publish_dir.is_some() {
+        "skipped"
+    } else {
+        "disabled"
+    });
+    publication::write_json(&root.join("summary.json"), &summary)?;
+    let names = [
+        "vrps.csv",
+        "vaps.csv",
+        "summary.json",
+        "stage-timing.json",
+        "analysis/timing.json",
+    ];
+    let mut files = Vec::new();
+    for name in names {
+        let source = root.join(name);
+        if !source.is_file() {
+            return Err(format!("missing expected artifact {name}"));
+        }
+        let dest = out.join(name);
+        publication::write_atomic(&dest, &fs::read(&source).map_err(|e| e.to_string())?)?;
+        files.push((dest, PathBuf::from(name)));
+    }
+    if let Some(dest) = &config.ccr_out {
+        publication::write_atomic(
+            dest,
+            &fs::read(root.join("result.ccr")).map_err(|e| e.to_string())?,
+        )?;
+        files.push((dest.clone(), PathBuf::from("result.ccr")));
+    }
+    // The completion marker lives in `out`; use paths relative to that directory
+    // for its local artifacts and retain an absolute path only for an explicitly
+    // configured CCR outside `out`.
+    let completed_files = files
+        .iter()
+        .map(|(path, relative)| {
+            if path.starts_with(out) {
+                relative.clone()
+            } else {
+                path.clone()
+            }
+        })
+        .collect::<Vec<_>>();
+    publication::complete(out, run_id, status, &completed_files)?;
+    if let Some(publish) = &config.publish_dir {
+        if status == "success" {
+            publication::publish(publish, run_id, &files, &summary)?;
+            summary["publicationStatus"] = json!("published");
+            if let Err(e) = publication::write_json(&out.join("summary.json"), &summary) {
+                crate::logging::emit(
+                    crate::logging::Level::Warn,
+                    "publication_bookkeeping_failed",
+                    || json!({"error":e}),
+                );
+            }
+        } else {
+            publication::attempt(publish, run_id, status)?;
+        }
+    }
+    crate::logging::emit(
+        crate::logging::Level::Info,
+        "validation_completed",
+        || json!({"elapsed_ms":start.elapsed().as_millis(),"result_status":status}),
+    );
+    Ok(if status == "partial" && config.require_complete {
+        3
+    } else {
+        0
+    })
 }
-
 fn directory_has_entries(path: &Path) -> Result<bool, String> {
     if !path.exists() {
         return Ok(false);
@@ -280,21 +279,3 @@ fn convert_vrp_csv(input: &Path, output: &Path) -> Result<(), String> {
         .flush()
         .map_err(|error| format!("flush Panda RPKI VRP CSV: {error}"))
 }
-
-#[cfg(test)]
-mod tests {
-    use super::resolve_setting;
-
-    #[test]
-    fn logging_setting_precedence_is_cli_then_env_then_default() {
-        assert_eq!(
-            resolve_setting(&["debug".to_string()], Some("error".to_string()), "info"),
-            "debug"
-        );
-        assert_eq!(
-            resolve_setting(&[], Some("json".to_string()), "text"),
-            "json"
-        );
-        assert_eq!(resolve_setting(&[], None, "info"), "info");
-    }
-}

src/daemon/mod.rs

@@ -2,10 +2,10 @@
 //! Reuses its atomic status/JSONL writers, serial loop and retention model.
 //! The adapter runs the public validate command with durable protocol state.
 mod metrics;
-mod process;
+pub(crate) mod process;
 mod storage;
 
-use crate::cli::{Options, VALIDATE_OPTIONS, VALIDATE_REPEATABLE};
+use crate::cli::config::Config as RuntimeConfig;
 use crate::logging::{Level, emit};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
@@ -15,20 +15,6 @@ use std::process::{Command, Stdio};
 use std::time::Instant;
 use storage::{append_json_line, write_json_pretty};
 
-const OWNED: &[&str] = &["out", "ccr-out", "rrdp-state-dir", "rrdp-sync-mode"];
-const CONTROLLER: &[&str] = &[
-    "state-root",
-    "interval-secs",
-    "max-runs",
-    "retain-runs",
-    "run-timeout-secs",
-    "shutdown-grace-secs",
-];
-
-pub(crate) fn usage() -> &'static str {
-    "DAEMON:\n  panda-rpki daemon --state-root <dir> [--interval-secs <n>] [--max-runs <n>] [--retain-runs <n>] [--run-timeout-secs <n>] [--shutdown-grace-secs <n>] -- --tal <file> --ta <file> [validation options]\n\nDefaults: interval=60s after each run, unlimited runs, retain=10, timeout=0 (disabled), shutdown grace=30s. Reuses protocol state across runs/restarts; automatically exports CCR and CSV per run. Output/state/sync-mode options are controller-owned."
-}
-
 struct Config {
     root: PathBuf,
     interval: u64,
@@ -36,86 +22,20 @@ struct Config {
     retain: usize,
     timeout: u64,
     grace: u64,
-    child_args: Vec<String>,
-    log_level: String,
-    log_format: String,
+    runtime: RuntimeConfig,
 }
-
-fn parse(args: &[String]) -> Result<Config, String> {
-    let split = args
-        .iter()
-        .position(|arg| arg == "--")
-        .ok_or("daemon requires -- before validation options")?;
-    let options = Options::parse(&args[..split], CONTROLLER, &[])?;
-    let child_args = args[split + 1..].to_vec();
-    let child = Options::parse(&child_args, VALIDATE_OPTIONS, VALIDATE_REPEATABLE)?;
-    for name in OWNED {
-        if !child.values(name).is_empty() {
-            return Err(format!("daemon manages --{name}; use --state-root"));
-        }
+impl Config {
+    fn new(runtime: RuntimeConfig) -> Result<Self, String> {
+        Ok(Self {
+            root: runtime.state_root.clone().ok_or("state-root required")?,
+            interval: runtime.interval_secs,
+            max_runs: runtime.max_runs,
+            retain: runtime.retain_runs,
+            timeout: runtime.run_timeout_secs,
+            grace: runtime.shutdown_grace_secs,
+            runtime,
+        })
     }
-    let count = child.values("tal").len();
-    if count == 0 || child.values("ta").len() != count {
-        return Err("--tal and --ta must be repeated as equal-length positional pairs".into());
-    }
-    if !child.values("tal-id").is_empty() && child.values("tal-id").len() != count {
-        return Err("--tal-id must be supplied once for each --tal when used".into());
-    }
-    for name in [
-        "parallel-phase2-object-workers",
-        "parallel-max-repo-sync-workers-global",
-        "parallel-phase2-worker-queue-capacity",
-        "parallel-repo-worker-queue-capacity",
-    ] {
-        if child.parse_or(name, 1usize)? == 0 {
-            return Err(format!("--{name} must be greater than zero"));
-        }
-    }
-    child.parse_or("max-ca-depth", 64usize)?;
-    child.parse_or("http-timeout-secs", 300u64)?;
-    for (name, variable, default) in [
-        ("log-level", "PANDA_RPKI_LOG_LEVEL", "info"),
-        ("log-format", "PANDA_RPKI_LOG_FORMAT", "text"),
-    ] {
-        let value = child
-            .values(name)
-            .first()
-            .cloned()
-            .or_else(|| std::env::var(variable).ok())
-            .unwrap_or(default.into());
-        if name == "log-level" {
-            crate::logging::parse_level(&value)?;
-        } else if !["text", "json"].contains(&value.as_str()) {
-            return Err(format!("invalid log format: {value}"));
-        }
-    }
-    let level = child
-        .values("log-level")
-        .first()
-        .cloned()
-        .or_else(|| std::env::var("PANDA_RPKI_LOG_LEVEL").ok())
-        .unwrap_or("info".into());
-    let format = child
-        .values("log-format")
-        .first()
-        .cloned()
-        .or_else(|| std::env::var("PANDA_RPKI_LOG_FORMAT").ok())
-        .unwrap_or("text".into());
-    let config = Config {
-        root: PathBuf::from(options.required("state-root")?),
-        interval: options.parse_or("interval-secs", 60u64)?,
-        max_runs: options.optional_parse("max-runs")?,
-        retain: options.parse_or("retain-runs", 10usize)?,
-        timeout: options.parse_or("run-timeout-secs", 0u64)?,
-        grace: options.parse_or("shutdown-grace-secs", 30u64)?,
-        child_args,
-        log_level: level,
-        log_format: format,
-    };
-    if config.root.as_os_str().is_empty() || config.max_runs == Some(0) || config.retain == 0 {
-        return Err("state-root must be non-empty; max-runs and retain-runs must be > 0".into());
-    }
-    Ok(config)
 }
 
 #[derive(Default, Serialize, Deserialize)]
@@ -187,9 +107,9 @@ fn apply_retention(runs: &Path, retain: usize) -> Result<Vec<String>, String> {
             && let Ok(summary) = read_json(&entry.path().join("run-summary.json"))
             && summary["controller"] == "panda-rpki-daemon-v1"
             && summary["run_seq"] == seq
-            && ["success", "failed"]
+            && ["success", "partial", "failed"]
                 .iter()
-                .any(|s| summary["status"] == *s)
+                .any(|s| summary["status"] == *s || summary["resultStatus"] == *s)
         {
             dirs.push((seq, entry.path(), name));
         }
@@ -197,7 +117,14 @@ fn apply_retention(runs: &Path, retain: usize) -> Result<Vec<String>, String> {
     dirs.sort_by_key(|x| x.0);
     let remove_count = dirs.len().saturating_sub(retain);
     let mut removed = Vec::new();
+    let latest = runs
+        .parent()
+        .and_then(|p| fs::read_link(p.join("latest")).ok())
+        .and_then(|p| p.file_name().map(|s| s.to_os_string()));
     for (_, path, name) in dirs.into_iter().take(remove_count) {
+        if latest.as_deref() == path.file_name() {
+            continue;
+        }
         fs::remove_dir_all(&path).map_err(|e| format!("remove old run {}: {e}", path.display()))?;
         removed.push(name);
     }
@@ -214,20 +141,8 @@ fn status(root: &Path, state: &str, seq: u64, completed: u64) -> Result<(), Stri
     )
 }
 
-pub(crate) fn run(args: &[String]) -> Result<(), String> {
-    if args.iter().any(|a| a == "--help" || a == "-h") {
-        return Err(crate::cli::usage());
-    }
-    let mut config = parse(args)?;
-    let progress_interval =
-        std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").unwrap_or_else(|_| "10".into());
-    Options::parse(&config.child_args, VALIDATE_OPTIONS, VALIDATE_REPEATABLE)?
-        .values("log-progress-interval-secs")
-        .first()
-        .unwrap_or(&progress_interval)
-        .parse::<u64>()
-        .map_err(|_| "invalid log-progress-interval-secs".to_string())?;
-    crate::logging::configure(&config.log_level, &config.log_format)?;
+pub(crate) fn run(runtime: RuntimeConfig) -> Result<i32, String> {
+    let mut config = Config::new(runtime)?;
     emit(Level::Info, "daemon_started", || {
         json!({"interval_secs": config.interval,
         "retain_runs": config.retain, "run_timeout_secs": config.timeout, "state_root": config.root})
@@ -256,7 +171,7 @@ pub(crate) fn run(args: &[String]) -> Result<(), String> {
     result
 }
 
-fn controller(config: &Config) -> Result<(), String> {
+fn controller(config: &Config) -> Result<i32, String> {
     let root = &config.root;
     let runs = root.join("runs");
     let state = root.join("state");
@@ -280,6 +195,7 @@ fn controller(config: &Config) -> Result<(), String> {
     let executable = std::env::current_exe().map_err(|e| e.to_string())?;
     let mut completed = 0u64;
     let mut failures = 0u64;
+    let mut partials = 0u64;
     status(root, "starting", lifecycle.last_seq, completed)?;
     emit(
         Level::Info,
@@ -333,18 +249,17 @@ fn controller(config: &Config) -> Result<(), String> {
         command.env("LC_ALL", "C");
         command
             .env("PANDA_RPKI_RUN_ID", &run_id)
+            .env("PANDA_RPKI_INTERNAL_CONTEXT", "daemon")
             .env("PANDA_RPKI_RUN_SEQ", seq.to_string());
+        let config_path = out.join("validator-config.toml");
+        crate::cli::publication::write_atomic(
+            &config_path,
+            config.runtime.child_config(&out, &state, mode)?.as_bytes(),
+        )?;
         command
             .arg("validate")
-            .args(&config.child_args)
-            .arg("--out")
-            .arg(&out)
-            .arg("--ccr-out")
-            .arg(out.join("result.ccr"))
-            .arg("--rrdp-state-dir")
-            .arg(&state)
-            .arg("--rrdp-sync-mode")
-            .arg(mode)
+            .arg("--config")
+            .arg(&config_path)
             .stdin(Stdio::null())
             .stdout(File::create(out.join("process-output.txt")).map_err(|e| e.to_string())?)
             .stderr(File::create(out.join("validator.log")).map_err(|e| e.to_string())?);
@@ -352,15 +267,27 @@ fn controller(config: &Config) -> Result<(), String> {
         let (exit_code, failure) = match result {
             Ok((exit, reason)) => (
                 exit.code(),
-                reason
-                    .map(str::to_string)
-                    .or_else(|| (!exit.success()).then(|| format!("validator exited: {exit}"))),
+                reason.map(str::to_string).or_else(|| {
+                    (!exit.success() && exit.code() != Some(3))
+                        .then(|| format!("validator exited: {exit}"))
+                }),
             ),
             Err(error) => (None, Some(error)),
         };
         let summary = read_json(&out.join("summary.json"));
         let failure = failure.or_else(|| summary.as_ref().err().cloned());
-        let success = failure.is_none();
+        let result_status = if failure.is_some() {
+            "failed"
+        } else {
+            summary
+                .as_ref()
+                .ok()
+                .and_then(|v| v["resultStatus"].as_str())
+                .unwrap_or("failed")
+        };
+        let success = result_status == "success";
+        let partial = result_status == "partial";
+        meta["resultStatus"] = json!(result_status);
         meta["status"] = json!(if success { "success" } else { "failed" });
         meta["finished_at"] = json!(now());
         meta["wall_ms"] = json!(start.elapsed().as_millis() as u64);
@@ -376,7 +303,25 @@ fn controller(config: &Config) -> Result<(), String> {
         lifecycle.last_success = success;
         write_json_pretty(&lifecycle_path, &lifecycle)?;
         completed += 1;
-        failures += u64::from(!success);
+        failures += u64::from(!success && !partial);
+        partials += u64::from(partial);
+        if success {
+            crate::cli::publication::switch_latest(
+                root,
+                &Path::new("runs").join(out.file_name().unwrap()),
+            )?;
+        }
+        crate::cli::publication::attempt(
+            root,
+            &run_id,
+            if success {
+                "success"
+            } else if partial {
+                "partial"
+            } else {
+                "failed"
+            },
+        )?;
         let removed = apply_retention(&runs, config.retain)?;
         if !removed.is_empty() {
             emit(
@@ -407,9 +352,13 @@ fn controller(config: &Config) -> Result<(), String> {
         || json!({"runs_completed": completed, "failed_runs": failures, "stop_requested": process::stopping()}),
     );
     if failures > 0 {
-        Err(format!("daemon completed with {failures} failed run(s)"))
+        Ok(2)
     } else {
-        Ok(())
+        Ok(if config.runtime.require_complete && partials > 0 {
+            3
+        } else {
+            0
+        })
     }
 }
 

src/daemon/process.rs

@@ -39,19 +39,24 @@ impl Drop for Signals {
     }
 }
 
-struct ChildGuard(Child);
+struct ChildGuard {
+    child: Child,
+    reaped: bool,
+}
 impl ChildGuard {
     #[allow(unsafe_code)]
     fn signal_group(&self, signal: libc::c_int) {
         // SAFETY: child is launched in its own group, with pgid equal to child PID.
-        unsafe { libc::kill(-(self.0.id() as libc::pid_t), signal) };
+        unsafe { libc::kill(-(self.child.id() as libc::pid_t), signal) };
     }
 }
 impl Drop for ChildGuard {
     fn drop(&mut self) {
         // Reap every child and terminate residual descendants on all error paths.
-        self.signal_group(libc::SIGKILL);
-        let _ = self.0.wait();
+        if !self.reaped {
+            self.signal_group(libc::SIGKILL);
+            let _ = self.child.wait();
+        }
     }
 }
 
@@ -62,16 +67,17 @@ pub(super) fn execute(
 ) -> Result<(ExitStatus, Option<&'static str>), String> {
     use std::os::unix::process::CommandExt;
     command.process_group(0);
-    let mut child = ChildGuard(
-        command
+    let mut child = ChildGuard {
+        child: command
             .spawn()
             .map_err(|e| format!("spawn validator: {e}"))?,
-    );
+        reaped: false,
+    };
     let start = Instant::now();
     crate::logging::emit(
         crate::logging::Level::Info,
         "daemon_child_started",
-        || serde_json::json!({"pid": child.0.id()}),
+        || serde_json::json!({"pid": child.child.id()}),
     );
     let mut stop_at = None;
     let mut terminated_at = None;
@@ -79,10 +85,11 @@ pub(super) fn execute(
     let mut killed = false;
     loop {
         if let Some(status) = child
-            .0
+            .child
             .try_wait()
             .map_err(|e| format!("wait validator: {e}"))?
         {
+            child.reaped = true;
             return Ok((status, reason));
         }
         if stopping() {
@@ -126,3 +133,71 @@ pub(super) fn sleep(seconds: u64) {
         std::thread::sleep(Duration::from_millis(100));
     }
 }
+
+/// Fork before starting validator worker threads; supervise the whole process group.
+#[allow(unsafe_code)]
+pub(crate) fn supervise(
+    timeout: u64,
+    run: impl FnOnce() -> Result<i32, String>,
+) -> Result<i32, String> {
+    let _signals = Signals::install()?;
+    // SAFETY: called only by the single-threaded CLI before validation workers start.
+    let pid = unsafe { libc::fork() };
+    if pid < 0 {
+        return Err(std::io::Error::last_os_error().to_string());
+    }
+    if pid == 0 {
+        unsafe {
+            libc::setpgid(0, 0);
+            libc::signal(libc::SIGTERM, libc::SIG_DFL);
+            libc::signal(libc::SIGINT, libc::SIG_DFL);
+        }
+        let code = match run() {
+            Ok(c) => c,
+            Err(e) => {
+                eprintln!("panda-rpki: {e}");
+                2
+            }
+        };
+        unsafe { libc::_exit(code) };
+    }
+    unsafe {
+        libc::setpgid(pid, pid);
+    }
+    let start = Instant::now();
+    let mut terminated = None;
+    let mut status = 0;
+    loop {
+        let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) };
+        if waited == pid {
+            if terminated.is_some() {
+                return Err("validation timed out or was interrupted".into());
+            }
+            return Ok(if libc::WIFEXITED(status) {
+                libc::WEXITSTATUS(status)
+            } else {
+                2
+            });
+        }
+        if waited < 0 {
+            let e = std::io::Error::last_os_error();
+            if e.kind() != std::io::ErrorKind::Interrupted {
+                return Err(e.to_string());
+            }
+        }
+        if terminated.is_none()
+            && (stopping() || (timeout > 0 && start.elapsed().as_secs() >= timeout))
+        {
+            unsafe {
+                libc::kill(-pid, libc::SIGTERM);
+            }
+            terminated = Some(Instant::now());
+        }
+        if terminated.is_some_and(|t| t.elapsed() >= Duration::from_secs(2)) {
+            unsafe {
+                libc::kill(-pid, libc::SIGKILL);
+            }
+        }
+        std::thread::sleep(Duration::from_millis(25));
+    }
+}

src/daemon/tests.rs

@@ -1,113 +1,76 @@
 use super::*;
 
-fn args(extra: &[&str]) -> Vec<String> {
-    [
-        vec!["--state-root", "data"],
-        extra.to_vec(),
-        vec!["--", "--tal", "a.tal", "--ta", "a.cer"],
-    ]
-    .concat()
-    .into_iter()
-    .map(str::to_owned)
-    .collect()
-}
-
 #[test]
-fn defaults_and_invalid_controller_options() {
-    let config = parse(&args(&[])).unwrap();
-    assert_eq!(
-        (
-            config.interval,
-            config.max_runs,
-            config.retain,
-            config.timeout,
-            config.grace
-        ),
-        (60, None, 10, 0, 30)
-    );
-    for bad in [
-        vec!["--max-runs", "0"],
-        vec!["--retain-runs", "0"],
-        vec!["--interval-secs", "-1"],
-        vec!["--run-timeout-secs", "bad"],
-        vec!["--interval-secs", "2", "--interval-secs", "3"],
-    ] {
-        assert!(parse(&args(&bad)).is_err());
-    }
-    let config = parse(&args(&["--interval-secs", "0", "--max-runs", "3"])).unwrap();
-    assert_eq!((config.interval, config.max_runs), (0, Some(3)));
-}
-
-#[test]
-fn rejects_child_overrides_and_unknown_flags() {
-    for flag in [
-        "--out",
-        "--rrdp-state-dir",
-        "--rrdp-sync-mode",
-        "--ccr-out",
-        "--unknown",
-    ] {
-        let mut argv = args(&[]);
-        argv.extend([flag.into(), "some-value".into()]);
-        assert!(parse(&argv).is_err(), "{flag}");
-    }
-    let mut argv = args(&[]);
-    argv.extend(["--parallel-phase2-object-workers".into(), "0".into()]);
-    assert!(parse(&argv).is_err());
-}
-
-#[test]
-fn protocol_state_selects_snapshot_delta_and_recovery() {
+fn sync_mode_selects_snapshot_delta_and_recovery() {
     let root = tempfile::tempdir().unwrap();
     assert_eq!(sync_mode(root.path(), false).unwrap(), "snapshot");
+
     let db = root.path().join("repository-db");
     fs::create_dir(&db).unwrap();
     assert_eq!(sync_mode(root.path(), true).unwrap(), "snapshot");
+
     fs::write(db.join("CURRENT"), b"x").unwrap();
     assert_eq!(sync_mode(root.path(), true).unwrap(), "delta");
     assert_eq!(sync_mode(root.path(), false).unwrap(), "auto");
 }
 
 #[test]
-fn retention_preserves_unrelated_incomplete_and_symlinked_runs() {
+fn retention_keeps_incomplete_and_latest_runs() {
     let root = tempfile::tempdir().unwrap();
-    for seq in 1..=3 {
-        let path = root.path().join(format!("run_{seq:06}"));
+    let runs = root.path().join("runs");
+    fs::create_dir(&runs).unwrap();
+    for (seq, result_status) in [(1, "success"), (2, "partial"), (3, "failed")] {
+        let path = runs.join(format!("run_{seq:06}"));
         fs::create_dir(&path).unwrap();
         write_json_pretty(
             &path.join("run-summary.json"),
             &json!({
-                "controller": "panda-rpki-daemon-v1", "run_seq": seq, "status": "success"
+                "controller": "panda-rpki-daemon-v1",
+                "run_seq": seq,
+                "status": if result_status == "success" { "success" } else { "failed" },
+                "resultStatus": result_status,
             }),
         )
         .unwrap();
     }
-    fs::create_dir(root.path().join("unrelated")).unwrap();
-    fs::create_dir(root.path().join("run_000004")).unwrap();
-    let outside = tempfile::tempdir().unwrap();
-    std::os::unix::fs::symlink(outside.path(), root.path().join("run_000005")).unwrap();
-    assert_eq!(
-        apply_retention(root.path(), 1).unwrap(),
-        ["run_000001", "run_000002"]
-    );
-    for name in ["run_000003", "run_000004", "run_000005", "unrelated"] {
-        assert!(root.path().join(name).exists());
-    }
-    assert!(regular_directory(&root.path().join("run_000005")).is_err());
+    std::os::unix::fs::symlink("runs/run_000001", root.path().join("latest")).unwrap();
+
+    // Only completed runs are candidates; partial is completed evidence and is retained
+    // alongside the latest pointer even when the retention limit is one.
+    assert_eq!(apply_retention(&runs, 1).unwrap(), ["run_000002"]);
+    assert!(runs.join("run_000001").exists());
+    assert!(!runs.join("run_000002").exists());
+    assert!(runs.join("run_000003").exists());
 }
 
 #[test]
-fn atomic_json_writers_and_state_lock() {
+fn retention_ignores_unrelated_paths_and_symlinks() {
+    let root = tempfile::tempdir().unwrap();
+    let runs = root.path().join("runs");
+    fs::create_dir(&runs).unwrap();
+    let path = runs.join("run_000001");
+    fs::create_dir(&path).unwrap();
+    write_json_pretty(
+        &path.join("run-summary.json"),
+        &json!({"controller": "other", "run_seq": 1, "status": "success"}),
+    )
+    .unwrap();
+    fs::create_dir(runs.join("unrelated")).unwrap();
+
+    let outside = tempfile::tempdir().unwrap();
+    std::os::unix::fs::symlink(outside.path(), runs.join("run_000002")).unwrap();
+
+    assert!(apply_retention(&runs, 0).unwrap().is_empty());
+    assert!(runs.join("run_000001").exists());
+    assert!(runs.join("run_000002").exists());
+    assert!(runs.join("unrelated").exists());
+}
+
+#[test]
+fn atomic_json_writer_replaces_content() {
     let root = tempfile::tempdir().unwrap();
     let path = root.path().join("state.json");
     write_json_pretty(&path, &json!({"seq": 1})).unwrap();
     write_json_pretty(&path, &json!({"seq": 2})).unwrap();
     assert_eq!(read_json(&path).unwrap()["seq"], 2);
-    let lines = root.path().join("summary.jsonl");
-    append_json_line(&lines, &json!({"seq": 1})).unwrap();
-    append_json_line(&lines, &json!({"seq": 2})).unwrap();
-    assert_eq!(fs::read_to_string(lines).unwrap().lines().count(), 2);
-    let file = File::open(&path).unwrap();
-    file.try_lock().unwrap();
-    assert!(File::open(path).unwrap().try_lock().is_err());
 }

src/main.rs

@@ -1,16 +1,10 @@
 fn main() {
-    match panda_rpki::cli::run(std::env::args().skip(1)) {
-        Ok(()) => {}
-        Err(error) if error == panda_rpki::cli::usage() => {
-            println!("{error}");
-        }
+    let code = match panda_rpki::cli::run(std::env::args().skip(1)) {
+        Ok(code) => code,
         Err(error) => {
-            panda_rpki::logging::emit(
-                panda_rpki::logging::Level::Error,
-                "command_failed",
-                || serde_json::json!({"error": error}),
-            );
-            std::process::exit(2);
+            eprintln!("panda-rpki: {error}");
+            2
         }
-    }
+    };
+    std::process::exit(code);
 }

src/repository/sync/rrdp/notification_sync.rs

@@ -106,15 +106,22 @@ fn sync_from_notification_inner(
         .as_ref()
         .filter(|s| s.session_id == notif.session_id.to_string());
     let log_selection = |mode: &str, reason: &str| {
-        crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({
-            "notify_uri": notification_uri, "mode": mode, "reason": reason,
-            "session_id": notif.session_id.to_string(), "serial": notif.serial,
-            "previous_serial": state.as_ref().map(|previous| previous.serial),
-            "previous_session_id": state.as_ref().map(|previous| &previous.session_id),
-        }));
+        crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || {
+            serde_json::json!({
+                "notify_uri": notification_uri, "mode": mode, "reason": reason,
+                "session_id": notif.session_id.to_string(), "serial": notif.serial,
+                "previous_serial": state.as_ref().map(|previous| previous.serial),
+                "previous_session_id": state.as_ref().map(|previous| &previous.session_id),
+            })
+        });
+    };
+    let mut snapshot_reason = if state.is_none() {
+        "no_local_state"
+    } else if same_session_state.is_none() {
+        "session_changed"
+    } else {
+        "delta_chain_unavailable"
     };
-    let mut snapshot_reason = if state.is_none() { "no_local_state" }
-        else if same_session_state.is_none() { "session_changed" } else { "delta_chain_unavailable" };
 
     if let Some(s) = same_session_state {
         if s.serial == notif.serial {
@@ -191,9 +198,12 @@ fn sync_from_notification_inner(
                         &notification_origin,
                     ) {
                         Ok(bytes) => {
-                            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
-                                "resource": "delta", "bytes": bytes.len(), "notify_uri": notification_uri,
-                            }));
+                            crate::logging::progress::emit!(
+                                "rrdp_download_completed",
+                                serde_json::json!({
+                                    "resource": "delta", "bytes": bytes.len(), "notify_uri": notification_uri,
+                                })
+                            );
                             if let Some(t) = timing.as_ref() {
                                 t.record_count("rrdp_delta_fetch_ok_total", 1);
                                 t.record_count("rrdp_delta_bytes_total", bytes.len() as u64);
@@ -243,7 +253,11 @@ fn sync_from_notification_inner(
                             Ok(n) => applied_total += n,
                             Err(error) => {
                                 snapshot_reason = "delta_apply_failed";
-                                crate::logging::emit(crate::logging::Level::Warn, "rrdp_delta_fallback", || serde_json::json!({"notify_uri": notification_uri, "error": error.to_string(), "serial": serial}));
+                                crate::logging::emit(
+                                    crate::logging::Level::Warn,
+                                    "rrdp_delta_fallback",
+                                    || serde_json::json!({"notify_uri": notification_uri, "error": error.to_string(), "serial": serial}),
+                                );
                                 ok = false;
                                 break;
                             }
@@ -320,9 +334,12 @@ fn sync_from_notification_inner(
         &notification_origin,
     ) {
         Ok(v) => {
-            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
-                "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
-            }));
+            crate::logging::progress::emit!(
+                "rrdp_download_completed",
+                serde_json::json!({
+                    "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
+                })
+            );
             if let Some(t) = timing.as_ref() {
                 t.record_count("rrdp_snapshot_fetch_ok_total", 1);
                 t.record_count("rrdp_snapshot_bytes_total", v.1);

src/repository/sync/rrdp/snapshot_sync.rs

@@ -80,10 +80,12 @@ fn sync_from_notification_snapshot_inner(
     };
     drop(_parse_step);
     drop(_parse_total);
-    crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({
-        "notify_uri": notification_uri, "mode": "snapshot", "reason": "snapshot_requested",
-        "session_id": notif.session_id.to_string(), "serial": notif.serial,
-    }));
+    crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || {
+        serde_json::json!({
+            "notify_uri": notification_uri, "mode": "snapshot", "reason": "snapshot_requested",
+            "session_id": notif.session_id.to_string(), "serial": notif.serial,
+        })
+    });
 
     let _fetch_step = timing
         .as_ref()
@@ -100,9 +102,12 @@ fn sync_from_notification_snapshot_inner(
         &notification_origin,
     ) {
         Ok(v) => {
-            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
-                "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
-            }));
+            crate::logging::progress::emit!(
+                "rrdp_download_completed",
+                serde_json::json!({
+                    "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
+                })
+            );
             if let Some(t) = timing.as_ref() {
                 t.record_count("rrdp_snapshot_fetch_ok_total", 1);
                 t.record_count("rrdp_snapshot_bytes_total", v.1);

src/runtime/post_validation.rs

@@ -40,26 +40,62 @@ fn print_summary_from_shared(validation_time: time::OffsetDateTime, shared: &Pos
         })
     });
     if warning_count > 0 {
-        let mut categories: std::collections::BTreeMap<&str, (usize, Vec<serde_json::Value>)> = std::collections::BTreeMap::new();
-        let warnings = shared.tree_warnings.iter().map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref()))
-            .chain(shared.publication_points.iter().flat_map(|pp| pp.warnings.iter()).map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref())));
+        let mut categories: std::collections::BTreeMap<&str, (usize, Vec<serde_json::Value>)> =
+            std::collections::BTreeMap::new();
+        let warnings = shared
+            .tree_warnings
+            .iter()
+            .map(|warning| {
+                (
+                    warning.category.as_str(),
+                    warning.message.as_str(),
+                    warning.context.as_deref(),
+                )
+            })
+            .chain(
+                shared
+                    .publication_points
+                    .iter()
+                    .flat_map(|pp| pp.warnings.iter())
+                    .map(|warning| {
+                        (
+                            warning.category.as_str(),
+                            warning.message.as_str(),
+                            warning.context.as_deref(),
+                        )
+                    }),
+            );
         for (category, message, context) in warnings {
             let entry = categories.entry(category).or_default();
             entry.0 += 1;
             let mut text = crate::logging::redact_text(message);
             let truncated = text.len() > 1024;
             let mut boundary = text.len().min(1024);
-            while !text.is_char_boundary(boundary) { boundary -= 1; }
+            while !text.is_char_boundary(boundary) {
+                boundary -= 1;
+            }
             text.truncate(boundary);
             if entry.1.len() < 3 {
-                entry.1.push(serde_json::json!({"message": text, "truncated": truncated}));
+                entry
+                    .1
+                    .push(serde_json::json!({"message": text, "truncated": truncated}));
             }
-            crate::logging::emit(crate::logging::Level::Debug, "validation_warning_detail", || serde_json::json!({
-                "category": category, "message": message, "context": context,
-            }));
+            crate::logging::emit(
+                crate::logging::Level::Debug,
+                "validation_warning_detail",
+                || {
+                    serde_json::json!({
+                        "category": category, "message": message, "context": context,
+                    })
+                },
+            );
         }
         for (category, (count, examples)) in categories {
-            crate::logging::emit(crate::logging::Level::Info, "validation_warning_category", || serde_json::json!({"category": category, "count": count, "examples": examples}));
+            crate::logging::emit(
+                crate::logging::Level::Info,
+                "validation_warning_category",
+                || serde_json::json!({"category": category, "count": count, "examples": examples}),
+            );
         }
         crate::logging::emit(crate::logging::Level::Warn, "validation_warnings", || {
             serde_json::json!({

src/runtime/run.rs

@@ -369,7 +369,9 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
             shared.publication_points.len() as u64,
         );
         let timing_json_path = out_dir.join("timing.json");
-        crate::logging::artifact::write("timing", &timing_json_path, || t.write_json(&timing_json_path, 20))?;
+        crate::logging::artifact::write("timing", &timing_json_path, || {
+            t.write_json(&timing_json_path, 20)
+        })?;
     }
 
     // Write the compatibility summary after analysis timing has been flushed.
@@ -378,13 +380,15 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
     // point audit, so the final summary can use those authoritative counts.
     if let Some(path) = args.summary_out_path.as_deref() {
         let analysis_counts = timing.as_ref().map(|(_, handle)| handle.counts_snapshot());
-        crate::logging::artifact::write("summary", path, || write_summary(
-            path,
-            &shared,
-            &args,
-            validation_ms,
-            analysis_counts.as_ref(),
-        ))?;
+        crate::logging::artifact::write("summary", path, || {
+            write_summary(
+                path,
+                &shared,
+                &args,
+                validation_ms,
+                analysis_counts.as_ref(),
+            )
+        })?;
     }
 
     print_summary_from_shared(validation_time, &shared);
@@ -393,6 +397,8 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
 
 #[derive(Serialize)]
 struct ValidationSummary {
+    #[serde(rename = "repositoryFailedCount")]
+    repository_failed_count: usize,
     backend: &'static str,
     worker_count: usize,
     repo_sync_worker_count: usize,
@@ -428,6 +434,17 @@ fn write_summary(
     use std::collections::BTreeSet;
 
     let mut summary = ValidationSummary {
+        repository_failed_count: shared
+            .publication_points
+            .iter()
+            .filter(|p| p.repo_sync_error.is_some())
+            .map(|p| {
+                p.rrdp_notification_uri
+                    .as_deref()
+                    .unwrap_or(&p.rsync_base_uri)
+            })
+            .collect::<BTreeSet<_>>()
+            .len(),
         backend: "panda-rpki",
         worker_count: args.parallel_phase2_config.object_workers,
         repo_sync_worker_count: args.parallel_phase1_config.max_repo_sync_workers_global,

src/ta_constraints/implementation.rs

@@ -143,25 +143,23 @@ impl TaConstraintsByTal {
         for spec in explicit_specs {
             let (tal_id, path) = spec
                 .split_once('=')
-                .ok_or_else(|| format!("--ta-constraints expects <tal-id>=<path>, got '{spec}'"))?;
+                .ok_or_else(|| format!("constraints expects <tal-id>=<path>, got '{spec}'"))?;
             let tal_id = tal_id.trim();
             let path = path.trim();
             if tal_id.is_empty() || path.is_empty() {
                 return Err(format!(
-                    "--ta-constraints expects non-empty <tal-id>=<path>, got '{spec}'"
+                    "constraints expects non-empty <tal-id>=<path>, got '{spec}'"
                 ));
             }
             if !tal_ids.contains(tal_id) {
-                return Err(format!(
-                    "--ta-constraints references unknown TAL id '{tal_id}'"
-                ));
+                return Err(format!("constraints references unknown TAL id '{tal_id}'"));
             }
             if explicit_paths
                 .insert(tal_id.to_string(), PathBuf::from(path))
                 .is_some()
             {
                 return Err(format!(
-                    "--ta-constraints specifies TAL id '{tal_id}' more than once"
+                    "constraints specifies TAL id '{tal_id}' more than once"
                 ));
             }
         }

src/validation/manifest/models_and_process.rs

@@ -212,9 +212,13 @@ impl ManifestFreshError {
         use crate::output::report::WarningCategory;
         match self {
             Self::Storage(_) | Self::RepoSyncFailed { .. } => WarningCategory::Unclassified,
-            Self::IssuerCaLoadFailed { .. } | Self::EeCertPath(_) | Self::EeResources(_) => WarningCategory::Certificate,
-            Self::ManifestOutsidePublicationPoint { .. } | Self::ManifestNumberNotIncreasing { .. }
-                | Self::ThisUpdateNotIncreasing { .. } | Self::HashMismatch { .. } => WarningCategory::SecurityPolicy,
+            Self::IssuerCaLoadFailed { .. } | Self::EeCertPath(_) | Self::EeResources(_) => {
+                WarningCategory::Certificate
+            }
+            Self::ManifestOutsidePublicationPoint { .. }
+            | Self::ManifestNumberNotIncreasing { .. }
+            | Self::ThisUpdateNotIncreasing { .. }
+            | Self::HashMismatch { .. } => WarningCategory::SecurityPolicy,
             _ => WarningCategory::ManifestCrl,
         }
     }

src/validation/run_tree_from_tal/discovery.rs

@@ -273,7 +273,11 @@ fn discover_multiple_roots_from_tal_inputs(
     let mut roots = Vec::with_capacity(tal_inputs.len());
     crate::logging::observer::phase("trust_anchor_discovery");
     for tal_input in tal_inputs {
-        crate::logging::emit(crate::logging::Level::Info, "trust_anchor_started", || serde_json::json!({"tal_id": tal_input.tal_id}));
+        crate::logging::emit(
+            crate::logging::Level::Info,
+            "trust_anchor_started",
+            || serde_json::json!({"tal_id": tal_input.tal_id}),
+        );
         let discovery = match root_discovery_from_tal_input(
             tal_input,
             http_fetcher,
@@ -295,7 +299,11 @@ fn discover_multiple_roots_from_tal_inputs(
                 continue;
             }
             Err(error) => {
-                crate::logging::emit(crate::logging::Level::Error, "trust_anchor_failed", || serde_json::json!({"tal_id": tal_input.tal_id, "error": error.to_string()}));
+                crate::logging::emit(
+                    crate::logging::Level::Error,
+                    "trust_anchor_failed",
+                    || serde_json::json!({"tal_id": tal_input.tal_id, "error": error.to_string()}),
+                );
                 return Err(error.into());
             }
         };
@@ -310,7 +318,11 @@ fn discover_multiple_roots_from_tal_inputs(
             discovery,
             root_handle,
         });
-        crate::logging::emit(crate::logging::Level::Info, "trust_anchor_completed", || serde_json::json!({"tal_id": tal_input.tal_id}));
+        crate::logging::emit(
+            crate::logging::Level::Info,
+            "trust_anchor_completed",
+            || serde_json::json!({"tal_id": tal_input.tal_id}),
+        );
     }
     if roots.is_empty() {
         return Err(RunTreeFromTalError::Setup(

src/validation/tree_parallel/phase2.rs

@@ -164,17 +164,19 @@ pub fn run_tree_parallel_phase2_audit_multi_root(
                     config,
                 );
 
-                crate::logging::observer::update(finished.len() as u64, || serde_json::json!({
-                    "publication_points_completed": finished.len(),
-                    "publication_points_started": instances_started,
-                    "ca_queue": ca_queue.len(), "ready_queue": ready_queue.len(),
-                    "waiting_repository_identities": ca_waiting_repo_by_identity.len(),
-                    "publication_points_inflight": inflight_publication_points.len(),
-                    "pending_object_dispatch": pending_roa_dispatch.len(),
-                    "pending_finalization": pending_finalization.len(),
-                    "finalization_inflight": finalize_inflight,
-                    "staging_inflight": staging_inflight,
-                }));
+                crate::logging::observer::update(finished.len() as u64, || {
+                    serde_json::json!({
+                        "publication_points_completed": finished.len(),
+                        "publication_points_started": instances_started,
+                        "ca_queue": ca_queue.len(), "ready_queue": ready_queue.len(),
+                        "waiting_repository_identities": ca_waiting_repo_by_identity.len(),
+                        "publication_points_inflight": inflight_publication_points.len(),
+                        "pending_object_dispatch": pending_roa_dispatch.len(),
+                        "pending_finalization": pending_finalization.len(),
+                        "finalization_inflight": finalize_inflight,
+                        "staging_inflight": staging_inflight,
+                    })
+                });
                 let repo_poll_timeout = event_poll_timeout(
                     &ca_queue,
                     &ready_queue,

src/validation/tree_runner/audit_projection.rs

@@ -127,7 +127,9 @@ fn build_publication_point_audit_from_failed_fetch(
         .map(AuditWarning::from)
         .collect::<Vec<_>>();
     warnings.push(AuditWarning::from(
-        &Warning::new(fresh_error.to_string()).with_category(fresh_error.warning_category()).with_context(&ca.manifest_rsync_uri),
+        &Warning::new(fresh_error.to_string())
+            .with_category(fresh_error.warning_category())
+            .with_context(&ca.manifest_rsync_uri),
     ));
     PublicationPointAudit {
         node_id: None,