panda-rpki-oss/src/validation/objects/parallel_processing.rs

/// Bounded parallel ROA validation.
///
/// Workers share only immutable certificate/resource state and an in-run CRL
/// verification map. Nothing in this path is persisted or reused by a later
/// validation run.
pub fn process_publication_point_for_issuer_parallel_roa<P: PublicationPointData>(
    publication_point: &P,
    policy: &Policy,
    issuer_ca_der: &[u8],
    issuer_ca_rsync_uri: Option<&str>,
    issuer_effective_ip: Option<&crate::model::rc::IpResourceSet>,
    issuer_effective_as: Option<&crate::model::rc::AsResourceSet>,
    validation_time: time::OffsetDateTime,
    timing: Option<&TimingHandle>,
    config: &ParallelPhase2Config,
) -> ObjectsOutput {
    if config.object_workers <= 1
        || policy.signed_object_failure_policy == SignedObjectFailurePolicy::DropPublicationPoint
    {
        return process_publication_point_for_issuer_with_ta_constraints(
            publication_point,
            policy,
            issuer_ca_der,
            issuer_ca_rsync_uri,
            issuer_effective_ip,
            issuer_effective_as,
            validation_time,
            timing,
            None,
        );
    }
    let Ok(pool) = ParallelRoaWorkerPool::new(config) else {
        return process_publication_point_for_issuer_with_ta_constraints(
            publication_point,
            policy,
            issuer_ca_der,
            issuer_ca_rsync_uri,
            issuer_effective_ip,
            issuer_effective_as,
            validation_time,
            timing,
            None,
        );
    };
    process_publication_point_for_issuer_parallel_roa_with_pool(
        publication_point,
        policy,
        issuer_ca_der,
        issuer_ca_rsync_uri,
        issuer_effective_ip,
        issuer_effective_as,
        validation_time,
        timing,
        &pool,
    )
}

pub fn process_publication_point_for_issuer_parallel_roa_with_pool<P: PublicationPointData>(
    publication_point: &P,
    policy: &Policy,
    issuer_ca_der: &[u8],
    issuer_ca_rsync_uri: Option<&str>,
    issuer_effective_ip: Option<&crate::model::rc::IpResourceSet>,
    issuer_effective_as: Option<&crate::model::rc::AsResourceSet>,
    validation_time: time::OffsetDateTime,
    timing: Option<&TimingHandle>,
    pool: &ParallelRoaWorkerPool,
) -> ObjectsOutput {
    if policy.signed_object_failure_policy == SignedObjectFailurePolicy::DropPublicationPoint {
        return process_publication_point_for_issuer_with_ta_constraints(
            publication_point,
            policy,
            issuer_ca_der,
            issuer_ca_rsync_uri,
            issuer_effective_ip,
            issuer_effective_as,
            validation_time,
            timing,
            None,
        );
    }
    process_publication_point_for_issuer_parallel_roa_inner(
        publication_point,
        policy,
        issuer_ca_der,
        issuer_ca_rsync_uri,
        issuer_effective_ip,
        issuer_effective_as,
        validation_time,
        timing,
        pool,
        None,
    )
    .unwrap_or_else(|_| {
        process_publication_point_for_issuer_with_ta_constraints(
            publication_point,
            policy,
            issuer_ca_der,
            issuer_ca_rsync_uri,
            issuer_effective_ip,
            issuer_effective_as,
            validation_time,
            timing,
            None,
        )
    })
}

#[derive(Clone)]
pub(crate) struct RoaTaskShared {
    locked_files: Arc<[PackFile]>,
    issuer_ca_der: Arc<[u8]>,
    issuer_ca: Arc<ResourceCertificate>,
    issuer_spki_der: Arc<[u8]>,
    issuer_ca_rsync_uri: Option<Arc<str>>,
    crl_states: Arc<Mutex<HashMap<String, IssuerCrlState>>>,
    issuer_resources_index: Arc<IssuerResourcesIndex>,
    issuer_effective_ip: Option<Arc<crate::model::rc::IpResourceSet>>,
    issuer_effective_as: Option<Arc<crate::model::rc::AsResourceSet>>,
    resource_validation_mode: ResourceValidationMode,
    ta_constraints: Option<Arc<crate::ta_constraints::TaConstraints>>,
}

