impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
    fn run_publication_point(
        &self,
        ca: &CaInstanceHandle,
    ) -> Result<PublicationPointRunResult, String> {
        let started = std::time::Instant::now();
        let _span = self
            .timing
            .as_ref()
            .map(|timing| timing.span_publication_point(&ca.manifest_rsync_uri));
        if let Some(timing) = self.timing.as_ref() {
            timing.record_count("publication_points_seen", 1);
        }
        crate::logging::progress::emit!(
            "publication_point_start",
            serde_json::json!({
                "manifest_rsync_uri": ca.manifest_rsync_uri,
                "publication_point_rsync_uri": ca.publication_point_rsync_uri,
                "rsync_base_uri": ca.rsync_base_uri,
                "rrdp_notification_uri": ca.rrdp_notification_uri,
            }),
        );

        let attempted_rrdp =
            self.policy.sync_preference == crate::validation::policy::SyncPreference::RrdpThenRsync;
        let repo_started = std::time::Instant::now();
        let (repo_sync_ok, repo_sync_error, repo_sync_source, repo_sync_phase, runtime_duration) =
            if let Some(runtime) = self.repo_sync_runtime.as_ref() {
                let outcome = runtime.sync_publication_point_repo(ca)?;
                (
                    outcome.repo_sync_ok,
                    outcome.repo_sync_err,
                    outcome.repo_sync_source,
                    outcome.repo_sync_phase,
                    Some(outcome.repo_sync_duration_ms),
                )
            } else {
                match sync_publication_point(
                    self.store,
                    self.policy,
                    if attempted_rrdp {
                        ca.rrdp_notification_uri.as_deref()
                    } else {
                        None
                    },
                    &ca.rsync_base_uri,
                    self.http_fetcher,
                    self.rsync_fetcher,
                    self.timing.as_ref(),
                    self.download_log.as_ref(),
                ) {
                    Ok(result) => (
                        true,
                        None,
                        Some(repo_sync_source_label(result.source).to_string()),
                        Some(repo_sync_phase_label(result.phase).to_string()),
                        None,
                    ),
                    Err(error) => (
                        false,
                        Some(error.to_string()),
                        None,
                        Some(
                            repo_sync_failure_phase_label(
                                attempted_rrdp,
                                ca.rrdp_notification_uri.as_deref(),
                                ca.rrdp_notification_uri.as_deref(),
                            )
                            .to_string(),
                        ),
                        None,
                    ),
                }
            };
        let repo_sync_duration_ms = effective_repo_sync_duration_ms(
            repo_started.elapsed().as_millis() as u64,
            runtime_duration,
            repo_sync_ok,
        );
        crate::logging::progress::emit!(
            "publication_point_repo_sync_done",
            serde_json::json!({
                "manifest_rsync_uri": ca.manifest_rsync_uri,
                "repo_sync_ok": repo_sync_ok,
                "repo_sync_source": repo_sync_source,
                "repo_sync_phase": repo_sync_phase,
                "repo_sync_error": repo_sync_error,
                "repo_sync_duration_ms": repo_sync_duration_ms,
            }),
        );

        let stage = self.stage_fresh_publication_point_after_repo_ready(
            ca,
            repo_sync_ok,
            repo_sync_error.as_deref(),
        );
        let stage = match stage {
            Ok(stage) => stage,
            Err(stage_error) => {
                let mut warnings = Vec::new();
                warnings.push(
                    Warning::new(format!(
                        "publication point processing failed: {}",
                        stage_error.error
                    ))
                    .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")])
                    .with_context(&ca.manifest_rsync_uri),
                );
                let audit = build_publication_point_audit_from_failed_fetch(
                    ca,
                    repo_sync_source.as_deref(),
                    repo_sync_phase.as_deref(),
                    Some(repo_sync_duration_ms),
                    repo_sync_error.as_deref(),
                    &warnings,
                    &stage_error.error,
                );
                crate::logging::progress::emit!(
                    "repo_terminal_failure",
                    serde_json::json!({
                        "manifest_rsync_uri": ca.manifest_rsync_uri,
                        "repo_sync_source": repo_sync_source,
                        "repo_sync_phase": repo_sync_phase,
                        "repo_sync_error": repo_sync_error,
                        "repo_sync_duration_ms": repo_sync_duration_ms,
                        "terminal_state": "failed_no_cache",
                        "error": stage_error.error.to_string(),
                    }),
                );
                return Ok(PublicationPointRunResult {
                    source: PublicationPointSource::FailedFetchNoCache,
                    snapshot: None,
                    warnings,
                    objects: empty_objects_output(),
                    audit,
                    discovered_children: Vec::new(),
                });
            }
        };

        let ta_constraints = self.policy.ta_constraints.for_tal(&ca.tal_id);
        let object_started = std::time::Instant::now();
        let mut objects = if let Some(constraints) = ta_constraints {
            crate::validation::objects::process_publication_point_for_issuer_with_ta_constraints(
                &stage.fresh_point,
                self.policy,
                stage.issuer_ca_der.as_ref(),
                ca.ca_certificate_rsync_uri.as_deref(),
                ca.effective_ip_resources.as_ref(),
                ca.effective_as_resources.as_ref(),
                self.validation_time,
                self.timing.as_ref(),
                Some(constraints),
            )
        } else if let Some(pool) = self.parallel_roa_worker_pool.as_ref() {
            process_publication_point_for_issuer_parallel_roa_with_pool(
                &stage.fresh_point,
                self.policy,
                stage.issuer_ca_der.as_ref(),
                ca.ca_certificate_rsync_uri.as_deref(),
                ca.effective_ip_resources.as_ref(),
                ca.effective_as_resources.as_ref(),
                self.validation_time,
                self.timing.as_ref(),
                pool,
            )
        } else if let Some(config) = self.parallel_phase2_config.as_ref() {
            process_publication_point_for_issuer_parallel_roa(
                &stage.fresh_point,
                self.policy,
                stage.issuer_ca_der.as_ref(),
                ca.ca_certificate_rsync_uri.as_deref(),
                ca.effective_ip_resources.as_ref(),
                ca.effective_as_resources.as_ref(),
                self.validation_time,
                self.timing.as_ref(),
                config,
            )
        } else {
            crate::validation::objects::process_publication_point_for_issuer(
                &stage.fresh_point,
                self.policy,
                stage.issuer_ca_der.as_ref(),
                ca.ca_certificate_rsync_uri.as_deref(),
                ca.effective_ip_resources.as_ref(),
                ca.effective_as_resources.as_ref(),
                self.validation_time,
                self.timing.as_ref(),
            )
        };
        let object_ms = object_started.elapsed().as_millis() as u64;
        self.record_publication_point_step_ms(
            &ca.manifest_rsync_uri,
            "objects_processing",
            object_ms,
        );
        objects
            .router_keys
            .extend(stage.discovered_router_keys.clone());

        let finalized = self.finalize_fresh_publication_point_from_reducer(
            ca,
            &stage.fresh_point,
            stage.warnings,
            objects,
            stage.child_audits,
            stage.discovered_children,
            repo_sync_source.as_deref(),
            repo_sync_phase.as_deref(),
            repo_sync_duration_ms,
            repo_sync_error.as_deref(),
        )?;
        let result = finalized.result;
        let total_ms = started.elapsed().as_millis() as u64;
        crate::logging::progress::emit!(
            "publication_point_finish",
            serde_json::json!({
                "manifest_rsync_uri": ca.manifest_rsync_uri,
                "repo_sync_source": repo_sync_source,
                "repo_sync_phase": repo_sync_phase,
                "repo_sync_duration_ms": repo_sync_duration_ms,
                "total_duration_ms": total_ms,
                "objects_processing_ms": object_ms,
                "warning_count": result.warnings.len(),
                "vrp_count": result.objects.vrps.len(),
                "aspa_count": result.objects.aspas.len(),
                "child_count": result.discovered_children.len(),
            }),
        );
        Ok(result)
    }
}