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, RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, RepoTransportTask, }; use crate::policy::SyncPreference; use crate::report::Warning; #[derive(Clone, Debug, PartialEq, Eq)] pub enum RepoRequestAction { Enqueued(RepoSyncTask), Waiting, Reused(RepoSyncResultRef), FailedReuse { detail: String }, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RepoCompletion { pub repo_key: RepoKey, pub released_requesters: Vec, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum TransportTaskState { Pending, Running, Finished, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct TransportInFlightEntry { pub state: TransportTaskState, pub task: RepoTransportTask, pub waiting_requesters: Vec, pub last_result: Option, pub started_at: Option, pub finished_at: Option, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RepoRuntimeRecord { pub identity: RepoIdentity, pub state: RepoRuntimeState, pub rrdp_notification_key: Option, pub rsync_scope_key: String, pub rsync_failure_scope_key: Option, pub requesters: Vec, pub validation_time: time::OffsetDateTime, pub priority: u8, pub last_success: Option, pub terminal_failure: Option, } #[derive(Clone, Debug, PartialEq, Eq)] pub enum TransportRequestAction { Enqueue(RepoTransportTask), Waiting { state: RepoRuntimeState }, ReusedSuccess(RepoTransportResultEnvelope), ReusedTerminalFailure(RepoTransportResultEnvelope), } #[derive(Clone, Debug, PartialEq, Eq)] pub struct TransportCompletion { pub released_requesters: Vec, pub follow_up_tasks: Vec, } #[derive(Default)] pub struct TransportStateTables { rrdp_inflight: HashMap, rsync_inflight: HashMap, rsync_failure_by_scope: HashMap, 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 { pub fn new() -> Self { 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) } pub fn finalized_runtime_records_for_transport( &self, dedup_key: &RepoDedupKey, ) -> Vec { self.runtime_records .values() .filter(|record| match dedup_key { RepoDedupKey::RrdpNotify { notification_uri } => { record.rrdp_notification_key.as_deref() == Some(notification_uri.as_str()) } RepoDedupKey::RsyncScope { rsync_scope_uri } => { record.rsync_scope_key == *rsync_scope_uri } }) .filter(|record| { matches!( record.state, RepoRuntimeState::RrdpOk | RepoRuntimeState::RsyncOk | RepoRuntimeState::FailedTerminal ) }) .cloned() .collect() } pub fn finalized_runtime_records_for_transport_result( &self, result: &RepoTransportResultEnvelope, ) -> Vec { let failure_scope = reusable_rsync_failure_scope(result); self.runtime_records .values() .filter(|record| { let exact_match = match &result.dedup_key { RepoDedupKey::RrdpNotify { notification_uri } => { record.rrdp_notification_key.as_deref() == Some(notification_uri.as_str()) } RepoDedupKey::RsyncScope { rsync_scope_uri } => { record.rsync_scope_key == *rsync_scope_uri } }; let failure_scope_match = failure_scope .map(|scope| record.rsync_failure_scope_key.as_deref() == Some(scope)) .unwrap_or(false); exact_match || failure_scope_match }) .filter(|record| { matches!( record.state, RepoRuntimeState::RrdpOk | RepoRuntimeState::RsyncOk | RepoRuntimeState::FailedTerminal ) }) .cloned() .collect() } pub fn reset_run_state(&mut self) { self.rrdp_inflight.clear(); self.rsync_inflight.clear(); self.rsync_failure_by_scope.clear(); self.rsync_failure_probe_inflight.clear(); self.rsync_failure_scope_reachable.clear(); self.runtime_records.clear(); } pub fn register_transport_request( &mut self, identity: RepoIdentity, requester: RepoRequester, validation_time: time::OffsetDateTime, priority: u8, rsync_scope_uri: String, rsync_failure_scope_uri: Option, sync_preference: SyncPreference, ) -> TransportRequestAction { if let Some(record) = self.runtime_records.get_mut(&identity) { record.requesters.push(requester.clone()); return match record.state { RepoRuntimeState::WaitingRrdp => { if let Some(key) = record.rrdp_notification_key.as_ref() { if let Some(entry) = self.rrdp_inflight.get_mut(key) { entry.waiting_requesters.push(requester); } } TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRrdp, } } RepoRuntimeState::RrdpOk | RepoRuntimeState::RsyncOk => { TransportRequestAction::ReusedSuccess( record .last_success .clone() .expect("success state must keep last_success"), ) } RepoRuntimeState::RrdpFailedPendingRsync | RepoRuntimeState::WaitingRsync => { if let Some(entry) = self.rsync_inflight.get_mut(&record.rsync_scope_key) { entry.waiting_requesters.push(requester); } record.state = RepoRuntimeState::WaitingRsync; TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync, } } RepoRuntimeState::FailedTerminal => TransportRequestAction::ReusedTerminalFailure( record .terminal_failure .clone() .expect("terminal failure must keep last result"), ), RepoRuntimeState::Init => TransportRequestAction::Waiting { state: RepoRuntimeState::Init, }, }; } // 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) { if let Some(result) = entry.last_result.clone() { return match result.result { RepoTransportResultKind::Success { .. } => { self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::RrdpOk, rrdp_notification_key: Some(notification_uri), rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: Some(result.clone()), terminal_failure: None, }, ); TransportRequestAction::ReusedSuccess(result) } RepoTransportResultKind::Failed { .. } => { self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity: identity.clone(), state: RepoRuntimeState::RrdpFailedPendingRsync, rrdp_notification_key: Some(notification_uri), rsync_scope_key: rsync_scope_uri.clone(), rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester.clone()], validation_time, priority, last_success: None, terminal_failure: None, }, ); self.register_rsync_request( identity, requester, validation_time, priority, rsync_scope_uri, rsync_failure_scope_uri, ) } }; } entry.waiting_requesters.push(requester.clone()); self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::WaitingRrdp, rrdp_notification_key: Some(notification_uri), rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: None, }, ); return TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRrdp, }; } let task = RepoTransportTask { dedup_key: RepoDedupKey::RrdpNotify { notification_uri: notification_uri.clone(), }, rsync_failure_scope_uri: None, repo_identity: identity.clone(), mode: RepoTransportMode::Rrdp, retry_short_timeout: false, tal_id: requester.tal_id.clone(), rir_id: requester.rir_id.clone(), validation_time, priority, requesters: vec![requester.clone()], }; self.rrdp_inflight.insert( notification_uri.clone(), TransportInFlightEntry { state: TransportTaskState::Pending, task: task.clone(), waiting_requesters: Vec::new(), last_result: None, started_at: None, finished_at: None, }, ); self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::WaitingRrdp, rrdp_notification_key: Some(notification_uri), rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: None, }, ); return TransportRequestAction::Enqueue(task); } } self.register_rsync_request( identity, requester, validation_time, priority, rsync_scope_uri, rsync_failure_scope_uri, ) } fn register_rsync_request( &mut self, identity: RepoIdentity, requester: RepoRequester, validation_time: time::OffsetDateTime, priority: u8, rsync_scope_uri: String, rsync_failure_scope_uri: Option, ) -> TransportRequestAction { if let Some(entry) = self.rsync_inflight.get_mut(&rsync_scope_uri) { if let Some(result) = entry.last_result.clone() { return match result.result { RepoTransportResultKind::Success { .. } => { self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::RsyncOk, rrdp_notification_key: None, rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: Some(result.clone()), terminal_failure: None, }, ); TransportRequestAction::ReusedSuccess(result) } RepoTransportResultKind::Failed { .. } => { 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.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: Some(result.clone()), }, ); TransportRequestAction::ReusedTerminalFailure(result) } }; } entry.waiting_requesters.push(requester.clone()); self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::WaitingRsync, rrdp_notification_key: None, rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: None, }, ); return TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync, }; } if let Some(failure_scope_uri) = rsync_failure_scope_uri.as_ref() { if let Some(result) = self.rsync_failure_by_scope.get(failure_scope_uri).cloned() { 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.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: Some(result.clone()), }, ); return TransportRequestAction::ReusedTerminalFailure(result); } if !self .rsync_failure_scope_reachable .contains(failure_scope_uri) { if self .rsync_failure_probe_inflight .get(failure_scope_uri) .is_some() { self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::WaitingRsync, rrdp_notification_key: None, rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: None, }, ); return TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync, }; } } } let task = RepoTransportTask { dedup_key: RepoDedupKey::RsyncScope { rsync_scope_uri: rsync_scope_uri.clone(), }, rsync_failure_scope_uri: rsync_failure_scope_uri.clone(), repo_identity: identity.clone(), mode: RepoTransportMode::Rsync, retry_short_timeout: false, tal_id: requester.tal_id.clone(), rir_id: requester.rir_id.clone(), validation_time, priority, requesters: vec![requester.clone()], }; self.rsync_inflight.insert( rsync_scope_uri.clone(), TransportInFlightEntry { state: TransportTaskState::Pending, task: task.clone(), waiting_requesters: Vec::new(), last_result: None, started_at: None, finished_at: None, }, ); if let Some(failure_scope_uri) = rsync_failure_scope_uri.as_ref() { if !self .rsync_failure_scope_reachable .contains(failure_scope_uri) { self.rsync_failure_probe_inflight .insert(failure_scope_uri.clone(), rsync_scope_uri.clone()); } } self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { identity, state: RepoRuntimeState::WaitingRsync, rrdp_notification_key: None, rsync_scope_key: rsync_scope_uri, rsync_failure_scope_key: rsync_failure_scope_uri.clone(), requesters: vec![requester], validation_time, priority, last_success: None, terminal_failure: None, }, ); TransportRequestAction::Enqueue(task) } fn schedule_rsync_for_record( record: &mut RepoRuntimeRecord, rsync_inflight: &mut HashMap, 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(); if let Some(entry) = rsync_inflight.get_mut(&rsync_scope_uri) { if let Some(result) = entry.last_result.clone() { match result.result { RepoTransportResultKind::Success { .. } => { record.state = RepoRuntimeState::RsyncOk; record.last_success = Some(result); } RepoTransportResultKind::Failed { .. } => { record.state = RepoRuntimeState::FailedTerminal; record.terminal_failure = Some(result); } } return; } entry.waiting_requesters.extend(record.requesters.clone()); record.state = RepoRuntimeState::WaitingRsync; return; } if let Some(failure_scope_uri) = record.rsync_failure_scope_key.as_ref() { if let Some(result) = rsync_failure_by_scope.get(failure_scope_uri).cloned() { record.state = RepoRuntimeState::FailedTerminal; record.terminal_failure = Some(result); return; } if !rsync_failure_scope_reachable.contains(failure_scope_uri) && rsync_failure_probe_inflight.contains_key(failure_scope_uri) { record.state = RepoRuntimeState::WaitingRsync; return; } } // 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() .expect("rsync record must keep at least one requester"); let task = RepoTransportTask { dedup_key: RepoDedupKey::RsyncScope { rsync_scope_uri: rsync_scope_uri.clone(), }, rsync_failure_scope_uri: record.rsync_failure_scope_key.clone(), repo_identity: record.identity.clone(), mode: RepoTransportMode::Rsync, retry_short_timeout: false, tal_id: first_requester.tal_id.clone(), rir_id: first_requester.rir_id.clone(), validation_time: record.validation_time, priority: record.priority, requesters: record.requesters.clone(), }; rsync_inflight.insert( rsync_scope_uri.clone(), TransportInFlightEntry { state: TransportTaskState::Pending, task: task.clone(), waiting_requesters: Vec::new(), last_result: None, started_at: None, finished_at: None, }, ); if let Some(failure_scope_uri) = record.rsync_failure_scope_key.as_ref() { if !rsync_failure_scope_reachable.contains(failure_scope_uri) { rsync_failure_probe_inflight.insert(failure_scope_uri.clone(), rsync_scope_uri); } } record.state = RepoRuntimeState::WaitingRsync; follow_up_tasks.push(task); } pub fn mark_transport_running( &mut self, dedup_key: &RepoDedupKey, started_at: time::OffsetDateTime, ) -> Result<(), String> { match dedup_key { RepoDedupKey::RrdpNotify { notification_uri } => { let entry = self .rrdp_inflight .get_mut(notification_uri) .ok_or_else(|| format!("rrdp transport not found: {notification_uri}"))?; entry.state = TransportTaskState::Running; entry.started_at = Some(started_at); } RepoDedupKey::RsyncScope { rsync_scope_uri } => { let entry = self .rsync_inflight .get_mut(rsync_scope_uri) .ok_or_else(|| format!("rsync transport not found: {rsync_scope_uri}"))?; entry.state = TransportTaskState::Running; entry.started_at = Some(started_at); } } Ok(()) } pub fn complete_transport_result( &mut self, result: RepoTransportResultEnvelope, finished_at: time::OffsetDateTime, ) -> Result { match (&result.dedup_key, &result.result) { ( RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Success { .. }, ) => { let entry = self .rrdp_inflight .get_mut(notification_uri) .ok_or_else(|| format!("rrdp transport not found: {notification_uri}"))?; entry.state = TransportTaskState::Finished; entry.finished_at = Some(finished_at); entry.last_result = Some(result.clone()); let released_requesters = std::mem::take(&mut entry.waiting_requesters); for record in self.runtime_records.values_mut() { if record.rrdp_notification_key.as_deref() == Some(notification_uri) && record.state == RepoRuntimeState::WaitingRrdp { record.state = RepoRuntimeState::RrdpOk; record.last_success = Some(result.clone()); } } Ok(TransportCompletion { released_requesters, follow_up_tasks: Vec::new(), }) } ( RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Failed { .. }, ) => { let entry = self .rrdp_inflight .get_mut(notification_uri) .ok_or_else(|| format!("rrdp transport not found: {notification_uri}"))?; entry.state = TransportTaskState::Finished; entry.finished_at = Some(finished_at); entry.last_result = Some(result.clone()); let mut follow_up_tasks = Vec::new(); for record in self.runtime_records.values_mut() { if record.rrdp_notification_key.as_deref() == Some(notification_uri) && record.state == RepoRuntimeState::WaitingRrdp { record.state = RepoRuntimeState::RrdpFailedPendingRsync; Self::schedule_rsync_for_record( record, &mut self.rsync_inflight, &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, ); } } Ok(TransportCompletion { released_requesters: Vec::new(), follow_up_tasks, }) } ( RepoDedupKey::RsyncScope { rsync_scope_uri }, RepoTransportResultKind::Success { .. }, ) => { let mut follow_up_tasks = Vec::new(); let entry = self .rsync_inflight .get_mut(rsync_scope_uri) .ok_or_else(|| format!("rsync transport not found: {rsync_scope_uri}"))?; entry.state = TransportTaskState::Finished; entry.finished_at = Some(finished_at); entry.last_result = Some(result.clone()); if let Some(failure_scope_uri) = result.rsync_failure_scope_uri.as_ref() { self.rsync_failure_probe_inflight.remove(failure_scope_uri); self.rsync_failure_scope_reachable .insert(failure_scope_uri.clone()); } let released_requesters = std::mem::take(&mut entry.waiting_requesters); for record in self.runtime_records.values_mut() { if record.rsync_scope_key == *rsync_scope_uri && matches!( record.state, RepoRuntimeState::WaitingRsync | RepoRuntimeState::RrdpFailedPendingRsync ) { record.state = RepoRuntimeState::RsyncOk; record.last_success = Some(result.clone()); } } if let Some(failure_scope_uri) = result.rsync_failure_scope_uri.as_ref() { for record in self.runtime_records.values_mut() { if record.rsync_scope_key != *rsync_scope_uri && record.rsync_failure_scope_key.as_deref() == Some(failure_scope_uri.as_str()) && matches!(record.state, RepoRuntimeState::WaitingRsync) { Self::schedule_rsync_for_record( record, &mut self.rsync_inflight, &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, ); } } } Ok(TransportCompletion { released_requesters, follow_up_tasks, }) } ( RepoDedupKey::RsyncScope { rsync_scope_uri }, RepoTransportResultKind::Failed { .. }, ) => { let entry = self .rsync_inflight .get_mut(rsync_scope_uri) .ok_or_else(|| format!("rsync transport not found: {rsync_scope_uri}"))?; entry.state = TransportTaskState::Finished; entry.finished_at = Some(finished_at); entry.last_result = Some(result.clone()); let reusable_failure_scope = reusable_rsync_failure_scope(&result).map(str::to_string); if let Some(failure_scope_uri) = result.rsync_failure_scope_uri.as_ref() { self.rsync_failure_probe_inflight.remove(failure_scope_uri); if reusable_failure_scope.as_deref() == Some(failure_scope_uri.as_str()) { self.rsync_failure_by_scope .insert(failure_scope_uri.clone(), result.clone()); } else { self.rsync_failure_scope_reachable .insert(failure_scope_uri.clone()); } } let released_requesters = std::mem::take(&mut entry.waiting_requesters); for record in self.runtime_records.values_mut() { if (record.rsync_scope_key == *rsync_scope_uri || reusable_failure_scope .as_deref() .map(|scope| record.rsync_failure_scope_key.as_deref() == Some(scope)) .unwrap_or(false)) && matches!( record.state, RepoRuntimeState::WaitingRsync | RepoRuntimeState::RrdpFailedPendingRsync ) { record.state = RepoRuntimeState::FailedTerminal; record.terminal_failure = Some(result.clone()); } } if let Some(failure_scope_uri) = result.rsync_failure_scope_uri.as_ref() { if reusable_failure_scope.as_deref() != Some(failure_scope_uri.as_str()) { let mut follow_up_tasks = Vec::new(); for record in self.runtime_records.values_mut() { if record.rsync_scope_key != *rsync_scope_uri && record.rsync_failure_scope_key.as_deref() == Some(failure_scope_uri.as_str()) && matches!(record.state, RepoRuntimeState::WaitingRsync) { Self::schedule_rsync_for_record( record, &mut self.rsync_inflight, &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, ); } } return Ok(TransportCompletion { released_requesters, follow_up_tasks, }); } } Ok(TransportCompletion { released_requesters, follow_up_tasks: Vec::new(), }) } } } } fn reusable_rsync_failure_scope(result: &RepoTransportResultEnvelope) -> Option<&str> { let scope = result.rsync_failure_scope_uri.as_deref()?; match (&result.mode, &result.result) { (RepoTransportMode::Rsync, RepoTransportResultKind::Failed { detail, .. }) if is_host_level_rsync_failure(detail) => { Some(scope) } _ => None, } } 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") || lower.contains("no route to host") || lower.contains("network is unreachable") || lower.contains("connection refused") || lower.contains("name or service not known") || lower.contains("temporary failure in name resolution") } #[derive(Default)] pub struct InFlightRepoTable { entries: HashMap, } impl InFlightRepoTable { pub fn new() -> Self { Self::default() } pub fn len(&self) -> usize { self.entries.len() } pub fn is_empty(&self) -> bool { self.entries.is_empty() } pub fn reset_run_state(&mut self) { self.entries.clear(); } pub fn get(&self, key: &RepoKey) -> Option<&InFlightRepoEntry> { self.entries.get(key) } pub fn last_result(&self, key: &RepoKey) -> Option<&RepoSyncResultEnvelope> { self.entries .get(key) .and_then(|entry| entry.last_result.as_ref()) } pub fn register_request( &mut self, repo_key: RepoKey, requester: RepoRequester, validation_time: time::OffsetDateTime, sync_preference: SyncPreference, priority: u8, ) -> RepoRequestAction { match self.entries.get_mut(&repo_key) { Some(entry) => match entry.state { RepoTaskState::Pending | RepoTaskState::Running => { entry.waiting_requesters.push(requester); RepoRequestAction::Waiting } RepoTaskState::Succeeded | RepoTaskState::Reused => { RepoRequestAction::Reused(entry.result_ref.clone().expect("result_ref exists")) } RepoTaskState::Failed => RepoRequestAction::FailedReuse { detail: entry .last_error .clone() .unwrap_or_else(|| "repo sync failed".to_string()), }, }, None => { let task = RepoSyncTask { repo_key: repo_key.clone(), validation_time, sync_preference, tal_id: requester.tal_id.clone(), rir_id: requester.rir_id.clone(), priority, requesters: vec![requester], }; self.entries.insert( repo_key, InFlightRepoEntry { state: RepoTaskState::Pending, task_ref: Some(task.clone()), waiting_requesters: Vec::new(), result_ref: None, last_result: None, last_error: None, started_at: None, finished_at: None, }, ); RepoRequestAction::Enqueued(task) } } } pub fn mark_running( &mut self, repo_key: &RepoKey, started_at: time::OffsetDateTime, ) -> Result<(), String> { let entry = self .entries .get_mut(repo_key) .ok_or_else(|| format!("repo not found: {}", repo_key.rsync_base_uri))?; match entry.state { RepoTaskState::Pending => { entry.state = RepoTaskState::Running; entry.started_at = Some(started_at); Ok(()) } other => Err(format!("repo cannot transition to running from {other:?}")), } } pub fn complete_success( &mut self, repo_key: &RepoKey, result: RepoSyncResultEnvelope, finished_at: time::OffsetDateTime, ) -> Result { let entry = self .entries .get_mut(repo_key) .ok_or_else(|| format!("repo not found: {}", repo_key.rsync_base_uri))?; match entry.state { RepoTaskState::Pending | RepoTaskState::Running => { let result_ref = match &result.result { RepoSyncResultKind::Success(result_ref) | RepoSyncResultKind::Reused(result_ref) => result_ref.clone(), RepoSyncResultKind::Failed { detail } => { return Err(format!( "success completion called with failure result: {detail}" )); } }; entry.state = RepoTaskState::Succeeded; entry.result_ref = Some(result_ref); entry.last_result = Some(result); entry.last_error = None; entry.finished_at = Some(finished_at); entry.task_ref = None; let released_requesters = std::mem::take(&mut entry.waiting_requesters); Ok(RepoCompletion { repo_key: repo_key.clone(), released_requesters, }) } other => Err(format!("repo cannot transition to success from {other:?}")), } } pub fn complete_failure( &mut self, repo_key: &RepoKey, result: RepoSyncResultEnvelope, finished_at: time::OffsetDateTime, ) -> Result { let entry = self .entries .get_mut(repo_key) .ok_or_else(|| format!("repo not found: {}", repo_key.rsync_base_uri))?; match entry.state { RepoTaskState::Pending | RepoTaskState::Running => { let detail = match &result.result { RepoSyncResultKind::Failed { detail } => detail.clone(), RepoSyncResultKind::Success(_) | RepoSyncResultKind::Reused(_) => { return Err("failure completion called with success result".to_string()); } }; entry.state = RepoTaskState::Failed; entry.result_ref = None; entry.last_result = Some(result); entry.last_error = Some(detail); entry.finished_at = Some(finished_at); entry.task_ref = None; let released_requesters = std::mem::take(&mut entry.waiting_requesters); Ok(RepoCompletion { repo_key: repo_key.clone(), released_requesters, }) } other => Err(format!("repo cannot transition to failure from {other:?}")), } } } #[cfg(test)] mod tests { use crate::parallel::repo_scheduler::{InFlightRepoTable, RepoRequestAction}; use crate::parallel::types::{ RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncResultKind, RepoSyncResultRef, RepoTaskState, }; use crate::policy::SyncPreference; fn requester(tal_id: &str, rir_id: &str, manifest: &str) -> RepoRequester { RepoRequester { tal_id: tal_id.to_string(), rir_id: rir_id.to_string(), parent_node_id: None, ca_instance_handle_id: format!("{tal_id}:{manifest}"), publication_point_rsync_uri: "rsync://example.test/repo/".to_string(), manifest_rsync_uri: manifest.to_string(), } } #[test] fn inflight_repo_table_stores_entries_by_repo_key() { let key = RepoKey::new("rsync://example.test/repo/", None); let mut table = InFlightRepoTable::new(); let action = table.register_request( key.clone(), requester("arin", "arin", "rsync://example.test/repo/root.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); assert!(matches!(action, RepoRequestAction::Enqueued(_))); assert_eq!(table.len(), 1); assert_eq!( table.get(&key).map(|entry| entry.state), Some(RepoTaskState::Pending) ); } #[test] fn running_repo_request_is_merged_into_waiting_list() { let key = RepoKey::new("rsync://example.test/repo/", None); let mut table = InFlightRepoTable::new(); let _ = table.register_request( key.clone(), requester("arin", "arin", "rsync://example.test/repo/root.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); table .mark_running(&key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark running"); let action = table.register_request( key.clone(), requester("apnic", "apnic", "rsync://example.test/repo/other.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); assert_eq!(action, RepoRequestAction::Waiting); assert_eq!(table.get(&key).unwrap().waiting_requesters.len(), 1); } #[test] fn complete_success_releases_waiting_requesters_and_reuses_result() { let key = RepoKey::new("rsync://example.test/repo/", None); let mut table = InFlightRepoTable::new(); let _ = table.register_request( key.clone(), requester("arin", "arin", "rsync://example.test/repo/root.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); table .mark_running(&key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark running"); let waiting = requester("ripe", "ripe", "rsync://example.test/repo/child.mft"); let _ = table.register_request( key.clone(), waiting.clone(), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); let result_ref = RepoSyncResultRef { repo_key: key.clone(), source: "rrdp".to_string(), }; let completion = table .complete_success( &key, RepoSyncResultEnvelope { repo_key: key.clone(), tal_id: "arin".to_string(), rir_id: "arin".to_string(), result: RepoSyncResultKind::Success(result_ref.clone()), phase: Some("rrdp_ok".to_string()), timing_ms: 10, warnings: Vec::new(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete success"); assert_eq!(completion.released_requesters, vec![waiting]); assert_eq!(table.get(&key).unwrap().state, RepoTaskState::Succeeded); let action = table.register_request( key, requester("afrinic", "afrinic", "rsync://example.test/repo/again.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); assert_eq!(action, RepoRequestAction::Reused(result_ref)); } #[test] fn complete_failure_releases_waiting_requesters_and_reuses_failure() { let key = RepoKey::new("rsync://example.test/repo/", None); let mut table = InFlightRepoTable::new(); let _ = table.register_request( key.clone(), requester("arin", "arin", "rsync://example.test/repo/root.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); table .mark_running(&key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark running"); let waiting = requester("ripe", "ripe", "rsync://example.test/repo/child.mft"); let _ = table.register_request( key.clone(), waiting.clone(), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); let completion = table .complete_failure( &key, RepoSyncResultEnvelope { repo_key: key.clone(), tal_id: "arin".to_string(), rir_id: "arin".to_string(), result: RepoSyncResultKind::Failed { detail: "network timeout".to_string(), }, phase: Some("rrdp_failed_rsync_failed".to_string()), timing_ms: 10, warnings: Vec::new(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete failure"); assert_eq!(completion.released_requesters, vec![waiting]); assert_eq!(table.get(&key).unwrap().state, RepoTaskState::Failed); let action = table.register_request( key, requester("afrinic", "afrinic", "rsync://example.test/repo/again.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); assert_eq!( action, RepoRequestAction::FailedReuse { detail: "network timeout".to_string() } ); } #[test] fn requesters_from_different_tals_do_not_lose_identity() { let key = RepoKey::new("rsync://shared.example/repo/", None); let mut table = InFlightRepoTable::new(); let _ = table.register_request( key.clone(), requester("arin", "arin", "rsync://shared.example/repo/a.mft"), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); table .mark_running(&key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark running"); let wait_apnic = requester("apnic", "apnic", "rsync://shared.example/repo/b.mft"); let wait_ripe = requester("ripe", "ripe", "rsync://shared.example/repo/c.mft"); let _ = table.register_request( key.clone(), wait_apnic.clone(), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); let _ = table.register_request( key.clone(), wait_ripe.clone(), time::OffsetDateTime::UNIX_EPOCH, SyncPreference::RrdpThenRsync, 0, ); let completion = table .complete_success( &key, RepoSyncResultEnvelope { repo_key: key.clone(), tal_id: "arin".to_string(), rir_id: "arin".to_string(), result: RepoSyncResultKind::Success(RepoSyncResultRef { repo_key: key.clone(), source: "rrdp".to_string(), }), phase: Some("rrdp_ok".to_string()), timing_ms: 10, warnings: Vec::new(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete success"); assert_eq!(completion.released_requesters, vec![wait_apnic, wait_ripe]); } } #[cfg(test)] mod transport_tests { use crate::parallel::repo_scheduler::{ TransportRequestAction, TransportStateTables, TransportTaskState, }; use crate::parallel::types::{ RepoDedupKey, RepoIdentity, RepoRequester, RepoRuntimeState, RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, }; use crate::policy::SyncPreference; fn requester(id: &str) -> RepoRequester { RepoRequester::with_tal_rir( "apnic", "apnic", format!("rsync://example.test/repo/{id}.mft"), "rsync://example.test/repo/".to_string(), format!("node:{id}"), ) } #[test] fn register_transport_request_enqueues_initial_rrdp_task() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/", ); let action = tables.register_transport_request( identity.clone(), requester("root"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); let task = match action { TransportRequestAction::Enqueue(task) => task, other => panic!("expected enqueue, got {other:?}"), }; assert_eq!(task.mode, RepoTransportMode::Rrdp); assert_eq!( task.dedup_key, RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string() } ); let key = "https://example.test/notify.xml".to_string(); assert_eq!( tables.rrdp_inflight.get(&key).unwrap().state, TransportTaskState::Pending ); assert_eq!( tables.runtime_records.get(&identity).unwrap().state, RepoRuntimeState::WaitingRrdp ); } #[test] fn register_transport_request_waits_on_existing_rrdp_task() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/a/", ); let _ = tables.register_transport_request( identity.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); let other_identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/b/", ); let action = tables.register_transport_request( other_identity.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); assert_eq!( action, TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRrdp } ); assert_eq!( tables.runtime_records.get(&other_identity).unwrap().state, RepoRuntimeState::WaitingRrdp ); assert_eq!( tables .rrdp_inflight .get("https://example.test/notify.xml") .unwrap() .waiting_requesters .len(), 1 ); } #[test] fn complete_rrdp_success_reuses_for_later_identity_requests() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/a/", ); let _ = tables.register_transport_request( identity.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); tables .mark_transport_running( &RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("mark running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, rsync_failure_scope_uri: None, repo_identity: identity.clone(), mode: RepoTransportMode::Rrdp, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 10, result: RepoTransportResultKind::Success { source: "rrdp".to_string(), warnings: Vec::new(), }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete success"); assert!(completion.follow_up_tasks.is_empty()); let later_identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/b/", ); let action = tables.register_transport_request( later_identity.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); assert!(matches!(action, TransportRequestAction::ReusedSuccess(_))); assert_eq!( tables.runtime_records.get(&later_identity).unwrap().state, RepoRuntimeState::RrdpOk ); let same_identity_action = tables.register_transport_request( later_identity.clone(), requester("b-again"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); assert!(matches!( same_identity_action, TransportRequestAction::ReusedSuccess(_) )); let finalized = tables.finalized_runtime_records_for_transport(&RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }); assert_eq!(finalized.len(), 2); } #[test] fn complete_rrdp_failure_enqueues_rsync_follow_up() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/a/", ); let _ = tables.register_transport_request( identity.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); tables .mark_transport_running( &RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("mark running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, rsync_failure_scope_uri: None, repo_identity: identity.clone(), mode: RepoTransportMode::Rrdp, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 10, result: RepoTransportResultKind::Failed { detail: "rrdp timeout".to_string(), warnings: Vec::new(), error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete failure"); assert_eq!(completion.follow_up_tasks.len(), 1); assert_eq!(completion.follow_up_tasks[0].mode, RepoTransportMode::Rsync); assert_eq!( tables.runtime_records.get(&identity).unwrap().state, RepoRuntimeState::WaitingRsync ); let same_identity_action = tables.register_transport_request( identity.clone(), requester("a-again"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); assert_eq!( same_identity_action, TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync } ); let later_identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/b/", ); let later_action = tables.register_transport_request( later_identity.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RrdpThenRsync, ); assert_eq!( later_action, TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync } ); assert_eq!( tables.runtime_records.get(&later_identity).unwrap().state, RepoRuntimeState::WaitingRsync ); } #[test] fn host_failure_scope_probes_once_then_reuses_terminal_failure() { let mut tables = TransportStateTables::new(); let identity_a = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/a/", ); let identity_b = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/b/", ); let _ = tables.register_transport_request( identity_a.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/a/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RrdpThenRsync, ); let _ = tables.register_transport_request( identity_b.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/b/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RrdpThenRsync, ); tables .mark_transport_running( &RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("mark running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: RepoDedupKey::RrdpNotify { notification_uri: "https://example.test/notify.xml".to_string(), }, rsync_failure_scope_uri: None, repo_identity: identity_a.clone(), mode: RepoTransportMode::Rrdp, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 10, result: RepoTransportResultKind::Failed { detail: "rrdp timeout".to_string(), warnings: Vec::new(), error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete rrdp failure"); assert_eq!(completion.follow_up_tasks.len(), 1); assert_eq!( completion.follow_up_tasks[0] .rsync_failure_scope_uri .as_deref(), Some("rsync://example.test/") ); let probe_task = completion.follow_up_tasks[0].clone(); tables .mark_transport_running(&probe_task.dedup_key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark rsync running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: probe_task.dedup_key, rsync_failure_scope_uri: probe_task.rsync_failure_scope_uri, repo_identity: probe_task.repo_identity, mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 15_000, 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, ) .expect("complete rsync failure"); assert!(completion.follow_up_tasks.is_empty()); assert_eq!( tables.runtime_records.get(&identity_a).unwrap().state, RepoRuntimeState::FailedTerminal ); assert_eq!( tables.runtime_records.get(&identity_b).unwrap().state, RepoRuntimeState::FailedTerminal ); let finalized = tables.finalized_runtime_records_for_transport_result( tables .runtime_records .get(&identity_a) .unwrap() .terminal_failure .as_ref() .unwrap(), ); assert_eq!(finalized.len(), 2); } #[test] fn complete_rsync_failure_reuses_terminal_failure() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let _ = tables.register_transport_request( identity.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); tables .mark_transport_running( &RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("mark running"); let _ = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string(), }, rsync_failure_scope_uri: None, repo_identity: identity.clone(), mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 20, result: RepoTransportResultKind::Failed { detail: "rsync timeout".to_string(), warnings: Vec::new(), error_class: crate::parallel::types::RepoTransportErrorClass::Unknown, }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete rsync failure"); let later_identity = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let action = tables.register_transport_request( later_identity, requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); assert!(matches!( action, TransportRequestAction::ReusedTerminalFailure(_) )); let finalized = tables.finalized_runtime_records_for_transport(&RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string(), }); assert_eq!(finalized.len(), 2); let same_identity_action = tables.register_transport_request( identity, requester("a-again"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); assert!(matches!( same_identity_action, TransportRequestAction::ReusedTerminalFailure(_) )); } #[test] fn complete_rsync_success_reuses_for_later_identity_requests() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let action = tables.register_transport_request( identity.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); assert!(matches!(action, TransportRequestAction::Enqueue(_))); tables .mark_transport_running( &RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string(), }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("mark running"); tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string(), }, rsync_failure_scope_uri: None, repo_identity: identity.clone(), mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 20, result: RepoTransportResultKind::Success { source: "rsync".to_string(), warnings: Vec::new(), }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete rsync success"); let later_identity = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let action = tables.register_transport_request( later_identity.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); assert!(matches!(action, TransportRequestAction::ReusedSuccess(_))); assert_eq!( tables.runtime_records.get(&later_identity).unwrap().state, RepoRuntimeState::RsyncOk ); } #[test] fn register_rsync_request_waits_on_existing_rsync_task() { let mut tables = TransportStateTables::new(); let identity_a = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let identity_b = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let _ = tables.register_transport_request( identity_a, requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); let action = tables.register_transport_request( identity_b.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); assert_eq!( action, TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync } ); assert_eq!( tables.runtime_records.get(&identity_b).unwrap().state, RepoRuntimeState::WaitingRsync ); assert_eq!( tables .rsync_inflight .get("rsync://example.test/module/") .unwrap() .waiting_requesters .len(), 1 ); } #[test] fn host_failure_scope_success_marks_host_reachable_and_schedules_waiters() { let mut tables = TransportStateTables::new(); let identity_a = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let identity_b = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let action_a = tables.register_transport_request( identity_a.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/a/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); let probe_task = match action_a { TransportRequestAction::Enqueue(task) => task, other => panic!("expected first host probe enqueue, got {other:?}"), }; let action_b = tables.register_transport_request( identity_b.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/b/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); assert_eq!( action_b, TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRsync } ); tables .mark_transport_running(&probe_task.dedup_key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark rsync running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: probe_task.dedup_key, rsync_failure_scope_uri: probe_task.rsync_failure_scope_uri, repo_identity: probe_task.repo_identity, mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 5, result: RepoTransportResultKind::Success { source: "rsync".to_string(), warnings: Vec::new(), }, }, time::OffsetDateTime::UNIX_EPOCH, ) .expect("complete rsync success"); assert_eq!( tables.runtime_records.get(&identity_a).unwrap().state, RepoRuntimeState::RsyncOk ); assert_eq!(completion.follow_up_tasks.len(), 1); assert_eq!( completion.follow_up_tasks[0].dedup_key, RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/repo/b/".to_string() } ); assert_eq!( tables.runtime_records.get(&identity_b).unwrap().state, RepoRuntimeState::WaitingRsync ); } #[test] fn non_host_level_rsync_failure_does_not_poison_host_scope() { let mut tables = TransportStateTables::new(); let identity_a = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let identity_b = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let action_a = tables.register_transport_request( identity_a.clone(), requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/a/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); let probe_task = match action_a { TransportRequestAction::Enqueue(task) => task, other => panic!("expected first host probe enqueue, got {other:?}"), }; let _ = tables.register_transport_request( identity_b.clone(), requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/b/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); tables .mark_transport_running(&probe_task.dedup_key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark rsync running"); let completion = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: probe_task.dedup_key, rsync_failure_scope_uri: probe_task.rsync_failure_scope_uri, repo_identity: probe_task.repo_identity, mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 5, 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, ) .expect("complete rsync failure"); assert_eq!( tables.runtime_records.get(&identity_a).unwrap().state, RepoRuntimeState::FailedTerminal ); assert_eq!(completion.follow_up_tasks.len(), 1); assert_eq!( completion.follow_up_tasks[0].dedup_key, RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/repo/b/".to_string() } ); } #[test] fn cached_host_level_failure_reuses_for_later_rsync_only_requests() { let mut tables = TransportStateTables::new(); let identity_a = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let action_a = tables.register_transport_request( identity_a, requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/a/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); let probe_task = match action_a { TransportRequestAction::Enqueue(task) => task, other => panic!("expected first host probe enqueue, got {other:?}"), }; tables .mark_transport_running(&probe_task.dedup_key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark rsync running"); let _ = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: probe_task.dedup_key, rsync_failure_scope_uri: probe_task.rsync_failure_scope_uri, repo_identity: probe_task.repo_identity, mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 15_000, 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, ) .expect("complete host failure"); let later_identity = RepoIdentity::new(None, "rsync://example.test/repo/c/"); let action = tables.register_transport_request( later_identity.clone(), requester("c"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/c/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); assert!(matches!( action, TransportRequestAction::ReusedTerminalFailure(_) )); assert_eq!( tables.runtime_records.get(&later_identity).unwrap().state, RepoRuntimeState::FailedTerminal ); } #[test] fn reset_run_state_clears_host_failure_scope_cache() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new(None, "rsync://example.test/repo/a/"); let action = tables.register_transport_request( identity, requester("a"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/a/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); let probe_task = match action { TransportRequestAction::Enqueue(task) => task, other => panic!("expected host probe enqueue, got {other:?}"), }; tables .mark_transport_running(&probe_task.dedup_key, time::OffsetDateTime::UNIX_EPOCH) .expect("mark rsync running"); let _ = tables .complete_transport_result( RepoTransportResultEnvelope { dedup_key: probe_task.dedup_key, rsync_failure_scope_uri: probe_task.rsync_failure_scope_uri, repo_identity: probe_task.repo_identity, mode: RepoTransportMode::Rsync, tal_id: "apnic".to_string(), rir_id: "apnic".to_string(), timing_ms: 15_000, 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, ) .expect("complete host failure"); tables.reset_run_state(); let later_identity = RepoIdentity::new(None, "rsync://example.test/repo/b/"); let action = tables.register_transport_request( later_identity, requester("b"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/repo/b/".to_string(), Some("rsync://example.test/".to_string()), SyncPreference::RsyncOnly, ); assert!(matches!(action, TransportRequestAction::Enqueue(_))); } #[test] fn register_transport_request_skips_rrdp_when_sync_preference_is_rsync_only() { let mut tables = TransportStateTables::new(); let identity = RepoIdentity::new( Some("https://example.test/notify.xml".to_string()), "rsync://example.test/repo/", ); let action = tables.register_transport_request( identity.clone(), requester("root"), time::OffsetDateTime::UNIX_EPOCH, 0, "rsync://example.test/module/".to_string(), None, SyncPreference::RsyncOnly, ); let task = match action { TransportRequestAction::Enqueue(task) => task, other => panic!("expected enqueue, got {other:?}"), }; assert_eq!(task.mode, RepoTransportMode::Rsync); assert_eq!( task.dedup_key, RepoDedupKey::RsyncScope { rsync_scope_uri: "rsync://example.test/module/".to_string() } ); assert!(tables.rrdp_inflight.is_empty()); assert_eq!( tables.runtime_records.get(&identity).unwrap().state, RepoRuntimeState::WaitingRsync ); } }