#[derive(Clone)]
pub(crate) struct OwnedRoaTask {
    pub(crate) publication_point_id: u64,
    index: usize,
    shared: Arc<RoaTaskShared>,
    validation_time: time::OffsetDateTime,
    strict_cms_der: bool,
    strict_name: bool,
    pub(crate) submitted_at: Option<Instant>,
}

#[derive(Clone)]
struct RoaTaskExecutor;

impl ObjectTaskExecutor<OwnedRoaTask, RoaTaskResult> for RoaTaskExecutor {
    fn execute(&self, worker_index: usize, task: OwnedRoaTask) -> RoaTaskResult {
        validate_owned_roa_task(worker_index, task)
    }
}

pub struct ParallelRoaWorkerPool {
    pool: Mutex<ObjectWorkerPool<OwnedRoaTask, RoaTaskResult, RoaTaskExecutor>>,
}

pub(crate) enum ParallelObjectsPrepare {
    Complete(ObjectsOutput),
    Staged(ParallelObjectsStage),
}

pub(crate) struct ParallelObjectsStage {
    pub(crate) publication_point_id: u64,
    pub(crate) shared: Arc<RoaTaskShared>,
    pub(crate) validation_time: time::OffsetDateTime,
    pub(crate) strict_cms_der: bool,
    pub(crate) strict_name: bool,
    pub(crate) roa_task_indices: Vec<usize>,
    pub(crate) ready_roa_results: Vec<RoaTaskResult>,
    pub(crate) warnings: Vec<Warning>,
    pub(crate) stats: ObjectsStats,
    pub(crate) audit: Vec<ObjectAuditEntry>,
}

