diff --git a/scripts/soak/portable-soak.env.example b/scripts/soak/portable-soak.env.example index 8bbfb17..766e3d6 100644 --- a/scripts/soak/portable-soak.env.example +++ b/scripts/soak/portable-soak.env.example @@ -66,6 +66,14 @@ PERIODIC_SNAPSHOT_MAX_DELTAS=100 # 每隔多少轮执行一次 db_stats --exact。设置为空或 0 表示关闭 exact 统计。 DB_STATS_EXACT_EVERY=3 +# 死 repo 传输层黑名单（#141，默认关）。设置状态文件路径即开启： +# 连续 N 轮传输层抓取失败的 repo 会被拉黑并在后续轮次跳过其传输尝试； +# daemon 会按间隔对黑名单条目做健康探针，恢复后自动摘除。 +# 留空表示完全关闭（行为与旧版本一致）。 +#DEAD_REPO_BLACKLIST_PATH=/var/lib/ours-rp/state/dead-repo-blacklist.json +#DEAD_REPO_BLACKLIST_FAIL_THRESHOLD=3 +#DEAD_REPO_HEALTH_CHECK_INTERVAL_SECS=600 + # 是否开启 ours RP progress log。1 表示开启。 RPKI_PROGRESS_LOG=1 diff --git a/scripts/soak/run_soak.sh b/scripts/soak/run_soak.sh index 350be4f..0d061bf 100755 --- a/scripts/soak/run_soak.sh +++ b/scripts/soak/run_soak.sh @@ -1331,6 +1331,17 @@ run_one_round() { daemon_args+=(--db-stats-exact-every "$DB_STATS_EXACT_EVERY") fi fi + # Dead-repo transport blacklist (#141): opt-in via env; the daemon forwards + # the enable flag and threshold to the child unless already present there. + if [[ -n "${DEAD_REPO_BLACKLIST_PATH:-}" ]]; then + daemon_args+=(--dead-repo-blacklist "$DEAD_REPO_BLACKLIST_PATH") + if [[ -n "${DEAD_REPO_BLACKLIST_FAIL_THRESHOLD:-}" ]]; then + daemon_args+=(--dead-repo-blacklist-fail-threshold "$DEAD_REPO_BLACKLIST_FAIL_THRESHOLD") + fi + if [[ -n "${DEAD_REPO_HEALTH_CHECK_INTERVAL_SECS:-}" ]]; then + daemon_args+=(--dead-repo-health-check-interval-secs "$DEAD_REPO_HEALTH_CHECK_INTERVAL_SECS") + fi + fi set +e env \ diff --git a/src/cli.rs b/src/cli.rs index a902a48..61be2f1 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -279,6 +279,13 @@ Options: --control-plane-stage-workers Experimental: Phase 2 ready publication point stage worker count; 0 disables the stage pool and keeps inline staging (default: 0) + --dead-repo-blacklist + Enable the dead-repo transport blacklist persisted at this JSON + path (default: disabled). Blacklisted rrdp repos skip straight to + rsync; dual-blacklisted repos terminate instantly. + --dead-repo-blacklist-fail-threshold + Consecutive runs with transport-class fetch failure before a + (repo, transport) entry is blacklisted (default: 3) --rsync-local-dir Use LocalDirRsyncFetcher rooted at this directory (offline tests) --disable-rrdp Disable RRDP and synchronize only via rsync @@ -313,6 +320,8 @@ pub fn parse_args(argv: &[String]) -> Result { let mut ta_paths: Vec = Vec::new(); let mut parallel_phase1_cfg = ParallelPhase1Config::default(); let mut parallel_phase2_cfg = ParallelPhase2Config::default(); + let mut dead_repo_blacklist_path: Option = None; + let mut dead_repo_blacklist_fail_threshold: Option = None; let mut db_path: Option = None; let mut raw_store_db: Option = None; @@ -531,6 +540,23 @@ pub fn parse_args(argv: &[String]) -> Result { .parse::() .map_err(|_| format!("invalid --control-plane-stage-workers: {v}"))?; } + "--dead-repo-blacklist" => { + i += 1; + let v = argv + .get(i) + .ok_or("--dead-repo-blacklist requires a value")?; + dead_repo_blacklist_path = Some(PathBuf::from(v)); + } + "--dead-repo-blacklist-fail-threshold" => { + i += 1; + let v = argv + .get(i) + .ok_or("--dead-repo-blacklist-fail-threshold requires a value")?; + dead_repo_blacklist_fail_threshold = Some( + v.parse::() + .map_err(|_| format!("invalid --dead-repo-blacklist-fail-threshold: {v}"))?, + ); + } "--db" => { i += 1; let v = argv.get(i).ok_or("--db requires a value")?; @@ -1122,6 +1148,25 @@ pub fn parse_args(argv: &[String]) -> Result { } } + if dead_repo_blacklist_fail_threshold.is_some() && dead_repo_blacklist_path.is_none() { + return Err( + "--dead-repo-blacklist-fail-threshold requires --dead-repo-blacklist".to_string() + ); + } + if let Some(path) = dead_repo_blacklist_path { + let fail_threshold = dead_repo_blacklist_fail_threshold.unwrap_or(3); + if fail_threshold == 0 { + return Err("--dead-repo-blacklist-fail-threshold must be >= 1".to_string()); + } + parallel_phase1_cfg.dead_repo_blacklist = + Some(crate::parallel::dead_repo_blacklist::DeadRepoBlacklistConfig { + path, + fail_threshold, + capacity: + crate::parallel::dead_repo_blacklist::DEAD_REPO_BLACKLIST_DEFAULT_CAPACITY, + }); + } + Ok(CliArgs { verification_only, verification_source_run_dir, diff --git a/src/parallel/config.rs b/src/parallel/config.rs index 52bec45..a4d8ff1 100644 --- a/src/parallel/config.rs +++ b/src/parallel/config.rs @@ -1,8 +1,13 @@ +use crate::parallel::dead_repo_blacklist::DeadRepoBlacklistConfig; + #[derive(Clone, Debug, PartialEq, Eq)] pub struct ParallelPhase1Config { pub max_repo_sync_workers_global: usize, pub max_inflight_snapshot_bytes_global: usize, pub max_pending_repo_results: usize, + /// Dead-repo transport blacklist (#141). `None` disables the feature + /// entirely (default, behavior unchanged). + pub dead_repo_blacklist: Option, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -42,6 +47,7 @@ impl Default for ParallelPhase1Config { max_repo_sync_workers_global: 4, max_inflight_snapshot_bytes_global: 512 * 1024 * 1024, max_pending_repo_results: 1024, + dead_repo_blacklist: None, } } } diff --git a/src/parallel/dead_repo_blacklist.rs b/src/parallel/dead_repo_blacklist.rs new file mode 100644 index 0000000..f02fe73 --- /dev/null +++ b/src/parallel/dead_repo_blacklist.rs @@ -0,0 +1,376 @@ +//! Persistent dead-repo transport blacklist (#141). +//! +//! Tracks per-(repo, transport) consecutive transport-layer fetch failures +//! across runs. Once an entry reaches the configured failure threshold it is +//! admitted to the blacklist; the live transport scheduler then skips the +//! dead transport (rrdp-blacklisted -> straight to rsync, dual-blacklisted -> +//! immediate terminal failure). Entries are removed either by a successful +//! fetch (self-heal) or by the daemon-side health probe. + +use std::collections::HashMap; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use crate::parallel::types::{RepoTransportErrorClass, RepoTransportMode}; + +pub const DEAD_REPO_BLACKLIST_SCHEMA_VERSION: u32 = 1; +pub const DEAD_REPO_BLACKLIST_DEFAULT_CAPACITY: usize = 256; +/// Probe backoff is capped at interval * 2^PROBE_BACKOFF_CAP_SHIFT. +pub const PROBE_BACKOFF_CAP_SHIFT: u32 = 4; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DeadRepoBlacklistConfig { + pub path: PathBuf, + pub fail_threshold: u32, + pub capacity: usize, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DeadRepoBlacklistEntry { + pub transport: RepoTransportMode, + pub uri: String, + /// Consecutive runs with transport-class fetch failure. Reset on success. + pub consecutive_failures: u32, + /// True once `consecutive_failures` reached the admission threshold. + #[serde(default)] + pub blacklisted: bool, + #[serde(default)] + pub first_failure_at_unix: u64, + #[serde(default)] + pub last_failure_at_unix: u64, + /// When the entry was admitted to the blacklist. + #[serde(default)] + pub added_at_unix: u64, + #[serde(default)] + pub last_probe_at_unix: Option, + #[serde(default)] + pub probe_failures: u32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DeadRepoFailureOutcome { + /// Failure counted, entry not (or not yet) blacklisted. + Counted { consecutive_failures: u32 }, + /// Entry just reached the threshold and was admitted. + Admitted, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DeadRepoSuccessOutcome { + /// A counting (not yet admitted) entry was reset. + CounterReset, + /// A blacklisted entry was removed (repo revived). + BlacklistRemoved, +} + +#[derive(Serialize, Deserialize)] +struct DeadRepoBlacklistFile { + schema_version: u32, + updated_at_unix: u64, + #[serde(default)] + entries: Vec, +} + +/// In-memory blacklist state for one writer at a time (child run, or daemon +/// supervisor between runs). Lookups go through a HashMap keyed by +/// (transport, uri). +#[derive(Clone, Debug, Default)] +pub struct DeadRepoBlacklist { + entries: HashMap<(RepoTransportMode, String), DeadRepoBlacklistEntry>, +} + +impl DeadRepoBlacklist { + pub fn new() -> Self { + Self::default() + } + + /// Load from disk. A missing file or an unreadable/incompatible file + /// degrades to an empty blacklist (logged by the caller via `warning`). + pub fn load(path: &Path) -> (Self, Option) { + let bytes = match fs::read(path) { + Ok(bytes) => bytes, + Err(err) if err.kind() == io::ErrorKind::NotFound => return (Self::new(), None), + Err(err) => { + return ( + Self::new(), + Some(format!( + "dead repo blacklist {} unreadable ({err}); starting empty", + path.display() + )), + ); + } + }; + match serde_json::from_slice::(&bytes) { + Ok(file) if file.schema_version == DEAD_REPO_BLACKLIST_SCHEMA_VERSION => { + let entries = file + .entries + .into_iter() + .map(|entry| ((entry.transport, entry.uri.clone()), entry)) + .collect(); + (Self { entries }, None) + } + Ok(file) => ( + Self::new(), + Some(format!( + "dead repo blacklist {} schema_version {} != {}; starting empty", + path.display(), + file.schema_version, + DEAD_REPO_BLACKLIST_SCHEMA_VERSION + )), + ), + Err(err) => ( + Self::new(), + Some(format!( + "dead repo blacklist {} unparsable ({err}); starting empty", + path.display() + )), + ), + } + } + + /// Persist atomically (tmp file + rename). + pub fn store_atomic(&self, path: &Path, now_unix: u64) -> io::Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let mut entries: Vec<&DeadRepoBlacklistEntry> = self.entries.values().collect(); + entries.sort_by(|a, b| a.uri.cmp(&b.uri).then(a.transport.cmp(&b.transport))); + let file = DeadRepoBlacklistFile { + schema_version: DEAD_REPO_BLACKLIST_SCHEMA_VERSION, + updated_at_unix: now_unix, + entries: entries.into_iter().cloned().collect(), + }; + let bytes = serde_json::to_vec_pretty(&file) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?; + let mut tmp_name = path.file_name().map(|name| name.to_os_string()).unwrap_or_default(); + tmp_name.push(".tmp"); + let tmp_path = path.with_file_name(tmp_name); + fs::write(&tmp_path, &bytes)?; + // Best-effort durability before the rename. + if let Ok(file) = fs::File::open(&tmp_path) { + let _ = file.sync_all(); + } + fs::rename(&tmp_path, path)?; + Ok(()) + } + + pub fn is_blacklisted(&self, transport: RepoTransportMode, uri: &str) -> bool { + self.entries + .get(&(transport, uri.to_string())) + .map(|entry| entry.blacklisted) + .unwrap_or(false) + } + + /// Record a transport-class fetch failure. Returns whether the failure + /// newly admitted the entry to the blacklist. + pub fn record_transport_failure( + &mut self, + transport: RepoTransportMode, + uri: &str, + now_unix: u64, + fail_threshold: u32, + capacity: usize, + ) -> DeadRepoFailureOutcome { + let key = (transport, uri.to_string()); + if !self.entries.contains_key(&key) { + self.evict_if_full(capacity); + } + let threshold = fail_threshold.max(1); + let entry = self.entries.entry(key).or_insert_with(|| DeadRepoBlacklistEntry { + transport, + uri: uri.to_string(), + consecutive_failures: 0, + blacklisted: false, + first_failure_at_unix: now_unix, + last_failure_at_unix: now_unix, + added_at_unix: 0, + last_probe_at_unix: None, + probe_failures: 0, + }); + entry.consecutive_failures = entry.consecutive_failures.saturating_add(1); + entry.last_failure_at_unix = now_unix; + if entry.first_failure_at_unix == 0 { + entry.first_failure_at_unix = now_unix; + } + if !entry.blacklisted && entry.consecutive_failures >= threshold { + entry.blacklisted = true; + entry.added_at_unix = now_unix; + // A fresh admission restarts probe bookkeeping. + entry.last_probe_at_unix = None; + entry.probe_failures = 0; + return DeadRepoFailureOutcome::Admitted; + } + DeadRepoFailureOutcome::Counted { + consecutive_failures: entry.consecutive_failures, + } + } + + /// Record a successful fetch: reset a counting entry, or remove a + /// blacklisted entry entirely (self-heal). + pub fn record_transport_success( + &mut self, + transport: RepoTransportMode, + uri: &str, + ) -> Option { + match self.entries.remove(&(transport, uri.to_string())) { + Some(entry) if entry.blacklisted => Some(DeadRepoSuccessOutcome::BlacklistRemoved), + Some(_) => Some(DeadRepoSuccessOutcome::CounterReset), + None => None, + } + } + + /// Blacklisted entries whose next probe is due at `now_unix`. + pub fn probe_due_entries( + &self, + now_unix: u64, + probe_interval_secs: u64, + ) -> Vec { + let mut due: Vec = self + .entries + .values() + .filter(|entry| entry.blacklisted) + .filter(|entry| { + let interval = probe_backoff_secs(probe_interval_secs, entry.probe_failures); + match entry.last_probe_at_unix { + None => true, + Some(last) => now_unix.saturating_sub(last) >= interval, + } + }) + .cloned() + .collect(); + due.sort_by(|a, b| a.uri.cmp(&b.uri)); + due + } + + /// Probe succeeded: remove the entry. Returns true if it was present. + pub fn record_probe_success(&mut self, transport: RepoTransportMode, uri: &str) -> bool { + self.entries.remove(&(transport, uri.to_string())).is_some() + } + + pub fn record_probe_failure(&mut self, transport: RepoTransportMode, uri: &str, now_unix: u64) { + if let Some(entry) = self.entries.get_mut(&(transport, uri.to_string())) { + entry.last_probe_at_unix = Some(now_unix); + entry.probe_failures = entry.probe_failures.saturating_add(1); + } + } + + pub fn len(&self) -> usize { + self.entries.len() + } + + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + pub fn blacklisted_len(&self) -> usize { + self.entries.values().filter(|entry| entry.blacklisted).count() + } + + /// Deterministic snapshot of all entries (for status/observability). + pub fn entries_sorted(&self) -> Vec { + let mut entries: Vec = self.entries.values().cloned().collect(); + entries.sort_by(|a, b| a.uri.cmp(&b.uri).then(a.transport.cmp(&b.transport))); + entries + } + + fn evict_if_full(&mut self, capacity: usize) { + let capacity = capacity.max(1); + if self.entries.len() < capacity { + return; + } + // Prefer evicting the stalest counting entry; fall back to the oldest + // blacklisted entry. + let victim = self + .entries + .values() + .filter(|entry| !entry.blacklisted) + .min_by_key(|entry| entry.last_failure_at_unix) + .map(|entry| (entry.transport, entry.uri.clone())) + .or_else(|| { + self.entries + .values() + .min_by_key(|entry| entry.added_at_unix) + .map(|entry| (entry.transport, entry.uri.clone())) + }); + if let Some(key) = victim { + self.entries.remove(&key); + } + } +} + +pub fn probe_backoff_secs(base_interval_secs: u64, probe_failures: u32) -> u64 { + // First retry after a probe failure still uses the base interval; + // exponential backoff kicks in from the second consecutive failure. + let shift = probe_failures.saturating_sub(1).min(PROBE_BACKOFF_CAP_SHIFT); + base_interval_secs.saturating_mul(1u64 << shift) +} + +/// Classify an HTTP fetch error string (formats from +/// `fetch/http.rs::BlockingHttpFetcher::fetch_bytes`). +pub fn classify_http_fetch_error(detail: &str) -> RepoTransportErrorClass { + if detail.starts_with("http status") { + // The peer answered: transport is fine, the failure is protocol-level. + RepoTransportErrorClass::Protocol + } else if detail.starts_with("http request failed") || detail.starts_with("http read body failed") { + RepoTransportErrorClass::TransportFetch + } else { + RepoTransportErrorClass::Unknown + } +} + +/// Classify an rsync fetch error string using the same host-level substring +/// set as the run-local rsync failure scope short-circuit. +pub fn classify_rsync_fetch_error(detail: &str) -> RepoTransportErrorClass { + if crate::parallel::repo_scheduler::is_host_level_rsync_failure(detail) + // Stall failures are transport death too: the fail-fast mechanism + // gives up when wall-clock windows pass with no (additional) progress + // (#141 M9). Content-level errors fail immediately and never carry + // this marker, so they still do not count. + || detail.contains("rsync fail-fast gave up") + { + RepoTransportErrorClass::TransportFetch + } else { + // Content/protocol-level rsync failures (digest mismatch, vanished + // files, partial transfers) never count toward the blacklist. + RepoTransportErrorClass::Unknown + } +} + +/// Map a structured RRDP sync error to a transport error class. +pub fn classify_rrdp_sync_error(err: &crate::sync::rrdp::RrdpSyncError) -> RepoTransportErrorClass { + match err { + crate::sync::rrdp::RrdpSyncError::Fetch(detail) => classify_http_fetch_error(detail), + crate::sync::rrdp::RrdpSyncError::Rrdp(_) => RepoTransportErrorClass::Protocol, + crate::sync::rrdp::RrdpSyncError::Storage(_) => RepoTransportErrorClass::Storage, + } +} + +/// Map a structured rsync-side repo sync error to a transport error class. +pub fn classify_rsync_repo_sync_error(err: &crate::sync::repo::RepoSyncError) -> RepoTransportErrorClass { + match err { + crate::sync::repo::RepoSyncError::Rsync(crate::fetch::rsync::RsyncFetchError::Fetch(detail)) => { + classify_rsync_fetch_error(detail) + } + crate::sync::repo::RepoSyncError::Storage(_) => RepoTransportErrorClass::Storage, + _ => RepoTransportErrorClass::Protocol, + } +} + +impl RepoTransportMode { + pub fn as_str(self) -> &'static str { + match self { + RepoTransportMode::Rrdp => "rrdp", + RepoTransportMode::Rsync => "rsync", + } + } +} + +// Keep the public surface free of internal key types; the (transport, uri) +// tuple key is an implementation detail of the lookup map. + +#[cfg(test)] +#[path = "dead_repo_blacklist_tests.rs"] +mod tests; diff --git a/src/parallel/dead_repo_blacklist_tests.rs b/src/parallel/dead_repo_blacklist_tests.rs new file mode 100644 index 0000000..8f3307f --- /dev/null +++ b/src/parallel/dead_repo_blacklist_tests.rs @@ -0,0 +1,360 @@ +use super::*; +use crate::parallel::types::RepoTransportMode; + +const RRDP_URI: &str = "https://dead.example.com/notification.xml"; +const RSYNC_URI: &str = "rsync://dead.example.com/repo/"; + +fn now() -> u64 { + 1_700_000_000 +} + +#[test] +fn admission_requires_threshold_consecutive_failures() { + let mut bl = DeadRepoBlacklist::new(); + assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + + let out1 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256); + assert_eq!( + out1, + DeadRepoFailureOutcome::Counted { + consecutive_failures: 1 + } + ); + assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + + let out2 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 1, 3, 256); + assert_eq!( + out2, + DeadRepoFailureOutcome::Counted { + consecutive_failures: 2 + } + ); + assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + + let out3 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 2, 3, 256); + assert_eq!(out3, DeadRepoFailureOutcome::Admitted); + assert!(bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + + // Further failures keep counting but do not re-admit. + let out4 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 3, 3, 256); + assert!(matches!( + out4, + DeadRepoFailureOutcome::Counted { + consecutive_failures: 4 + } + )); +} + +#[test] +fn threshold_of_one_admits_immediately() { + let mut bl = DeadRepoBlacklist::new(); + let out = bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now(), 1, 256); + assert_eq!(out, DeadRepoFailureOutcome::Admitted); +} + +#[test] +fn success_resets_counting_entry() { + let mut bl = DeadRepoBlacklist::new(); + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256); + assert_eq!(bl.len(), 1); + let out = bl.record_transport_success(RepoTransportMode::Rrdp, RRDP_URI); + assert_eq!(out, Some(DeadRepoSuccessOutcome::CounterReset)); + assert!(bl.is_empty()); + // Next failure starts from zero again. + let out = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256); + assert_eq!( + out, + DeadRepoFailureOutcome::Counted { + consecutive_failures: 1 + } + ); +} + +#[test] +fn success_on_blacklisted_entry_self_heals() { + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now() + i, 3, 256); + } + assert!(bl.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI)); + let out = bl.record_transport_success(RepoTransportMode::Rsync, RSYNC_URI); + assert_eq!(out, Some(DeadRepoSuccessOutcome::BlacklistRemoved)); + assert!(!bl.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI)); + assert!(bl.is_empty()); +} + +#[test] +fn success_on_unknown_entry_is_none() { + let mut bl = DeadRepoBlacklist::new(); + assert_eq!( + bl.record_transport_success(RepoTransportMode::Rrdp, RRDP_URI), + None + ); +} + +#[test] +fn entries_are_per_transport() { + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256); + } + assert!(bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + assert!(!bl.is_blacklisted(RepoTransportMode::Rsync, RRDP_URI)); +} + +#[test] +fn capacity_evicts_stale_counting_entries_first() { + let mut bl = DeadRepoBlacklist::new(); + // One blacklisted entry (old) and two counting entries. + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://a.example/n.xml", 100 + i, 1, 3); + } + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://b.example/n.xml", 200, 3, 3); + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://c.example/n.xml", 300, 3, 3); + assert_eq!(bl.len(), 3); + assert_eq!(bl.blacklisted_len(), 1); + + // Inserting a fourth entry evicts the stalest counting entry (b). + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://d.example/n.xml", 400, 3, 3); + assert_eq!(bl.len(), 3); + assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://a.example/n.xml")); + assert!(!bl.entries_sorted().iter().any(|e| e.uri == "https://b.example/n.xml")); + assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://c.example/n.xml")); + assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://d.example/n.xml")); +} + +#[test] +fn capacity_evicts_oldest_blacklisted_when_no_counting_entries() { + let mut bl = DeadRepoBlacklist::new(); + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://a.example/n.xml", 100, 1, 2); + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://b.example/n.xml", 200, 1, 2); + assert_eq!(bl.blacklisted_len(), 2); + bl.record_transport_failure(RepoTransportMode::Rrdp, "https://c.example/n.xml", 300, 1, 2); + assert_eq!(bl.len(), 2); + assert!(!bl.entries_sorted().iter().any(|e| e.uri == "https://a.example/n.xml")); + assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://b.example/n.xml")); + assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://c.example/n.xml")); +} + +#[test] +fn store_load_roundtrip() { + let dir = std::env::temp_dir().join(format!( + "dead_repo_blacklist_test_roundtrip_{}_{}", + std::process::id(), + now() + )); + let path = dir.join("state").join("dead-repo-blacklist.json"); + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256); + } + bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now(), 3, 256); + bl.store_atomic(&path, now() + 10).unwrap(); + + let (loaded, warning) = DeadRepoBlacklist::load(&path); + assert!(warning.is_none()); + assert!(loaded.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI)); + assert!(!loaded.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI)); + assert_eq!(loaded.len(), 2); + let entry = loaded + .entries_sorted() + .into_iter() + .find(|e| e.uri == RRDP_URI) + .unwrap(); + assert_eq!(entry.consecutive_failures, 3); + assert!(entry.blacklisted); + assert_eq!(entry.first_failure_at_unix, now()); + + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn load_missing_file_is_empty_without_warning() { + let path = std::env::temp_dir().join(format!( + "dead_repo_blacklist_test_missing_{}_{}.json", + std::process::id(), + now() + )); + let (bl, warning) = DeadRepoBlacklist::load(&path); + assert!(bl.is_empty()); + assert!(warning.is_none()); +} + +#[test] +fn load_corrupted_file_degrades_to_empty_with_warning() { + let path = std::env::temp_dir().join(format!( + "dead_repo_blacklist_test_corrupt_{}_{}.json", + std::process::id(), + now() + )); + std::fs::write(&path, b"{not json").unwrap(); + let (bl, warning) = DeadRepoBlacklist::load(&path); + assert!(bl.is_empty()); + assert!(warning.is_some()); + let _ = std::fs::remove_file(&path); +} + +#[test] +fn load_wrong_schema_version_degrades_to_empty() { + let path = std::env::temp_dir().join(format!( + "dead_repo_blacklist_test_schema_{}_{}.json", + std::process::id(), + now() + )); + std::fs::write( + &path, + r#"{"schema_version": 99, "updated_at_unix": 0, "entries": []}"#, + ) + .unwrap(); + let (bl, warning) = DeadRepoBlacklist::load(&path); + assert!(bl.is_empty()); + assert!(warning.unwrap().contains("schema_version")); + let _ = std::fs::remove_file(&path); +} + +#[test] +fn probe_due_uses_exponential_backoff() { + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256); + } + // Never probed: due immediately. + let due = bl.probe_due_entries(now() + 100, 600); + assert_eq!(due.len(), 1); + + // Probe failure: not due until base interval has passed. + bl.record_probe_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 100); + assert!(bl.probe_due_entries(now() + 100 + 599, 600).is_empty()); + assert_eq!(bl.probe_due_entries(now() + 100 + 600, 600).len(), 1); + + // Second probe failure: backoff doubles to 1200s. + bl.record_probe_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 700); + assert!(bl.probe_due_entries(now() + 700 + 1199, 600).is_empty()); + assert_eq!(bl.probe_due_entries(now() + 700 + 1200, 600).len(), 1); +} + +#[test] +fn probe_success_removes_entry() { + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now() + i, 3, 256); + } + assert!(bl.record_probe_success(RepoTransportMode::Rsync, RSYNC_URI)); + assert!(bl.is_empty()); + assert!(!bl.record_probe_success(RepoTransportMode::Rsync, RSYNC_URI)); +} + +#[test] +fn counting_entries_are_not_probed() { + let mut bl = DeadRepoBlacklist::new(); + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256); + assert!(bl.probe_due_entries(now() + 10_000, 600).is_empty()); +} + +#[test] +fn probe_backoff_caps_at_max_shift() { + assert_eq!(probe_backoff_secs(600, 0), 600); + assert_eq!(probe_backoff_secs(600, 1), 600); + assert_eq!(probe_backoff_secs(600, 2), 1200); + assert_eq!(probe_backoff_secs(600, 5), 9600); + assert_eq!(probe_backoff_secs(600, 100), 9600); +} + +#[test] +fn http_fetch_error_classification() { + use crate::parallel::types::RepoTransportErrorClass; + assert_eq!( + classify_http_fetch_error("http request failed: reqwest::Error { kind: Connect, .. }"), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_http_fetch_error("http read body failed: connection closed; status=200"), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_http_fetch_error("http status 404 Not Found; content_type=; .."), + RepoTransportErrorClass::Protocol + ); + assert_eq!( + classify_http_fetch_error("something unexpected"), + RepoTransportErrorClass::Unknown + ); +} + +#[test] +fn rsync_fetch_error_classification() { + use crate::parallel::types::RepoTransportErrorClass; + assert_eq!( + classify_rsync_fetch_error("rsync error: timeout waiting for daemon connection"), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_rsync_fetch_error("rsync: failed to connect to host: Connection refused"), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_rsync_fetch_error("temporary failure in name resolution"), + RepoTransportErrorClass::TransportFetch + ); + // Content-level failures must not count. + assert_eq!( + classify_rsync_fetch_error("rsync file digest mismatch after download"), + RepoTransportErrorClass::Unknown + ); + // Stall failures (#141 M9): fail-fast give-up means the host delivered no + // bytes inside the wall-clock window — transport death, must count. + assert_eq!( + classify_rsync_fetch_error( + "rsync fail-fast gave up after 1 attempts with no progress: rsync wall-clock window expired" + ), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_rsync_fetch_error( + "rsync fail-fast gave up after 3 attempts with no additional progress: rsync wall-clock window expired" + ), + RepoTransportErrorClass::TransportFetch + ); + // Content-level give-up text without the fail-fast marker stays unknown. + assert_eq!( + classify_rsync_fetch_error("rsync error: some files vanished before transfer"), + RepoTransportErrorClass::Unknown + ); +} + +#[test] +fn rrdp_sync_error_classification() { + use crate::parallel::types::RepoTransportErrorClass; + use crate::sync::rrdp::RrdpSyncError; + assert_eq!( + classify_rrdp_sync_error(&RrdpSyncError::Fetch( + "http request failed: connect timeout".to_string() + )), + RepoTransportErrorClass::TransportFetch + ); + assert_eq!( + classify_rrdp_sync_error(&RrdpSyncError::Fetch("http status 500".to_string())), + RepoTransportErrorClass::Protocol + ); + assert_eq!( + classify_rrdp_sync_error(&RrdpSyncError::Storage("db full".to_string())), + RepoTransportErrorClass::Storage + ); +} + +#[test] +fn readmission_after_probe_removal_starts_fresh() { + let mut bl = DeadRepoBlacklist::new(); + for i in 0..3 { + bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256); + } + assert!(bl.record_probe_success(RepoTransportMode::Rrdp, RRDP_URI)); + // No half-open: needs the full threshold again. + let out = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 10, 3, 256); + assert_eq!( + out, + DeadRepoFailureOutcome::Counted { + consecutive_failures: 1 + } + ); +} diff --git a/src/parallel/mod.rs b/src/parallel/mod.rs index 109319f..06174cd 100644 --- a/src/parallel/mod.rs +++ b/src/parallel/mod.rs @@ -1,4 +1,5 @@ pub mod config; +pub mod dead_repo_blacklist; pub mod object_worker; pub mod phase2_scheduler; pub mod repo_runtime; diff --git a/src/parallel/repo_runtime.rs b/src/parallel/repo_runtime.rs index f1a61ca..ac7a4a2 100644 --- a/src/parallel/repo_runtime.rs +++ b/src/parallel/repo_runtime.rs @@ -103,6 +103,17 @@ pub trait RepoSyncRuntime: Send + Sync { ) -> Result; fn transport_prefetch_snapshot(&self) -> TransportPrefetchSnapshot; + + /// Dead-repo blacklist working copy (#141) for run-end persistence. + /// `None` when the feature is disabled. + fn dead_repo_blacklist_state( + &self, + ) -> Option<( + crate::parallel::dead_repo_blacklist::DeadRepoBlacklistConfig, + crate::parallel::dead_repo_blacklist::DeadRepoBlacklist, + )> { + None + } } pub struct Phase1RepoSyncRuntime { @@ -634,6 +645,18 @@ impl RepoSyncRuntime for Phase1RepoSyncRuntime { }) .unwrap_or_else(|| TransportPrefetchSnapshot::new(self.sync_preference, Vec::new())) } + + fn dead_repo_blacklist_state( + &self, + ) -> Option<( + crate::parallel::dead_repo_blacklist::DeadRepoBlacklistConfig, + crate::parallel::dead_repo_blacklist::DeadRepoBlacklist, + )> { + let coordinator = self.coordinator.lock().expect("coordinator lock poisoned"); + let config = coordinator.config.dead_repo_blacklist.clone()?; + let blacklist = coordinator.dead_repo_blacklist().cloned()?; + Some((config, blacklist)) + } } fn outcome_from_transport_result( @@ -666,7 +689,9 @@ fn outcome_from_transport_result( } } ( - RepoTransportResultKind::Failed { detail, warnings }, + RepoTransportResultKind::Failed { + detail, warnings, .. + }, RepoRuntimeState::FailedTerminal, ) => RepoSyncRuntimeOutcome { repo_sync_ok: false, @@ -798,6 +823,7 @@ mod tests { result: RepoTransportResultKind::Failed { detail: "rrdp failed".to_string(), warnings: vec![Warning::new("rrdp failed")], + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, } } @@ -842,6 +868,7 @@ mod tests { result: RepoTransportResultKind::Failed { detail: "rrdp failed".to_string(), warnings: vec![Warning::new("rrdp failed")], + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, } } @@ -858,6 +885,7 @@ mod tests { result: RepoTransportResultKind::Failed { detail: "rsync failed".to_string(), warnings: vec![Warning::new("rsync failed")], + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, } } diff --git a/src/parallel/repo_scheduler.rs b/src/parallel/repo_scheduler.rs index 8666a23..f14d19f 100644 --- a/src/parallel/repo_scheduler.rs +++ b/src/parallel/repo_scheduler.rs @@ -1,11 +1,14 @@ use std::collections::{HashMap, HashSet}; +use crate::parallel::dead_repo_blacklist::DeadRepoBlacklist; use crate::parallel::types::{ InFlightRepoEntry, RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoRuntimeState, RepoSyncResultEnvelope, RepoSyncResultKind, RepoSyncResultRef, RepoSyncTask, RepoTaskState, - RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, RepoTransportTask, + RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope, + RepoTransportResultKind, RepoTransportTask, }; use crate::policy::SyncPreference; +use crate::report::Warning; #[derive(Clone, Debug, PartialEq, Eq)] pub enum RepoRequestAction { @@ -74,6 +77,9 @@ pub struct TransportStateTables { rsync_failure_probe_inflight: HashMap, rsync_failure_scope_reachable: HashSet, runtime_records: HashMap, + /// Frozen run-start snapshot of the dead-repo blacklist (#141). Entries + /// admitted during this run only take effect from the next run. + dead_repo_blacklist: Option, } impl TransportStateTables { @@ -81,6 +87,44 @@ impl TransportStateTables { Self::default() } + pub fn set_dead_repo_blacklist(&mut self, blacklist: DeadRepoBlacklist) { + self.dead_repo_blacklist = Some(blacklist); + } + + fn dead_repo_terminal_envelope( + identity: &RepoIdentity, + requesters: &[RepoRequester], + rsync_scope_uri: &str, + rsync_failure_scope_uri: Option, + ) -> RepoTransportResultEnvelope { + let first_requester = requesters + .first() + .expect("blacklist terminal record must keep at least one requester"); + RepoTransportResultEnvelope { + dedup_key: RepoDedupKey::RsyncScope { + rsync_scope_uri: rsync_scope_uri.to_string(), + }, + rsync_failure_scope_uri, + repo_identity: identity.clone(), + mode: RepoTransportMode::Rsync, + tal_id: first_requester.tal_id.clone(), + rir_id: first_requester.rir_id.clone(), + timing_ms: 0, + result: RepoTransportResultKind::Failed { + detail: format!( + "dead repo blacklist: rsync transport persistently unreachable: {}", + identity.rsync_base_uri + ), + warnings: vec![Warning::new(format!( + "dead_repo_blacklist_skip_all: repository {} skipped after repeated transport failures", + identity.rsync_base_uri + )) + .with_context(identity.rsync_base_uri.clone())], + error_class: RepoTransportErrorClass::Unknown, + }, + } + } + pub fn runtime_record(&self, identity: &RepoIdentity) -> Option<&RepoRuntimeRecord> { self.runtime_records.get(identity) } @@ -205,6 +249,82 @@ impl TransportStateTables { }; } + // Dead-repo blacklist fast paths (#141), consulted only for fresh + // identities; existing records already encode their terminal state. + if let Some(blacklist) = self.dead_repo_blacklist.as_ref() { + let rrdp_wanted = sync_preference == SyncPreference::RrdpThenRsync + && identity.notification_uri.is_some(); + let rrdp_dead = rrdp_wanted + && identity + .notification_uri + .as_deref() + .map(|uri| blacklist.is_blacklisted(RepoTransportMode::Rrdp, uri)) + .unwrap_or(false); + let rsync_dead = blacklist.is_blacklisted( + RepoTransportMode::Rsync, + identity.rsync_base_uri.as_str(), + ); + if rsync_dead && (!rrdp_wanted || rrdp_dead) { + crate::progress_log::emit( + "dead_repo_blacklist_skip_all", + serde_json::json!({ + "repo_key_notification_uri": identity.notification_uri, + "repo_key_rsync_base_uri": identity.rsync_base_uri, + }), + ); + let envelope = Self::dead_repo_terminal_envelope( + &identity, + std::slice::from_ref(&requester), + &rsync_scope_uri, + rsync_failure_scope_uri.clone(), + ); + self.runtime_records.insert( + identity.clone(), + RepoRuntimeRecord { + identity, + state: RepoRuntimeState::FailedTerminal, + rrdp_notification_key: None, + rsync_scope_key: rsync_scope_uri, + rsync_failure_scope_key: rsync_failure_scope_uri, + requesters: vec![requester], + validation_time, + priority, + last_success: None, + terminal_failure: Some(envelope.clone()), + }, + ); + return TransportRequestAction::ReusedTerminalFailure(envelope); + } + if rrdp_dead { + crate::progress_log::emit( + "dead_repo_blacklist_skip_rrdp", + serde_json::json!({ + "repo_key_notification_uri": identity.notification_uri, + "repo_key_rsync_base_uri": identity.rsync_base_uri, + }), + ); + let mut action = self.register_rsync_request( + identity, + requester, + validation_time, + priority, + rsync_scope_uri, + rsync_failure_scope_uri, + ); + // A blacklisted rrdp transport is a known-persistent failure, + // so the rsync fallback must keep the short retry profile + // (#141): the prefetch layer records no real rrdp failure in + // skipped runs, which would otherwise silently drop the task + // back to the default 15s window from the second skipped run + // on. The coordinator wrapper only ever sets the flag to true, + // so setting it here survives registration. + if let TransportRequestAction::Enqueue(task) = &mut action { + task.retry_short_timeout = true; + } + return action; + } + } + if sync_preference == SyncPreference::RrdpThenRsync { if let Some(notification_uri) = identity.notification_uri.clone() { if let Some(entry) = self.rrdp_inflight.get_mut(¬ification_uri) { @@ -512,6 +632,7 @@ impl TransportStateTables { rsync_failure_by_scope: &HashMap, rsync_failure_probe_inflight: &mut HashMap, rsync_failure_scope_reachable: &HashSet, + dead_repo_blacklist: Option<&DeadRepoBlacklist>, follow_up_tasks: &mut Vec, ) { let rsync_scope_uri = record.rsync_scope_key.clone(); @@ -548,6 +669,34 @@ impl TransportStateTables { } } + // Dead-repo blacklist (#141): the rsync transport of this repo is + // persistently unreachable; terminate instantly instead of enqueueing + // another doomed fetch (rrdp was already attempted or skipped). + if let Some(blacklist) = dead_repo_blacklist { + if blacklist.is_blacklisted( + RepoTransportMode::Rsync, + record.identity.rsync_base_uri.as_str(), + ) { + crate::progress_log::emit( + "dead_repo_blacklist_skip_all", + serde_json::json!({ + "repo_key_notification_uri": record.identity.notification_uri, + "repo_key_rsync_base_uri": record.identity.rsync_base_uri, + "after_rrdp_failure": true, + }), + ); + let envelope = Self::dead_repo_terminal_envelope( + &record.identity, + &record.requesters, + &rsync_scope_uri, + record.rsync_failure_scope_key.clone(), + ); + record.state = RepoRuntimeState::FailedTerminal; + record.terminal_failure = Some(envelope); + return; + } + } + let first_requester = record .requesters .first() @@ -666,6 +815,7 @@ impl TransportStateTables { &self.rsync_failure_by_scope, &mut self.rsync_failure_probe_inflight, &self.rsync_failure_scope_reachable, + self.dead_repo_blacklist.as_ref(), &mut follow_up_tasks, ); } @@ -718,6 +868,7 @@ impl TransportStateTables { &self.rsync_failure_by_scope, &mut self.rsync_failure_probe_inflight, &self.rsync_failure_scope_reachable, + self.dead_repo_blacklist.as_ref(), &mut follow_up_tasks, ); } @@ -783,6 +934,7 @@ impl TransportStateTables { &self.rsync_failure_by_scope, &mut self.rsync_failure_probe_inflight, &self.rsync_failure_scope_reachable, + self.dead_repo_blacklist.as_ref(), &mut follow_up_tasks, ); } @@ -814,7 +966,7 @@ fn reusable_rsync_failure_scope(result: &RepoTransportResultEnvelope) -> Option< } } -fn is_host_level_rsync_failure(detail: &str) -> bool { +pub(crate) fn is_host_level_rsync_failure(detail: &str) -> bool { let lower = detail.to_ascii_lowercase(); lower.contains("timeout waiting for daemon connection") || lower.contains("failed to connect") @@ -1462,6 +1614,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rrdp timeout".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -1567,6 +1720,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rrdp timeout".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -1597,6 +1751,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rsync error: timeout waiting for daemon connection".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -1659,6 +1814,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rsync timeout".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -1920,6 +2076,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rsync file digest mismatch after download".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -1972,6 +2129,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "rsync error: failed to connect to daemon".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, @@ -2031,6 +2189,7 @@ mod transport_tests { result: RepoTransportResultKind::Failed { detail: "temporary failure in name resolution".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, diff --git a/src/parallel/repo_worker.rs b/src/parallel/repo_worker.rs index 435de94..ec226ae 100644 --- a/src/parallel/repo_worker.rs +++ b/src/parallel/repo_worker.rs @@ -115,19 +115,24 @@ impl RepoTransportExecutor for LiveRrdpTransportExecutor RepoTransportResultEnvelope { - dedup_key: task.dedup_key, - rsync_failure_scope_uri: task.rsync_failure_scope_uri, - repo_identity: task.repo_identity, - mode: RepoTransportMode::Rrdp, - tal_id: task.tal_id, - rir_id: task.rir_id, - timing_ms: started.elapsed().as_millis() as u64, - result: RepoTransportResultKind::Failed { - detail: err.to_string(), - warnings: Vec::new(), - }, - }, + Err(err) => { + let error_class = + crate::parallel::dead_repo_blacklist::classify_rrdp_sync_error(&err); + RepoTransportResultEnvelope { + dedup_key: task.dedup_key, + rsync_failure_scope_uri: task.rsync_failure_scope_uri, + repo_identity: task.repo_identity, + mode: RepoTransportMode::Rrdp, + tal_id: task.tal_id, + rir_id: task.rir_id, + timing_ms: started.elapsed().as_millis() as u64, + result: RepoTransportResultKind::Failed { + detail: err.to_string(), + warnings: Vec::new(), + error_class, + }, + } + } } } } @@ -206,19 +211,24 @@ impl RepoTransportExecutor for LiveRsyncTransportExec warnings: Vec::new(), }, }, - Err(err) => RepoTransportResultEnvelope { - dedup_key: task.dedup_key, - rsync_failure_scope_uri: task.rsync_failure_scope_uri, - repo_identity: task.repo_identity, - mode: RepoTransportMode::Rsync, - tal_id: task.tal_id, - rir_id: task.rir_id, - timing_ms: started.elapsed().as_millis() as u64, - result: RepoTransportResultKind::Failed { - detail: err.to_string(), - warnings: Vec::new(), - }, - }, + Err(err) => { + let error_class = + crate::parallel::dead_repo_blacklist::classify_rsync_repo_sync_error(&err); + RepoTransportResultEnvelope { + dedup_key: task.dedup_key, + rsync_failure_scope_uri: task.rsync_failure_scope_uri, + repo_identity: task.repo_identity, + mode: RepoTransportMode::Rsync, + tal_id: task.tal_id, + rir_id: task.rir_id, + timing_ms: started.elapsed().as_millis() as u64, + result: RepoTransportResultKind::Failed { + detail: err.to_string(), + warnings: Vec::new(), + error_class, + }, + } + } } } } diff --git a/src/parallel/run_coordinator.rs b/src/parallel/run_coordinator.rs index 2596654..3c63134 100644 --- a/src/parallel/run_coordinator.rs +++ b/src/parallel/run_coordinator.rs @@ -2,6 +2,9 @@ use std::collections::VecDeque; use crate::current_repo_index::{CurrentRepoIndex, CurrentRepoIndexHandle}; use crate::parallel::config::ParallelPhase1Config; +use crate::parallel::dead_repo_blacklist::{ + DeadRepoBlacklist, DeadRepoFailureOutcome, DeadRepoSuccessOutcome, +}; use crate::parallel::repo_scheduler::{ InFlightRepoTable, RepoCompletion, RepoRequestAction, TransportCompletion, TransportRequestAction, TransportStateTables, @@ -9,7 +12,8 @@ use crate::parallel::repo_scheduler::{ use crate::parallel::stats::ParallelRunStats; use crate::parallel::types::{ RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncTask, - RepoTransportResultEnvelope, RepoTransportTask, TalInputSpec, + RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope, + RepoTransportResultKind, RepoTransportTask, TalInputSpec, }; use crate::policy::SyncPreference; @@ -22,19 +26,42 @@ pub struct GlobalRunCoordinator { pub pending_repo_tasks: VecDeque, pub pending_transport_tasks: VecDeque, pub stats: ParallelRunStats, + /// Mutable working copy of the dead-repo blacklist (#141): failure/success + /// counters update during the run and persist at run end. Skip decisions + /// inside `transport_tables` use the frozen run-start snapshot. + pub dead_repo_blacklist: Option, } impl GlobalRunCoordinator { pub fn new(config: ParallelPhase1Config, tal_inputs: Vec) -> Self { + let dead_repo_blacklist = config + .dead_repo_blacklist + .as_ref() + .map(|blacklist_config| { + let (blacklist, warning) = DeadRepoBlacklist::load(&blacklist_config.path); + if let Some(warning) = warning { + crate::progress_log::emit( + "dead_repo_blacklist_load_warning", + serde_json::json!({ "warning": warning }), + ); + } + blacklist + }); + let mut transport_tables = TransportStateTables::new(); + if let Some(blacklist) = dead_repo_blacklist.as_ref() { + // Frozen run-start snapshot for skip decisions. + transport_tables.set_dead_repo_blacklist(blacklist.clone()); + } Self { config, tal_inputs, current_repo_index: CurrentRepoIndex::shared(), in_flight_repos: InFlightRepoTable::new(), - transport_tables: TransportStateTables::new(), + transport_tables, pending_repo_tasks: VecDeque::new(), pending_transport_tasks: VecDeque::new(), stats: ParallelRunStats::default(), + dead_repo_blacklist, } } @@ -193,9 +220,92 @@ impl GlobalRunCoordinator { { self.stats.repo_tasks_failed += 1; } + self.update_dead_repo_blacklist(&result, finished_at); Ok(completion) } + /// Update dead-repo blacklist counters (#141) from a real worker result. + /// Runs on the single pump thread; synthesized terminal envelopes never + /// pass through here, so no double counting is possible. + fn update_dead_repo_blacklist( + &mut self, + result: &RepoTransportResultEnvelope, + finished_at: time::OffsetDateTime, + ) { + let Some(blacklist_config) = self.config.dead_repo_blacklist.clone() else { + return; + }; + let Some(blacklist) = self.dead_repo_blacklist.as_mut() else { + return; + }; + let now_unix = finished_at.unix_timestamp().max(0) as u64; + let key: Option<(RepoTransportMode, String)> = match (&result.dedup_key, &result.result) { + (RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Success { .. }) => { + Some((RepoTransportMode::Rrdp, notification_uri.clone())) + } + (RepoDedupKey::RsyncScope { .. }, RepoTransportResultKind::Success { .. }) => { + Some((RepoTransportMode::Rsync, result.repo_identity.rsync_base_uri.clone())) + } + ( + RepoDedupKey::RrdpNotify { notification_uri }, + RepoTransportResultKind::Failed { + error_class: RepoTransportErrorClass::TransportFetch, + .. + }, + ) => Some((RepoTransportMode::Rrdp, notification_uri.clone())), + ( + RepoDedupKey::RsyncScope { .. }, + RepoTransportResultKind::Failed { + error_class: RepoTransportErrorClass::TransportFetch, + .. + }, + ) => Some((RepoTransportMode::Rsync, result.repo_identity.rsync_base_uri.clone())), + _ => None, + }; + let Some((transport, uri)) = key else { + return; + }; + match &result.result { + RepoTransportResultKind::Success { .. } => { + if let Some(outcome) = blacklist.record_transport_success(transport, &uri) { + crate::progress_log::emit( + "dead_repo_blacklist_remove", + serde_json::json!({ + "transport": transport.as_str(), + "uri": uri, + "reason": match outcome { + DeadRepoSuccessOutcome::CounterReset => "counter_reset", + DeadRepoSuccessOutcome::BlacklistRemoved => "fetch_success", + }, + }), + ); + } + } + RepoTransportResultKind::Failed { .. } => { + if let DeadRepoFailureOutcome::Admitted = blacklist.record_transport_failure( + transport, + &uri, + now_unix, + blacklist_config.fail_threshold, + blacklist_config.capacity, + ) { + crate::progress_log::emit( + "dead_repo_blacklist_admit", + serde_json::json!({ + "transport": transport.as_str(), + "uri": uri, + "fail_threshold": blacklist_config.fail_threshold, + }), + ); + } + } + } + } + + pub fn dead_repo_blacklist(&self) -> Option<&DeadRepoBlacklist> { + self.dead_repo_blacklist.as_ref() + } + pub fn runtime_record( &self, identity: &RepoIdentity, @@ -465,4 +575,288 @@ mod tests { assert_eq!(coordinator.stats.repo_tasks_total, 1); assert_eq!(coordinator.stats.repo_tasks_reused, 1); } + + // ---- Dead-repo blacklist (#141) integration tests ---- + + use crate::parallel::dead_repo_blacklist::{DeadRepoBlacklist, DeadRepoBlacklistConfig}; + use crate::parallel::types::{ + RepoDedupKey, RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope, + RepoTransportResultKind, + }; + + const DEAD_NOTIFY: &str = "https://dead.example/notification.xml"; + const DEAD_BASE: &str = "rsync://dead.example/repo/"; + + fn blacklist_config(fail_threshold: u32) -> DeadRepoBlacklistConfig { + DeadRepoBlacklistConfig { + path: std::env::temp_dir().join(format!( + "dead_repo_blacklist_coordinator_test_{}.json", + std::process::id() + )), + fail_threshold, + capacity: 256, + } + } + + fn coordinator_with_blacklist( + fail_threshold: u32, + seed: DeadRepoBlacklist, + ) -> GlobalRunCoordinator { + let mut config = ParallelPhase1Config::default(); + config.dead_repo_blacklist = Some(blacklist_config(fail_threshold)); + let mut coordinator = GlobalRunCoordinator::new( + config, + vec![TalInputSpec::from_url("https://example.test/arin.tal")], + ); + coordinator.dead_repo_blacklist = Some(seed.clone()); + coordinator.transport_tables.set_dead_repo_blacklist(seed); + coordinator + } + + fn dead_identity() -> RepoIdentity { + RepoIdentity::new(Some(DEAD_NOTIFY.to_string()), DEAD_BASE) + } + + fn dead_requester() -> RepoRequester { + requester("arin", "arin", "rsync://dead.example/repo/root.mft") + } + + fn transport_envelope( + mode: RepoTransportMode, + error_class: RepoTransportErrorClass, + ) -> RepoTransportResultEnvelope { + let (dedup_key, detail) = match mode { + RepoTransportMode::Rrdp => ( + RepoDedupKey::RrdpNotify { + notification_uri: DEAD_NOTIFY.to_string(), + }, + "http request failed: connect timeout".to_string(), + ), + RepoTransportMode::Rsync => ( + RepoDedupKey::RsyncScope { + rsync_scope_uri: DEAD_BASE.to_string(), + }, + "rsync error: timeout waiting for daemon connection".to_string(), + ), + }; + RepoTransportResultEnvelope { + dedup_key, + rsync_failure_scope_uri: None, + repo_identity: dead_identity(), + mode, + tal_id: "arin".to_string(), + rir_id: "arin".to_string(), + timing_ms: 1, + result: RepoTransportResultKind::Failed { + detail, + warnings: Vec::new(), + error_class, + }, + } + } + + fn register_dead( + coordinator: &mut GlobalRunCoordinator, + ) -> crate::parallel::repo_scheduler::TransportRequestAction { + coordinator.register_transport_request( + dead_identity(), + dead_requester(), + time::OffsetDateTime::UNIX_EPOCH, + 0, + DEAD_BASE.to_string(), + None, + SyncPreference::RrdpThenRsync, + false, + ) + } + + #[test] + fn blacklist_admission_after_threshold_transport_failures() { + let mut coordinator = coordinator_with_blacklist(2, DeadRepoBlacklist::new()); + for _ in 0..2 { + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + coordinator + .complete_transport_result( + transport_envelope( + RepoTransportMode::Rrdp, + RepoTransportErrorClass::TransportFetch, + ), + time::OffsetDateTime::UNIX_EPOCH, + ) + .expect("complete"); + // Simulate the next run: run-local scheduler state resets, the + // persistent blacklist working copy carries over. + coordinator.reset_run_state(); + } + let blacklist = coordinator.dead_repo_blacklist().expect("blacklist"); + assert!(blacklist.is_blacklisted(RepoTransportMode::Rrdp, DEAD_NOTIFY)); + // Admission mid-run does not affect this run's frozen snapshot: the + // scheduler still enqueues rrdp for a fresh identity registration. + coordinator.reset_run_state(); + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + } + + #[test] + fn blacklist_ignores_protocol_and_unknown_failures() { + let mut coordinator = coordinator_with_blacklist(1, DeadRepoBlacklist::new()); + for error_class in [ + RepoTransportErrorClass::Protocol, + RepoTransportErrorClass::Unknown, + RepoTransportErrorClass::Storage, + ] { + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + coordinator + .complete_transport_result( + transport_envelope(RepoTransportMode::Rrdp, error_class), + time::OffsetDateTime::UNIX_EPOCH, + ) + .expect("complete"); + coordinator.reset_run_state(); + } + assert!( + coordinator + .dead_repo_blacklist() + .expect("blacklist") + .is_empty() + ); + } + + #[test] + fn blacklist_skip_rrdp_routes_straight_to_rsync() { + let mut seed = DeadRepoBlacklist::new(); + seed.record_transport_failure(RepoTransportMode::Rrdp, DEAD_NOTIFY, 100, 1, 256); + let mut coordinator = coordinator_with_blacklist(3, seed); + let action = register_dead(&mut coordinator); + let task = match action { + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(task) => task, + other => panic!("expected rsync enqueue, got {other:?}"), + }; + assert_eq!(task.mode, RepoTransportMode::Rsync); + // register_dead passes retry_short_timeout=false (mirroring a prefetch + // snapshot that recorded no real rrdp failure for skipped runs); the + // skip path must still keep the short retry profile (#141). + assert!(task.retry_short_timeout); + } + + #[test] + fn blacklist_dual_dead_terminates_immediately() { + let mut seed = DeadRepoBlacklist::new(); + seed.record_transport_failure(RepoTransportMode::Rrdp, DEAD_NOTIFY, 100, 1, 256); + seed.record_transport_failure(RepoTransportMode::Rsync, DEAD_BASE, 100, 1, 256); + let mut coordinator = coordinator_with_blacklist(3, seed); + let action = register_dead(&mut coordinator); + let envelope = match action { + crate::parallel::repo_scheduler::TransportRequestAction::ReusedTerminalFailure( + envelope, + ) => envelope, + other => panic!("expected terminal failure, got {other:?}"), + }; + match &envelope.result { + RepoTransportResultKind::Failed { + detail, warnings, .. + } => { + assert!(detail.contains("dead repo blacklist")); + assert!(warnings + .iter() + .any(|warning| warning.message.contains("dead_repo_blacklist_skip_all"))); + } + other => panic!("expected failed result, got {other:?}"), + } + let record = coordinator + .runtime_record(&dead_identity()) + .expect("record"); + assert_eq!( + record.state, + crate::parallel::types::RepoRuntimeState::FailedTerminal + ); + assert_eq!(coordinator.stats.repo_tasks_total, 0); + assert_eq!(coordinator.stats.repo_tasks_reused, 1); + } + + #[test] + fn blacklist_rsync_dead_terminates_after_rrdp_failure() { + // Only rsync blacklisted: rrdp is attempted; when it fails, the rsync + // fallback is short-circuited into an instant terminal failure. + let mut seed = DeadRepoBlacklist::new(); + seed.record_transport_failure(RepoTransportMode::Rsync, DEAD_BASE, 100, 1, 256); + let mut coordinator = coordinator_with_blacklist(3, seed); + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + let completion = coordinator + .complete_transport_result( + transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::TransportFetch), + time::OffsetDateTime::UNIX_EPOCH, + ) + .expect("complete"); + assert!(completion.follow_up_tasks.is_empty()); + let record = coordinator + .runtime_record(&dead_identity()) + .expect("record"); + assert_eq!( + record.state, + crate::parallel::types::RepoRuntimeState::FailedTerminal + ); + let terminal = record.terminal_failure.as_ref().expect("terminal failure"); + match &terminal.result { + RepoTransportResultKind::Failed { detail, .. } => { + assert!(detail.contains("dead repo blacklist")); + } + other => panic!("expected failed result, got {other:?}"), + } + } + + #[test] + fn blacklist_success_resets_failure_counter() { + let mut coordinator = coordinator_with_blacklist(3, DeadRepoBlacklist::new()); + // One transport failure starts counting. + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + coordinator + .complete_transport_result( + transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::TransportFetch), + time::OffsetDateTime::UNIX_EPOCH, + ) + .expect("complete"); + assert_eq!(coordinator.dead_repo_blacklist().expect("blacklist").len(), 1); + + // A later success clears the entry. + coordinator.reset_run_state(); + let action = register_dead(&mut coordinator); + assert!(matches!( + action, + crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_) + )); + let mut success = transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::Unknown); + success.result = RepoTransportResultKind::Success { + source: "rrdp".to_string(), + warnings: Vec::new(), + }; + coordinator + .complete_transport_result(success, time::OffsetDateTime::UNIX_EPOCH) + .expect("complete"); + assert!( + coordinator + .dead_repo_blacklist() + .expect("blacklist") + .is_empty() + ); + } } diff --git a/src/parallel/transport_prefetch.rs b/src/parallel/transport_prefetch.rs index 85c471d..fac75e5 100644 --- a/src/parallel/transport_prefetch.rs +++ b/src/parallel/transport_prefetch.rs @@ -484,6 +484,7 @@ mod tests { result: crate::parallel::types::RepoTransportResultKind::Failed { detail: "timeout".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }); recorder.record_result(&crate::parallel::types::RepoTransportResultEnvelope { @@ -499,6 +500,7 @@ mod tests { result: crate::parallel::types::RepoTransportResultKind::Failed { detail: "timeout".to_string(), warnings: Vec::new(), + error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }); diff --git a/src/parallel/types.rs b/src/parallel/types.rs index 48a6a71..bf0195e 100644 --- a/src/parallel/types.rs +++ b/src/parallel/types.rs @@ -1,5 +1,7 @@ use std::path::{Path, PathBuf}; +use serde::{Deserialize, Serialize}; + use crate::policy::SyncPreference; use crate::report::Warning; @@ -96,7 +98,8 @@ pub enum RepoDedupKey { RsyncScope { rsync_scope_uri: String }, } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum RepoTransportMode { Rrdp, Rsync, @@ -116,6 +119,26 @@ pub struct RepoTransportTask { pub requesters: Vec, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RepoTransportErrorClass { + /// Transport-layer fetch failure (connect timeout/refused/TLS/DNS/broken + /// connection). Only this class counts toward the dead-repo blacklist. + TransportFetch, + /// Protocol-level failure (HTTP status, XML/parse, content mismatch). + Protocol, + /// Local storage failure. + Storage, + /// Unclassified (synthesized or legacy results); never counted. + Unknown, +} + +impl Default for RepoTransportErrorClass { + fn default() -> Self { + RepoTransportErrorClass::Unknown + } +} + #[derive(Clone, Debug, PartialEq, Eq)] pub enum RepoTransportResultKind { Success { @@ -125,6 +148,7 @@ pub enum RepoTransportResultKind { Failed { detail: String, warnings: Vec, + error_class: RepoTransportErrorClass, }, } @@ -479,6 +503,7 @@ mod tests { result: RepoTransportResultKind::Failed { detail: "timeout".to_string(), warnings: vec![Warning::new("timeout")], + error_class: super::RepoTransportErrorClass::Unknown, }, }; assert!(matches!(ok.result, RepoTransportResultKind::Success { .. })); diff --git a/src/tools/rpki_artifact_metrics.rs b/src/tools/rpki_artifact_metrics.rs index b435fa9..2bf6709 100644 --- a/src/tools/rpki_artifact_metrics.rs +++ b/src/tools/rpki_artifact_metrics.rs @@ -433,6 +433,8 @@ struct LatestRunMetrics { download_bytes: Option, artifact_sizes: BTreeMap, state_path_sizes: BTreeMap, + dead_repo_blacklist_entries: Option, + dead_repo_blacklist_blacklisted: Option, } #[derive(Clone, Debug, Default, Serialize)] @@ -1377,6 +1379,10 @@ fn build_latest_metrics(record: &RunRecord, snapshot: &mut MetricsSnapshot) { latest.download_bytes = json_u64(summary, &["stageTiming", "download_bytes_total"]); latest.artifact_sizes = extract_artifact_sizes(summary.get("artifacts")); latest.state_path_sizes = extract_path_sizes(summary.get("pathStats")); + latest.dead_repo_blacklist_entries = + json_u64(summary, &["deadRepoBlacklist", "entries"]); + latest.dead_repo_blacklist_blacklisted = + json_u64(summary, &["deadRepoBlacklist", "blacklisted"]); } parse_report(&record.path.join("report.json"), snapshot, &mut latest); @@ -2419,6 +2425,22 @@ fn render_latest_metrics(writer: &mut PromWriter<'_>, instance: &str, latest: &L value as f64, ); } + if let Some(value) = latest.dead_repo_blacklist_entries { + writer.gauge( + "ours_rp_dead_repo_blacklist_entries", + "Dead-repo transport blacklist entries (latest run)", + &[label("instance", instance), label("state", "total")], + value as f64, + ); + } + if let Some(value) = latest.dead_repo_blacklist_blacklisted { + writer.gauge( + "ours_rp_dead_repo_blacklist_entries", + "Dead-repo transport blacklist entries (latest run)", + &[label("instance", instance), label("state", "blacklisted")], + value as f64, + ); + } for (stage, value) in &latest.stage_seconds { writer.gauge( "ours_rp_run_stage_duration_seconds", @@ -3893,4 +3915,37 @@ mod tests { }); encode_content_info(&ci).expect("encode ccr") } + + #[test] + fn dead_repo_blacklist_gauges_render_from_latest_summary() { + let td = TempDir::new().expect("tempdir"); + write_success_run(td.path(), "run_0001", 1); + // Rewrite the summary with the blacklist section present. + let run = td.path().join("runs").join("run_0001"); + fs::write( + run.join("run-summary.json"), + r#"{"runSeq":1,"runId":"run_0001","startedAtRfc3339Utc":"2026-07-21T00:00:00Z","finishedAtRfc3339Utc":"2026-07-21T00:00:10Z","wallMs":10000,"status":"success","exitCode":0,"deadRepoBlacklist":{"path":"/state/dead-repo-blacklist.json","entries":5,"blacklisted":3}}"#, + ) + .expect("summary"); + let snapshot = scan_run_root(td.path(), "test").expect("scan"); + let latest = snapshot.latest_run.as_ref().expect("latest run"); + assert_eq!(latest.dead_repo_blacklist_entries, Some(5)); + assert_eq!(latest.dead_repo_blacklist_blacklisted, Some(3)); + let metrics = render_metrics(&snapshot); + assert!(metrics.contains( + r#"ours_rp_dead_repo_blacklist_entries{instance="test",state="total"} 5"# + )); + assert!(metrics.contains( + r#"ours_rp_dead_repo_blacklist_entries{instance="test",state="blacklisted"} 3"# + )); + } + + #[test] + fn dead_repo_blacklist_gauges_absent_without_section() { + let td = TempDir::new().expect("tempdir"); + write_success_run(td.path(), "run_0001", 1); + let snapshot = scan_run_root(td.path(), "test").expect("scan"); + let metrics = render_metrics(&snapshot); + assert!(!metrics.contains("ours_rp_dead_repo_blacklist_entries")); + } } diff --git a/src/tools/rpki_daemon.rs b/src/tools/rpki_daemon.rs index 3d2d8b8..68c955d 100644 --- a/src/tools/rpki_daemon.rs +++ b/src/tools/rpki_daemon.rs @@ -6,6 +6,9 @@ use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::time::Duration; +use crate::parallel::dead_repo_blacklist::DeadRepoBlacklist; +use crate::parallel::types::RepoTransportMode; + #[derive(Clone, Debug, PartialEq, Eq)] struct Args { state_root: PathBuf, @@ -21,6 +24,12 @@ struct Args { db_stats_bin: Option, db_stats_exact_every: Option, time_bin: Option, + /// Dead-repo blacklist (#141): enables daemon-side health checks and + /// auto-injects the child `--dead-repo-blacklist` flag when absent. + dead_repo_blacklist: Option, + dead_repo_blacklist_fail_threshold: Option, + dead_repo_health_check_interval_secs: u64, + dead_repo_probe_rsync_bin: PathBuf, child_args: Vec, } @@ -52,6 +61,23 @@ struct DaemonStatus { current_run_seq: Option, current_run_id: Option, last_run_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + dead_repo_blacklist: Option, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +struct DeadRepoBlacklistStatus { + path: String, + entries: usize, + blacklisted: usize, + last_health_check_at_rfc3339_utc: Option, +} + +/// Mutable daemon-side health check bookkeeping (#141). +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct DeadRepoHealthRuntime { + last_health_check_at: Option, } #[derive(Clone, Debug, Serialize, PartialEq, Eq)] @@ -145,6 +171,16 @@ struct RunSummary { db_stats: Vec, retention_deleted_runs: Vec, artifacts: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + dead_repo_blacklist: Option, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +struct DeadRepoBlacklistRunSummary { + path: String, + entries: usize, + blacklisted: usize, } fn usage() -> String { @@ -170,6 +206,17 @@ Options: Run db_stats --exact every n runs (default: disabled) --time-bin GNU time binary for child process metrics (default: /usr/bin/time when present) --no-time-wrapper Disable GNU time wrapper + --dead-repo-blacklist + Enable the dead-repo transport blacklist (#141): daemon probes + blacklisted entries between runs and removes revived repos; + auto-injects the child flag when not already present + --dead-repo-blacklist-fail-threshold + Admission threshold forwarded to the child when its args do not + already set it (default: child default 3) + --dead-repo-health-check-interval-secs + Minimum seconds between health check sweeps (default: 600) + --dead-repo-probe-rsync-bin + rsync binary used for health probes (default: rsync) --help Show this help Child argument placeholders: @@ -211,12 +258,16 @@ fn parse_args(argv: &[String]) -> Result { let mut db_stats_exact_every = None; let mut time_bin = default_time_bin(); let mut no_time_wrapper = false; + let mut dead_repo_blacklist: Option = None; + let mut dead_repo_blacklist_fail_threshold: Option = None; + let mut dead_repo_health_check_interval_secs = 600u64; + let mut dead_repo_probe_rsync_bin = PathBuf::from("rsync"); let mut i = 1usize; while i < argv.len() { match argv[i].as_str() { "--" => { - let child_args = argv[i + 1..].to_vec(); + let mut child_args = argv[i + 1..].to_vec(); let state_root = state_root.ok_or_else(|| format!("--state-root is required\n\n{}", usage()))?; let work_db = work_db.unwrap_or_else(|| state_root.join("state").join("work-db")); @@ -225,6 +276,23 @@ fn parse_args(argv: &[String]) -> Result { if no_time_wrapper { time_bin = None; } + // Dead-repo blacklist (#141): forward enablement to the child + // unless its args already carry the flag explicitly. + if let Some(path) = dead_repo_blacklist.as_ref() { + if !child_args.iter().any(|arg| arg == "--dead-repo-blacklist") { + child_args.push("--dead-repo-blacklist".to_string()); + child_args.push(path_string(path)); + } + if let Some(threshold) = dead_repo_blacklist_fail_threshold { + if !child_args + .iter() + .any(|arg| arg == "--dead-repo-blacklist-fail-threshold") + { + child_args.push("--dead-repo-blacklist-fail-threshold".to_string()); + child_args.push(threshold.to_string()); + } + } + } let args = Args { state_root, rpki_bin: rpki_bin @@ -240,6 +308,10 @@ fn parse_args(argv: &[String]) -> Result { db_stats_bin, db_stats_exact_every, time_bin, + dead_repo_blacklist, + dead_repo_blacklist_fail_threshold, + dead_repo_health_check_interval_secs, + dead_repo_probe_rsync_bin, child_args, }; return validate_args(args); @@ -315,6 +387,33 @@ fn parse_args(argv: &[String]) -> Result { "--no-time-wrapper" => { no_time_wrapper = true; } + "--dead-repo-blacklist" => { + i += 1; + dead_repo_blacklist = Some(PathBuf::from(value_at(argv, i, "--dead-repo-blacklist")?)); + } + "--dead-repo-blacklist-fail-threshold" => { + i += 1; + let parsed = parse_u64( + value_at(argv, i, "--dead-repo-blacklist-fail-threshold")?, + "--dead-repo-blacklist-fail-threshold", + )?; + if parsed == 0 || parsed > u32::MAX as u64 { + return Err("--dead-repo-blacklist-fail-threshold must be in 1..=u32::MAX".to_string()); + } + dead_repo_blacklist_fail_threshold = Some(parsed as u32); + } + "--dead-repo-health-check-interval-secs" => { + i += 1; + dead_repo_health_check_interval_secs = parse_u64( + value_at(argv, i, "--dead-repo-health-check-interval-secs")?, + "--dead-repo-health-check-interval-secs", + )?; + } + "--dead-repo-probe-rsync-bin" => { + i += 1; + dead_repo_probe_rsync_bin = + PathBuf::from(value_at(argv, i, "--dead-repo-probe-rsync-bin")?); + } other => return Err(format!("unknown argument: {other}\n\n{}", usage())), } i += 1; @@ -330,6 +429,11 @@ fn validate_args(args: Args) -> Result { usage() )); } + if args.dead_repo_blacklist_fail_threshold.is_some() && args.dead_repo_blacklist.is_none() { + return Err( + "--dead-repo-blacklist-fail-threshold requires --dead-repo-blacklist".to_string(), + ); + } Ok(args) } @@ -387,10 +491,19 @@ fn write_json_pretty(path: &Path, value: &T) -> Result<(), String> fs::create_dir_all(parent) .map_err(|e| format!("create parent dir failed: {}: {e}", parent.display()))?; } - let file = - File::create(path).map_err(|e| format!("create json failed: {}: {e}", path.display()))?; - serde_json::to_writer_pretty(file, value) - .map_err(|e| format!("write json failed: {}: {e}", path.display())) + let bytes = serde_json::to_vec_pretty(value) + .map_err(|e| format!("serialize json failed: {}: {e}", path.display()))?; + // Atomic write (tmp + rename) so concurrent readers never see a torn file. + let mut tmp_name = path + .file_name() + .map(|name| name.to_os_string()) + .unwrap_or_default(); + tmp_name.push(".tmp"); + let tmp_path = path.with_file_name(tmp_name); + fs::write(&tmp_path, &bytes) + .map_err(|e| format!("write json tmp failed: {}: {e}", tmp_path.display()))?; + fs::rename(&tmp_path, path) + .map_err(|e| format!("rename json tmp failed: {} -> {}: {e}", tmp_path.display(), path.display())) } fn append_json_line(path: &Path, value: &T) -> Result<(), String> { @@ -409,12 +522,30 @@ fn append_json_line(path: &Path, value: &T) -> Result<(), String> .map_err(|e| format!("flush jsonl failed: {}: {e}", path.display())) } +fn dead_repo_blacklist_status( + args: &Args, + health: &DeadRepoHealthRuntime, +) -> Option { + let path = args.dead_repo_blacklist.as_ref()?; + let (blacklist, _) = DeadRepoBlacklist::load(path); + let last_health_check_at_rfc3339_utc = health + .last_health_check_at + .and_then(|t| format_rfc3339(t).ok()); + Some(DeadRepoBlacklistStatus { + path: path_string(path), + entries: blacklist.len(), + blacklisted: blacklist.blacklisted_len(), + last_health_check_at_rfc3339_utc, + }) +} + fn write_status( args: &Args, state: DaemonState, runs_completed: u64, current: Option<&RunContext>, last_run_id: Option, + health: &DeadRepoHealthRuntime, ) -> Result<(), String> { let updated_at_rfc3339_utc = format_rfc3339(utc_now())?; let status = DaemonStatus { @@ -425,10 +556,131 @@ fn write_status( current_run_seq: current.map(|ctx| ctx.seq), current_run_id: current.map(|ctx| ctx.run_id.clone()), last_run_id, + dead_repo_blacklist: dead_repo_blacklist_status(args, health), }; write_json_pretty(&status_path(args), &status) } +/// Short-timeout HTTP GET of the RRDP notification file. Any HTTP status +/// (even 4xx) proves the transport is alive; only transport errors keep the +/// entry blacklisted. +fn probe_rrdp_transport(uri: &str) -> bool { + let config = crate::fetch::http::HttpFetcherConfig { + connect_timeout: Duration::from_secs(3), + timeout: Duration::from_secs(6), + large_body_timeout: Duration::from_secs(6), + ..Default::default() + }; + let fetcher = match crate::fetch::http::BlockingHttpFetcher::new(config) { + Ok(fetcher) => fetcher, + Err(_) => return false, + }; + match fetcher.fetch_bytes(uri) { + Ok(_) => true, + Err(err) => err.starts_with("http status"), + } +} + +/// rsync list-only probe: connecting to the daemon and listing the module +/// root is enough to prove the transport is alive. +fn probe_rsync_transport(rsync_bin: &Path, base_uri: &str) -> bool { + Command::new(rsync_bin) + .arg("--contimeout=5") + .arg("--timeout=8") + .arg(base_uri) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false) +} + +/// Probe due blacklist entries between runs (#141). Runs in the supervisor +/// loop while no child is active, preserving the single-writer assumption. +fn maybe_run_dead_repo_health_check(args: &Args, health: &mut DeadRepoHealthRuntime) { + let Some(path) = args.dead_repo_blacklist.clone() else { + return; + }; + let now = utc_now(); + if let Some(last) = health.last_health_check_at { + let elapsed = (now - last).whole_seconds(); + if elapsed >= 0 && (elapsed as u64) < args.dead_repo_health_check_interval_secs { + return; + } + } + let (mut blacklist, warning) = DeadRepoBlacklist::load(&path); + if let Some(warning) = warning { + eprintln!("[dead-repo-health] {warning}"); + } + let now_unix = now.unix_timestamp().max(0) as u64; + let due = blacklist.probe_due_entries(now_unix, args.dead_repo_health_check_interval_secs); + if due.is_empty() { + health.last_health_check_at = Some(now); + return; + } + // Probe in parallel: in the docker soak the daemon runs once per run + // (--max-runs 1), so a serial sweep of ~12 dead entries would add up to a + // minute of latency to every run. + let rsync_bin = args.dead_repo_probe_rsync_bin.clone(); + let probe_results: Vec<(RepoTransportMode, String, bool)> = due + .iter() + .map(|entry| (entry.transport, entry.uri.clone())) + .collect::>() + .into_iter() + .map(|(transport, uri)| { + let rsync_bin = rsync_bin.clone(); + std::thread::spawn(move || { + let alive = match transport { + RepoTransportMode::Rrdp => probe_rrdp_transport(&uri), + RepoTransportMode::Rsync => probe_rsync_transport(&rsync_bin, &uri), + }; + (transport, uri, alive) + }) + }) + .filter_map(|handle| handle.join().ok()) + .collect(); + let mut removed = 0usize; + let mut still_dead = 0usize; + for (transport, uri, alive) in &probe_results { + if *alive { + blacklist.record_probe_success(*transport, uri); + removed += 1; + eprintln!("[dead-repo-health] removed {} {} (probe ok)", transport.as_str(), uri); + crate::progress_log::emit( + "dead_repo_blacklist_remove", + serde_json::json!({ + "transport": transport.as_str(), + "uri": uri, + "reason": "health_probe_ok", + }), + ); + } else { + blacklist.record_probe_failure(*transport, uri, now_unix); + still_dead += 1; + eprintln!( + "[dead-repo-health] still dead {} {} (probe failed)", + transport.as_str(), + uri + ); + } + } + if let Err(err) = blacklist.store_atomic(&path, now_unix) { + eprintln!( + "[dead-repo-health] persist blacklist failed: {}: {err}", + path.display() + ); + } + crate::progress_log::emit( + "dead_repo_health_check", + serde_json::json!({ + "probed": probe_results.len(), + "removed": removed, + "still_dead": still_dead, + }), + ); + health.last_health_check_at = Some(now); +} + fn render_child_args(args: &[String], daemon_args: &Args, ctx: &RunContext) -> Vec { args.iter() .map(|arg| { @@ -634,6 +886,7 @@ fn run_child_once(args: &Args, ctx: &RunContext) -> Result { db_stats: Vec::new(), retention_deleted_runs: Vec::new(), artifacts, + dead_repo_blacklist: None, }; Ok(summary) } @@ -1031,6 +1284,14 @@ fn collect_post_run_metrics(args: &Args, ctx: &RunContext, summary: &mut RunSumm summary.path_stats = collect_state_path_stats(args, ctx); summary.db_stats = collect_db_stats(args, ctx); summary.artifacts = collect_artifacts(&ctx.run_dir).unwrap_or_default(); + if let Some(path) = args.dead_repo_blacklist.as_ref() { + let (blacklist, _) = DeadRepoBlacklist::load(path); + summary.dead_repo_blacklist = Some(DeadRepoBlacklistRunSummary { + path: path_string(path), + entries: blacklist.len(), + blacklisted: blacklist.blacklisted_len(), + }); + } } fn run_daemon(args: &Args) -> Result<(), String> { @@ -1050,13 +1311,17 @@ fn run_daemon(args: &Args) -> Result<(), String> { let mut runs_completed = 0u64; let mut next_seq = 1u64; let mut last_run_id = None; + let mut health = DeadRepoHealthRuntime::default(); write_status( args, DaemonState::Starting, runs_completed, None, last_run_id.clone(), + &health, )?; + // Probe once at startup so revived repos are removed before the first run. + maybe_run_dead_repo_health_check(args, &mut health); loop { if args.max_runs.is_some_and(|max| runs_completed >= max) { @@ -1069,6 +1334,7 @@ fn run_daemon(args: &Args) -> Result<(), String> { runs_completed, None, last_run_id.clone(), + &health, )?; let ctx = make_run_context(args, next_seq, utc_now()); write_status( @@ -1077,6 +1343,7 @@ fn run_daemon(args: &Args) -> Result<(), String> { runs_completed, Some(&ctx), last_run_id.clone(), + &health, )?; let mut summary = run_child_once(args, &ctx)?; write_status( @@ -1085,6 +1352,7 @@ fn run_daemon(args: &Args) -> Result<(), String> { runs_completed, Some(&ctx), last_run_id.clone(), + &health, )?; collect_post_run_metrics(args, &ctx, &mut summary); let removed = apply_retention(&args.state_root.join("runs"), args.retain_runs)?; @@ -1099,19 +1367,29 @@ fn run_daemon(args: &Args) -> Result<(), String> { if args.max_runs.is_some_and(|max| runs_completed >= max) { break; } + // Health probes run only while no child is active (single-writer). + maybe_run_dead_repo_health_check(args, &mut health); write_status( args, DaemonState::Sleeping, runs_completed, None, last_run_id.clone(), + &health, )?; if args.interval_secs > 0 { std::thread::sleep(Duration::from_secs(args.interval_secs)); } } - write_status(args, DaemonState::Exited, runs_completed, None, last_run_id) + write_status( + args, + DaemonState::Exited, + runs_completed, + None, + last_run_id, + &health, + ) } pub fn main_entry() -> i32 { @@ -1154,6 +1432,10 @@ mod tests { db_stats_bin: None, db_stats_exact_every: None, time_bin: None, + dead_repo_blacklist: None, + dead_repo_blacklist_fail_threshold: None, + dead_repo_health_check_interval_secs: 600, + dead_repo_probe_rsync_bin: PathBuf::from("rsync"), child_args: vec!["--version".to_string()], } } @@ -1330,6 +1612,10 @@ mod tests { db_stats_bin: None, db_stats_exact_every: None, time_bin: None, + dead_repo_blacklist: None, + dead_repo_blacklist_fail_threshold: None, + dead_repo_health_check_interval_secs: 600, + dead_repo_probe_rsync_bin: PathBuf::from("rsync"), child_args: vec![ "{state_root}/state/work-db".to_string(), "{run_out}/result.ccr".to_string(), @@ -1652,6 +1938,10 @@ mod tests { db_stats_bin: None, db_stats_exact_every: None, time_bin: None, + dead_repo_blacklist: None, + dead_repo_blacklist_fail_threshold: None, + dead_repo_health_check_interval_secs: 600, + dead_repo_probe_rsync_bin: PathBuf::from("rsync"), child_args: vec![ "-c".to_string(), "echo stdout-{run_seq}; echo stderr-{run_seq} >&2; echo marker > {run_out}/marker.txt".to_string(), @@ -1743,6 +2033,10 @@ mod tests { db_stats_bin: Some(db_stats_bin), db_stats_exact_every: Some(1), time_bin: None, + dead_repo_blacklist: None, + dead_repo_blacklist_fail_threshold: None, + dead_repo_health_check_interval_secs: 600, + dead_repo_probe_rsync_bin: PathBuf::from("rsync"), child_args: vec![ "-c".to_string(), "mkdir -p {state_root}/state/work-db {state_root}/state/repo-bytes.db; \ @@ -1781,4 +2075,130 @@ mod tests { .any(|item| item["label"] == "work_db" && item["exists"] == true) ); } + + #[test] + fn dead_repo_blacklist_flag_is_injected_into_child_args() { + let argv = vec![ + "rpki_daemon".to_string(), + "--state-root".to_string(), + "/tmp/daemon".to_string(), + "--rpki-bin".to_string(), + "/bin/echo".to_string(), + "--dead-repo-blacklist".to_string(), + "/tmp/daemon/state/dead-repo-blacklist.json".to_string(), + "--dead-repo-blacklist-fail-threshold".to_string(), + "2".to_string(), + "--".to_string(), + "--db".to_string(), + "{state_root}/state/work-db".to_string(), + ]; + let args = parse_args(&argv).expect("parse"); + assert_eq!( + args.child_args, + vec![ + "--db".to_string(), + "{state_root}/state/work-db".to_string(), + "--dead-repo-blacklist".to_string(), + "/tmp/daemon/state/dead-repo-blacklist.json".to_string(), + "--dead-repo-blacklist-fail-threshold".to_string(), + "2".to_string(), + ] + ); + } + + #[test] + fn dead_repo_blacklist_injection_respects_existing_child_flags() { + let argv = vec![ + "rpki_daemon".to_string(), + "--state-root".to_string(), + "/tmp/daemon".to_string(), + "--rpki-bin".to_string(), + "/bin/echo".to_string(), + "--dead-repo-blacklist".to_string(), + "/tmp/daemon/state/dead-repo-blacklist.json".to_string(), + "--".to_string(), + "--dead-repo-blacklist".to_string(), + "/custom/path.json".to_string(), + ]; + let args = parse_args(&argv).expect("parse"); + assert_eq!( + args.child_args, + vec![ + "--dead-repo-blacklist".to_string(), + "/custom/path.json".to_string(), + ] + ); + } + + #[test] + fn dead_repo_threshold_without_blacklist_is_rejected() { + let argv = vec![ + "rpki_daemon".to_string(), + "--state-root".to_string(), + "/tmp/daemon".to_string(), + "--rpki-bin".to_string(), + "/bin/echo".to_string(), + "--dead-repo-blacklist-fail-threshold".to_string(), + "2".to_string(), + "--".to_string(), + "--version".to_string(), + ]; + let err = parse_args(&argv).expect_err("parse should fail"); + assert!(err.contains("requires --dead-repo-blacklist"), "{err}"); + } + + #[test] + fn health_check_keeps_dead_entry_and_removes_revived_entry() { + use std::io::{Read as _, Write as _}; + use std::net::TcpListener; + + // Revived RRDP host: one-shot HTTP server answering 200. + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let port = listener.local_addr().expect("local addr").port(); + std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf); + let body = b""; + let _ = stream.write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .as_bytes(), + ); + let _ = stream.write_all(body); + } + }); + + let td = tempfile::tempdir().expect("tempdir"); + let blacklist_path = td.path().join("dead-repo-blacklist.json"); + let revived_uri = format!("http://127.0.0.1:{port}/notification.xml"); + let dead_uri = "http://127.0.0.1:1/notification.xml".to_string(); + let mut blacklist = DeadRepoBlacklist::new(); + for uri in [&revived_uri, &dead_uri] { + blacklist.record_transport_failure(RepoTransportMode::Rrdp, uri, 100, 1, 256); + } + blacklist + .store_atomic(&blacklist_path, 100) + .expect("store blacklist"); + + let mut args = test_args(td.path().join("daemon")); + args.dead_repo_blacklist = Some(blacklist_path.clone()); + args.dead_repo_health_check_interval_secs = 0; + let mut health = DeadRepoHealthRuntime::default(); + maybe_run_dead_repo_health_check(&args, &mut health); + + let (after, warning) = DeadRepoBlacklist::load(&blacklist_path); + assert!(warning.is_none()); + assert!(!after.is_blacklisted(RepoTransportMode::Rrdp, &revived_uri)); + assert!(after.is_blacklisted(RepoTransportMode::Rrdp, &dead_uri)); + assert!(health.last_health_check_at.is_some()); + + // Status JSON exposes the blacklist section. + let status = dead_repo_blacklist_status(&args, &health).expect("status"); + assert_eq!(status.entries, 1); + assert_eq!(status.blacklisted, 1); + assert!(status.last_health_check_at_rfc3339_utc.is_some()); + } } diff --git a/src/validation/run_tree_from_tal.rs b/src/validation/run_tree_from_tal.rs index 4de925f..09524c6 100644 --- a/src/validation/run_tree_from_tal.rs +++ b/src/validation/run_tree_from_tal.rs @@ -345,6 +345,56 @@ fn persist_transport_request_prefetch_snapshot( Ok(()) } +/// Persist the dead-repo blacklist working copy at run end (#141). Failures +/// degrade to a progress event; a broken blacklist file must never fail a run. +fn persist_dead_repo_blacklist( + runtime: &Arc, + timing: Option<&TimingHandle>, +) -> Result<(), RunTreeFromTalError> { + let Some((blacklist_config, blacklist)) = runtime.dead_repo_blacklist_state() else { + return Ok(()); + }; + let persist_started = std::time::Instant::now(); + let now_unix = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or(0); + match blacklist.store_atomic(&blacklist_config.path, now_unix) { + Ok(()) => { + let persist_ms = persist_started.elapsed().as_millis() as u64; + record_transport_prefetch_count( + timing, + "dead_repo_blacklist_entries", + blacklist.len() as u64, + ); + record_transport_prefetch_count( + timing, + "dead_repo_blacklist_blacklisted", + blacklist.blacklisted_len() as u64, + ); + crate::progress_log::emit( + "dead_repo_blacklist_persisted", + serde_json::json!({ + "path": blacklist_config.path.display().to_string(), + "entries": blacklist.len(), + "blacklisted": blacklist.blacklisted_len(), + "persist_ms": persist_ms, + }), + ); + } + Err(err) => { + crate::progress_log::emit( + "dead_repo_blacklist_persist_error", + serde_json::json!({ + "path": blacklist_config.path.display().to_string(), + "error": err.to_string(), + }), + ); + } + } + Ok(()) +} + fn root_discovery_from_tal_input( tal_input: &TalInputSpec, http_fetcher: &dyn Fetcher, @@ -813,6 +863,7 @@ where run_tree_serial_audit(root, &runner, config)? }; persist_transport_request_prefetch_snapshot(store.as_ref(), &runtime, config, timing.as_ref())?; + persist_dead_repo_blacklist(&runtime, timing.as_ref())?; let downloads = download_log.snapshot_events(); let download_stats = DownloadLogHandle::stats_from_events(&downloads); Ok(RunTreeFromTalAuditOutput { @@ -936,6 +987,7 @@ where run_tree_serial_audit_multi_root(root_handles, &runner, config)? }; persist_transport_request_prefetch_snapshot(store.as_ref(), &runtime, config, timing.as_ref())?; + persist_dead_repo_blacklist(&runtime, timing.as_ref())?; let downloads = download_log.snapshot_events(); let download_stats = DownloadLogHandle::stats_from_events(&downloads); Ok(RunTreeFromTalAuditOutput {