NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260605 ROA增量验证和rsync host失败短路

Browse Source
This commit is contained in:
yuyr
2026-06-06 15:16:38 +08:00
parent e597c7c124
commit 4045f9a3a5
27 changed files with 2850 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/parallel/repo_runtime.rs
View File

@@ -76,6 +76,7 @@ pub struct Phase1RepoSyncRuntime<E: RepoTransportExecutor> {
coordinator: Mutex<GlobalRunCoordinator>,
worker_pool: Mutex<RepoTransportWorkerPool<E>>,
rsync_scope_resolver: Arc<dyn Fn(&str) -> String + Send + Sync>,
rsync_failure_scope_resolver: Arc<dyn Fn(&str) -> Option<String> + Send + Sync>,
sync_preference: SyncPreference,
}
@@ -85,11 +86,28 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
worker_pool: RepoTransportWorkerPool<E>,
rsync_scope_resolver: Arc<dyn Fn(&str) -> String + Send + Sync>,
sync_preference: SyncPreference,
) -> Self {
Self::new_with_failure_scope(
coordinator,
worker_pool,
rsync_scope_resolver,
Arc::new(|_base: &str| None),
sync_preference,
)
}
pub fn new_with_failure_scope(
coordinator: GlobalRunCoordinator,
worker_pool: RepoTransportWorkerPool<E>,
rsync_scope_resolver: Arc<dyn Fn(&str) -> String + Send + Sync>,
rsync_failure_scope_resolver: Arc<dyn Fn(&str) -> Option<String> + Send + Sync>,
sync_preference: SyncPreference,
) -> Self {
Self {
coordinator: Mutex::new(coordinator),
worker_pool: Mutex::new(worker_pool),
rsync_scope_resolver,
rsync_failure_scope_resolver,
sync_preference,
}
}
@@ -117,6 +135,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
let identity = Self::build_identity(ca);
let requester = Self::build_requester(ca);
let rsync_scope_uri = (self.rsync_scope_resolver)(&identity.rsync_base_uri);
let rsync_failure_scope_uri = (self.rsync_failure_scope_resolver)(&identity.rsync_base_uri);
let action = {
let mut coordinator = self.coordinator.lock().expect("coordinator lock poisoned");
coordinator.register_transport_request(
@@ -125,6 +144,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
time::OffsetDateTime::now_utc(),
priority,
rsync_scope_uri,
rsync_failure_scope_uri,
self.sync_preference,
)
};
@@ -137,6 +157,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
"manifest_rsync_uri": ca.manifest_rsync_uri,
"publication_point_rsync_uri": ca.publication_point_rsync_uri,
"repo_key_rsync_base_uri": task.repo_identity.rsync_base_uri,
"rsync_failure_scope_uri": task.rsync_failure_scope_uri,
"repo_key_notification_uri": task.repo_identity.notification_uri,
"priority": priority,
"transport_mode": match task.mode {
@@ -160,6 +181,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
"manifest_rsync_uri": ca.manifest_rsync_uri,
"publication_point_rsync_uri": ca.publication_point_rsync_uri,
"repo_key_rsync_base_uri": identity.rsync_base_uri,
"rsync_failure_scope_uri": (self.rsync_failure_scope_resolver)(&identity.rsync_base_uri),
"repo_key_notification_uri": identity.notification_uri,
"priority": priority,
"runtime_state": format!("{state:?}"),
@@ -175,6 +197,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
"manifest_rsync_uri": ca.manifest_rsync_uri,
"publication_point_rsync_uri": ca.publication_point_rsync_uri,
"repo_key_rsync_base_uri": identity.rsync_base_uri,
"rsync_failure_scope_uri": result.rsync_failure_scope_uri,
"repo_key_notification_uri": identity.notification_uri,
"priority": priority,
"transport_mode": match result.mode {
@@ -213,6 +236,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
"phase1_repo_task_dispatched",
serde_json::json!({
"repo_key_rsync_base_uri": task.repo_identity.rsync_base_uri,
"rsync_failure_scope_uri": task.rsync_failure_scope_uri,
"repo_key_notification_uri": task.repo_identity.notification_uri,
"requester_count": task.requesters.len(),
"priority": task.priority,
@@ -240,11 +264,12 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
return Ok(None);
};
let transport_identity = envelope.repo_identity.clone();
let completed_dedup_key = envelope.dedup_key.clone();
let completed_envelope = envelope.clone();
crate::progress_log::emit(
"phase1_repo_task_result",
serde_json::json!({
"repo_key_rsync_base_uri": envelope.repo_identity.rsync_base_uri,
"rsync_failure_scope_uri": envelope.rsync_failure_scope_uri,
"repo_key_notification_uri": envelope.repo_identity.notification_uri,
"timing_ms": envelope.timing_ms,
"transport_mode": match envelope.mode {
@@ -283,7 +308,7 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
let completions = {
let coordinator = self.coordinator.lock().expect("coordinator lock poisoned");
coordinator
.finalized_runtime_records_for_transport(&completed_dedup_key)
.finalized_runtime_records_for_transport_result(&completed_envelope)
.into_iter()
.filter_map(|record| {
let outcome = match record.state {
@@ -506,6 +531,7 @@ mod tests {
fn execute_transport(&self, task: RepoTransportTask) -> RepoTransportResultEnvelope {
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: task.mode,
tal_id: task.tal_id,
@@ -531,6 +557,7 @@ mod tests {
self.count.fetch_add(1, Ordering::SeqCst);
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: task.mode,
tal_id: task.tal_id,
@@ -559,6 +586,7 @@ mod tests {
self.rrdp_count.fetch_add(1, Ordering::SeqCst);
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: RepoTransportMode::Rrdp,
tal_id: task.tal_id,
@@ -574,6 +602,7 @@ mod tests {
self.rsync_count.fetch_add(1, Ordering::SeqCst);
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: RepoTransportMode::Rsync,
tal_id: task.tal_id,
@@ -601,6 +630,7 @@ mod tests {
self.rrdp_count.fetch_add(1, Ordering::SeqCst);
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: RepoTransportMode::Rrdp,
tal_id: task.tal_id,
@@ -616,6 +646,7 @@ mod tests {
self.rsync_count.fetch_add(1, Ordering::SeqCst);
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri.clone(),
repo_identity: task.repo_identity,
mode: RepoTransportMode::Rsync,
tal_id: task.tal_id,

878
src/parallel/repo_scheduler.rs
View File

File diff suppressed because it is too large Load Diff

7
src/parallel/repo_worker.rs
View File

@@ -85,6 +85,7 @@ impl<H: Fetcher + 'static> RepoTransportExecutor for LiveRrdpTransportExecutor<H
) {
Ok(_) => 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,
@@ -97,6 +98,7 @@ impl<H: Fetcher + 'static> RepoTransportExecutor for LiveRrdpTransportExecutor<H
},
Err(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,
@@ -151,6 +153,7 @@ impl<R: RsyncFetcher + 'static> RepoTransportExecutor for LiveRsyncTransportExec
) {
Ok(_) => 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,
@@ -163,6 +166,7 @@ impl<R: RsyncFetcher + 'static> RepoTransportExecutor for LiveRsyncTransportExec
},
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,
@@ -660,6 +664,7 @@ mod tests {
dedup_key: RepoDedupKey::RrdpNotify {
notification_uri: notification_uri.to_string(),
},
rsync_failure_scope_uri: None,
repo_identity: RepoIdentity::new(Some(notification_uri.to_string()), rsync_base_uri),
mode: RepoTransportMode::Rrdp,
tal_id: "arin".to_string(),
@@ -684,6 +689,7 @@ mod tests {
dedup_key: RepoDedupKey::RsyncScope {
rsync_scope_uri: rsync_scope_uri.to_string(),
},
rsync_failure_scope_uri: None,
repo_identity: RepoIdentity::new(None, rsync_base_uri),
mode: RepoTransportMode::Rsync,
tal_id: "arin".to_string(),
@@ -873,6 +879,7 @@ mod tests {
fn execute_transport(&self, task: RepoTransportTask) -> RepoTransportResultEnvelope {
RepoTransportResultEnvelope {
dedup_key: task.dedup_key,
rsync_failure_scope_uri: task.rsync_failure_scope_uri,
repo_identity: task.repo_identity,
mode: task.mode,
tal_id: task.tal_id,

11
src/parallel/run_coordinator.rs
View File

@@ -123,6 +123,7 @@ impl GlobalRunCoordinator {
validation_time: time::OffsetDateTime,
priority: u8,
rsync_scope_uri: String,
rsync_failure_scope_uri: Option<String>,
sync_preference: SyncPreference,
) -> TransportRequestAction {
let action = self.transport_tables.register_transport_request(
@@ -131,6 +132,7 @@ impl GlobalRunCoordinator {
validation_time,
priority,
rsync_scope_uri,
rsync_failure_scope_uri,
sync_preference,
);
match &action {
@@ -205,6 +207,14 @@ impl GlobalRunCoordinator {
.finalized_runtime_records_for_transport(dedup_key)
}
pub fn finalized_runtime_records_for_transport_result(
&self,
result: &RepoTransportResultEnvelope,
) -> Vec<crate::parallel::repo_scheduler::RepoRuntimeRecord> {
self.transport_tables
.finalized_runtime_records_for_transport_result(result)
}
pub fn reset_run_state(&mut self) {
self.in_flight_repos.reset_run_state();
self.transport_tables.reset_run_state();
@@ -296,6 +306,7 @@ mod tests {
time::OffsetDateTime::UNIX_EPOCH,
0,
"rsync://example.test/repo/".to_string(),
None,
SyncPreference::RrdpThenRsync,
);
assert!(matches!(

5
src/parallel/types.rs
View File

@@ -105,6 +105,7 @@ pub enum RepoTransportMode {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RepoTransportTask {
pub dedup_key: RepoDedupKey,
pub rsync_failure_scope_uri: Option<String>,
pub repo_identity: RepoIdentity,
pub mode: RepoTransportMode,
pub tal_id: String,
@@ -129,6 +130,7 @@ pub enum RepoTransportResultKind {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RepoTransportResultEnvelope {
pub dedup_key: RepoDedupKey,
pub rsync_failure_scope_uri: Option<String>,
pub repo_identity: RepoIdentity,
pub mode: RepoTransportMode,
pub tal_id: String,
@@ -218,6 +220,7 @@ impl RepoSyncTask {
) -> RepoTransportTask {
RepoTransportTask {
dedup_key,
rsync_failure_scope_uri: None,
repo_identity: self.repo_key.as_identity(),
mode,
tal_id: self.tal_id.clone(),
@@ -450,6 +453,7 @@ mod tests {
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: "arin".to_string(),
@@ -464,6 +468,7 @@ mod tests {
dedup_key: RepoDedupKey::RsyncScope {
rsync_scope_uri: "rsync://example.test/module/".to_string(),
},
rsync_failure_scope_uri: None,
repo_identity: identity,
mode: RepoTransportMode::Rsync,
tal_id: "arin".to_string(),