From 3d7e18a2385b80a62c9a77f6e884dd3d44bd4384 Mon Sep 17 00:00:00 2001 From: yuyr Date: Sat, 1 Aug 2026 20:37:53 +0800 Subject: [PATCH] =?UTF-8?q?20260801=20=E5=B9=B6=E8=A1=8C=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E9=9D=A2=EF=BC=9Astage=20worker=E6=B1=A0+CurrentRepoIndex=20Rw?= =?UTF-8?q?Lock=E5=8C=96=EF=BC=8C--control-plane-stage-workers=E5=BC=80?= =?UTF-8?q?=E5=85=B3=E9=BB=98=E8=AE=A4=E5=85=B3=E9=97=AD(#138)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/cli.rs | 12 + src/current_repo_index.rs | 13 +- src/parallel/config.rs | 4 + src/parallel/object_worker.rs | 212 +++- src/parallel/repo_worker.rs | 4 +- src/parallel/run_coordinator.rs | 8 +- src/sync/repo.rs | 2 +- src/sync/rrdp.rs | 6 +- src/sync/rrdp/snapshot_apply.rs | 4 +- src/sync/rrdp/tests.rs | 4 +- src/validation/manifest.rs | 6 +- src/validation/run_tree_from_tal.rs | 6 +- src/validation/tree_parallel.rs | 1713 ++++++++++++++++++++++----- src/validation/tree_runner.rs | 4 +- 14 files changed, 1653 insertions(+), 345 deletions(-) diff --git a/src/cli.rs b/src/cli.rs index a33d74b..01019e9 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -276,6 +276,9 @@ Options: Legacy Phase 2 scheduler finalize time budget; dedicated finalize worker ignores it (default: 100) --parallel-phase2-finalize-queue-capacity Phase 2 dedicated finalize worker queue capacity (default: 32768) + --control-plane-stage-workers + Experimental: Phase 2 ready publication point stage worker count; + 0 disables the stage pool and keeps inline staging (default: 0) --rsync-local-dir Use LocalDirRsyncFetcher rooted at this directory (offline tests) --disable-rrdp Disable RRDP and synchronize only via rsync @@ -519,6 +522,15 @@ pub fn parse_args(argv: &[String]) -> Result { format!("invalid --parallel-phase2-finalize-queue-capacity: {v}") })?; } + "--control-plane-stage-workers" => { + i += 1; + let v = argv + .get(i) + .ok_or("--control-plane-stage-workers requires a value")?; + parallel_phase2_cfg.stage_workers = v + .parse::() + .map_err(|_| format!("invalid --control-plane-stage-workers: {v}"))?; + } "--db" => { i += 1; let v = argv.get(i).ok_or("--db requires a value")?; diff --git a/src/current_repo_index.rs b/src/current_repo_index.rs index 4214845..1dc6e79 100644 --- a/src/current_repo_index.rs +++ b/src/current_repo_index.rs @@ -1,5 +1,5 @@ use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, RwLock}; use crate::storage::{RepositoryViewEntry, RepositoryViewState}; @@ -25,7 +25,12 @@ pub struct CurrentRepoIndex { by_uri: HashMap, } -pub type CurrentRepoIndexHandle = Arc>; +/// Shared handle to the run-wide current repository index. Readers (phase 2 +/// publication point staging, cache lookups, output snapshots) take +/// `.read()` and no longer exclude each other; writers (repo sync transports +/// applying repository view entries, run-state reset) take `.write()` and +/// stay exclusive. +pub type CurrentRepoIndexHandle = Arc>; impl CurrentRepoIndex { pub fn new() -> Self { @@ -33,7 +38,7 @@ impl CurrentRepoIndex { } pub fn shared() -> CurrentRepoIndexHandle { - Arc::new(Mutex::new(Self::new())) + Arc::new(RwLock::new(Self::new())) } pub fn get_by_uri(&self, rsync_uri: &str) -> Option<&CurrentRepoEntry> { @@ -205,7 +210,7 @@ mod tests { #[test] fn current_repo_index_snapshot_objects_and_counts_are_sorted() { let handle = CurrentRepoIndex::shared(); - let mut index = handle.lock().expect("lock index"); + let mut index = handle.write().expect("write-lock index"); index .apply_repository_view_entries(&[ present( diff --git a/src/parallel/config.rs b/src/parallel/config.rs index df8127c..52bec45 100644 --- a/src/parallel/config.rs +++ b/src/parallel/config.rs @@ -15,6 +15,9 @@ pub struct ParallelPhase2Config { pub publication_point_finalize_batch_size: usize, pub publication_point_finalize_wall_time_budget_ms: u64, pub publication_point_finalize_queue_capacity: usize, + /// Experimental: workers of the ready publication point stage pool. `0` + /// disables the pool and keeps the inline compute+apply staging path. + pub stage_workers: usize, } impl Default for ParallelPhase2Config { @@ -28,6 +31,7 @@ impl Default for ParallelPhase2Config { publication_point_finalize_batch_size: 256, publication_point_finalize_wall_time_budget_ms: 100, publication_point_finalize_queue_capacity: 32768, + stage_workers: 0, } } } diff --git a/src/parallel/object_worker.rs b/src/parallel/object_worker.rs index 708f4cd..c32980e 100644 --- a/src/parallel/object_worker.rs +++ b/src/parallel/object_worker.rs @@ -1,9 +1,13 @@ +use std::marker::PhantomData; use std::sync::Arc; use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TrySendError}; -use std::thread::{self, JoinHandle}; +use std::thread::{self, JoinHandle, Scope}; use std::time::Duration; -pub trait ObjectTaskExecutor: Send + Sync + 'static { +/// Executor shared by every worker of a pool. The `'static` bound required to +/// move the executor into detached threads is expressed on the pool types +/// instead of this trait so scoped pools can borrow their environment. +pub trait ObjectTaskExecutor: Send + Sync { fn execute(&self, worker_index: usize, task: T) -> R; } @@ -22,7 +26,7 @@ pub struct ObjectWorkerPool where T: Send + 'static, R: Send + 'static, - E: ObjectTaskExecutor, + E: ObjectTaskExecutor + 'static, { task_txs: Vec>>, result_rx: Receiver, @@ -35,7 +39,7 @@ impl ObjectWorkerPool where T: Send + 'static, R: Send + 'static, - E: ObjectTaskExecutor, + E: ObjectTaskExecutor + 'static, { pub fn new(worker_count: usize, queue_capacity: usize, executor: E) -> Result { if worker_count == 0 { @@ -140,7 +144,7 @@ impl Drop for ObjectWorkerPool where T: Send + 'static, R: Send + 'static, - E: ObjectTaskExecutor, + E: ObjectTaskExecutor + 'static, { fn drop(&mut self) { let _ = self.shutdown_inner(); @@ -153,8 +157,8 @@ fn object_worker_loop( result_tx: mpsc::Sender, executor: Arc, ) where - T: Send + 'static, - R: Send + 'static, + T: Send, + R: Send, E: ObjectTaskExecutor, { loop { @@ -170,6 +174,120 @@ fn object_worker_loop( } } +/// Scoped variant of `ObjectWorkerPool`: workers are spawned on a +/// `std::thread::Scope`, so tasks, results and the executor may borrow their +/// environment (`'env`) instead of being `'static`. The pool never sends +/// `Shutdown`; workers exit when every task sender is dropped, which happens +/// when the pool itself is dropped ahead of the scope join. +pub struct ScopedObjectWorkerPool<'scope, 'env, T, R, E> +where + T: Send + 'env, + R: Send + 'env, + E: ObjectTaskExecutor + 'env, +{ + task_txs: Vec>>, + result_rx: Receiver, + next_worker_idx: usize, + _executor: Arc, + // Join handles are intentionally not stored: dropping a `ScopedJoinHandle` + // detaches the worker and the enclosing scope joins it on exit, after the + // dropped task senders have made every worker return. + _marker: PhantomData<(&'scope (), &'env ())>, +} + +impl<'scope, 'env, T, R, E> ScopedObjectWorkerPool<'scope, 'env, T, R, E> +where + T: Send + 'env, + R: Send + 'env, + E: ObjectTaskExecutor + 'env, +{ + pub fn new( + scope: &'scope Scope<'scope, 'env>, + worker_count: usize, + queue_capacity: usize, + executor: E, + ) -> Result { + if worker_count == 0 { + return Err("ScopedObjectWorkerPool requires at least one worker".to_string()); + } + if queue_capacity == 0 { + return Err("ScopedObjectWorkerPool requires queue_capacity > 0".to_string()); + } + + let executor = Arc::new(executor); + let (result_tx, result_rx) = mpsc::channel::(); + let mut task_txs = Vec::with_capacity(worker_count); + + for worker_index in 0..worker_count { + let (task_tx, task_rx) = mpsc::sync_channel::>(queue_capacity); + let result_tx = result_tx.clone(); + let executor = Arc::clone(&executor); + thread::Builder::new() + .name(format!("object-validation-worker-{worker_index}")) + .spawn_scoped(scope, move || { + object_worker_loop(worker_index, task_rx, result_tx, executor) + }) + .map_err(|e| format!("spawn scoped object worker failed: {e}"))?; + task_txs.push(task_tx); + } + + Ok(Self { + task_txs, + result_rx, + next_worker_idx: 0, + _executor: executor, + _marker: PhantomData, + }) + } + + pub fn worker_count(&self) -> usize { + self.task_txs.len() + } + + pub fn try_submit_round_robin(&mut self, task: T) -> Result> { + let worker_index = self.next_worker_idx % self.task_txs.len(); + match self.task_txs[worker_index].try_send(ObjectWorkerMessage::Task(task)) { + Ok(()) => { + self.next_worker_idx = (worker_index + 1) % self.task_txs.len(); + Ok(worker_index) + } + Err(TrySendError::Full(ObjectWorkerMessage::Task(task))) => { + Err(ObjectWorkerSubmitError::QueueFull { worker_index, task }) + } + Err(TrySendError::Disconnected(ObjectWorkerMessage::Task(task))) => { + Err(ObjectWorkerSubmitError::Disconnected { worker_index, task }) + } + Err(TrySendError::Full(ObjectWorkerMessage::Shutdown)) + | Err(TrySendError::Disconnected(ObjectWorkerMessage::Shutdown)) => { + unreachable!("shutdown is never submitted via try_submit_round_robin") + } + } + } + + pub fn recv_result_timeout(&self, timeout: Duration) -> Result, String> { + match self.result_rx.recv_timeout(timeout) { + Ok(result) => Ok(Some(result)), + Err(RecvTimeoutError::Timeout) => Ok(None), + Err(RecvTimeoutError::Disconnected) => { + Err("scoped object worker result channel disconnected".to_string()) + } + } + } +} + +impl<'scope, 'env, T, R, E> Drop for ScopedObjectWorkerPool<'scope, 'env, T, R, E> +where + T: Send + 'env, + R: Send + 'env, + E: ObjectTaskExecutor + 'env, +{ + fn drop(&mut self) { + // Close every worker input queue so blocked `recv` calls return and + // the scoped workers exit before the enclosing scope joins them. + self.task_txs.clear(); + } +} + #[cfg(test)] mod tests { use super::{ObjectTaskExecutor, ObjectWorkerPool, ObjectWorkerSubmitError}; @@ -284,4 +402,84 @@ mod tests { Some(2) ); } + + struct BorrowingEchoExecutor<'a> { + base: &'a u32, + } + + impl<'a> ObjectTaskExecutor for BorrowingEchoExecutor<'a> { + fn execute(&self, _worker_index: usize, task: u32) -> u32 { + task + *self.base + } + } + + #[test] + fn scoped_object_worker_pool_borrows_environment_and_processes_tasks() { + let base = 100u32; + std::thread::scope(|scope| { + let mut pool = super::ScopedObjectWorkerPool::new( + scope, + 2, + 2, + BorrowingEchoExecutor { base: &base }, + ) + .expect("scoped pool"); + assert_eq!(pool.worker_count(), 2); + pool.try_submit_round_robin(1).expect("submit 1"); + pool.try_submit_round_robin(2).expect("submit 2"); + let mut results = Vec::new(); + for _ in 0..2 { + results.push( + pool.recv_result_timeout(Duration::from_secs(1)) + .expect("result channel") + .expect("result"), + ); + } + results.sort(); + assert_eq!(results, vec![101, 102]); + // Dropping the pool inside the scope closes the task queues; the + // workers exit on their own and the scope join below must not hang. + drop(pool); + }); + } + + #[test] + fn scoped_object_worker_pool_reports_full_queue() { + std::thread::scope(|scope| { + let barrier = Arc::new(Barrier::new(2)); + let started = Arc::new(AtomicBool::new(false)); + let mut pool = super::ScopedObjectWorkerPool::new( + scope, + 1, + 1, + BlockingExecutor { + barrier: Arc::clone(&barrier), + started: Arc::clone(&started), + }, + ) + .expect("scoped pool"); + pool.try_submit_round_robin(1).expect("first task"); + let deadline = std::time::Instant::now() + Duration::from_secs(1); + while !started.load(Ordering::SeqCst) { + assert!( + std::time::Instant::now() < deadline, + "scoped worker did not start first task" + ); + std::thread::sleep(Duration::from_millis(1)); + } + pool.try_submit_round_robin(2).expect("queued task"); + match pool.try_submit_round_robin(3) { + Err(ObjectWorkerSubmitError::QueueFull { worker_index, task }) => { + assert_eq!(worker_index, 0); + assert_eq!(task, 3); + } + other => panic!("expected queue full, got {other:?}"), + } + // Dropping the pool closes the task queues; releasing the barrier + // afterwards lets the blocked worker finish task 1, observe the + // closed result channel, and exit before the scope join. + drop(pool); + barrier.wait(); + }); + } } diff --git a/src/parallel/repo_worker.rs b/src/parallel/repo_worker.rs index 3e5526d..435de94 100644 --- a/src/parallel/repo_worker.rs +++ b/src/parallel/repo_worker.rs @@ -1160,7 +1160,7 @@ mod tests { matches!(result.result, RepoTransportResultKind::Success { .. }), "{result:?}" ); - let index = current_repo_index.lock().expect("index lock"); + let index = current_repo_index.read().expect("index read lock"); assert!( index .get_by_uri("rsync://example.test/repo/a.roa") @@ -1217,7 +1217,7 @@ mod tests { result.result, RepoTransportResultKind::Success { .. } )); - let index = current_repo_index.lock().expect("index lock"); + let index = current_repo_index.read().expect("index read lock"); assert!( index .get_by_uri("rsync://example.test/repo/a.roa") diff --git a/src/parallel/run_coordinator.rs b/src/parallel/run_coordinator.rs index db5ffec..2596654 100644 --- a/src/parallel/run_coordinator.rs +++ b/src/parallel/run_coordinator.rs @@ -225,7 +225,7 @@ impl GlobalRunCoordinator { self.pending_repo_tasks.clear(); self.pending_transport_tasks.clear(); self.stats = ParallelRunStats::default(); - if let Ok(mut index) = self.current_repo_index.lock() { + if let Ok(mut index) = self.current_repo_index.write() { index.clear(); } } @@ -322,7 +322,7 @@ mod tests { assert_eq!(coordinator.pending_transport_tasks.len(), 1); { - let mut index = coordinator.current_repo_index.lock().expect("index lock"); + let mut index = coordinator.current_repo_index.write().expect("index write lock"); index .apply_repository_view_entries(&[RepositoryViewEntry { rsync_uri: "rsync://example.test/repo/a.roa".to_string(), @@ -343,8 +343,8 @@ mod tests { assert_eq!( coordinator .current_repo_index - .lock() - .expect("index lock") + .read() + .expect("index read lock") .active_uri_count(), 0 ); diff --git a/src/sync/repo.rs b/src/sync/repo.rs index 9aba46d..827c03b 100644 --- a/src/sync/repo.rs +++ b/src/sync/repo.rs @@ -672,7 +672,7 @@ fn rsync_sync_into_current_store( .map_err(|e| RepoSyncError::Storage(e.to_string()))?; if let Some(index) = current_repo_index { index - .lock() + .write() .map_err(|_| RepoSyncError::Storage("current repo index lock poisoned".to_string()))? .apply_repository_view_entries(&repository_view_entries) .map_err(RepoSyncError::Storage)?; diff --git a/src/sync/rrdp.rs b/src/sync/rrdp.rs index 5d2ccda..8b13435 100644 --- a/src/sync/rrdp.rs +++ b/src/sync/rrdp.rs @@ -1029,7 +1029,7 @@ fn hydrate_current_repo_index_from_rrdp_members( } index - .lock() + .write() .map_err(|_| RrdpSyncError::Storage("current repo index lock poisoned".to_string()))? .apply_repository_view_entries(&entries) .map_err(RrdpSyncError::Storage)?; @@ -1184,7 +1184,7 @@ fn apply_delta( ¤t_hash, ); index - .lock() + .write() .map_err(|_| { RrdpSyncError::Storage("current repo index lock poisoned".to_string()) })? @@ -1240,7 +1240,7 @@ fn apply_delta( Some(previous_hash.clone()), ); index - .lock() + .write() .map_err(|_| { RrdpSyncError::Storage( "current repo index lock poisoned".to_string(), diff --git a/src/sync/rrdp/snapshot_apply.rs b/src/sync/rrdp/snapshot_apply.rs index 731febf..9360b89 100644 --- a/src/sync/rrdp/snapshot_apply.rs +++ b/src/sync/rrdp/snapshot_apply.rs @@ -306,7 +306,7 @@ pub(super) fn apply_snapshot_from_bufread( .map_err(|e| RrdpSyncError::Storage(e.to_string()))?; if let Some(index) = current_repo_index { index - .lock() + .write() .map_err(|_| RrdpSyncError::Storage("current repo index lock poisoned".to_string()))? .apply_repository_view_entries(&repository_view_entries) .map_err(RrdpSyncError::Storage)?; @@ -365,7 +365,7 @@ fn flush_snapshot_publish_batch( .map_err(|e| RrdpSyncError::Storage(e.to_string()))?; if let Some(index) = current_repo_index { index - .lock() + .write() .map_err(|_| RrdpSyncError::Storage("current repo index lock poisoned".to_string()))? .apply_repository_view_entries(&repository_view_entries) .map_err(RrdpSyncError::Storage)?; diff --git a/src/sync/rrdp/tests.rs b/src/sync/rrdp/tests.rs index 5dea86f..39a4c85 100644 --- a/src/sync/rrdp/tests.rs +++ b/src/sync/rrdp/tests.rs @@ -1025,7 +1025,7 @@ fn sync_from_notification_same_serial_hydrates_current_repo_index() { .expect("same serial no-op"); assert_eq!(applied, 0); - let index = index.lock().expect("lock index"); + let index = index.read().expect("read-lock index"); assert_eq!(index.active_uri_count(), 2); assert!(index.get_by_uri(uri_a).is_some()); assert!(index.get_by_uri(uri_b).is_some()); @@ -1089,7 +1089,7 @@ fn sync_from_notification_delta_hydrates_unchanged_current_repo_entries() { .expect("delta sync"); assert_eq!(applied, 1); - let index = index.lock().expect("lock index"); + let index = index.read().expect("read-lock index"); assert_eq!(index.active_uri_count(), 3); assert!( index.get_by_uri(uri_a).is_some(), diff --git a/src/validation/manifest.rs b/src/validation/manifest.rs index f36a0b3..90ceb94 100644 --- a/src/validation/manifest.rs +++ b/src/validation/manifest.rs @@ -639,7 +639,7 @@ pub(crate) fn try_build_fresh_publication_point_with_timing( > { let mut timing = FreshPublicationPointTimingBreakdown::default(); let current_index_lock_started = std::time::Instant::now(); - let current_index_guard = current_repo_index.and_then(|handle| handle.lock().ok()); + let current_index_guard = current_repo_index.and_then(|handle| handle.read().ok()); timing.current_index_lock_ms = current_index_lock_started.elapsed().as_millis() as u64; if !rsync_uri_is_under_publication_point(manifest_rsync_uri, publication_point_rsync_uri) { @@ -1445,8 +1445,8 @@ mod tests { } current_index - .lock() - .expect("index lock") + .write() + .expect("index write lock") .apply_repository_view_entries(&entries) .expect("apply current index"); diff --git a/src/validation/run_tree_from_tal.rs b/src/validation/run_tree_from_tal.rs index c709fcc..33271c0 100644 --- a/src/validation/run_tree_from_tal.rs +++ b/src/validation/run_tree_from_tal.rs @@ -119,7 +119,7 @@ fn snapshot_current_repo_objects( return Vec::new(); } current_repo_index - .and_then(|handle| handle.lock().ok().map(|idx| idx.snapshot_objects())) + .and_then(|handle| handle.read().ok().map(|idx| idx.snapshot_objects())) .unwrap_or_default() } @@ -2392,8 +2392,8 @@ mod multi_tal_tests { fn snapshot_current_repo_objects_is_on_demand() { let handle = CurrentRepoIndex::shared(); handle - .lock() - .expect("lock index") + .write() + .expect("write-lock index") .apply_repository_view_entries(&[RepositoryViewEntry { rsync_uri: "rsync://example.test/repo/a.roa".to_string(), current_hash: Some("11".repeat(32)), diff --git a/src/validation/tree_parallel.rs b/src/validation/tree_parallel.rs index ba6eea2..43d7bef 100644 --- a/src/validation/tree_parallel.rs +++ b/src/validation/tree_parallel.rs @@ -4,7 +4,9 @@ use std::time::{Duration, Instant}; use crate::audit::{DiscoveredFrom, PublicationPointAudit}; use crate::cir::CirInputAccumulator; -use crate::parallel::object_worker::ObjectWorkerSubmitError; +use crate::parallel::object_worker::{ + ObjectTaskExecutor, ObjectWorkerSubmitError, ScopedObjectWorkerPool, +}; use crate::parallel::repo_runtime::{RepoSyncRequestStatus, RepoSyncRuntimeOutcome}; use crate::parallel::types::RepoIdentity; use crate::policy::SignedObjectFailurePolicy; @@ -44,7 +46,7 @@ struct ReadyCaInstance { struct InflightPublicationPoint { node: QueuedCaInstance, fresh_stage: FreshPublicationPointStage, - objects_stage: ParallelObjectsStage, + objects_prepare: ParallelObjectsPrepare, repo_outcome: RepoSyncRuntimeOutcome, warnings: Vec, started_at: Instant, @@ -104,11 +106,140 @@ struct FinalizeTask { state: InflightPublicationPoint, } +/// Outcome of the pure compute phase for one ready publication point. +/// +/// `compute_ready_publication_point_stage` only performs read-only validation +/// work (publication point cache lookup, fresh snapshot staging, ROA prepare) +/// and returns this enum. `apply_ready_publication_point_stage` then performs +/// every write to control-loop state (`ca_queue`/`next_id`, `finished`, +/// `pending_roa_dispatch`, `pending_finalization`, `inflight_publication_points`) +/// in the same per-publication-point order the monolithic staging function did. +/// Each variant payload is boxed so the enum itself stays small. +enum StageOutcome { + /// Publication point cache hit: the run result is fully built in memory and + /// only needs child enqueueing plus a `finished` entry. + CacheHit(Box), + /// Fresh staging failed: apply runs the existing blocking + /// `run_publication_point` fallback inline on the control thread. + FreshError(Box), + /// Fresh staging succeeded and ROA prepare returned complete objects (no + /// ROA tasks to dispatch): apply hands the publication point to the + /// finalize worker through `pending_finalization` instead of running the + /// finalize synchronously on the control thread. + Complete(Box), + /// Fresh staging succeeded with a staged objects plan that contains zero + /// ROA tasks: apply queues the finalize task directly. + ZeroTask(Box), + /// Fresh staging succeeded with ROA tasks to dispatch: apply appends the + /// tasks to `pending_roa_dispatch` and registers the inflight publication + /// point. + Fresh(Box), +} + +struct CacheHitOutcome { + ready: ReadyCaInstance, + publication_point_started: Instant, + result: PublicationPointRunResult, +} + +struct FreshErrorOutcome { + ready: ReadyCaInstance, + publication_point_started: Instant, +} + +struct CompleteOutcome { + ready: ReadyCaInstance, + publication_point_started: Instant, + fresh_stage: FreshPublicationPointStage, + warnings: Vec, + objects: ObjectsOutput, +} + +struct StagedOutcome { + ready: ReadyCaInstance, + publication_point_started: Instant, + fresh_stage: FreshPublicationPointStage, + warnings: Vec, + objects_stage: ParallelObjectsStage, +} + struct FinalizeWorkerResult { finished: FinishedPublicationPoint, metrics: FinalizePublicationPointMetrics, } +/// Task submitted to the experimental ready-stage worker pool: everything +/// `compute_ready_publication_point_stage` needs for one ready publication +/// point. +struct ReadyStageTask { + ready: ReadyCaInstance, + ready_queue_len_after_pop: usize, + submitted_at: Instant, +} + +/// Result drained from the ready-stage worker pool: the compute outcome and +/// its metrics plus per-task pool timing for the `phase2_stage_pool_stats` +/// observability event. +struct ReadyStageWorkerResult { + outcome: StageOutcome, + metrics: ReadyStageMetrics, + queue_wait_ms: u64, + worker_ms: u64, +} + +/// Executor borrowing the publication point runner so stage workers can run +/// the read-only compute phase off the control thread. The runner is shared +/// with the finalize worker and the ROA pool in the same way; the scoped pool +/// guarantees all borrows end before the enclosing `std::thread::scope`. +struct ReadyStageTaskExecutor<'a> { + runner: &'a Rpkiv1PublicationPointRunner<'a>, +} + +impl<'a> ObjectTaskExecutor for ReadyStageTaskExecutor<'a> { + fn execute(&self, _worker_index: usize, task: ReadyStageTask) -> ReadyStageWorkerResult { + let worker_started = Instant::now(); + let queue_wait_ms = worker_started + .saturating_duration_since(task.submitted_at) + .as_millis() as u64; + let (outcome, metrics) = compute_ready_publication_point_stage( + self.runner, + task.ready, + task.ready_queue_len_after_pop, + ); + ReadyStageWorkerResult { + outcome, + metrics, + queue_wait_ms, + worker_ms: elapsed_ms(worker_started), + } + } +} + +type ReadyStagePool<'scope, 'env> = ScopedObjectWorkerPool< + 'scope, + 'env, + ReadyStageTask, + ReadyStageWorkerResult, + ReadyStageTaskExecutor<'env>, +>; + +#[derive(Default)] +struct StageDispatchMetrics { + submitted: usize, + queue_full: bool, + duration_ms: u64, +} + +#[derive(Default)] +struct StageDrainMetrics { + results_drained: usize, + queue_wait_ms_total: u64, + queue_wait_ms_max: u64, + worker_ms_total: u64, + worker_ms_max: u64, + duration_ms: u64, +} + #[derive(Default)] struct ReadyStageMetrics { manifest_rsync_uri: Option, @@ -551,6 +682,14 @@ pub fn run_tree_parallel_phase2_audit_multi_root( .map(|cfg| cfg.publication_point_finalize_queue_capacity) .unwrap_or(32768) .max(1); + // Experimental ready-stage pool: `stage_workers == 0` keeps the inline + // compute+apply staging path byte-for-byte; any positive value moves the + // compute phase to a scoped worker pool. + let stage_worker_count = phase2_config.map(|cfg| cfg.stage_workers).unwrap_or(0); + let stage_queue_capacity = phase2_config + .map(|cfg| cfg.worker_queue_capacity) + .unwrap_or(256) + .max(1); let (finalize_task_tx, finalize_task_rx) = mpsc::sync_channel::(publication_point_finalize_queue_capacity); @@ -567,9 +706,35 @@ pub fn run_tree_parallel_phase2_audit_multi_root( ) }); + // The stage pool borrows the runner like the finalize worker does; it + // lives inside this scope and is dropped before the scope joins. + let mut stage_pool = if stage_worker_count > 0 { + Some( + ReadyStagePool::new( + scope, + stage_worker_count, + stage_queue_capacity, + ReadyStageTaskExecutor { runner }, + ) + .map_err(TreeRunError::Runner)?, + ) + } else { + None + }; + // Submitted-but-not-yet-drained stage tasks. The drain loop collects + // every available result each turn, so `staging_inflight == 0` also + // implies the stage result channel is empty. + let mut staging_inflight = 0usize; + let run_result: Result<(), TreeRunError> = (|| { loop { let control_loop_started = Instant::now(); + // With the stage pool enabled the batch wall clock covers the + // whole turn (turn-head drain + dispatch + apply); the inline + // path keeps its historical start point at the ready batch. + let turn_stage_started = Instant::now(); + let mut ready_batch_metrics = ReadyStageBatchMetrics::default(); + let mut stage_drain_metrics = StageDrainMetrics::default(); drain_finalize_results_with_progress( &finalize_result_rx, &mut finished, @@ -591,6 +756,22 @@ pub fn run_tree_parallel_phase2_audit_multi_root( pending_roa_dispatch.len(), object_result_drain_batch_size, )?; + if let Some(pool) = stage_pool.as_ref() { + drain_stage_results( + pool, + runner, + &mut next_id, + &mut ca_queue, + &mut pending_roa_dispatch, + &mut inflight_publication_points, + &mut pending_finalization, + &mut finished, + &mut staging_inflight, + &mut ready_batch_metrics, + &mut stage_drain_metrics, + config, + )?; + } submit_pending_finalization_with_progress( &finalize_task_tx, &mut pending_finalization, @@ -618,6 +799,7 @@ pub fn run_tree_parallel_phase2_audit_multi_root( &inflight_publication_points, &pending_finalization, finalize_inflight, + staging_inflight, instances_started, config, ); @@ -642,14 +824,24 @@ pub fn run_tree_parallel_phase2_audit_multi_root( } let ready_batch_started = Instant::now(); - let mut ready_batch_metrics = ReadyStageBatchMetrics::default(); let mut ready_time_budget_exhausted = false; - while ready_batch_metrics.ready_count < ready_batch_size { - let Some(ready) = ready_queue.pop_front() else { - break; - }; - let ready_queue_len_after_pop = ready_queue.len(); - let metrics = stage_ready_publication_point( + let mut stage_dispatch_metrics = StageDispatchMetrics::default(); + if let Some(pool) = stage_pool.as_mut() { + // Pool path: the ready batch becomes a dispatch loop that + // submits compute tasks and applies whatever results are + // already available; backpressure requeues for next turn. + stage_dispatch_metrics = submit_ready_batch_to_stage_pool( + pool, + &mut ready_queue, + &mut staging_inflight, + ready_batch_size, + ready_batch_wall_time_budget, + )?; + ready_time_budget_exhausted = stage_dispatch_metrics.queue_full + || (!ready_queue.is_empty() + && ready_batch_started.elapsed() >= ready_batch_wall_time_budget); + drain_stage_results( + pool, runner, &mut next_id, &mut ca_queue, @@ -657,21 +849,50 @@ pub fn run_tree_parallel_phase2_audit_multi_root( &mut inflight_publication_points, &mut pending_finalization, &mut finished, - ready, - ready_queue_len_after_pop, + &mut staging_inflight, + &mut ready_batch_metrics, + &mut stage_drain_metrics, config, - config.compact_audit, - ); - ready_batch_metrics.record(metrics); - if ready_batch_metrics.ready_count > 0 - && ready_batch_started.elapsed() >= ready_batch_wall_time_budget - { - ready_time_budget_exhausted = !ready_queue.is_empty(); - break; + )?; + } else { + while ready_batch_metrics.ready_count < ready_batch_size { + let Some(ready) = ready_queue.pop_front() else { + break; + }; + let ready_queue_len_after_pop = ready_queue.len(); + let (outcome, metrics) = compute_ready_publication_point_stage( + runner, + ready, + ready_queue_len_after_pop, + ); + let metrics = apply_ready_publication_point_stage( + runner, + &mut next_id, + &mut ca_queue, + &mut pending_roa_dispatch, + &mut inflight_publication_points, + &mut pending_finalization, + &mut finished, + outcome, + metrics, + config, + config.compact_audit, + ); + ready_batch_metrics.record(metrics); + if ready_batch_metrics.ready_count > 0 + && ready_batch_started.elapsed() >= ready_batch_wall_time_budget + { + ready_time_budget_exhausted = !ready_queue.is_empty(); + break; + } } } if ready_batch_metrics.ready_count > 0 { - ready_batch_metrics.total_ms = elapsed_ms(ready_batch_started); + ready_batch_metrics.total_ms = if stage_pool.is_some() { + elapsed_ms(turn_stage_started) + } else { + elapsed_ms(ready_batch_started) + }; let ready_count_budget_exhausted = ready_batch_metrics.ready_count >= ready_batch_size && !ready_queue.is_empty(); @@ -733,6 +954,29 @@ pub fn run_tree_parallel_phase2_audit_multi_root( }), ); } + if stage_pool.is_some() + && (stage_dispatch_metrics.submitted > 0 + || stage_drain_metrics.results_drained > 0 + || stage_dispatch_metrics.queue_full) + { + crate::progress_log::emit( + "phase2_stage_pool_stats", + serde_json::json!({ + "stage_workers": stage_worker_count, + "submitted": stage_dispatch_metrics.submitted, + "results_drained": stage_drain_metrics.results_drained, + "queue_full": stage_dispatch_metrics.queue_full, + "staging_inflight": staging_inflight, + "ready_queue_len": ready_queue.len(), + "queue_wait_ms_total": stage_drain_metrics.queue_wait_ms_total, + "queue_wait_ms_max": stage_drain_metrics.queue_wait_ms_max, + "worker_ms_total": stage_drain_metrics.worker_ms_total, + "worker_ms_max": stage_drain_metrics.worker_ms_max, + "dispatch_duration_ms": stage_dispatch_metrics.duration_ms, + "drain_duration_ms": stage_drain_metrics.duration_ms, + }), + ); + } flush_pending_roa_dispatch_with_progress( runner, @@ -786,6 +1030,7 @@ pub fn run_tree_parallel_phase2_audit_multi_root( &inflight_publication_points, &pending_finalization, finalize_inflight, + staging_inflight, instances_started, config, ) { @@ -800,6 +1045,9 @@ pub fn run_tree_parallel_phase2_audit_multi_root( })(); drop(finalize_task_tx); + // Dropping the pool closes the stage task queues so the scoped stage + // workers exit before the scope joins them below. + drop(stage_pool); let worker_result = finalize_worker .join() .map_err(|_| TreeRunError::Runner("phase2 finalize worker panicked".to_string()))?; @@ -948,19 +1196,11 @@ fn start_queued_ca_instances( } } -fn stage_ready_publication_point( +fn compute_ready_publication_point_stage( runner: &Rpkiv1PublicationPointRunner<'_>, - next_id: &mut u64, - ca_queue: &mut VecDeque, - pending_roa_dispatch: &mut VecDeque, - inflight_publication_points: &mut HashMap, - pending_finalization: &mut VecDeque, - finished: &mut Vec, ready: ReadyCaInstance, ready_queue_len_after_pop: usize, - config: &TreeRunConfig, - compact_audit: bool, -) -> ReadyStageMetrics { +) -> (StageOutcome, ReadyStageMetrics) { let publication_point_started = Instant::now(); let ready_queue_wait_ms = publication_point_started .saturating_duration_since(ready.ready_enqueued_at) @@ -985,38 +1225,14 @@ fn stage_ready_publication_point( ) { metrics.complete_count = 1; metrics.discovered_children = result.discovered_children.len(); - let child_enqueue_started = Instant::now(); - enqueue_discovered_children( - runner, - next_id, - ca_queue, - &ready.node, - config, - result.discovered_children.clone(), + return ( + StageOutcome::CacheHit(Box::new(CacheHitOutcome { + ready, + publication_point_started, + result, + })), + metrics, ); - metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); - runner.record_publication_point_step_ms( - metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), - "publication_point_cache_child_enqueue", - metrics.child_enqueue_ms, - ); - finished.push(FinishedPublicationPoint { - node: FinishedPublicationPointNode::from_queued(ready.node), - result: compact_phase2_finished_result(result, compact_audit), - }); - metrics.total_ms = elapsed_ms(publication_point_started); - emit_ready_publication_point_control_slow( - metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), - metrics - .publication_point_rsync_uri - .as_deref() - .unwrap_or_default(), - &repo_outcome, - &metrics, - "publication_point_cache", - false, - ); - return metrics; } let stage_fresh_started = Instant::now(); @@ -1046,40 +1262,15 @@ fn stage_ready_publication_point( }), ); } - metrics.fallback_count = 1; - let fallback_started = Instant::now(); - let fallback = runner.run_publication_point(&ready.node.handle); - metrics.fallback_full_run_ms = elapsed_ms(fallback_started); - if let Ok(result) = fallback.as_ref() { - metrics.discovered_children = result.discovered_children.len(); - let child_enqueue_started = Instant::now(); - enqueue_discovered_children( - runner, - next_id, - ca_queue, - &ready.node, - config, - result.discovered_children.clone(), - ); - metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); - } - finished.push(FinishedPublicationPoint { - node: FinishedPublicationPointNode::from_queued(ready.node), - result: compact_phase2_finished_result_result(fallback, compact_audit), - }); - metrics.total_ms = elapsed_ms(publication_point_started); - emit_ready_publication_point_control_slow( - metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), - metrics - .publication_point_rsync_uri - .as_deref() - .unwrap_or_default(), - &repo_outcome, - &metrics, - "fallback", - true, + // The blocking `run_publication_point` fallback stays on the control + // thread; it is executed by the apply phase for this outcome. + return ( + StageOutcome::FreshError(Box::new(FreshErrorOutcome { + ready, + publication_point_started, + })), + metrics, ); - return metrics; } }; metrics.snapshot_prepare_ms = fresh_stage.snapshot_prepare_ms; @@ -1138,16 +1329,6 @@ fn stage_ready_publication_point( warnings.extend(fresh_stage.warnings.clone()); metrics.discovered_children = fresh_stage.discovered_children.len(); - let child_enqueue_started = Instant::now(); - enqueue_discovered_children( - runner, - next_id, - ca_queue, - &ready.node, - config, - fresh_stage.discovered_children.clone(), - ); - metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); let prepare_started = Instant::now(); let roa_presence_scan_started = Instant::now(); @@ -1217,25 +1398,16 @@ fn stage_ready_publication_point( &objects.router_keys, ), ); - let direct_finalize_started = Instant::now(); - let finalize_metrics = finalize_ready_objects( - runner, - ready.node, - fresh_stage, - warnings, - objects, - repo_outcome.clone(), - finished, - compact_audit, - ); - metrics.direct_finalize_ms = finalize_metrics - .finalize_ms - .max(elapsed_ms(direct_finalize_started)); - runner.record_publication_point_step_ms( - &metrics.manifest_rsync_uri.clone().unwrap_or_default(), - "fresh_direct_finalize", - metrics.direct_finalize_ms, - ); + ( + StageOutcome::Complete(Box::new(CompleteOutcome { + ready, + publication_point_started, + fresh_stage, + warnings, + objects, + })), + metrics, + ) } ParallelObjectsPrepare::Staged(objects_stage) => { metrics.prepare_ms = elapsed_ms(prepare_started); @@ -1247,6 +1419,208 @@ fn stage_ready_publication_point( metrics.staged_count = 1; metrics.locked_files = objects_stage.locked_file_count(); metrics.aspa_objects = objects_stage.aspa_task_count(); + let task_count = objects_stage.roa_task_count(); + metrics.roa_tasks = task_count; + let outcome = StagedOutcome { + ready, + publication_point_started, + fresh_stage, + warnings, + objects_stage, + }; + if task_count == 0 { + metrics.zero_task_count = 1; + (StageOutcome::ZeroTask(Box::new(outcome)), metrics) + } else { + (StageOutcome::Fresh(Box::new(outcome)), metrics) + } + } + } +} + +fn apply_ready_publication_point_stage( + runner: &Rpkiv1PublicationPointRunner<'_>, + next_id: &mut u64, + ca_queue: &mut VecDeque, + pending_roa_dispatch: &mut VecDeque, + inflight_publication_points: &mut HashMap, + pending_finalization: &mut VecDeque, + finished: &mut Vec, + outcome: StageOutcome, + mut metrics: ReadyStageMetrics, + config: &TreeRunConfig, + compact_audit: bool, +) -> ReadyStageMetrics { + match outcome { + StageOutcome::CacheHit(outcome) => { + let CacheHitOutcome { + ready, + publication_point_started, + result, + } = *outcome; + let repo_outcome = ready.repo_outcome.clone(); + let child_enqueue_started = Instant::now(); + enqueue_discovered_children( + runner, + next_id, + ca_queue, + &ready.node, + config, + result.discovered_children.clone(), + ); + metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); + runner.record_publication_point_step_ms( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + "publication_point_cache_child_enqueue", + metrics.child_enqueue_ms, + ); + finished.push(FinishedPublicationPoint { + node: FinishedPublicationPointNode::from_queued(ready.node), + result: compact_phase2_finished_result(result, compact_audit), + }); + metrics.total_ms = elapsed_ms(publication_point_started); + emit_ready_publication_point_control_slow( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + metrics + .publication_point_rsync_uri + .as_deref() + .unwrap_or_default(), + &repo_outcome, + &metrics, + "publication_point_cache", + false, + ); + metrics + } + StageOutcome::FreshError(outcome) => { + let FreshErrorOutcome { + ready, + publication_point_started, + } = *outcome; + let repo_outcome = ready.repo_outcome.clone(); + metrics.fallback_count = 1; + let fallback_started = Instant::now(); + let fallback = runner.run_publication_point(&ready.node.handle); + metrics.fallback_full_run_ms = elapsed_ms(fallback_started); + if let Ok(result) = fallback.as_ref() { + metrics.discovered_children = result.discovered_children.len(); + let child_enqueue_started = Instant::now(); + enqueue_discovered_children( + runner, + next_id, + ca_queue, + &ready.node, + config, + result.discovered_children.clone(), + ); + metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); + } + finished.push(FinishedPublicationPoint { + node: FinishedPublicationPointNode::from_queued(ready.node), + result: compact_phase2_finished_result_result(fallback, compact_audit), + }); + metrics.total_ms = elapsed_ms(publication_point_started); + emit_ready_publication_point_control_slow( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + metrics + .publication_point_rsync_uri + .as_deref() + .unwrap_or_default(), + &repo_outcome, + &metrics, + "fallback", + true, + ); + metrics + } + StageOutcome::Complete(outcome) => { + let CompleteOutcome { + ready, + publication_point_started, + fresh_stage, + warnings, + objects, + } = *outcome; + let repo_outcome = ready.repo_outcome.clone(); + let child_enqueue_started = Instant::now(); + enqueue_discovered_children( + runner, + next_id, + ca_queue, + &ready.node, + config, + fresh_stage.discovered_children.clone(), + ); + metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); + // The finalize no longer runs synchronously here: the publication + // point is queued for the shared finalize worker through the same + // pending_finalization path (queue capacity backpressure and + // finalize_inflight accounting included) as zero-task staging. The + // per-publication-point total timing is recorded by the finalize + // worker, exactly like the zero-task and staged paths. + let direct_finalize_started = Instant::now(); + pending_finalization.push_back(FinalizeTask { + state: InflightPublicationPoint { + node: ready.node, + fresh_stage, + objects_prepare: ParallelObjectsPrepare::Complete(objects), + repo_outcome: repo_outcome.clone(), + warnings, + started_at: publication_point_started, + objects_started_at: Instant::now(), + task_count: 0, + tasks_submitted: 0, + first_task_submitted_at: None, + last_task_submitted_at: None, + first_result_at: None, + last_result_at: None, + worker_ms_total: 0, + worker_ms_max: 0, + queue_wait_ms_total: 0, + queue_wait_ms_max: 0, + finalize_enqueued_at: Some(Instant::now()), + results: Vec::new(), + }, + }); + metrics.direct_finalize_ms = elapsed_ms(direct_finalize_started); + runner.record_publication_point_step_ms( + &metrics.manifest_rsync_uri.clone().unwrap_or_default(), + "fresh_direct_finalize", + metrics.direct_finalize_ms, + ); + metrics.total_ms = elapsed_ms(publication_point_started); + emit_ready_publication_point_control_slow( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + metrics + .publication_point_rsync_uri + .as_deref() + .unwrap_or_default(), + &repo_outcome, + &metrics, + "complete", + false, + ); + metrics + } + StageOutcome::ZeroTask(outcome) => { + let StagedOutcome { + ready, + publication_point_started, + fresh_stage, + warnings, + objects_stage, + } = *outcome; + let repo_outcome = ready.repo_outcome.clone(); + let child_enqueue_started = Instant::now(); + enqueue_discovered_children( + runner, + next_id, + ca_queue, + &ready.node, + config, + fresh_stage.discovered_children.clone(), + ); + metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); let build_tasks_started = Instant::now(); objects_stage.append_roa_tasks_to(pending_roa_dispatch); metrics.build_roa_tasks_ms = elapsed_ms(build_tasks_started); @@ -1256,85 +1630,110 @@ fn stage_ready_publication_point( metrics.build_roa_tasks_ms, ); let task_count = objects_stage.roa_task_count(); - metrics.roa_tasks = task_count; - if task_count == 0 { - metrics.zero_task_count = 1; - pending_finalization.push_back(FinalizeTask { - state: InflightPublicationPoint { - node: ready.node, - fresh_stage, - objects_stage, - repo_outcome: repo_outcome.clone(), - warnings, - started_at: publication_point_started, - objects_started_at: Instant::now(), - task_count, - tasks_submitted: 0, - first_task_submitted_at: None, - last_task_submitted_at: None, - first_result_at: None, - last_result_at: None, - worker_ms_total: 0, - worker_ms_max: 0, - queue_wait_ms_total: 0, - queue_wait_ms_max: 0, - finalize_enqueued_at: Some(Instant::now()), - results: Vec::new(), - }, - }); - } else { - inflight_publication_points.insert( - ready.node.id, - InflightPublicationPoint { - node: ready.node, - fresh_stage, - objects_stage, - repo_outcome: repo_outcome.clone(), - warnings, - started_at: publication_point_started, - objects_started_at: Instant::now(), - task_count, - tasks_submitted: 0, - first_task_submitted_at: None, - last_task_submitted_at: None, - first_result_at: None, - last_result_at: None, - worker_ms_total: 0, - worker_ms_max: 0, - queue_wait_ms_total: 0, - queue_wait_ms_max: 0, - finalize_enqueued_at: None, - results: Vec::with_capacity(task_count), - }, - ); - } + pending_finalization.push_back(FinalizeTask { + state: InflightPublicationPoint { + node: ready.node, + fresh_stage, + objects_prepare: ParallelObjectsPrepare::Staged(objects_stage), + repo_outcome: repo_outcome.clone(), + warnings, + started_at: publication_point_started, + objects_started_at: Instant::now(), + task_count, + tasks_submitted: 0, + first_task_submitted_at: None, + last_task_submitted_at: None, + first_result_at: None, + last_result_at: None, + worker_ms_total: 0, + worker_ms_max: 0, + queue_wait_ms_total: 0, + queue_wait_ms_max: 0, + finalize_enqueued_at: Some(Instant::now()), + results: Vec::new(), + }, + }); + metrics.total_ms = elapsed_ms(publication_point_started); + emit_ready_publication_point_control_slow( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + metrics + .publication_point_rsync_uri + .as_deref() + .unwrap_or_default(), + &repo_outcome, + &metrics, + "zero_task", + false, + ); + metrics + } + StageOutcome::Fresh(outcome) => { + let StagedOutcome { + ready, + publication_point_started, + fresh_stage, + warnings, + objects_stage, + } = *outcome; + let repo_outcome = ready.repo_outcome.clone(); + let child_enqueue_started = Instant::now(); + enqueue_discovered_children( + runner, + next_id, + ca_queue, + &ready.node, + config, + fresh_stage.discovered_children.clone(), + ); + metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); + let build_tasks_started = Instant::now(); + objects_stage.append_roa_tasks_to(pending_roa_dispatch); + metrics.build_roa_tasks_ms = elapsed_ms(build_tasks_started); + runner.record_publication_point_step_ms( + &ready.node.handle.manifest_rsync_uri, + "fresh_build_roa_tasks", + metrics.build_roa_tasks_ms, + ); + let task_count = objects_stage.roa_task_count(); + inflight_publication_points.insert( + ready.node.id, + InflightPublicationPoint { + node: ready.node, + fresh_stage, + objects_prepare: ParallelObjectsPrepare::Staged(objects_stage), + repo_outcome: repo_outcome.clone(), + warnings, + started_at: publication_point_started, + objects_started_at: Instant::now(), + task_count, + tasks_submitted: 0, + first_task_submitted_at: None, + last_task_submitted_at: None, + first_result_at: None, + last_result_at: None, + worker_ms_total: 0, + worker_ms_max: 0, + queue_wait_ms_total: 0, + queue_wait_ms_max: 0, + finalize_enqueued_at: None, + results: Vec::with_capacity(task_count), + }, + ); + metrics.total_ms = elapsed_ms(publication_point_started); + emit_ready_publication_point_control_slow( + metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), + metrics + .publication_point_rsync_uri + .as_deref() + .unwrap_or_default(), + &repo_outcome, + &metrics, + "staged", + false, + ); + metrics } } - metrics.total_ms = elapsed_ms(publication_point_started); - if metrics.complete_count > 0 { - runner.record_publication_point_total_ms( - metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), - metrics.total_ms, - ); - } - emit_ready_publication_point_control_slow( - metrics.manifest_rsync_uri.as_deref().unwrap_or_default(), - metrics - .publication_point_rsync_uri - .as_deref() - .unwrap_or_default(), - &repo_outcome, - &metrics, - if metrics.complete_count > 0 { - "complete" - } else if metrics.zero_task_count > 0 { - "zero_task" - } else { - "staged" - }, - false, - ); - metrics } fn emit_ready_publication_point_control_slow( @@ -1448,75 +1847,6 @@ fn enqueue_discovered_children( } } -fn finalize_ready_objects( - runner: &Rpkiv1PublicationPointRunner<'_>, - node: QueuedCaInstance, - fresh_stage: FreshPublicationPointStage, - warnings: Vec, - objects: ObjectsOutput, - repo_outcome: RepoSyncRuntimeOutcome, - finished: &mut Vec, - compact_audit: bool, -) -> FinalizePublicationPointMetrics { - let locked_files = fresh_stage.fresh_point.files().len(); - let finalize_started = Instant::now(); - let finalized = runner.finalize_fresh_publication_point_from_reducer( - &node.handle, - &fresh_stage.fresh_point, - warnings, - objects, - fresh_stage.child_audits, - fresh_stage.discovered_children, - repo_outcome.repo_sync_source.as_deref(), - repo_outcome.repo_sync_phase.as_deref(), - repo_outcome.repo_sync_duration_ms, - repo_outcome.repo_sync_err.as_deref(), - ); - let finalize_ms = elapsed_ms(finalize_started); - let (result, metrics) = match finalized { - Ok(output) => { - let metrics = finalize_metrics_from_output( - &output, - 0, - finalize_ms, - None, - finalize_ms, - locked_files, - ); - emit_finalize_breakdown( - "phase2_direct_finalize_breakdown", - &node.handle.manifest_rsync_uri, - &node.handle.publication_point_rsync_uri, - &metrics, - ); - ( - compact_phase2_finished_result(output.result, compact_audit), - metrics, - ) - } - Err(err) => { - let metrics = FinalizePublicationPointMetrics { - finalize_ms, - finalize_worker_ms: finalize_ms, - locked_files, - ..FinalizePublicationPointMetrics::default() - }; - emit_finalize_breakdown( - "phase2_direct_finalize_breakdown", - &node.handle.manifest_rsync_uri, - &node.handle.publication_point_rsync_uri, - &metrics, - ); - (FinishedPublicationPointResult::Err(err), metrics) - } - }; - finished.push(FinishedPublicationPoint { - node: FinishedPublicationPointNode::from_queued(node), - result, - }); - metrics -} - fn finalize_metrics_from_output( output: &FreshPublicationPointFinalizeOutput, reduce_ms: u64, @@ -1740,6 +2070,127 @@ fn drain_object_results_with_progress( Ok(()) } +/// Submission endpoint of the stage pool, abstracted so the dispatch loop can +/// be tested with a deterministic backpressure source. +trait ReadyStageSubmitter { + // The error must hand the task back so the dispatch loop can requeue it; + // boxing it away would only shuffle the same bytes around. + #[allow(clippy::result_large_err)] + fn try_submit_ready_stage( + &mut self, + task: ReadyStageTask, + ) -> Result<(), ObjectWorkerSubmitError>; +} + +impl ReadyStageSubmitter for ReadyStagePool<'_, '_> { + fn try_submit_ready_stage( + &mut self, + task: ReadyStageTask, + ) -> Result<(), ObjectWorkerSubmitError> { + self.try_submit_round_robin(task).map(|_| ()) + } +} + +/// Dispatch loop of the pool-enabled ready batch: pop ready publication +/// points and submit their compute tasks to the stage pool, following the +/// `flush_pending_roa_dispatch` backpressure pattern — on a full worker queue +/// the publication point goes back to the head of the ready queue so the next +/// turn retries it first (nothing lost, nothing duplicated). +fn submit_ready_batch_to_stage_pool( + stage_pool: &mut impl ReadyStageSubmitter, + ready_queue: &mut VecDeque, + staging_inflight: &mut usize, + ready_batch_size: usize, + ready_batch_wall_time_budget: Duration, +) -> Result { + let started = Instant::now(); + let mut metrics = StageDispatchMetrics::default(); + while metrics.submitted < ready_batch_size { + let Some(ready) = ready_queue.pop_front() else { + break; + }; + let task = ReadyStageTask { + ready_queue_len_after_pop: ready_queue.len(), + ready, + submitted_at: Instant::now(), + }; + match stage_pool.try_submit_ready_stage(task) { + Ok(_) => { + *staging_inflight += 1; + metrics.submitted += 1; + } + Err(ObjectWorkerSubmitError::QueueFull { task, .. }) => { + ready_queue.push_front(task.ready); + metrics.queue_full = true; + break; + } + Err(ObjectWorkerSubmitError::Disconnected { .. }) => { + return Err(TreeRunError::Runner( + "ready stage worker queue disconnected".to_string(), + )); + } + } + if metrics.submitted > 0 && started.elapsed() >= ready_batch_wall_time_budget { + break; + } + } + metrics.duration_ms = elapsed_ms(started); + Ok(metrics) +} + +/// Collect every available stage result without blocking and run the apply +/// phase for each on the control thread, exactly like the inline path would +/// have done right after compute. Drained metrics aggregate into the same +/// `ReadyStageBatchMetrics`, keeping the `phase2_ready_queue_*` events +/// unchanged. +#[allow(clippy::too_many_arguments)] +fn drain_stage_results( + stage_pool: &ReadyStagePool<'_, '_>, + runner: &Rpkiv1PublicationPointRunner<'_>, + next_id: &mut u64, + ca_queue: &mut VecDeque, + pending_roa_dispatch: &mut VecDeque, + inflight_publication_points: &mut HashMap, + pending_finalization: &mut VecDeque, + finished: &mut Vec, + staging_inflight: &mut usize, + batch_metrics: &mut ReadyStageBatchMetrics, + pool_metrics: &mut StageDrainMetrics, + config: &TreeRunConfig, +) -> Result<(), TreeRunError> { + let started = Instant::now(); + loop { + let Some(result) = stage_pool + .recv_result_timeout(Duration::from_millis(0)) + .map_err(TreeRunError::Runner)? + else { + break; + }; + pool_metrics.results_drained += 1; + pool_metrics.queue_wait_ms_total += result.queue_wait_ms; + pool_metrics.queue_wait_ms_max = pool_metrics.queue_wait_ms_max.max(result.queue_wait_ms); + pool_metrics.worker_ms_total += result.worker_ms; + pool_metrics.worker_ms_max = pool_metrics.worker_ms_max.max(result.worker_ms); + *staging_inflight = staging_inflight.saturating_sub(1); + let metrics = apply_ready_publication_point_stage( + runner, + next_id, + ca_queue, + pending_roa_dispatch, + inflight_publication_points, + pending_finalization, + finished, + result.outcome, + result.metrics, + config, + config.compact_audit, + ); + batch_metrics.record(metrics); + } + pool_metrics.duration_ms += elapsed_ms(started); + Ok(()) +} + fn submit_pending_finalization( finalize_task_tx: &SyncSender, pending_finalization: &mut VecDeque, @@ -1939,7 +2390,7 @@ fn finalize_publication_point_state( let InflightPublicationPoint { node, fresh_stage, - objects_stage, + objects_prepare, repo_outcome, warnings, started_at, @@ -1968,28 +2419,91 @@ fn finalize_publication_point_state( "fresh_objects_processing_lifetime", objects_processing_ms, ); - let reduce_started = Instant::now(); - let locked_files = objects_stage.locked_file_count(); - let reduce_result = reduce_parallel_roa_stage(objects_stage, results, runner.timing.as_ref()); - let reduce_ms = elapsed_ms(reduce_started); - runner.record_publication_point_step_ms( - &node.handle.manifest_rsync_uri, - "fresh_roa_reduce", - reduce_ms, - ); - let (result, mut metrics) = match reduce_result { - Ok(mut objects) => { - let finalize_started = Instant::now(); - objects - .router_keys - .extend(fresh_stage.discovered_router_keys.clone()); - objects.local_outputs_cache.extend( - crate::validation::tree_runner::build_router_key_local_outputs( - &node.handle, - &objects.router_keys, - ), + let (result, mut metrics, reduce_ms) = match objects_prepare { + ParallelObjectsPrepare::Staged(objects_stage) => { + let reduce_started = Instant::now(); + let locked_files = objects_stage.locked_file_count(); + let reduce_result = + reduce_parallel_roa_stage(objects_stage, results, runner.timing.as_ref()); + let reduce_ms = elapsed_ms(reduce_started); + runner.record_publication_point_step_ms( + &node.handle.manifest_rsync_uri, + "fresh_roa_reduce", + reduce_ms, ); + + let (result, metrics) = match reduce_result { + Ok(mut objects) => { + let finalize_started = Instant::now(); + objects + .router_keys + .extend(fresh_stage.discovered_router_keys.clone()); + objects.local_outputs_cache.extend( + crate::validation::tree_runner::build_router_key_local_outputs( + &node.handle, + &objects.router_keys, + ), + ); + let finalized = runner.finalize_fresh_publication_point_from_reducer( + &node.handle, + &fresh_stage.fresh_point, + warnings, + objects, + fresh_stage.child_audits, + fresh_stage.discovered_children, + repo_outcome.repo_sync_source.as_deref(), + repo_outcome.repo_sync_phase.as_deref(), + repo_outcome.repo_sync_duration_ms, + repo_outcome.repo_sync_err.as_deref(), + ); + let finalize_ms = elapsed_ms(finalize_started); + match finalized { + Ok(output) => { + let metrics = finalize_metrics_from_output( + &output, + reduce_ms, + finalize_ms, + finalize_queue_wait_ms, + 0, + locked_files, + ); + ( + compact_phase2_finished_result(output.result, compact_audit), + metrics, + ) + } + Err(err) => ( + FinishedPublicationPointResult::Err(err), + FinalizePublicationPointMetrics { + reduce_ms, + finalize_ms, + finalize_queue_wait_ms, + locked_files, + ..FinalizePublicationPointMetrics::default() + }, + ), + } + } + Err(err) => ( + FinishedPublicationPointResult::Err(err), + FinalizePublicationPointMetrics { + reduce_ms, + finalize_queue_wait_ms, + locked_files, + ..FinalizePublicationPointMetrics::default() + }, + ), + }; + (result, metrics, reduce_ms) + } + ParallelObjectsPrepare::Complete(objects) => { + // ROA prepare already produced complete objects for this publication + // point, so there is nothing to reduce; finalize directly. This is + // the former control-thread "direct finalize", now running on the + // finalize worker through the regular task queue. + let locked_files = fresh_stage.fresh_point.files().len(); + let finalize_started = Instant::now(); let finalized = runner.finalize_fresh_publication_point_from_reducer( &node.handle, &fresh_stage.fresh_point, @@ -2003,11 +2517,11 @@ fn finalize_publication_point_state( repo_outcome.repo_sync_err.as_deref(), ); let finalize_ms = elapsed_ms(finalize_started); - match finalized { + let (result, metrics) = match finalized { Ok(output) => { let metrics = finalize_metrics_from_output( &output, - reduce_ms, + 0, finalize_ms, finalize_queue_wait_ms, 0, @@ -2021,24 +2535,15 @@ fn finalize_publication_point_state( Err(err) => ( FinishedPublicationPointResult::Err(err), FinalizePublicationPointMetrics { - reduce_ms, finalize_ms, finalize_queue_wait_ms, locked_files, ..FinalizePublicationPointMetrics::default() }, ), - } + }; + (result, metrics, 0) } - Err(err) => ( - FinishedPublicationPointResult::Err(err), - FinalizePublicationPointMetrics { - reduce_ms, - finalize_queue_wait_ms, - locked_files, - ..FinalizePublicationPointMetrics::default() - }, - ), }; let finalize_worker_ms = elapsed_ms(finalize_worker_started); metrics.finalize_worker_ms = finalize_worker_ms; @@ -2168,13 +2673,17 @@ fn event_poll_timeout( inflight_publication_points: &HashMap, pending_finalization: &VecDeque, finalize_inflight: usize, + staging_inflight: usize, instances_started: usize, config: &TreeRunConfig, ) -> Duration { + // Stage results pending collection must not let the loop sleep: with the + // stage pool enabled a 50ms nap per turn would collapse throughput. if !ready_queue.is_empty() || !pending_roa_dispatch.is_empty() || !inflight_publication_points.is_empty() || !pending_finalization.is_empty() + || staging_inflight > 0 || (!ca_queue.is_empty() && can_start_more(instances_started, config)) { Duration::from_millis(0) @@ -2193,9 +2702,13 @@ fn is_complete( inflight_publication_points: &HashMap, pending_finalization: &VecDeque, finalize_inflight: usize, + staging_inflight: usize, instances_started: usize, config: &TreeRunConfig, ) -> bool { + // `staging_inflight == 0` additionally implies the stage result channel is + // drained empty: the drain loop collects every available result each turn + // and only decrements the counter while collecting. let ca_queue_done = ca_queue.is_empty() || !can_start_more(instances_started, config); ca_queue_done && ready_queue.is_empty() @@ -2204,6 +2717,7 @@ fn is_complete( && inflight_publication_points.is_empty() && pending_finalization.is_empty() && finalize_inflight == 0 + && staging_inflight == 0 } fn build_tree_output(mut finished: Vec) -> TreeRunAuditOutput { @@ -2286,20 +2800,39 @@ pub fn run_tree_parallel_phase2_audit( #[cfg(test)] mod tests { use super::{ - FinishedPublicationPointResult, compact_phase2_finished_result, - compact_phase2_finished_result_result, finalize_metrics_from_output, + CacheHitOutcome, CompleteOutcome, FinishedPublicationPointResult, InflightPublicationPoint, + QueuedCaInstance, ReadyCaInstance, ReadyStageMetrics, ReadyStageSubmitter, StageOutcome, + apply_ready_publication_point_stage, compact_phase2_finished_result, + compact_phase2_finished_result_result, compute_ready_publication_point_stage, + event_poll_timeout, finalize_metrics_from_output, finalize_publication_point_state, + is_complete, submit_ready_batch_to_stage_pool, }; use crate::audit::{ - AuditObjectKind, AuditObjectResult, ObjectAuditEntry, PublicationPointAudit, + AuditObjectKind, AuditObjectResult, DiscoveredFrom, ObjectAuditEntry, + PublicationPointAudit, }; - use crate::storage::PackTime; - use crate::validation::manifest::PublicationPointSource; - use crate::validation::objects::{ObjectsOutput, ObjectsStats}; + use crate::fetch::rsync::{RsyncFetchError, RsyncFetcher}; + use crate::parallel::repo_runtime::RepoSyncRuntimeOutcome; + use crate::policy::{CaFailedFetchPolicy, Policy, SyncPreference}; + use crate::storage::{PackTime, RocksStore}; + use crate::sync::rrdp::Fetcher; + use crate::validation::manifest::{ + FreshPublicationPointTimingBreakdown, FreshValidatedPublicationPoint, + PublicationPointSource, + }; + use crate::validation::objects::{ObjectsOutput, ObjectsStats, ParallelObjectsPrepare}; use crate::validation::publication_point::PublicationPointSnapshot; - use crate::validation::tree::PublicationPointRunResult; - use crate::validation::tree_runner::{ - BuildVcirTimingBreakdown, FreshPublicationPointFinalizeOutput, PersistVcirTimingBreakdown, + use crate::validation::tree::{ + CaCertificateRef, CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult, + TreeRunConfig, }; + use crate::validation::tree_runner::{ + BuildVcirTimingBreakdown, FreshPublicationPointFinalizeOutput, FreshPublicationPointStage, + PersistVcirTimingBreakdown, Rpkiv1PublicationPointRunner, + }; + use std::collections::{HashMap, VecDeque}; + use std::sync::Mutex; + use std::time::{Duration, Instant}; fn sample_snapshot() -> PublicationPointSnapshot { PublicationPointSnapshot { @@ -2495,4 +3028,560 @@ mod tests { assert_eq!(metrics.vap_count, 1); assert_eq!(metrics.audit_object_count, 1); } + + struct NeverHttpFetcher; + impl Fetcher for NeverHttpFetcher { + fn fetch(&self, _uri: &str) -> Result, String> { + Err("http fetch disabled in test".to_string()) + } + } + + struct FailingRsyncFetcher; + impl RsyncFetcher for FailingRsyncFetcher { + fn fetch_objects( + &self, + _rsync_base_uri: &str, + ) -> Result)>, RsyncFetchError> { + Err(RsyncFetchError::Fetch("rsync disabled in test".to_string())) + } + } + + fn stage_test_runner<'a>( + store: &'a RocksStore, + policy: &'a Policy, + ) -> Rpkiv1PublicationPointRunner<'a> { + Rpkiv1PublicationPointRunner { + store, + policy, + http_fetcher: &NeverHttpFetcher, + rsync_fetcher: &FailingRsyncFetcher, + validation_time: time::OffsetDateTime::now_utc(), + timing: None, + download_log: None, + replay_archive_index: None, + replay_delta_index: None, + rrdp_dedup: false, + rrdp_repo_cache: Mutex::new(HashMap::new()), + rsync_dedup: false, + rsync_repo_cache: Mutex::new(HashMap::new()), + current_repo_index: None, + repo_sync_runtime: None, + parallel_phase2_config: None, + parallel_roa_worker_pool: None, + ccr_accumulator: None, + persist_vcir: true, + enable_roa_validation_cache: false, + enable_child_certificate_validation_cache: false, + publication_point_cache_observe_only: false, + enable_publication_point_validation_cache: false, + } + } + + fn stage_test_handle(manifest_rsync_uri: &str) -> CaInstanceHandle { + CaInstanceHandle { + tal_id: "test".to_string(), + ca_certificate: CaCertificateRef::inline_der(vec![1]), + ca_certificate_rsync_uri: Some("rsync://example.test/repo/ca.cer".to_string()), + effective_ip_resources: None, + effective_as_resources: None, + manifest_rsync_uri: manifest_rsync_uri.to_string(), + publication_point_rsync_uri: "rsync://example.test/repo/".to_string(), + rsync_base_uri: "rsync://example.test/repo/".to_string(), + rrdp_notification_uri: None, + parent_manifest_rsync_uri: None, + depth: 0, + } + } + + fn stage_test_ready(id: u64) -> ReadyCaInstance { + ReadyCaInstance { + node: QueuedCaInstance { + id, + handle: stage_test_handle("rsync://example.test/repo/example.mft"), + parent_id: None, + discovered_from: None, + }, + repo_outcome: stage_test_repo_outcome(), + ready_enqueued_at: Instant::now(), + } + } + + fn stage_test_repo_outcome() -> RepoSyncRuntimeOutcome { + RepoSyncRuntimeOutcome { + repo_sync_ok: true, + repo_sync_err: None, + repo_sync_source: Some("rsync".to_string()), + repo_sync_phase: Some("rsync".to_string()), + repo_sync_duration_ms: 0, + warnings: Vec::new(), + } + } + + fn stage_test_child( + manifest_rsync_uri: &str, + ca_certificate_rsync_uri: &str, + ) -> DiscoveredChildCaInstance { + DiscoveredChildCaInstance { + handle: CaInstanceHandle { + ca_certificate_rsync_uri: Some(ca_certificate_rsync_uri.to_string()), + ..stage_test_handle(manifest_rsync_uri) + }, + discovered_from: DiscoveredFrom { + parent_manifest_rsync_uri: "rsync://example.test/repo/example.mft".to_string(), + child_ca_certificate_rsync_uri: ca_certificate_rsync_uri.to_string(), + child_ca_certificate_sha256_hex: "55".repeat(32), + }, + child_entry_projection: None, + } + } + + fn stage_test_fresh_stage() -> FreshPublicationPointStage { + FreshPublicationPointStage { + fresh_point: FreshValidatedPublicationPoint { + manifest_rsync_uri: "rsync://example.test/repo/example.mft".to_string(), + publication_point_rsync_uri: "rsync://example.test/repo/".to_string(), + manifest_number_be: vec![1], + this_update: PackTime { + rfc3339_utc: "2026-04-21T00:00:00Z".to_string(), + }, + next_update: PackTime { + rfc3339_utc: "2026-04-22T00:00:00Z".to_string(), + }, + verified_at: PackTime { + rfc3339_utc: "2026-04-21T00:00:01Z".to_string(), + }, + manifest_bytes: vec![1, 2, 3], + files: Vec::new(), + }, + issuer_ca_der: vec![1u8].into(), + snapshot_prepare_timing: FreshPublicationPointTimingBreakdown::default(), + snapshot_prepare_ms: 0, + discovered_children: Vec::new(), + child_audits: Vec::new(), + discovered_router_keys: Vec::new(), + child_discovery_ms: 0, + warnings: Vec::new(), + } + } + + fn stage_test_metrics() -> ReadyStageMetrics { + ReadyStageMetrics { + ready_count: 1, + manifest_rsync_uri: Some("rsync://example.test/repo/example.mft".to_string()), + publication_point_rsync_uri: Some("rsync://example.test/repo/".to_string()), + ..ReadyStageMetrics::default() + } + } + + #[test] + fn compute_apply_fresh_error_runs_inline_fallback_and_finishes_err() { + let store_dir = tempfile::tempdir().expect("store dir"); + let store = RocksStore::open(store_dir.path()).expect("open rocksdb"); + let policy = Policy { + sync_preference: SyncPreference::RsyncOnly, + ca_failed_fetch_policy: CaFailedFetchPolicy::StopAllOutput, + ..Policy::default() + }; + let runner = stage_test_runner(&store, &policy); + let config = TreeRunConfig::default(); + let mut next_id = 1u64; + let mut ca_queue = VecDeque::new(); + let mut pending_roa_dispatch = VecDeque::new(); + let mut inflight_publication_points = HashMap::new(); + let mut pending_finalization = VecDeque::new(); + let mut finished = Vec::new(); + + let (outcome, metrics) = + compute_ready_publication_point_stage(&runner, stage_test_ready(0), 0); + assert!( + matches!(outcome, StageOutcome::FreshError(_)), + "fresh staging against an empty store must fail" + ); + let metrics = apply_ready_publication_point_stage( + &runner, + &mut next_id, + &mut ca_queue, + &mut pending_roa_dispatch, + &mut inflight_publication_points, + &mut pending_finalization, + &mut finished, + outcome, + metrics, + &config, + false, + ); + + assert_eq!(metrics.ready_count, 1); + assert_eq!(metrics.fallback_count, 1); + assert_eq!(metrics.complete_count, 0); + assert_eq!(metrics.staged_count, 0); + assert_eq!(metrics.zero_task_count, 0); + assert_eq!(finished.len(), 1); + match &finished[0].result { + FinishedPublicationPointResult::Err(_) => {} + FinishedPublicationPointResult::Ok { .. } => { + panic!("fallback without repository data must fail") + } + } + assert!(ca_queue.is_empty()); + assert!(pending_roa_dispatch.is_empty()); + assert!(pending_finalization.is_empty()); + assert!(inflight_publication_points.is_empty()); + assert_eq!(next_id, 1); + } + + #[test] + fn apply_cache_hit_enqueues_children_sorted_and_finishes() { + let store_dir = tempfile::tempdir().expect("store dir"); + let store = RocksStore::open(store_dir.path()).expect("open rocksdb"); + let policy = Policy::default(); + let runner = stage_test_runner(&store, &policy); + let config = TreeRunConfig::default(); + let mut next_id = 42u64; + let mut ca_queue = VecDeque::new(); + let mut pending_roa_dispatch = VecDeque::new(); + let mut inflight_publication_points = HashMap::new(); + let mut pending_finalization = VecDeque::new(); + let mut finished = Vec::new(); + + let mut result = sample_result(); + result.discovered_children = vec![ + stage_test_child( + "rsync://example.test/repo/z.mft", + "rsync://example.test/repo/z.cer", + ), + stage_test_child( + "rsync://example.test/repo/a.mft", + "rsync://example.test/repo/a.cer", + ), + ]; + let outcome = StageOutcome::CacheHit(Box::new(CacheHitOutcome { + ready: stage_test_ready(7), + publication_point_started: Instant::now(), + result, + })); + let mut metrics = stage_test_metrics(); + metrics.complete_count = 1; + metrics.discovered_children = 2; + let metrics = apply_ready_publication_point_stage( + &runner, + &mut next_id, + &mut ca_queue, + &mut pending_roa_dispatch, + &mut inflight_publication_points, + &mut pending_finalization, + &mut finished, + outcome, + metrics, + &config, + false, + ); + + assert_eq!(finished.len(), 1); + match &finished[0].result { + FinishedPublicationPointResult::Ok { .. } => {} + FinishedPublicationPointResult::Err(err) => { + panic!("cache hit must finish ok: {err}") + } + } + assert_eq!(finished[0].node.id, 7); + // Children are enqueued sorted by manifest URI with sequential ids + // taken from next_id, exactly like the monolithic staging did. + assert_eq!(ca_queue.len(), 2); + assert_eq!( + ca_queue[0].handle.manifest_rsync_uri, + "rsync://example.test/repo/a.mft" + ); + assert_eq!( + ca_queue[1].handle.manifest_rsync_uri, + "rsync://example.test/repo/z.mft" + ); + assert_eq!(ca_queue[0].id, 42); + assert_eq!(ca_queue[1].id, 43); + assert_eq!(ca_queue[0].parent_id, Some(7)); + assert_eq!(ca_queue[1].parent_id, Some(7)); + assert_eq!(next_id, 44); + assert!(pending_roa_dispatch.is_empty()); + assert!(pending_finalization.is_empty()); + assert!(inflight_publication_points.is_empty()); + assert_eq!(metrics.complete_count, 1); + assert_eq!(metrics.discovered_children, 2); + } + + #[test] + fn apply_complete_enqueues_finalize_task_instead_of_finishing() { + let store_dir = tempfile::tempdir().expect("store dir"); + let store = RocksStore::open(store_dir.path()).expect("open rocksdb"); + let policy = Policy::default(); + let runner = stage_test_runner(&store, &policy); + let config = TreeRunConfig::default(); + let mut next_id = 1u64; + let mut ca_queue = VecDeque::new(); + let mut pending_roa_dispatch = VecDeque::new(); + let mut inflight_publication_points = HashMap::new(); + let mut pending_finalization = VecDeque::new(); + let mut finished = Vec::new(); + + let outcome = StageOutcome::Complete(Box::new(CompleteOutcome { + ready: stage_test_ready(9), + publication_point_started: Instant::now(), + fresh_stage: stage_test_fresh_stage(), + warnings: Vec::new(), + objects: sample_result().objects, + })); + let mut metrics = stage_test_metrics(); + metrics.complete_count = 1; + let metrics = apply_ready_publication_point_stage( + &runner, + &mut next_id, + &mut ca_queue, + &mut pending_roa_dispatch, + &mut inflight_publication_points, + &mut pending_finalization, + &mut finished, + outcome, + metrics, + &config, + false, + ); + + assert!( + finished.is_empty(), + "complete staging must defer the finalize to the worker queue" + ); + assert_eq!(pending_finalization.len(), 1); + let task = pending_finalization.pop_front().expect("finalize task"); + let state = task.state; + assert_eq!(state.node.id, 9); + assert_eq!(state.task_count, 0); + assert!(state.finalize_enqueued_at.is_some()); + assert!(state.results.is_empty()); + assert!(matches!( + state.objects_prepare, + ParallelObjectsPrepare::Complete(_) + )); + assert!(inflight_publication_points.is_empty()); + assert!(pending_roa_dispatch.is_empty()); + assert!(ca_queue.is_empty()); + assert_eq!(next_id, 1); + assert_eq!(metrics.complete_count, 1); + } + + #[test] + fn finalize_worker_complete_arm_finalizes_without_reduce() { + let store_dir = tempfile::tempdir().expect("store dir"); + let store = RocksStore::open(store_dir.path()).expect("open rocksdb"); + let policy = Policy::default(); + let mut runner = stage_test_runner(&store, &policy); + runner.persist_vcir = false; + + let state = InflightPublicationPoint { + node: QueuedCaInstance { + id: 3, + handle: stage_test_handle("rsync://example.test/repo/example.mft"), + parent_id: None, + discovered_from: None, + }, + fresh_stage: stage_test_fresh_stage(), + objects_prepare: ParallelObjectsPrepare::Complete(sample_result().objects), + repo_outcome: stage_test_repo_outcome(), + warnings: Vec::new(), + started_at: Instant::now(), + objects_started_at: Instant::now(), + task_count: 0, + tasks_submitted: 0, + first_task_submitted_at: None, + last_task_submitted_at: None, + first_result_at: None, + last_result_at: None, + worker_ms_total: 0, + worker_ms_max: 0, + queue_wait_ms_total: 0, + queue_wait_ms_max: 0, + finalize_enqueued_at: Some(Instant::now()), + results: Vec::new(), + }; + let result = finalize_publication_point_state(&runner, state, false); + + match result.finished.result { + FinishedPublicationPointResult::Ok { source, .. } => { + assert_eq!(source, PublicationPointSource::Fresh); + } + FinishedPublicationPointResult::Err(err) => { + panic!("complete finalize must succeed: {err}") + } + } + assert_eq!(result.finished.node.id, 3); + assert_eq!(result.metrics.reduce_ms, 0); + assert_eq!(result.metrics.locked_files, 0); + } + + #[test] + fn is_complete_waits_for_staging_inflight() { + let ca_queue = VecDeque::new(); + let ready_queue = VecDeque::new(); + let ca_waiting_repo_by_identity = HashMap::new(); + let pending_roa_dispatch = VecDeque::new(); + let inflight_publication_points = HashMap::new(); + let pending_finalization = VecDeque::new(); + let config = TreeRunConfig::default(); + + // Everything drained and no staging in flight: the loop may exit. + assert!(is_complete( + &ca_queue, + &ready_queue, + &ca_waiting_repo_by_identity, + &pending_roa_dispatch, + &inflight_publication_points, + &pending_finalization, + 0, + 0, + 0, + &config, + )); + // A submitted stage task whose result has not been collected yet must + // keep the loop alive, otherwise its publication point would be lost. + assert!(!is_complete( + &ca_queue, + &ready_queue, + &ca_waiting_repo_by_identity, + &pending_roa_dispatch, + &inflight_publication_points, + &pending_finalization, + 0, + 1, + 0, + &config, + )); + } + + #[test] + fn event_poll_timeout_stays_awake_while_staging() { + let ca_queue = VecDeque::new(); + let ready_queue = VecDeque::new(); + let pending_roa_dispatch = VecDeque::new(); + let inflight_publication_points = HashMap::new(); + let pending_finalization = VecDeque::new(); + let config = TreeRunConfig::default(); + + // Staging in flight: poll must not sleep, or stage results pile up. + assert_eq!( + event_poll_timeout( + &ca_queue, + &ready_queue, + &pending_roa_dispatch, + &inflight_publication_points, + &pending_finalization, + 0, + 1, + 0, + &config, + ), + Duration::from_millis(0) + ); + // Without staging the historical tiers are unchanged. + assert_eq!( + event_poll_timeout( + &ca_queue, + &ready_queue, + &pending_roa_dispatch, + &inflight_publication_points, + &pending_finalization, + 1, + 0, + 0, + &config, + ), + Duration::from_millis(10) + ); + assert_eq!( + event_poll_timeout( + &ca_queue, + &ready_queue, + &pending_roa_dispatch, + &inflight_publication_points, + &pending_finalization, + 0, + 0, + 0, + &config, + ), + Duration::from_millis(50) + ); + } + + struct MockStageSubmitter { + capacity: usize, + submitted_ids: Vec, + } + + impl ReadyStageSubmitter for MockStageSubmitter { + fn try_submit_ready_stage( + &mut self, + task: super::ReadyStageTask, + ) -> Result<(), crate::parallel::object_worker::ObjectWorkerSubmitError> + { + if self.submitted_ids.len() < self.capacity { + self.submitted_ids.push(task.ready.node.id); + Ok(()) + } else { + Err( + crate::parallel::object_worker::ObjectWorkerSubmitError::QueueFull { + worker_index: 0, + task, + }, + ) + } + } + } + + #[test] + fn submit_ready_batch_requeues_on_backpressure_without_loss_or_duplication() { + let mut ready_queue = VecDeque::new(); + for id in [10, 11, 12] { + ready_queue.push_back(stage_test_ready(id)); + } + let mut staging_inflight = 0usize; + + // First turn: only two tasks fit, the third must be returned to the + // head of the ready queue. + let mut submitter = MockStageSubmitter { + capacity: 2, + submitted_ids: Vec::new(), + }; + let metrics = submit_ready_batch_to_stage_pool( + &mut submitter, + &mut ready_queue, + &mut staging_inflight, + 256, + Duration::from_secs(60), + ) + .expect("dispatch"); + assert_eq!(metrics.submitted, 2); + assert!(metrics.queue_full); + assert_eq!(staging_inflight, 2); + assert_eq!(submitter.submitted_ids, vec![10, 11]); + assert_eq!(ready_queue.len(), 1); + assert_eq!(ready_queue[0].node.id, 12); + + // Next turn retries the requeued publication point; across both turns + // every id is submitted exactly once. + let mut submitter = MockStageSubmitter { + capacity: 8, + submitted_ids: Vec::new(), + }; + let metrics = submit_ready_batch_to_stage_pool( + &mut submitter, + &mut ready_queue, + &mut staging_inflight, + 256, + Duration::from_secs(60), + ) + .expect("dispatch"); + assert_eq!(metrics.submitted, 1); + assert!(!metrics.queue_full); + assert_eq!(staging_inflight, 3); + assert_eq!(submitter.submitted_ids, vec![12]); + assert!(ready_queue.is_empty()); + } } diff --git a/src/validation/tree_runner.rs b/src/validation/tree_runner.rs index cfd07e9..d63c7db 100644 --- a/src/validation/tree_runner.rs +++ b/src/validation/tree_runner.rs @@ -511,7 +511,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { fn current_hash_for_uri(&self, uri: &str) -> Option<[u8; 32]> { if let Some(index) = self.current_repo_index.as_ref() { - if let Ok(index) = index.lock() { + if let Ok(index) = index.read() { if let Some(entry) = index.get_by_uri(uri) { return Some(entry.current_hash); } @@ -760,7 +760,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { fn current_manifest_hash_hex_for_audit(&self, ca: &CaInstanceHandle) -> Option { if let Some(index_handle) = self.current_repo_index.as_ref() - && let Ok(index) = index_handle.lock() + && let Ok(index) = index_handle.read() && let Some(entry) = index.get_by_uri(&ca.manifest_rsync_uri) { return Some(entry.current_hash_hex.clone());