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

20260416_2 并行优化phase1后进行snapshot fast path优化，通过四方面关键优化技术消除了验证主链路上snapshot构建是的db访问性能热点问题，性能热点转移到ROA验证处理本身，目前APNIC+ARIN全量同步从500秒压缩到212秒，离rpkclient 112秒差距变小

Browse Source
This commit is contained in:
yuyr
2026-04-17 14:58:47 +08:00
parent 38421b1ae7
commit 224ae10052
36 changed files with 2935 additions and 937 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/parallel/repo_runtime.rs
View File

@@ -5,8 +5,8 @@ use crate::parallel::repo_scheduler::TransportRequestAction;
use crate::parallel::repo_worker::{RepoTransportExecutor, RepoTransportWorkerPool};
use crate::parallel::run_coordinator::GlobalRunCoordinator;
use crate::parallel::types::{
RepoIdentity, RepoRuntimeState, RepoTransportMode, RepoTransportResultEnvelope,
RepoTransportResultKind, RepoRequester,
RepoIdentity, RepoRequester, RepoRuntimeState, RepoTransportMode, RepoTransportResultEnvelope,
RepoTransportResultKind,
};
use crate::policy::SyncPreference;
use crate::report::Warning;
@@ -160,7 +160,8 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
};
{
let mut coordinator = self.coordinator.lock().expect("coordinator lock poisoned");
coordinator.mark_transport_running(&task.dedup_key, time::OffsetDateTime::now_utc())?;
coordinator
.mark_transport_running(&task.dedup_key, time::OffsetDateTime::now_utc())?;
}
crate::progress_log::emit(
"phase1_repo_task_dispatched",
@@ -233,7 +234,9 @@ impl<E: RepoTransportExecutor> Phase1RepoSyncRuntime<E> {
fn runtime_state_for_identity(&self, identity: &RepoIdentity) -> Option<RepoRuntimeState> {
let coordinator = self.coordinator.lock().expect("coordinator lock poisoned");
coordinator.runtime_record(identity).map(|record| record.state)
coordinator
.runtime_record(identity)
.map(|record| record.state)
}
fn resolved_outcome_for_identity(
@@ -313,20 +316,21 @@ fn outcome_from_transport_result(
warnings: warnings.clone(),
}
}
(RepoTransportResultKind::Failed { detail, warnings }, RepoRuntimeState::FailedTerminal) => {
RepoSyncRuntimeOutcome {
repo_sync_ok: false,
repo_sync_err: Some(detail.clone()),
repo_sync_source: None,
repo_sync_phase: Some(if envelope.repo_identity.notification_uri.is_some() {
"rrdp_failed_rsync_failed".to_string()
} else {
"rsync_failed".to_string()
}),
repo_sync_duration_ms: envelope.timing_ms,
warnings: warnings.clone(),
}
}
(
RepoTransportResultKind::Failed { detail, warnings },
RepoRuntimeState::FailedTerminal,
) => RepoSyncRuntimeOutcome {
repo_sync_ok: false,
repo_sync_err: Some(detail.clone()),
repo_sync_source: None,
repo_sync_phase: Some(if envelope.repo_identity.notification_uri.is_some() {
"rrdp_failed_rsync_failed".to_string()
} else {
"rsync_failed".to_string()
}),
repo_sync_duration_ms: envelope.timing_ms,
warnings: warnings.clone(),
},
_ => RepoSyncRuntimeOutcome {
repo_sync_ok: false,
repo_sync_err: Some("repo runtime state unresolved".to_string()),
@@ -344,7 +348,6 @@ mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use crate::policy::SyncPreference;
use crate::parallel::config::ParallelPhase1Config;
use crate::parallel::repo_runtime::{Phase1RepoSyncRuntime, RepoSyncRuntime};
use crate::parallel::repo_worker::{
@@ -352,9 +355,10 @@ mod tests {
};
use crate::parallel::run_coordinator::GlobalRunCoordinator;
use crate::parallel::types::{
RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind,
RepoTransportTask, TalInputSpec,
RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, RepoTransportTask,
TalInputSpec,
};
use crate::policy::SyncPreference;
use crate::report::Warning;
use crate::validation::tree::{CaInstanceHandle, DiscoveredChildCaInstance};
@@ -492,7 +496,10 @@ mod tests {
.expect("sync repo");
assert!(outcome.repo_sync_ok);
assert_eq!(outcome.repo_sync_source.as_deref(), Some("rsync"));
assert_eq!(outcome.repo_sync_phase.as_deref(), Some("rrdp_failed_rsync_ok"));
assert_eq!(
outcome.repo_sync_phase.as_deref(),
Some("rrdp_failed_rsync_ok")
);
assert_eq!(rrdp_count.load(Ordering::SeqCst), 1);
assert_eq!(rsync_count.load(Ordering::SeqCst), 1);
}

65
src/parallel/repo_scheduler.rs
View File

@@ -1,11 +1,11 @@
use std::collections::HashMap;
use crate::policy::SyncPreference;
use crate::parallel::types::{
InFlightRepoEntry, RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoRuntimeState,
RepoSyncResultEnvelope, RepoSyncResultKind, RepoSyncResultRef, RepoSyncTask, RepoTaskState,
RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, RepoTransportTask,
};
use crate::policy::SyncPreference;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RepoRequestAction {
@@ -241,7 +241,13 @@ impl TransportStateTables {
}
}
self.register_rsync_request(identity, requester, validation_time, priority, rsync_scope_uri)
self.register_rsync_request(
identity,
requester,
validation_time,
priority,
rsync_scope_uri,
)
}
fn register_rsync_request(
@@ -384,7 +390,10 @@ impl TransportStateTables {
finished_at: time::OffsetDateTime,
) -> Result<TransportCompletion, String> {
match (&result.dedup_key, &result.result) {
(RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Success { .. }) => {
(
RepoDedupKey::RrdpNotify { notification_uri },
RepoTransportResultKind::Success { .. },
) => {
let entry = self
.rrdp_inflight
.get_mut(notification_uri)
@@ -406,7 +415,10 @@ impl TransportStateTables {
follow_up_tasks: Vec::new(),
})
}
(RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Failed { .. }) => {
(
RepoDedupKey::RrdpNotify { notification_uri },
RepoTransportResultKind::Failed { .. },
) => {
let entry = self
.rrdp_inflight
.get_mut(notification_uri)
@@ -422,7 +434,9 @@ impl TransportStateTables {
record.state = RepoRuntimeState::RrdpFailedPendingRsync;
let rsync_scope_uri = record.rsync_scope_key.clone();
if let Some(existing) = self.rsync_inflight.get_mut(&rsync_scope_uri) {
existing.waiting_requesters.extend(record.requesters.clone());
existing
.waiting_requesters
.extend(record.requesters.clone());
record.state = RepoRuntimeState::WaitingRsync;
} else {
let task = RepoTransportTask {
@@ -458,7 +472,10 @@ impl TransportStateTables {
follow_up_tasks,
})
}
(RepoDedupKey::RsyncScope { rsync_scope_uri }, RepoTransportResultKind::Success { .. }) => {
(
RepoDedupKey::RsyncScope { rsync_scope_uri },
RepoTransportResultKind::Success { .. },
) => {
let entry = self
.rsync_inflight
.get_mut(rsync_scope_uri)
@@ -471,7 +488,8 @@ impl TransportStateTables {
if record.rsync_scope_key == *rsync_scope_uri
&& matches!(
record.state,
RepoRuntimeState::WaitingRsync | RepoRuntimeState::RrdpFailedPendingRsync
RepoRuntimeState::WaitingRsync
| RepoRuntimeState::RrdpFailedPendingRsync
)
{
record.state = RepoRuntimeState::RsyncOk;
@@ -483,7 +501,10 @@ impl TransportStateTables {
follow_up_tasks: Vec::new(),
})
}
(RepoDedupKey::RsyncScope { rsync_scope_uri }, RepoTransportResultKind::Failed { .. }) => {
(
RepoDedupKey::RsyncScope { rsync_scope_uri },
RepoTransportResultKind::Failed { .. },
) => {
let entry = self
.rsync_inflight
.get_mut(rsync_scope_uri)
@@ -496,7 +517,8 @@ impl TransportStateTables {
if record.rsync_scope_key == *rsync_scope_uri
&& matches!(
record.state,
RepoRuntimeState::WaitingRsync | RepoRuntimeState::RrdpFailedPendingRsync
RepoRuntimeState::WaitingRsync
| RepoRuntimeState::RrdpFailedPendingRsync
)
{
record.state = RepoRuntimeState::FailedTerminal;
@@ -512,7 +534,6 @@ impl TransportStateTables {
}
}
#[derive(Default)]
pub struct InFlightRepoTable {
entries: HashMap<RepoKey, InFlightRepoEntry>,
@@ -536,7 +557,9 @@ impl InFlightRepoTable {
}
pub fn last_result(&self, key: &RepoKey) -> Option<&RepoSyncResultEnvelope> {
self.entries.get(key).and_then(|entry| entry.last_result.as_ref())
self.entries
.get(key)
.and_then(|entry| entry.last_result.as_ref())
}
pub fn register_request(
@@ -626,7 +649,9 @@ impl InFlightRepoTable {
RepoSyncResultKind::Success(result_ref)
| RepoSyncResultKind::Reused(result_ref) => result_ref.clone(),
RepoSyncResultKind::Failed { detail } => {
return Err(format!("success completion called with failure result: {detail}"));
return Err(format!(
"success completion called with failure result: {detail}"
));
}
};
entry.state = RepoTaskState::Succeeded;
@@ -912,12 +937,14 @@ mod tests {
#[cfg(test)]
mod transport_tests {
use crate::policy::SyncPreference;
use crate::parallel::repo_scheduler::{TransportRequestAction, TransportStateTables, TransportTaskState};
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(
@@ -956,7 +983,10 @@ mod transport_tests {
}
);
let key = "https://example.test/notify.xml".to_string();
assert_eq!(tables.rrdp_inflight.get(&key).unwrap().state, TransportTaskState::Pending);
assert_eq!(
tables.rrdp_inflight.get(&key).unwrap().state,
TransportTaskState::Pending
);
assert_eq!(
tables.runtime_records.get(&identity).unwrap().state,
RepoRuntimeState::WaitingRrdp
@@ -1171,7 +1201,10 @@ mod transport_tests {
"rsync://example.test/module/".to_string(),
SyncPreference::RsyncOnly,
);
assert!(matches!(action, TransportRequestAction::ReusedTerminalFailure(_)));
assert!(matches!(
action,
TransportRequestAction::ReusedTerminalFailure(_)
));
}
#[test]

160
src/parallel/repo_worker.rs
View File

@@ -10,10 +10,13 @@ use super::types::{
};
use crate::analysis::timing::TimingHandle;
use crate::audit_downloads::DownloadLogHandle;
use crate::current_repo_index::CurrentRepoIndexHandle;
use crate::fetch::rsync::RsyncFetcher;
use crate::policy::Policy;
use crate::storage::RocksStore;
use crate::sync::repo::{run_rrdp_transport, run_rsync_transport, sync_publication_point, RepoSyncSource};
use crate::sync::repo::{
RepoSyncSource, run_rrdp_transport, run_rsync_transport, sync_publication_point,
};
use crate::sync::rrdp::Fetcher;
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -39,6 +42,7 @@ pub trait RepoTransportExecutor: Send + Sync + 'static {
pub struct LiveRrdpTransportExecutor<H: Fetcher> {
store: Arc<RocksStore>,
current_repo_index: CurrentRepoIndexHandle,
http_fetcher: Arc<H>,
timing: Option<TimingHandle>,
download_log: Option<DownloadLogHandle>,
@@ -47,12 +51,14 @@ pub struct LiveRrdpTransportExecutor<H: Fetcher> {
impl<H: Fetcher> LiveRrdpTransportExecutor<H> {
pub fn new(
store: Arc<RocksStore>,
current_repo_index: CurrentRepoIndexHandle,
http_fetcher: Arc<H>,
timing: Option<TimingHandle>,
download_log: Option<DownloadLogHandle>,
) -> Self {
Self {
store,
current_repo_index,
http_fetcher,
timing,
download_log,
@@ -72,6 +78,7 @@ impl<H: Fetcher + 'static> RepoTransportExecutor for LiveRrdpTransportExecutor<H
match run_rrdp_transport(
self.store.as_ref(),
notification_uri,
Some(&self.current_repo_index),
self.http_fetcher.as_ref(),
self.timing.as_ref(),
self.download_log.as_ref(),
@@ -106,6 +113,7 @@ impl<H: Fetcher + 'static> RepoTransportExecutor for LiveRrdpTransportExecutor<H
pub struct LiveRsyncTransportExecutor<R: RsyncFetcher> {
store: Arc<RocksStore>,
current_repo_index: CurrentRepoIndexHandle,
rsync_fetcher: Arc<R>,
timing: Option<TimingHandle>,
download_log: Option<DownloadLogHandle>,
@@ -114,12 +122,14 @@ pub struct LiveRsyncTransportExecutor<R: RsyncFetcher> {
impl<R: RsyncFetcher> LiveRsyncTransportExecutor<R> {
pub fn new(
store: Arc<RocksStore>,
current_repo_index: CurrentRepoIndexHandle,
rsync_fetcher: Arc<R>,
timing: Option<TimingHandle>,
download_log: Option<DownloadLogHandle>,
) -> Self {
Self {
store,
current_repo_index,
rsync_fetcher,
timing,
download_log,
@@ -134,6 +144,7 @@ impl<R: RsyncFetcher + 'static> RepoTransportExecutor for LiveRsyncTransportExec
match run_rsync_transport(
self.store.as_ref(),
&task.repo_identity.rsync_base_uri,
Some(&self.current_repo_index),
self.rsync_fetcher.as_ref(),
self.timing.as_ref(),
self.download_log.as_ref(),
@@ -174,6 +185,7 @@ pub struct LiveRepoTransportExecutor<H: Fetcher, R: RsyncFetcher> {
impl<H: Fetcher, R: RsyncFetcher> LiveRepoTransportExecutor<H, R> {
pub fn new(
store: Arc<RocksStore>,
current_repo_index: CurrentRepoIndexHandle,
http_fetcher: Arc<H>,
rsync_fetcher: Arc<R>,
timing: Option<TimingHandle>,
@@ -182,11 +194,18 @@ impl<H: Fetcher, R: RsyncFetcher> LiveRepoTransportExecutor<H, R> {
Self {
rrdp: LiveRrdpTransportExecutor::new(
Arc::clone(&store),
current_repo_index.clone(),
http_fetcher,
timing.clone(),
download_log.clone(),
),
rsync: LiveRsyncTransportExecutor::new(store, rsync_fetcher, timing, download_log),
rsync: LiveRsyncTransportExecutor::new(
store,
current_repo_index,
rsync_fetcher,
timing,
download_log,
),
}
}
}
@@ -231,7 +250,9 @@ impl<H: Fetcher, R: RsyncFetcher> LiveRepoSyncExecutor<H, R> {
}
}
impl<H: Fetcher + 'static, R: RsyncFetcher + 'static> RepoSyncExecutor for LiveRepoSyncExecutor<H, R> {
impl<H: Fetcher + 'static, R: RsyncFetcher + 'static> RepoSyncExecutor
for LiveRepoSyncExecutor<H, R>
{
fn execute(&self, task: RepoSyncTask) -> RepoSyncResultEnvelope {
let started = std::time::Instant::now();
crate::progress_log::emit(
@@ -269,10 +290,12 @@ impl<H: Fetcher + 'static, R: RsyncFetcher + 'static> RepoSyncExecutor for LiveR
repo_key: task.repo_key.clone(),
tal_id: task.tal_id,
rir_id: task.rir_id,
result: super::types::RepoSyncResultKind::Success(super::types::RepoSyncResultRef {
repo_key: task.repo_key,
source: repo_sync_source_label(res.source).to_string(),
}),
result: super::types::RepoSyncResultKind::Success(
super::types::RepoSyncResultRef {
repo_key: task.repo_key,
source: repo_sync_source_label(res.source).to_string(),
},
),
phase: Some(repo_sync_phase_label(res.phase).to_string()),
timing_ms,
warnings: res.warnings,
@@ -595,15 +618,16 @@ mod tests {
RepoSyncExecutor, RepoTransportExecutor, RepoTransportWorkerPool, RepoWorkerPool,
RepoWorkerPoolConfig,
};
use crate::current_repo_index::CurrentRepoIndex;
use crate::fetch::rsync::{
LocalDirRsyncFetcher, RsyncFetchError, RsyncFetchResult, RsyncFetcher,
};
use crate::parallel::config::ParallelPhase1Config;
use crate::parallel::types::{
RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope,
RepoSyncResultKind, RepoSyncResultRef, RepoSyncTask, RepoTransportMode,
RepoTransportResultEnvelope, RepoTransportResultKind, RepoTransportTask,
};
use crate::fetch::rsync::{
LocalDirRsyncFetcher, RsyncFetchError, RsyncFetchResult, RsyncFetcher,
};
use crate::policy::SyncPreference;
use crate::report::Warning;
use crate::storage::RocksStore;
@@ -628,7 +652,10 @@ mod tests {
}
}
fn sample_rrdp_transport_task(notification_uri: &str, rsync_base_uri: &str) -> RepoTransportTask {
fn sample_rrdp_transport_task(
notification_uri: &str,
rsync_base_uri: &str,
) -> RepoTransportTask {
RepoTransportTask {
dedup_key: RepoDedupKey::RrdpNotify {
notification_uri: notification_uri.to_string(),
@@ -649,7 +676,10 @@ mod tests {
}
}
fn sample_rsync_transport_task(rsync_scope_uri: &str, rsync_base_uri: &str) -> RepoTransportTask {
fn sample_rsync_transport_task(
rsync_scope_uri: &str,
rsync_base_uri: &str,
) -> RepoTransportTask {
RepoTransportTask {
dedup_key: RepoDedupKey::RsyncScope {
rsync_scope_uri: rsync_scope_uri.to_string(),
@@ -774,7 +804,10 @@ mod tests {
}
fn rrdp_notification_xml(session_id: &str, serial: u64, snapshot_uri: &str) -> String {
let snapshot_body = rrdp_snapshot_xml(session_id, &[("rsync://example.test/repo/a.roa", b"a".as_ref())]);
let snapshot_body = rrdp_snapshot_xml(
session_id,
&[("rsync://example.test/repo/a.roa", b"a".as_ref())],
);
let snapshot_hash = sha256_hex(snapshot_body.as_bytes());
format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -785,13 +818,11 @@ mod tests {
}
fn rrdp_snapshot_xml(session_id: &str, objects: &[(&str, &[u8])]) -> String {
let mut body = String::from(
&format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
let mut body = String::from(&format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
<snapshot xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{session_id}" serial="1">
"#
),
);
));
for (uri, bytes) in objects {
body.push_str(&format!(
" <publish uri=\"{uri}\">{}</publish>\n",
@@ -896,9 +927,11 @@ mod tests {
);
assert_eq!(results.len(), 2);
assert!(results
.iter()
.all(|res| matches!(res.result, RepoSyncResultKind::Success(_))));
assert!(
results
.iter()
.all(|res| matches!(res.result, RepoSyncResultKind::Success(_)))
);
}
#[test]
@@ -977,8 +1010,14 @@ mod tests {
.recv_result_timeout(Duration::from_secs(1))
.expect("recv second")
.expect("second result");
assert!(matches!(first.result, RepoTransportResultKind::Success { .. }));
assert!(matches!(second.result, RepoTransportResultKind::Success { .. }));
assert!(matches!(
first.result,
RepoTransportResultKind::Success { .. }
));
assert!(matches!(
second.result,
RepoTransportResultKind::Success { .. }
));
}
#[test]
@@ -1031,10 +1070,12 @@ mod tests {
result.result,
RepoSyncResultKind::Failed { ref detail } if detail.contains("rsync fallback failed")
));
assert!(store
.get_repository_view_entry("rsync://example.test/repo/a.roa")
.expect("read view")
.is_none());
assert!(
store
.get_repository_view_entry("rsync://example.test/repo/a.roa")
.expect("read view")
.is_none()
);
}
#[test]
@@ -1044,16 +1085,22 @@ mod tests {
let http = MockHttpFetcher::new();
let notify = "https://example.test/notification.xml";
let snapshot = "https://example.test/snapshot.xml";
let snapshot_bytes = rrdp_snapshot_xml("123e4567-e89b-12d3-a456-426614174000", &[(
"rsync://example.test/repo/a.roa",
b"a".as_ref(),
)]);
let snapshot_bytes = rrdp_snapshot_xml(
"123e4567-e89b-12d3-a456-426614174000",
&[("rsync://example.test/repo/a.roa", b"a".as_ref())],
);
let notification_bytes =
rrdp_notification_xml("123e4567-e89b-12d3-a456-426614174000", 1, snapshot);
http.insert(notify, notification_bytes.into_bytes());
http.insert(snapshot, snapshot_bytes.into_bytes());
let executor =
LiveRrdpTransportExecutor::new(Arc::clone(&store), Arc::new(http), None, None);
let current_repo_index = CurrentRepoIndex::shared();
let executor = LiveRrdpTransportExecutor::new(
Arc::clone(&store),
current_repo_index.clone(),
Arc::new(http),
None,
None,
);
let result = executor.execute_transport(sample_rrdp_transport_task(
notify,
"rsync://example.test/repo/",
@@ -1062,6 +1109,16 @@ mod tests {
matches!(result.result, RepoTransportResultKind::Success { .. }),
"{result:?}"
);
let index = current_repo_index.lock().expect("index lock");
assert!(
index
.get_by_uri("rsync://example.test/repo/a.roa")
.is_some()
);
assert_eq!(
index.list_scope_uris(notify),
vec!["rsync://example.test/repo/a.roa".to_string()]
);
}
#[test]
@@ -1070,6 +1127,7 @@ mod tests {
let store = Arc::new(RocksStore::open(td.path()).expect("open store"));
let executor = LiveRrdpTransportExecutor::new(
Arc::clone(&store),
CurrentRepoIndex::shared(),
Arc::new(ErrorHttpFetcher),
None,
None,
@@ -1078,7 +1136,10 @@ mod tests {
"https://example.test/notification.xml",
"rsync://example.test/repo/",
));
assert!(matches!(result.result, RepoTransportResultKind::Failed { .. }));
assert!(matches!(
result.result,
RepoTransportResultKind::Failed { .. }
));
}
#[test]
@@ -1089,8 +1150,10 @@ mod tests {
fs::write(td.path().join("nested").join("b.cer"), b"b").expect("write");
let store_dir = tempfile::tempdir().expect("store tempdir");
let store = Arc::new(RocksStore::open(store_dir.path()).expect("open store"));
let current_repo_index = CurrentRepoIndex::shared();
let executor = LiveRsyncTransportExecutor::new(
Arc::clone(&store),
current_repo_index.clone(),
Arc::new(LocalDirRsyncFetcher::new(td.path())),
None,
None,
@@ -1099,7 +1162,28 @@ mod tests {
"rsync://example.test/repo/",
"rsync://example.test/repo/",
));
assert!(matches!(result.result, RepoTransportResultKind::Success { .. }));
assert!(matches!(
result.result,
RepoTransportResultKind::Success { .. }
));
let index = current_repo_index.lock().expect("index lock");
assert!(
index
.get_by_uri("rsync://example.test/repo/a.roa")
.is_some()
);
assert!(
index
.get_by_uri("rsync://example.test/repo/nested/b.cer")
.is_some()
);
assert_eq!(
index.list_scope_uris("rsync://example.test/repo/"),
vec![
"rsync://example.test/repo/a.roa".to_string(),
"rsync://example.test/repo/nested/b.cer".to_string()
]
);
}
#[test]
@@ -1108,6 +1192,7 @@ mod tests {
let store = Arc::new(RocksStore::open(store_dir.path()).expect("open store"));
let executor = LiveRsyncTransportExecutor::new(
Arc::clone(&store),
CurrentRepoIndex::shared(),
Arc::new(FailingRsyncFetcher),
None,
None,
@@ -1116,6 +1201,9 @@ mod tests {
"rsync://example.test/module/",
"rsync://example.test/repo/",
));
assert!(matches!(result.result, RepoTransportResultKind::Failed { .. }));
assert!(matches!(
result.result,
RepoTransportResultKind::Failed { .. }
));
}
}

28
src/parallel/run_coordinator.rs
View File

@@ -1,20 +1,22 @@
use std::collections::VecDeque;
use crate::current_repo_index::{CurrentRepoIndex, CurrentRepoIndexHandle};
use crate::parallel::config::ParallelPhase1Config;
use crate::parallel::repo_scheduler::{
InFlightRepoTable, RepoCompletion, RepoRequestAction, TransportCompletion, TransportRequestAction,
TransportStateTables,
InFlightRepoTable, RepoCompletion, RepoRequestAction, TransportCompletion,
TransportRequestAction, TransportStateTables,
};
use crate::parallel::stats::ParallelRunStats;
use crate::parallel::types::{
RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncTask, RepoTransportResultEnvelope,
RepoTransportTask, TalInputSpec,
RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncTask,
RepoTransportResultEnvelope, RepoTransportTask, TalInputSpec,
};
use crate::policy::SyncPreference;
pub struct GlobalRunCoordinator {
pub config: ParallelPhase1Config,
pub tal_inputs: Vec<TalInputSpec>,
pub current_repo_index: CurrentRepoIndexHandle,
pub in_flight_repos: InFlightRepoTable,
pub transport_tables: TransportStateTables,
pub pending_repo_tasks: VecDeque<RepoSyncTask>,
@@ -27,6 +29,7 @@ impl GlobalRunCoordinator {
Self {
config,
tal_inputs,
current_repo_index: CurrentRepoIndex::shared(),
in_flight_repos: InFlightRepoTable::new(),
transport_tables: TransportStateTables::new(),
pending_repo_tasks: VecDeque::new(),
@@ -35,6 +38,10 @@ impl GlobalRunCoordinator {
}
}
pub fn current_repo_index_handle(&self) -> CurrentRepoIndexHandle {
self.current_repo_index.clone()
}
pub fn register_repo_request(
&mut self,
repo_key: RepoKey,
@@ -158,7 +165,8 @@ impl GlobalRunCoordinator {
dedup_key: &crate::parallel::types::RepoDedupKey,
started_at: time::OffsetDateTime,
) -> Result<(), String> {
self.transport_tables.mark_transport_running(dedup_key, started_at)?;
self.transport_tables
.mark_transport_running(dedup_key, started_at)?;
self.stats.repo_tasks_running += 1;
Ok(())
}
@@ -168,10 +176,14 @@ impl GlobalRunCoordinator {
result: RepoTransportResultEnvelope,
finished_at: time::OffsetDateTime,
) -> Result<TransportCompletion, String> {
let completion = self.transport_tables.complete_transport_result(result.clone(), finished_at)?;
let completion = self
.transport_tables
.complete_transport_result(result.clone(), finished_at)?;
self.stats.repo_tasks_running = self.stats.repo_tasks_running.saturating_sub(1);
if matches!(result.result, crate::parallel::types::RepoTransportResultKind::Failed { .. })
&& result.mode == crate::parallel::types::RepoTransportMode::Rsync
if matches!(
result.result,
crate::parallel::types::RepoTransportResultKind::Failed { .. }
) && result.mode == crate::parallel::types::RepoTransportMode::Rsync
{
self.stats.repo_tasks_failed += 1;
}

23
src/parallel/types.rs
View File

@@ -12,7 +12,10 @@ pub enum TalSource {
ta_der: Vec<u8>,
},
FilePath(PathBuf),
FilePathWithTa { tal_path: PathBuf, ta_path: PathBuf },
FilePathWithTa {
tal_path: PathBuf,
ta_path: PathBuf,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -57,11 +60,7 @@ impl TalInputSpec {
}
}
pub fn from_ta_der(
tal_url: impl Into<String>,
tal_bytes: Vec<u8>,
ta_der: Vec<u8>,
) -> Self {
pub fn from_ta_der(tal_url: impl Into<String>, tal_bytes: Vec<u8>, ta_der: Vec<u8>) -> Self {
let tal_url = tal_url.into();
let tal_id = derive_tal_id_from_url_like(&tal_url);
Self {
@@ -317,10 +316,9 @@ mod tests {
use crate::report::Warning;
use super::{
derive_tal_id_from_path, derive_tal_id_from_url_like, RepoDedupKey, RepoIdentity,
RepoKey, RepoRequester, RepoRuntimeState, RepoSyncTask, RepoTaskState,
RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind, TalInputSpec,
TalSource,
RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoRuntimeState, RepoSyncTask,
RepoTaskState, RepoTransportMode, RepoTransportResultEnvelope, RepoTransportResultKind,
TalInputSpec, TalSource, derive_tal_id_from_path, derive_tal_id_from_url_like,
};
#[test]
@@ -477,7 +475,10 @@ mod tests {
},
};
assert!(matches!(ok.result, RepoTransportResultKind::Success { .. }));
assert!(matches!(fail.result, RepoTransportResultKind::Failed { .. }));
assert!(matches!(
fail.result,
RepoTransportResultKind::Failed { .. }
));
}
#[test]