#[derive(Debug, thiserror::Error)]
pub(crate) enum ObjectValidateError {
    #[error("object bytes load failed: {0}")]
    BytesLoad(String),
    #[error("ROA decode failed: {0}")]
    RoaDecode(#[from] RoaDecodeError),
    #[error("ROA embedded EE resource validation failed: {0}")]
    RoaEeResources(#[from] RoaValidateError),
    #[error("ASPA decode failed: {0}")]
    AspaDecode(#[from] AspaDecodeError),
    #[error("ASPA embedded EE resource validation failed: {0}")]
    AspaEeResources(#[from] AspaValidateError),
    #[error("CMS signature verification failed: {0}")]
    Signature(#[from] SignedObjectVerifyError),
    #[error("EE certificate path validation failed: {0}")]
    CertPath(#[from] CertPathError),
    #[error("certificate CRLDistributionPoints URIs missing (RFC 6487 §4.8.6)")]
    MissingCrlDpUris,
    #[error("no CRL available in publication point snapshot (RFC 9286 §7)")]
    MissingCrlInPack,
    #[error("CRL referenced by CRLDistributionPoints not found in publication point snapshot: {0}")]
    CrlNotFound(String),
    #[error("issuer effective IP resources missing")]
    MissingIssuerEffectiveIp,
    #[error("issuer effective AS resources missing")]
    MissingIssuerEffectiveAs,
    #[error("EE certificate resources are not a subset of issuer effective resources")]
    EeResourcesNotSubset,
    #[error("EE certificate violates locally configured TA constraints: {0}")]
    TaConstraints(#[from] crate::ta_constraints::TaConstraintsViolation),
}

impl ParallelRoaWorkerPool {
    pub fn new(config: &ParallelPhase2Config) -> Result<Self, String> {
        if config.object_workers <= 1 {
            return Err("parallel ROA worker pool requires object_workers > 1".to_string());
        }
        Ok(Self {
            pool: Mutex::new(ObjectWorkerPool::new(
                config.object_workers,
                config.worker_queue_capacity,
                RoaTaskExecutor,
            )?),
        })
    }

    pub(crate) fn try_submit_round_robin(
        &self,
        task: OwnedRoaTask,
    ) -> Result<usize, ObjectWorkerSubmitError<OwnedRoaTask>> {
        self.pool
            .lock()
            .expect("parallel ROA worker pool lock")
            .try_submit_round_robin(task)
    }

    pub(crate) fn recv_result_timeout(
        &self,
        timeout: Duration,
    ) -> Result<Option<RoaTaskResult>, String> {
        self.pool
            .lock()
            .expect("parallel ROA worker pool lock")
            .recv_result_timeout(timeout)
    }
}

fn validate_owned_roa_task(worker_index: usize, task: OwnedRoaTask) -> RoaTaskResult {
    let worker_started = Instant::now();
    let queue_wait_ms = task
        .submitted_at
        .map(|submitted_at| worker_started.saturating_duration_since(submitted_at))
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0);
    let shared = task.shared.as_ref();
    let file = shared
        .locked_files
        .get(task.index)
        .expect("ROA task index must reference locked file");
    let outcome = match SubjectPublicKeyInfo::from_der(shared.issuer_spki_der.as_ref()) {
        Ok(([], issuer_spki)) => process_roa_with_issuer_parallel(
            file,
            shared.issuer_ca_der.as_ref(),
            shared.issuer_ca.as_ref(),
            &issuer_spki,
            shared.issuer_ca_rsync_uri.as_deref(),
            shared.crl_states.as_ref(),
            shared.issuer_resources_index.as_ref(),
            shared.issuer_effective_ip.as_deref(),
            shared.issuer_effective_as.as_deref(),
            task.validation_time,
            None,
            task.strict_cms_der,
            task.strict_name,
            shared.resource_validation_mode,
            shared.ta_constraints.as_deref(),
        )
        .map(|vrps| RoaTaskOk { vrps }),
        Ok((rem, _)) => Err(ObjectValidateError::CertPath(
            CertPathError::IssuerSpkiTrailingBytes(rem.len()),
        )),
        Err(error) => Err(ObjectValidateError::CertPath(
            CertPathError::IssuerSpkiParse(error.to_string()),
        )),
    };
    RoaTaskResult {
        publication_point_id: task.publication_point_id,
        index: task.index,
        worker_index,
        queue_wait_ms,
        worker_ms: worker_started.elapsed().as_millis() as u64,
        outcome,
    }
}

fn process_publication_point_for_issuer_parallel_roa_inner<P: PublicationPointData>(
    publication_point: &P,
    policy: &Policy,
    issuer_ca_der: &[u8],
    issuer_ca_rsync_uri: Option<&str>,
    issuer_effective_ip: Option<&crate::model::rc::IpResourceSet>,
    issuer_effective_as: Option<&crate::model::rc::AsResourceSet>,
    validation_time: time::OffsetDateTime,
    timing: Option<&TimingHandle>,
    pool: &ParallelRoaWorkerPool,
    ta_constraints: Option<Arc<crate::ta_constraints::TaConstraints>>,
) -> Result<ObjectsOutput, String> {
    let stage = match prepare_publication_point_for_parallel_roa_and_ta_constraints(
        0,
        publication_point,
        policy,
        issuer_ca_der,
        issuer_ca_rsync_uri,
        issuer_effective_ip,
        issuer_effective_as,
        validation_time,
        ta_constraints,
    ) {
        ParallelObjectsPrepare::Complete(output) => return Ok(output),
        ParallelObjectsPrepare::Staged(stage) => stage,
    };
    let task_count = stage.roa_task_count();
    let mut pending = std::collections::VecDeque::with_capacity(task_count);
    stage.append_roa_tasks_to(&mut pending);
    while let Some(mut task) = pending.pop_front() {
        task.submitted_at = Some(Instant::now());
        match pool.try_submit_round_robin(task) {
            Ok(_) => {}
            Err(ObjectWorkerSubmitError::QueueFull { task, .. }) => {
                pending.push_front(task);
                std::thread::yield_now();
            }
            Err(ObjectWorkerSubmitError::Disconnected { .. }) => {
                return Err("parallel ROA worker queue disconnected".to_string());
            }
        }
    }
    let mut results = Vec::with_capacity(task_count);
    while results.len() < task_count {
        let Some(result) = pool.recv_result_timeout(Duration::from_secs(30))? else {
            return Err("parallel ROA worker timed out".to_string());
        };
        results.push(result);
    }
    reduce_parallel_roa_stage(stage, results, timing)
}